public class MpmcAtomicArrayQueue<E> extends AbstractQueue<E>
ConcurrentCircularArrayQueue
. This
implies that any and all threads may call the offer/poll/peek methods and correctness is maintained. Tradeoffs to keep in mind:
MessagePassingQueue.Consumer<T>, MessagePassingQueue.ExitCondition, MessagePassingQueue.Supplier<T>, MessagePassingQueue.WaitStrategy
Modifier and Type | Field and Description |
---|---|
protected AtomicReferenceArray<E> |
buffer |
protected int |
mask |
static int |
MAX_LOOK_AHEAD_STEP |
protected AtomicLongArray |
sequenceBuffer |
UNBOUNDED_CAPACITY
Constructor and Description |
---|
MpmcAtomicArrayQueue(int capacity) |
Modifier and Type | Method and Description |
---|---|
protected long |
calcSequenceOffset(long index) |
protected static int |
calcSequenceOffset(long index,
int mask) |
int |
capacity() |
void |
clear()
Removes all items from the queue.
|
long |
currentConsumerIndex()
This method has no concurrent visibility semantics.
|
long |
currentProducerIndex()
This method has no concurrent visibility semantics.
|
int |
drain(MessagePassingQueue.Consumer<E> c)
Remove all available item from the queue and hand to consume.
|
int |
drain(MessagePassingQueue.Consumer<E> c,
int limit)
Remove up to limit elements from the queue and hand to consume.
|
void |
drain(MessagePassingQueue.Consumer<E> c,
MessagePassingQueue.WaitStrategy w,
MessagePassingQueue.ExitCondition exit)
Remove elements from the queue and hand to consume forever.
|
int |
fill(MessagePassingQueue.Supplier<E> s)
Stuff the queue with elements from the supplier.
|
int |
fill(MessagePassingQueue.Supplier<E> s,
int limit)
Stuff the queue with up to limit elements from the supplier.
|
void |
fill(MessagePassingQueue.Supplier<E> s,
MessagePassingQueue.WaitStrategy wait,
MessagePassingQueue.ExitCondition exit)
Stuff the queue with elements from the supplier forever.
|
boolean |
isEmpty()
This method's accuracy is subject to concurrent modifications happening as the observation is carried
out.
|
Iterator<E> |
iterator()
Get an iterator for this queue.
|
long |
lvConsumerIndex() |
long |
lvProducerIndex() |
protected long |
lvSequence(AtomicLongArray buffer,
int offset) |
boolean |
offer(E e)
Called from a producer thread subject to the restrictions appropriate to the implementation and
according to the
Queue.offer(Object) interface. |
E |
peek()
Called from the consumer thread subject to the restrictions appropriate to the implementation and
according to the
Queue.peek() interface. |
E |
poll()
Called from the consumer thread subject to the restrictions appropriate to the implementation and
according to the
Queue.poll() interface. |
boolean |
relaxedOffer(E e)
Called from a producer thread subject to the restrictions appropriate to the implementation.
|
E |
relaxedPeek()
Called from the consumer thread subject to the restrictions appropriate to the implementation.
|
E |
relaxedPoll()
Called from the consumer thread subject to the restrictions appropriate to the implementation.
|
int |
size()
This method's accuracy is subject to concurrent modifications happening as the size is estimated and as
such is a best effort rather than absolute value.
|
protected void |
soSequence(AtomicLongArray buffer,
int offset,
long e) |
String |
toString() |
add, addAll, element, remove
contains, containsAll, remove, removeAll, retainAll, toArray, toArray
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
contains, containsAll, equals, hashCode, parallelStream, remove, removeAll, removeIf, retainAll, spliterator, stream, toArray, toArray
public static final int MAX_LOOK_AHEAD_STEP
protected final AtomicLongArray sequenceBuffer
protected final AtomicReferenceArray<E> buffer
protected final int mask
public boolean offer(E e)
MessagePassingQueue
Queue.offer(Object)
interface.e
- not null
, will throw NPE if it ispublic E poll()
Queue.poll()
interface.
Because return null indicates queue is empty we cannot simply rely on next element visibility for poll and must test producer index when next element is not visible.
null
iff emptypublic E peek()
MessagePassingQueue
Queue.peek()
interface.null
iff emptypublic boolean relaxedOffer(E e)
MessagePassingQueue
Queue.offer(Object)
this method may return false without the queue being full.e
- not null
, will throw NPE if it ispublic E relaxedPoll()
MessagePassingQueue
Queue.poll()
this method may return null
without the queue being empty.null
if unable to pollpublic E relaxedPeek()
MessagePassingQueue
Queue.peek()
this method may return null
without the queue being empty.null
if unable to peekpublic int drain(MessagePassingQueue.Consumer<E> c, int limit)
MessagePassingQueue
M m;
int i = 0;
for(;i < limit && (m = relaxedPoll()) != null; i++){
c.accept(m);
}
return i;
There's no strong commitment to the queue being empty at the end of a drain. Called from a consumer thread subject to the restrictions appropriate to the implementation.
WARNING: Explicit assumptions are made with regards to MessagePassingQueue.Consumer.accept(T)
make sure you have read
and understood these before using this method.
public int fill(MessagePassingQueue.Supplier<E> s, int limit)
MessagePassingQueue
for(int i=0; i < limit && relaxedOffer(s.get()); i++);
There's no strong commitment to the queue being full at the end of a fill. Called from a producer
thread subject to the restrictions appropriate to the implementation.
WARNING: Explicit assumptions are made with regards to MessagePassingQueue.Supplier.get()
make sure you have read
and understood these before using this method.
public int drain(MessagePassingQueue.Consumer<E> c)
MessagePassingQueue
M m; while((m = relaxedPoll()) != null){ c.accept(m); }There's no strong commitment to the queue being empty at the end of a drain. Called from a consumer thread subject to the restrictions appropriate to the implementation.
WARNING: Explicit assumptions are made with regards to MessagePassingQueue.Consumer.accept(T)
make sure you have read
and understood these before using this method.
public int fill(MessagePassingQueue.Supplier<E> s)
MessagePassingQueue
while(relaxedOffer(s.get());There's no strong commitment to the queue being full at the end of a fill. Called from a producer thread subject to the restrictions appropriate to the implementation.
Unbounded queues will fill up the queue with a fixed amount rather than fill up to oblivion.
WARNING: Explicit assumptions are made with regards to MessagePassingQueue.Supplier.get()
make sure you have read
and understood these before using this method.
public void drain(MessagePassingQueue.Consumer<E> c, MessagePassingQueue.WaitStrategy w, MessagePassingQueue.ExitCondition exit)
MessagePassingQueue
int idleCounter = 0; while (exit.keepRunning()) { E e = relaxedPoll(); if(e==null){ idleCounter = wait.idle(idleCounter); continue; } idleCounter = 0; c.accept(e); }
Called from a consumer thread subject to the restrictions appropriate to the implementation.
WARNING: Explicit assumptions are made with regards to MessagePassingQueue.Consumer.accept(T)
make sure you have read
and understood these before using this method.
public void fill(MessagePassingQueue.Supplier<E> s, MessagePassingQueue.WaitStrategy wait, MessagePassingQueue.ExitCondition exit)
MessagePassingQueue
int idleCounter = 0;
while (exit.keepRunning()) {
E e = s.get();
while (!relaxedOffer(e)) {
idleCounter = wait.idle(idleCounter);
continue;
}
idleCounter = 0;
}
Called from a producer thread subject to the restrictions appropriate to the implementation. The main difference
being that implementors MUST assure room in the queue is available BEFORE calling MessagePassingQueue.Supplier.get()
.
WARNING: Explicit assumptions are made with regards to MessagePassingQueue.Supplier.get()
make sure you have read
and understood these before using this method.
public final long lvConsumerIndex()
public final long lvProducerIndex()
protected final long calcSequenceOffset(long index)
protected static int calcSequenceOffset(long index, int mask)
protected final void soSequence(AtomicLongArray buffer, int offset, long e)
protected final long lvSequence(AtomicLongArray buffer, int offset)
public String toString()
toString
in class AbstractCollection<E>
public void clear()
MessagePassingQueue
Collection.clear()
interface.clear
in interface MessagePassingQueue<E>
clear
in interface Collection<E>
clear
in class AbstractQueue<E>
public final int capacity()
capacity
in interface IndexedQueueSizeUtil.IndexedQueue
capacity
in interface MessagePassingQueue<E>
MessagePassingQueue.UNBOUNDED_CAPACITY
if not boundedpublic final int size()
size
in interface MessagePassingQueue<E>
size
in interface Collection<E>
size
in class AbstractCollection<E>
Integer.MAX_VALUE
but less or equals to
capacity (if bounded).public final boolean isEmpty()
MessagePassingQueue
isEmpty
in interface MessagePassingQueue<E>
isEmpty
in interface Collection<E>
isEmpty
in class AbstractCollection<E>
public final long currentProducerIndex()
QueueProgressIndicators
currentProducerIndex
in interface QueueProgressIndicators
public final long currentConsumerIndex()
QueueProgressIndicators
currentConsumerIndex
in interface QueueProgressIndicators
public final Iterator<E> iterator()
The iterator provides a best-effort snapshot of the elements in the queue. The returned iterator is not guaranteed to return elements in queue order, and races with the consumer thread may cause gaps in the sequence of returned elements. Like {link #relaxedPoll}, the iterator may not immediately return newly inserted elements.
iterator
in interface Iterable<E>
iterator
in interface Collection<E>
iterator
in class AbstractCollection<E>
Copyright © 2008–2021 The Netty Project. All rights reserved.