rx.internal.util.unsafe

Class MpmcArrayQueue<E>

  • Type Parameters:
    E - type of the element stored in the Queue
    All Implemented Interfaces:
    Iterable<E>, Collection<E>, Queue<E> 

    public class MpmcArrayQueue<E>
extends ConcurrentSequencedCircularArrayQueue<E>
    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.

    • Constructor Detail

      • MpmcArrayQueue

        public MpmcArrayQueue(int capacity)

    • Method Detail

      • offer

        public boolean offer(E 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.

      • peek

        public E peek()

      • size

        public int size()

      • isEmpty

        public boolean isEmpty()

      • 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)