Top Description Fields Constructors Methods
rx.internal.util.unsafe

public Class MpmcArrayQueue<E>

Additional top-level classes in compilation unit: MpmcArrayQueueL1Pad, MpmcArrayQueueProducerField, MpmcArrayQueueL2Pad, MpmcArrayQueueConsumerField.

extends MpmcArrayQueueConsumerField<E>
Class Inheritance
Annotations
@SuppressAnimalSniffer
Type Parameters
<E>
type of the element stored in the java.util.Queue
Static Imports
rx.internal.util.unsafe.UnsafeAccess.UNSAFE
A Multi-Producer-Multi-Consumer queue based on a ConcurrentCircularArrayQueue. This implies that any and all threads may call the offer/poll/peek methods and correctness is maintained.
This implementation follows patterns documented on the package level for False Sharing protection.
The algorithm for offer/poll is an adaptation of the one put forward by D. Vyukov (See here). The original algorithm uses an array of structs which should offer nice locality properties but is sadly not possible in Java (waiting on Value Types or similar). The alternative explored here utilizes 2 arrays, one for each field of the struct. There is a further alternative in the experimental project which uses iteration phase markers to achieve the same algorithm and is closer structurally to the original, but sadly does not perform as well as this implementation.
Trade-offs to keep in mind:
  1. Padding for false sharing: counter fields and queue fields are all padded as well as either side of both arrays. We are trading memory to avoid false sharing(active and passive).
  2. 2 arrays instead of one: The algorithm requires an extra array of longs matching the size of the elements array. This is doubling/tripling the memory allocated for the buffer.
  3. Power of 2 capacity: Actual elements buffer (and sequence buffer) is the closest power of 2 larger or equal to the requested capacity.

Field Summary

Modifier and TypeField and Description
pack-priv long
p30

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p30.

pack-priv long
p31

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p31.

pack-priv long
p32

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p32.

pack-priv long
p33

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p33.

pack-priv long
p34

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p34.

pack-priv long
p35

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p35.

pack-priv long
p36

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p36.

pack-priv long
p37

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p37.

pack-priv long
p40

pack-priv long
p41

pack-priv long
p42

pack-priv long
p43

pack-priv long
p44

pack-priv long
p45

pack-priv long
p46

Constructor Summary

AccessConstructor and Description
public
MpmcArrayQueue(final int capacity)

Method Summary

Modifier and TypeMethod and Description
public boolean
isEmpty()

Overrides java.util.AbstractCollection.isEmpty.

Implements rx.internal.util.unsafe.MessagePassingQueue.isEmpty, java.util.Collection.isEmpty.

This method's accuracy is subject to concurrent modifications happening as the observation is carried out.
public boolean
offer(final E
the Object to enqueue, not null
e)

Implements rx.internal.util.unsafe.MessagePassingQueue.offer, java.util.Queue.offer.

Called from a producer thread subject to the restrictions appropriate to the implementation and according to the Queue#offer(Object) interface.
public E
peek()

Implements rx.internal.util.unsafe.MessagePassingQueue.peek, java.util.Queue.peek.

Called from the consumer thread subject to the restrictions appropriate to the implementation and according to the Queue#peek() interface.
public E
poll()

Implements rx.internal.util.unsafe.MessagePassingQueue.poll, java.util.Queue.poll.

Called from the consumer thread subject to the restrictions appropriate to the implementation and according to the Queue#poll() interface.
public int
size()

Implements abstract java.util.AbstractCollection.size.

Implements rx.internal.util.unsafe.MessagePassingQueue.size, java.util.Collection.size.

This method's accuracy is subject to concurrent modifications happening as the size is estimated and as such is a best effort rather than absolute value.
Inherited from rx.internal.util.unsafe.MpmcArrayQueueConsumerField:
casConsumerIndexlvConsumerIndex

Field Detail

p30back to summary
pack-priv long p30

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p30.

p31back to summary
pack-priv long p31

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p31.

p32back to summary
pack-priv long p32

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p32.

p33back to summary
pack-priv long p33

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p33.

p34back to summary
pack-priv long p34

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p34.

p35back to summary
pack-priv long p35

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p35.

p36back to summary
pack-priv long p36

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p36.

p37back to summary
pack-priv long p37

Hides rx.internal.util.unsafe.MpmcArrayQueueL2Pad.p37.

p40back to summary
pack-priv long p40
p41back to summary
pack-priv long p41
p42back to summary
pack-priv long p42
p43back to summary
pack-priv long p43
p44back to summary
pack-priv long p44
p45back to summary
pack-priv long p45
p46back to summary
pack-priv long p46

Constructor Detail

MpmcArrayQueueback to summary
public MpmcArrayQueue(final int capacity)

Method Detail

isEmptyback to summary
public boolean isEmpty()

Overrides java.util.AbstractCollection.isEmpty.

Implements rx.internal.util.unsafe.MessagePassingQueue.isEmpty, java.util.Collection.isEmpty.

Doc from rx.internal.util.unsafe.MessagePassingQueue.isEmpty.

This method's accuracy is subject to concurrent modifications happening as the observation is carried out.

Returns:boolean

true if empty, false otherwise

Annotations
@Override
offerback to summary
public boolean offer(final E e)

Implements rx.internal.util.unsafe.MessagePassingQueue.offer, java.util.Queue.offer.

Doc from rx.internal.util.unsafe.MessagePassingQueue.offer.

Called from a producer thread subject to the restrictions appropriate to the implementation and according to the Queue#offer(Object) interface.

Parameters
e:E

the Object to enqueue, not null

Returns:boolean

true if element was inserted into the queue, false iff full

Annotations
@Override
peekback to summary
public E peek()

Implements rx.internal.util.unsafe.MessagePassingQueue.peek, java.util.Queue.peek.

Doc from rx.internal.util.unsafe.MessagePassingQueue.peek.

Called from the consumer thread subject to the restrictions appropriate to the implementation and according to the Queue#peek() interface.

Returns:E

a message from the queue if one is available, null iff empty

Annotations
@Override
pollback to summary
public E poll()

Implements rx.internal.util.unsafe.MessagePassingQueue.poll, java.util.Queue.poll.

Doc from rx.internal.util.unsafe.MessagePassingQueue.poll.

Called from the consumer thread subject to the restrictions appropriate to the implementation and according to the Queue#poll() interface.

Because return null indicates queue is empty we cannot simply rely on next element visibility for poll and must test producer index when next element is not visible.

Returns:E

a message from the queue if one is available, null iff empty

Annotations
@Override
sizeback to summary
public int size()

Implements abstract java.util.AbstractCollection.size.

Implements rx.internal.util.unsafe.MessagePassingQueue.size, java.util.Collection.size.

Doc from rx.internal.util.unsafe.MessagePassingQueue.size.

This method's accuracy is subject to concurrent modifications happening as the size is estimated and as such is a best effort rather than absolute value. For some implementations this method may be O(n) rather than O(1).

Returns:int

number of messages in the queue, between 0 and queue capacity or Integer#MAX_VALUE if not bounded

Annotations
@Override