Class SingleConsumerQueue<E>
- java.lang.Object
-
- java.util.AbstractCollection<E>
-
- java.util.AbstractQueue<E>
-
- com.github.benmanes.caffeine.SCQHeader.PadHead<E>
-
- com.github.benmanes.caffeine.SCQHeader.HeadRef<E>
-
- com.github.benmanes.caffeine.SCQHeader.PadHeadAndTail<E>
-
- com.github.benmanes.caffeine.SCQHeader.HeadAndTailRef<E>
-
- com.github.benmanes.caffeine.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 ofnull
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
-
-
Nested Class Summary
Nested Classes Modifier and Type Class Description (package private) static class
SingleConsumerQueue.LinearizableNode<E>
(package private) static class
SingleConsumerQueue.Node<E>
(package private) static class
SingleConsumerQueue.SerializationProxy<E>
A proxy that is serialized instead of the queue.
-
Field Summary
Fields Modifier and Type Field Description (package private) java.util.concurrent.atomic.AtomicReference<SingleConsumerQueue.Node<E>>[]
arena
(package private) static int
ARENA_LENGTH
The number of slots in the elimination array.(package private) static int
ARENA_MASK
The mask value for indexing into the arena.(package private) java.util.function.Function<E,SingleConsumerQueue.Node<E>>
factory
(package private) static int
NCPU
The number of CPUs(package private) static java.util.function.Function<?,?>
OPTIMISIC
The factory for creating an optimistic node.(package private) static long
PROBE
The offset to the thread-specific probe field.(package private) static long
serialVersionUID
(package private) static int
SPINS
The number of times to spin (doing nothing except polling a memory location) before giving up while waiting to eliminate an operation.-
Fields inherited from class com.github.benmanes.caffeine.SCQHeader.HeadAndTailRef
tail, TAIL_OFFSET
-
Fields inherited from class com.github.benmanes.caffeine.SCQHeader.PadHeadAndTail
p20, p21, p22, p23, p24, p25, p26, p27, p30, p31, p32, p33, p34, p35, p36
-
Fields inherited from class com.github.benmanes.caffeine.SCQHeader.HeadRef
head
-
-
Constructor Summary
Constructors Modifier Constructor Description private
SingleConsumerQueue(java.util.function.Function<E,SingleConsumerQueue.Node<E>> factory)
-
Method Summary
All Methods Static Methods Instance Methods Concrete Methods Modifier and Type Method Description boolean
add(E e)
boolean
addAll(java.util.Collection<? extends E> c)
(package private) void
append(SingleConsumerQueue.Node<E> first, SingleConsumerQueue.Node<E> last)
Adds the linked list of nodes to the queue.(package private) static int
ceilingPowerOfTwo(int x)
boolean
contains(java.lang.Object o)
(package private) static <E> SingleConsumerQueue.Node<E>
findLast(SingleConsumerQueue.Node<E> node)
Returns the last node in the linked list.(package private) static int
index()
Returns the arena index for the current thread.boolean
isEmpty()
java.util.Iterator<E>
iterator()
static <E> SingleConsumerQueue<E>
linearizable()
Creates a queue with a linearizable backoff strategy.boolean
offer(E e)
static <E> SingleConsumerQueue<E>
optimistic()
Creates a queue with an optimistic backoff strategy.E
peek()
E
poll()
private void
readObject(java.io.ObjectInputStream stream)
int
size()
(package private) SingleConsumerQueue.Node<E>
transferOrCombine(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.(package private) java.lang.Object
writeReplace()
-
Methods inherited from class com.github.benmanes.caffeine.SCQHeader.HeadAndTailRef
casTail, lazySetTail
-
Methods inherited from class java.util.AbstractCollection
containsAll, remove, removeAll, retainAll, toArray, toArray, toString
-
Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
-
-
-
-
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.
-
arena
final java.util.concurrent.atomic.AtomicReference<SingleConsumerQueue.Node<E>>[] arena
-
factory
final java.util.function.Function<E,SingleConsumerQueue.Node<E>> factory
-
serialVersionUID
static final long serialVersionUID
- See Also:
- Constant Field Values
-
-
Constructor Detail
-
SingleConsumerQueue
private SingleConsumerQueue(java.util.function.Function<E,SingleConsumerQueue.Node<E>> factory)
-
-
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()
-
size
public int size()
-
contains
public boolean contains(java.lang.Object o)
-
add
public boolean add(E e)
-
addAll
public boolean addAll(java.util.Collection<? extends E> c)
-
append
void append(@Nonnull SingleConsumerQueue.Node<E> first, @Nonnull SingleConsumerQueue.Node<E> last)
Adds the linked list of nodes to the queue.
-
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 transferlast
- 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.
-
findLast
@Nonnull static <E> SingleConsumerQueue.Node<E> findLast(@Nonnull SingleConsumerQueue.Node<E> node)
Returns the last node in the linked list.
-
iterator
public java.util.Iterator<E> iterator()
-
writeReplace
java.lang.Object writeReplace()
-
readObject
private void readObject(java.io.ObjectInputStream stream) throws java.io.InvalidObjectException
- Throws:
java.io.InvalidObjectException
-
-