Class SingleConsumerQueue<E>

  • Type Parameters:
    E - the type of elements held in this collection
    All Implemented Interfaces:
    java.io.Serializable, java.lang.Iterable<E>, java.util.Collection<E>, java.util.Queue<E>

    public final class SingleConsumerQueue<E>
    extends SCQHeader.HeadAndTailRef<E>
    implements java.util.Queue<E>, java.io.Serializable
    A lock-free unbounded queue based on linked nodes that supports concurrent producers and is restricted to a single consumer. This queue orders elements FIFO (first-in-first-out). The head of the queue is that element that has been on the queue the longest time. The tail of the queue is that element that has been on the queue the shortest time. New elements are inserted at the tail of the queue, and the queue retrieval operations obtain elements at the head of the queue. Like most other concurrent collection implementations, this class does not permit the use of null elements.

    A SingleConsumerQueue is an appropriate choice when many producer threads will share access to a common collection and a single consumer thread drains it. This collection is useful in scenarios such as implementing flat combining, actors, or lock amortization.

    This implementation employs combination to transfer elements between threads that are producing concurrently. This approach avoids contention on the queue by combining colliding operations that have identical semantics. When a pair of producers collide, the task of performing the combined set of operations is delegated to one of the threads and the other thread optionally waits for its operation to be completed. This decision of whether to wait for completion is determined by constructing either a linearizable or optimistic queue.

    Iterators are weakly consistent, returning elements reflecting the state of the queue at some point at or since the creation of the iterator. They do not throw ConcurrentModificationException, and may proceed concurrently with other operations. Elements contained in the queue since the creation of the iterator will be returned exactly once.

    Beware that it is the responsibility of the caller to ensure that a consumer has exclusive read access to the queue. This implementation does not include fail-fast behavior to guard against incorrect consumer usage.

    Beware that, unlike in most collections, the size method is NOT a constant-time operation. Because of the asynchronous nature of these queues, determining the current number of elements requires a traversal of the elements, and so may report inaccurate results if this collection is modified during traversal.

    Warning: This class is scheduled for removal in version 3.0.0.

    See Also:
    Serialized Form
    • Field Detail

      • NCPU

        static final int NCPU
        The number of CPUs
      • ARENA_LENGTH

        static final int ARENA_LENGTH
        The number of slots in the elimination array.
      • ARENA_MASK

        static final int ARENA_MASK
        The mask value for indexing into the arena.
      • OPTIMISIC

        static final java.util.function.Function<?,​?> OPTIMISIC
        The factory for creating an optimistic node.
      • SPINS

        static final int SPINS
        The number of times to spin (doing nothing except polling a memory location) before giving up while waiting to eliminate an operation. Should be zero on uniprocessors. On multiprocessors, this value should be large enough so that two threads exchanging items as fast as possible block only when one of them is stalled (due to GC or preemption), but not much longer, to avoid wasting CPU resources. Seen differently, this value is a little over half the number of cycles of an average context switch time on most systems. The value here is approximately the average of those across a range of tested systems.
      • PROBE

        static final long PROBE
        The offset to the thread-specific probe field.
    • Constructor Detail

    • Method Detail

      • ceilingPowerOfTwo

        static int ceilingPowerOfTwo​(int x)
      • optimistic

        public static <E> SingleConsumerQueue<E> optimistic()
        Creates a queue with an optimistic backoff strategy. A thread completes its operation without waiting after it successfully hands off the additional element(s) to another producing thread for batch insertion. This optimistic behavior may result in additions not appearing in FIFO order due to the backoff strategy trying to compensate for queue contention.
        Type Parameters:
        E - the type of elements held in this collection
        Returns:
        a new queue where producers complete their operation immediately if combined with another producing thread's
      • linearizable

        public static <E> SingleConsumerQueue<E> linearizable()
        Creates a queue with a linearizable backoff strategy. A thread waits for a completion signal if it successfully hands off the additional element(s) to another producing thread for batch insertion.
        Type Parameters:
        E - the type of elements held in this collection
        Returns:
        a new queue where producers wait for a completion signal after combining its addition with another producing thread's
      • isEmpty

        public boolean isEmpty()
        Specified by:
        isEmpty in interface java.util.Collection<E>
        Overrides:
        isEmpty in class java.util.AbstractCollection<E>
      • size

        public int size()
        Specified by:
        size in interface java.util.Collection<E>
        Specified by:
        size in class java.util.AbstractCollection<E>
      • contains

        public boolean contains​(java.lang.Object o)
        Specified by:
        contains in interface java.util.Collection<E>
        Overrides:
        contains in class java.util.AbstractCollection<E>
      • peek

        public E peek()
        Specified by:
        peek in interface java.util.Queue<E>
      • offer

        public boolean offer​(E e)
        Specified by:
        offer in interface java.util.Queue<E>
      • poll

        public E poll()
        Specified by:
        poll in interface java.util.Queue<E>
      • add

        public boolean add​(E e)
        Specified by:
        add in interface java.util.Collection<E>
        Specified by:
        add in interface java.util.Queue<E>
        Overrides:
        add in class java.util.AbstractQueue<E>
      • addAll

        public boolean addAll​(java.util.Collection<? extends E> c)
        Specified by:
        addAll in interface java.util.Collection<E>
        Overrides:
        addAll in class java.util.AbstractQueue<E>
      • transferOrCombine

        @Nullable
        SingleConsumerQueue.Node<E> transferOrCombine​(@Nonnull
                                                      SingleConsumerQueue.Node<E> first,
                                                      SingleConsumerQueue.Node<E> last)
        Attempts to receive a linked list from a waiting producer or transfer the specified linked list to an arriving producer.
        Parameters:
        first - the first node in the linked list to try to transfer
        last - the last node in the linked list to try to transfer
        Returns:
        either null if the element was transferred, the first node if neither a transfer nor receive were successful, or the received last element from a producer
      • index

        static int index()
        Returns the arena index for the current thread.
      • iterator

        public java.util.Iterator<E> iterator()
        Specified by:
        iterator in interface java.util.Collection<E>
        Specified by:
        iterator in interface java.lang.Iterable<E>
        Specified by:
        iterator in class java.util.AbstractCollection<E>
      • writeReplace

        java.lang.Object writeReplace()
      • readObject

        private void readObject​(java.io.ObjectInputStream stream)
                         throws java.io.InvalidObjectException
        Throws:
        java.io.InvalidObjectException