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>
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
FieldsModifier and TypeFieldDescription(package private) AtomicReferenceArray
<Object> (package private) final AtomicLong
(package private) final int
private static final Object
(package private) static final int
(package private) AtomicReferenceArray
<Object> (package private) final AtomicLong
(package private) long
(package private) int
(package private) final int
-
Constructor Summary
ConstructorsConstructorDescriptionSpscLinkedArrayQueue
(int bufferSize) Constructs a linked array-based queue instance with the given island size rounded up to the next power of 2. -
Method Summary
Modifier and TypeMethodDescriptionprivate 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
private long
private long
private static Object
lvElement
(AtomicReferenceArray<Object> buffer, int offset) private AtomicReferenceArray
<Object> lvNextBufferAndUnlink
(AtomicReferenceArray<Object> curr, int nextIndex) private long
private T
newBufferPeek
(AtomicReferenceArray<Object> nextBuffer, long index, int mask) private T
newBufferPoll
(AtomicReferenceArray<Object> nextBuffer, long index, int mask) boolean
Atomically enqueue a single value.boolean
Offer two elements at the same time.peek()
Returns the next element in this queue without removing it ornull
if this queue is emptypoll()
Tries to dequeue a value (non-null) or returns null if the queue is empty.private void
resize
(AtomicReferenceArray<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
(AtomicReferenceArray<Object> buffer, int offset, Object e) private void
soNext
(AtomicReferenceArray<Object> curr, AtomicReferenceArray<Object> next) private void
soProducerIndex
(long v) private boolean
writeToQueue
(AtomicReferenceArray<Object> buffer, T e, long index, int offset)
-
Field Details
-
MAX_LOOK_AHEAD_STEP
static final int MAX_LOOK_AHEAD_STEP -
producerIndex
-
producerLookAheadStep
int producerLookAheadStep -
producerLookAhead
long producerLookAhead -
producerMask
final int producerMask -
producerBuffer
AtomicReferenceArray<Object> producerBuffer -
consumerMask
final int consumerMask -
consumerBuffer
AtomicReferenceArray<Object> consumerBuffer -
consumerIndex
-
HAS_NEXT
-
-
Constructor Details
-
SpscLinkedArrayQueue
public SpscLinkedArrayQueue(int bufferSize) Constructs a linked array-based queue instance with the given island size rounded up to the next power of 2.- Parameters:
bufferSize
- the maximum number of elements per island
-
-
Method Details
-
offer
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
-
resize
private void resize(AtomicReferenceArray<Object> oldBuffer, long currIndex, int offset, T e, long mask) -
soNext
-
lvNextBufferAndUnlink
private AtomicReferenceArray<Object> lvNextBufferAndUnlink(AtomicReferenceArray<Object> curr, int nextIndex) -
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
-
peek
Returns the next element in this queue without removing it ornull
if this queue is empty- Returns:
- the next element or
null
-
newBufferPeek
-
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
-
lvElement
-
offer
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
-