Package io.reactivex.rxjava3.operators
Class SpscLinkedArrayQueue<T>
- java.lang.Object
-
- io.reactivex.rxjava3.operators.SpscLinkedArrayQueue<T>
-
- Type Parameters:
T
- the contained value type
- All Implemented Interfaces:
SimplePlainQueue<T>
,SimpleQueue<T>
public final class SpscLinkedArrayQueue<T> extends java.lang.Object implements SimplePlainQueue<T>
A single-producer single-consumer array-backed queue which can allocate new arrays in case the consumer is slower than the producer.- Since:
- 3.1.1
-
-
Field Summary
Fields Modifier and Type Field Description (package private) java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object>
consumerBuffer
(package private) java.util.concurrent.atomic.AtomicLong
consumerIndex
(package private) int
consumerMask
private static java.lang.Object
HAS_NEXT
(package private) static int
MAX_LOOK_AHEAD_STEP
(package private) java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object>
producerBuffer
(package private) java.util.concurrent.atomic.AtomicLong
producerIndex
(package private) long
producerLookAhead
(package private) int
producerLookAheadStep
(package private) int
producerMask
-
Constructor Summary
Constructors Constructor Description SpscLinkedArrayQueue(int bufferSize)
Constructs a linked array-based queue instance with the given island size rounded up to the next power of 2.
-
Method Summary
All Methods Static Methods Instance Methods Concrete Methods Modifier and Type Method Description private void
adjustLookAheadStep(int capacity)
private static int
calcDirectOffset(int index)
private static int
calcWrappedOffset(long index, int mask)
void
clear()
Removes all enqueued items from this queue.boolean
isEmpty()
Returns true if the queue is empty.private long
lpConsumerIndex()
private long
lpProducerIndex()
private long
lvConsumerIndex()
private static java.lang.Object
lvElement(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> buffer, int offset)
private java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object>
lvNextBufferAndUnlink(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> curr, int nextIndex)
private long
lvProducerIndex()
private T
newBufferPeek(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> nextBuffer, long index, int mask)
private T
newBufferPoll(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> nextBuffer, long index, int mask)
boolean
offer(T e)
Atomically enqueue a single value.boolean
offer(T first, T second)
Offer two elements at the same time.T
peek()
Returns the next element in this queue without removing it ornull
if this queue is emptyT
poll()
Tries to dequeue a value (non-null) or returns null if the queue is empty.private void
resize(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> oldBuffer, long currIndex, int offset, T e, long mask)
int
size()
Returns the number of elements in the queue.private void
soConsumerIndex(long v)
private static void
soElement(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> buffer, int offset, java.lang.Object e)
private void
soNext(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> curr, java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> next)
private void
soProducerIndex(long v)
private boolean
writeToQueue(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> buffer, T e, long index, int offset)
-
-
-
Field Detail
-
MAX_LOOK_AHEAD_STEP
static final int MAX_LOOK_AHEAD_STEP
-
producerIndex
final java.util.concurrent.atomic.AtomicLong producerIndex
-
producerLookAheadStep
int producerLookAheadStep
-
producerLookAhead
long producerLookAhead
-
producerMask
final int producerMask
-
producerBuffer
java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> producerBuffer
-
consumerMask
final int consumerMask
-
consumerBuffer
java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> consumerBuffer
-
consumerIndex
final java.util.concurrent.atomic.AtomicLong consumerIndex
-
HAS_NEXT
private static final java.lang.Object HAS_NEXT
-
-
Method Detail
-
offer
public boolean offer(T e)
Atomically enqueue a single value.This implementation is correct for single producer thread use only.
- Specified by:
offer
in interfaceSimpleQueue<T>
- Parameters:
e
- the value to enqueue, not null- Returns:
- true if successful, false if the value was not enqueued likely due to reaching the queue capacity)
-
writeToQueue
private boolean writeToQueue(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> buffer, T e, long index, int offset)
-
resize
private void resize(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> oldBuffer, long currIndex, int offset, T e, long mask)
-
soNext
private void soNext(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> curr, java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> next)
-
lvNextBufferAndUnlink
private java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> lvNextBufferAndUnlink(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> curr, int nextIndex)
-
poll
@Nullable public T poll()
Tries to dequeue a value (non-null) or returns null if the queue is empty.If the producer uses
SimpleQueue.offer(Object, Object)
and when polling in pairs, if the first poll() returns a non-null item, the second poll() is guaranteed to return a non-null item as well.This implementation is correct for single consumer thread use only.
- Specified by:
poll
in interfaceSimplePlainQueue<T>
- Specified by:
poll
in interfaceSimpleQueue<T>
- Returns:
- the item or null to indicate an empty queue
-
newBufferPoll
private T newBufferPoll(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> nextBuffer, long index, int mask)
-
peek
@Nullable public T peek()
Returns the next element in this queue without removing it ornull
if this queue is empty- Returns:
- the next element or
null
-
newBufferPeek
private T newBufferPeek(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> nextBuffer, long index, int mask)
-
clear
public void clear()
Description copied from interface:SimpleQueue
Removes all enqueued items from this queue.- Specified by:
clear
in interfaceSimpleQueue<T>
-
size
public int size()
Returns the number of elements in the queue.- Returns:
- the number of elements in the queue
-
isEmpty
public boolean isEmpty()
Description copied from interface:SimpleQueue
Returns true if the queue is empty.Note however that due to potential fused functions in
SimpleQueue.poll()
it is possible this method returns false but then poll() returns null because the fused function swallowed the available item(s).- Specified by:
isEmpty
in interfaceSimpleQueue<T>
- Returns:
- true if the queue is empty
-
adjustLookAheadStep
private void adjustLookAheadStep(int capacity)
-
lvProducerIndex
private long lvProducerIndex()
-
lvConsumerIndex
private long lvConsumerIndex()
-
lpProducerIndex
private long lpProducerIndex()
-
lpConsumerIndex
private long lpConsumerIndex()
-
soProducerIndex
private void soProducerIndex(long v)
-
soConsumerIndex
private void soConsumerIndex(long v)
-
calcWrappedOffset
private static int calcWrappedOffset(long index, int mask)
-
calcDirectOffset
private static int calcDirectOffset(int index)
-
soElement
private static void soElement(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> buffer, int offset, java.lang.Object e)
-
lvElement
private static java.lang.Object lvElement(java.util.concurrent.atomic.AtomicReferenceArray<java.lang.Object> buffer, int offset)
-
offer
public boolean offer(T first, T second)
Offer two elements at the same time.Don't use the regular offer() with this at all!
- Specified by:
offer
in interfaceSimpleQueue<T>
- Parameters:
first
- the first value, not nullsecond
- the second value, not null- Returns:
- true if the queue accepted the two new values
-
-