org.jctools.queues

Class MpmcArrayQueue<E>

java.lang.Object

java.util.AbstractCollection<E>

java.util.AbstractQueue<E>

org.jctools.queues.ConcurrentCircularArrayQueue<E>

org.jctools.queues.ConcurrentSequencedCircularArrayQueue<E>

org.jctools.queues.MpmcArrayQueue<E>

Type Parameters:

E - type of the element stored in the Queue

All Implemented Interfaces:

Iterable<E>, Collection<E>, Queue<E>, QueueProgressIndicators

public class MpmcArrayQueue<E>
extends ConcurrentSequencedCircularArrayQueue<E>
implements QueueProgressIndicators

A Multi-Producer-Multi-Consumer queue based on a ConcurrentCircularArrayQueue. This implies that any and all threads may call the offer/poll/peek methods and correctness is maintained.
This implementation follows patterns documented on the package level for False Sharing protection.
The algorithm for offer/poll is an adaptation of the one put forward by D. Vyukov (See here). The original algorithm uses an array of structs which should offer nice locality properties but is sadly not possible in Java (waiting on Value Types or similar). The alternative explored here utilizes 2 arrays, one for each field of the struct. There is a further alternative in the experimental project which uses iteration phase markers to achieve the same algo and is closer structurally to the original, but sadly does not perform as well as this implementation.
Tradeoffs to keep in mind:

1. Padding for false sharing: counter fields and queue fields are all padded as well as either side of both arrays. We are trading memory to avoid false sharing(active and passive).
2. 2 arrays instead of one: The algorithm requires an extra array of longs matching the size of the elements array. This is doubling/tripling the memory allocated for the buffer.
3. Power of 2 capacity: Actual elements buffer (and sequence buffer) is the closest power of 2 larger or equal to the requested capacity.

Field Summary

Fields inherited from class org.jctools.queues.ConcurrentSequencedCircularArrayQueue

SEQ_BUFFER_PAD, sequenceBuffer

Fields inherited from class org.jctools.queues.ConcurrentCircularArrayQueue

buffer, mask, REF_BUFFER_PAD, SPARSE_SHIFT

Constructor Summary

Constructors

Constructor and Description
MpmcArrayQueue(int capacity)

Method Summary

Modifier and Type	Method and Description
protected boolean	casConsumerIndex(long expect, long newValue)
protected boolean	casProducerIndex(long expect, long newValue)
long	currentConsumerIndex() This method has no concurrent visibility semantics.
long	currentProducerIndex() This method has no concurrent visibility semantics.
boolean	isEmpty()
protected long	lvConsumerIndex()
protected long	lvProducerIndex()
boolean	offer(E e)
E	peek()
E	poll()
int	size()

Methods inherited from class org.jctools.queues.ConcurrentSequencedCircularArrayQueue

calcSequenceOffset, calcSequenceOffset, lvSequence, soSequence

Methods inherited from class org.jctools.queues.ConcurrentCircularArrayQueue

calcElementOffset, calcElementOffset, clear, iterator, lpElement, lpElement, lvElement, lvElement, soElement, soElement, spElement, spElement

Methods inherited from class java.util.AbstractQueue

add, addAll, element, remove

Methods inherited from class java.util.AbstractCollection

contains, containsAll, remove, removeAll, retainAll, toArray, toArray, toString

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface java.util.Collection

contains, containsAll, equals, hashCode, parallelStream, remove, removeAll, removeIf, retainAll, spliterator, stream, toArray, toArray

Methods inherited from interface java.lang.Iterable

forEach

Constructor Detail

MpmcArrayQueue

public MpmcArrayQueue(int capacity)

Method Detail

offer

public boolean offer(E e)

Specified by:

offer in interface Queue<E>

poll

public E poll()

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.

Specified by:

poll in interface Queue<E>

peek

public E peek()

Specified by:

peek in interface Queue<E>

size

public int size()

Specified by:

size in interface Collection<E>

Specified by:

size in class AbstractCollection<E>

isEmpty

public boolean isEmpty()

Specified by:

isEmpty in interface Collection<E>

Overrides:

isEmpty in class AbstractCollection<E>

currentProducerIndex

public long currentProducerIndex()

Description copied from interface: QueueProgressIndicators

This method has no concurrent visibility semantics. The value returned may be negative. Under normal circumstances 2 consecutive calls to this method can offer an idea of progress made by producer threads by subtracting the 2 results though in extreme cases (if producers have progressed by more than 2^64) this may also fail.
This value will normally indicate number of elements passed into the queue, but may under some circumstances be a derivative of that figure. This method should not be used to derive size or emptiness.

Specified by:

currentProducerIndex in interface QueueProgressIndicators

Returns:

the current value of the producer progress index

currentConsumerIndex

public long currentConsumerIndex()

Description copied from interface: QueueProgressIndicators

This method has no concurrent visibility semantics. The value returned may be negative. Under normal circumstances 2 consecutive calls to this method can offer an idea of progress made by consumer threads by subtracting the 2 results though in extreme cases (if consumers have progressed by more than 2^64) this may also fail.
This value will normally indicate number of elements taken out of the queue, but may under some circumstances be a derivative of that figure. This method should not be used to derive size or emptiness.

Specified by:

currentConsumerIndex in interface QueueProgressIndicators

Returns:

the current value of the consumer progress index

lvConsumerIndex

protected final long lvConsumerIndex()

casConsumerIndex

protected final boolean casConsumerIndex(long expect,
                                         long newValue)

lvProducerIndex

protected final long lvProducerIndex()

casProducerIndex

protected final boolean casProducerIndex(long expect,
                                         long newValue)