Top Description Fields Constructors Methods
io.reactivex.internal.queue

public final Class SpscArrayQueue<E>

extends AtomicReferenceArray<E>
implements SimplePlainQueue<E>
Class Inheritance
All Implemented Interfaces
io.reactivex.internal.fuseable.SimplePlainQueue, io.reactivex.internal.fuseable.SimpleQueue
Type Parameters
<E>
the element type of the queue
Imports
java.util.concurrent.atomic.*, io.reactivex.annotations.Nullable, io.reactivex.internal.fuseable.SimplePlainQueue, io.reactivex.internal.util.Pow2
A Single-Producer-Single-Consumer queue backed by a pre-allocated buffer.

This implementation is a mashup of the Fast Flow algorithm with an optimization of the offer method taken from the BQueue algorithm (a variation on Fast Flow), and adjusted to comply with Queue.offer semantics with regards to capacity.
For convenience the relevant papers are available in the resources folder:
2010 - Pisa - SPSC Queues on Shared Cache Multi-Core Systems.pdf
2012 - Junchang- BQueue- Efficient and Practical Queuing.pdf
This implementation is wait free.

Field Summary

Modifier and TypeField and Description
pack-priv final AtomicLong
consumerIndex

pack-priv final int
lookAheadStep

pack-priv final int
mask

private static final Integer
MAX_LOOK_AHEAD_STEP

pack-priv final AtomicLong
producerIndex

pack-priv long
producerLookAhead

private static final long
serialVersionUID

Hides java.util.concurrent.atomic.AtomicReferenceArray.serialVersionUID.

Constructor Summary

AccessConstructor and Description
public
SpscArrayQueue(int capacity)

Method Summary

Modifier and TypeMethod and Description
pack-priv int
calcElementOffset(long index, int mask)

pack-priv int
calcElementOffset(long index)

public void
clear()

Implements io.reactivex.internal.fuseable.SimpleQueue.clear.

Removes all enqueued items from this queue.
public boolean
isEmpty()

Implements io.reactivex.internal.fuseable.SimpleQueue.isEmpty.

Returns true if the queue is empty.
pack-priv E
lvElement(int offset)

public boolean
offer(E
the value to enqueue, not null
e)

Implements io.reactivex.internal.fuseable.SimpleQueue.offer.

Atomically enqueue a single.
public boolean
offer(E
the first value to enqueue, not null
v1, E
the second value to enqueue, not null
v2)

Implements io.reactivex.internal.fuseable.SimpleQueue.offer.

Atomically enqueue two values.
public E
poll()

Implements io.reactivex.internal.fuseable.SimplePlainQueue.poll.

Tries to dequeue a value (non-null) or returns null if the queue is empty.
pack-priv void
soConsumerIndex(long newIndex)

pack-priv void
soElement(int offset, E value)

pack-priv void
soProducerIndex(long newIndex)

Inherited from java.util.concurrent.atomic.AtomicReferenceArray:
accumulateAndGetcompareAndExchangecompareAndExchangeAcquirecompareAndExchangeReleasecompareAndSetgetgetAcquiregetAndAccumulategetAndSetgetAndUpdategetOpaquegetPlainlazySetlengthsetsetOpaquesetPlainsetReleasetoStringupdateAndGetweakCompareAndSetweakCompareAndSetAcquireweakCompareAndSetPlainweakCompareAndSetReleaseweakCompareAndSetVolatile

Field Detail

consumerIndexback to summary
pack-priv final AtomicLong consumerIndex
lookAheadStepback to summary
pack-priv final int lookAheadStep
maskback to summary
pack-priv final int mask
MAX_LOOK_AHEAD_STEPback to summary
private static final Integer MAX_LOOK_AHEAD_STEP
producerIndexback to summary
pack-priv final AtomicLong producerIndex
producerLookAheadback to summary
pack-priv long producerLookAhead
serialVersionUIDback to summary
private static final long serialVersionUID

Hides java.util.concurrent.atomic.AtomicReferenceArray.serialVersionUID.

Constructor Detail

SpscArrayQueueback to summary
public SpscArrayQueue(int capacity)

Method Detail

calcElementOffsetback to summary
pack-priv int calcElementOffset(long index, int mask)
calcElementOffsetback to summary
pack-priv int calcElementOffset(long index)
clearback to summary
public void clear()

Implements io.reactivex.internal.fuseable.SimpleQueue.clear.

Doc from io.reactivex.internal.fuseable.SimpleQueue.clear.

Removes all enqueued items from this queue.

Annotations
@Override
isEmptyback to summary
public boolean isEmpty()

Implements io.reactivex.internal.fuseable.SimpleQueue.isEmpty.

Doc from io.reactivex.internal.fuseable.SimpleQueue.isEmpty.

Returns true if the queue is empty.

Note however that due to potential fused functions in poll() it is possible this method returns false but then poll() returns null because the fused function swallowed the available item(s).

Returns:boolean

true if the queue is empty

Annotations
@Override
lvElementback to summary
pack-priv E lvElement(int offset)
offerback to summary
public boolean offer(E e)

Implements io.reactivex.internal.fuseable.SimpleQueue.offer.

Doc from io.reactivex.internal.fuseable.SimpleQueue.offer.

Atomically enqueue a single.

Parameters
e:E

the value to enqueue, not null

Returns:boolean

true if successful, false if the value was not enqueued likely due to reaching the queue capacity)

Annotations
@Override
offerback to summary
public boolean offer(E v1, E v2)

Implements io.reactivex.internal.fuseable.SimpleQueue.offer.

Doc from io.reactivex.internal.fuseable.SimpleQueue.offer.

Atomically enqueue two values.

Parameters
v1:E

the first value to enqueue, not null

v2:E

the second value to enqueue, not null

Returns:boolean

true if successful, false if the value was not enqueued likely due to reaching the queue capacity)

Annotations
@Override
pollback to summary
public E poll()

Implements io.reactivex.internal.fuseable.SimplePlainQueue.poll.

Doc from io.reactivex.internal.fuseable.SimpleQueue.poll.

Tries to dequeue a value (non-null) or returns null if the queue is empty.

If the producer uses 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.

Returns:E

the item or null to indicate an empty queue

Annotations
@Nullable
@Override
soConsumerIndexback to summary
pack-priv void soConsumerIndex(long newIndex)
soElementback to summary
pack-priv void soElement(int offset, E value)
soProducerIndexback to summary
pack-priv void soProducerIndex(long newIndex)