org.jctools.queues.atomic

Class MpscLinkedAtomicQueue<E>

java.lang.Object

java.util.AbstractCollection<E>

java.util.AbstractQueue<E>

org.jctools.queues.atomic.BaseLinkedAtomicQueue<E>

org.jctools.queues.atomic.MpscLinkedAtomicQueue<E>

Type Parameters:

E -

All Implemented Interfaces:

java.lang.Iterable<E>, java.util.Collection<E>, java.util.Queue<E>

public final class MpscLinkedAtomicQueue<E>
extends BaseLinkedAtomicQueue<E>

This is a direct Java port of the MPSC algorithm as presented on 1024 Cores by D. Vyukov. The original has been adapted to Java and it's quirks with regards to memory model and layout:

1. Use XCHG functionality provided by AtomicReference (which is better in JDK 8+).

The queue is initialized with a stub node which is set to both the producer and consumer node references. From this point follow the notes on offer/poll.

Constructor Summary

Constructors

Constructor and Description
MpscLinkedAtomicQueue()

Method Summary

Modifier and Type	Method and Description
boolean	offer(E e)
E	peek()
E	poll()

Methods inherited from class org.jctools.queues.atomic.BaseLinkedAtomicQueue

getSingleConsumerNodeValue, isEmpty, iterator, lpConsumerNode, lpProducerNode, lvConsumerNode, lvProducerNode, size, spConsumerNode, spProducerNode, xchgProducerNode

Methods inherited from class java.util.AbstractQueue

add, addAll, clear, element, remove

Methods inherited from class java.util.AbstractCollection

contains, containsAll, remove, removeAll, retainAll, toArray, toArray, toString

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface java.util.Collection

contains, containsAll, equals, hashCode, parallelStream, remove, removeAll, removeIf, retainAll, spliterator, stream, toArray, toArray

Methods inherited from interface java.lang.Iterable

forEach

Constructor Detail

MpscLinkedAtomicQueue

public MpscLinkedAtomicQueue()

Method Detail

offer

public final boolean offer(E e)

IMPLEMENTATION NOTES:
Offer is allowed from multiple threads.
Offer allocates a new node and:

1. Swaps it atomically with current producer node (only one producer 'wins')
2. Sets the new node as the node following from the swapped producer node

This works because each producer is guaranteed to 'plant' a new node and link the old node. No 2 producers can get the same producer node as part of XCHG guarantee.

See Also:

MessagePassingQueue.offer(Object), Queue.offer(java.lang.Object)

poll

public final E poll()

IMPLEMENTATION NOTES:
Poll is allowed from a SINGLE thread.
Poll reads the next node from the consumerNode and:

1. If it is null, the queue is assumed empty (though it might not be).
2. If it is not null set it as the consumer node and return it's now evacuated value.

This means the consumerNode.value is always null, which is also the starting point for the queue. Because null values are not allowed to be offered this is the only node with it's value set to null at any one time.

See Also:

MessagePassingQueue.poll(), Queue.poll()

peek

public final E peek()