E - type of the element stored in the Queuepublic 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 |
protected AtomicLongArray |
sequenceBuffer |
UNBOUNDED_CAPACITY| Constructor and Description |
|---|
MpmcAtomicArrayQueue(int capacity) |
| Modifier and Type | Method and Description |
|---|---|
protected int |
calcElementOffset(long index) |
protected int |
calcElementOffset(long index,
int mask) |
protected long |
calcSequenceOffset(long index) |
protected static int |
calcSequenceOffset(long index,
int mask) |
int |
capacity() |
protected boolean |
casConsumerIndex(long expect,
long newValue) |
protected boolean |
casProducerIndex(long expect,
long newValue) |
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 w,
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() |
static <E> E |
lpElement(AtomicReferenceArray<E> buffer,
int offset) |
protected E |
lpElement(int offset) |
long |
lvConsumerIndex() |
static <E> E |
lvElement(AtomicReferenceArray<E> buffer,
int offset) |
protected E |
lvElement(int offset) |
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.
|
static <E> void |
soElement(AtomicReferenceArray<E> buffer,
int offset,
E value) |
protected void |
soElement(int offset,
E value) |
protected void |
soSequence(AtomicLongArray buffer,
int offset,
long e) |
static <E> void |
spElement(AtomicReferenceArray<E> buffer,
int offset,
E value) |
protected void |
spElement(int offset,
E value) |
static <E> void |
svElement(AtomicReferenceArray<E> buffer,
int offset,
E value) |
String |
toString() |
add, addAll, element, removecontains, containsAll, remove, removeAll, retainAll, toArray, toArrayclone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, waitcontains, containsAll, equals, hashCode, parallelStream, remove, removeAll, removeIf, retainAll, spliterator, stream, toArray, toArrayprotected final AtomicLongArray sequenceBuffer
protected final AtomicReferenceArray<E> buffer
protected final int mask
public boolean offer(E e)
MessagePassingQueueQueue.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.
public E peek()
MessagePassingQueueQueue.peek() interface.public boolean relaxedOffer(E e)
MessagePassingQueueQueue.offer(Object) this method may return false without the queue being full.e - not null, will throw NPE if it ispublic E relaxedPoll()
MessagePassingQueueQueue.poll() this method may return null without the queue being empty.public E relaxedPeek()
MessagePassingQueueQueue.peek() this method may return null without the queue being empty.public int drain(MessagePassingQueue.Consumer<E> c)
MessagePassingQueue
M m;
while((m = relaxedPoll()) != null){
c.accept(m);
}
public int fill(MessagePassingQueue.Supplier<E> s)
MessagePassingQueue
while(relaxedOffer(s.get());
public 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.
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.
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.
public void fill(MessagePassingQueue.Supplier<E> s, MessagePassingQueue.WaitStrategy w, 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.
public final long lvConsumerIndex()
protected final boolean casConsumerIndex(long expect,
long newValue)
public final long lvProducerIndex()
protected final boolean casProducerIndex(long expect,
long newValue)
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 Iterator<E> iterator()
iterator in interface Iterable<E>iterator in interface Collection<E>iterator in class AbstractCollection<E>public String toString()
toString in class AbstractCollection<E>public void clear()
MessagePassingQueueCollection.clear() interface.clear in interface Collection<E>clear in interface MessagePassingQueue<E>clear in class AbstractQueue<E>protected final int calcElementOffset(long index,
int mask)
protected final int calcElementOffset(long index)
public static <E> E lvElement(AtomicReferenceArray<E> buffer, int offset)
public static <E> E lpElement(AtomicReferenceArray<E> buffer, int offset)
protected final E lpElement(int offset)
public static <E> void spElement(AtomicReferenceArray<E> buffer, int offset, E value)
protected final void spElement(int offset,
E value)
public static <E> void soElement(AtomicReferenceArray<E> buffer, int offset, E value)
protected final void soElement(int offset,
E value)
public static <E> void svElement(AtomicReferenceArray<E> buffer, int offset, E value)
protected final E lvElement(int offset)
public final int capacity()
capacity in interface MessagePassingQueue<E>MessagePassingQueue.UNBOUNDED_CAPACITY if not boundedpublic final int size()
size in interface Collection<E>size in interface MessagePassingQueue<E>size in class AbstractCollection<E>Integer.MAX_VALUE but less or equals to
capacity (if bounded).public final boolean isEmpty()
MessagePassingQueueisEmpty in interface Collection<E>isEmpty in interface MessagePassingQueue<E>isEmpty in class AbstractCollection<E>public final long currentProducerIndex()
QueueProgressIndicatorscurrentProducerIndex in interface QueueProgressIndicatorspublic final long currentConsumerIndex()
QueueProgressIndicatorscurrentConsumerIndex in interface QueueProgressIndicatorsCopyright © 2013–2018. All rights reserved.