E
- public class SpscArrayQueue<E> extends ConcurrentCircularArrayQueue<E> implements QueueProgressIndicators
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.
MessagePassingQueue.Consumer<T>, MessagePassingQueue.ExitCondition, MessagePassingQueue.Supplier<T>, MessagePassingQueue.WaitStrategy
Modifier and Type | Field and Description |
---|---|
protected static long |
C_INDEX_OFFSET |
protected long |
consumerIndex |
protected int |
lookAheadStep |
static int |
MAX_LOOK_AHEAD_STEP |
protected static long |
P_INDEX_OFFSET |
protected long |
producerIndex |
protected long |
producerLimit |
buffer, mask
UNBOUNDED_CAPACITY
Constructor and Description |
---|
SpscArrayQueue(int capacity) |
Modifier and Type | Method and Description |
---|---|
int |
drain(MessagePassingQueue.Consumer<E> c)
Remove all available item from the queue and hand to consume.
|
int |
drain(MessagePassingQueue.Consumer<E> c,
int limit)
Remove up to limit elements from the queue and hand to consume.
|
void |
drain(MessagePassingQueue.Consumer<E> c,
MessagePassingQueue.WaitStrategy w,
MessagePassingQueue.ExitCondition exit)
Remove elements from the queue and hand to consume forever.
|
int |
fill(MessagePassingQueue.Supplier<E> s)
Stuff the queue with elements from the supplier.
|
int |
fill(MessagePassingQueue.Supplier<E> s,
int limit)
Stuff the queue with up to limit elements from the supplier.
|
void |
fill(MessagePassingQueue.Supplier<E> s,
MessagePassingQueue.WaitStrategy w,
MessagePassingQueue.ExitCondition e)
Stuff the queue with elements from the supplier forever.
|
long |
lvConsumerIndex() |
long |
lvProducerIndex() |
boolean |
offer(E e)
Called from a producer thread subject to the restrictions appropriate to the implementation and
according to the
Queue.offer(Object) interface. |
E |
peek()
Called from the consumer thread subject to the restrictions appropriate to the implementation and
according to the
Queue.peek() interface. |
E |
poll()
Called from the consumer thread subject to the restrictions appropriate to the implementation and
according to the
Queue.poll() interface. |
boolean |
relaxedOffer(E message)
Called from a producer thread subject to the restrictions appropriate to the implementation.
|
E |
relaxedPeek()
Called from the consumer thread subject to the restrictions appropriate to the implementation.
|
E |
relaxedPoll()
Called from the consumer thread subject to the restrictions appropriate to the implementation.
|
calcElementOffset, calcElementOffset, capacity, clear, currentConsumerIndex, currentProducerIndex, isEmpty, iterator, size, toString
add, addAll, element, remove
contains, containsAll, remove, removeAll, retainAll, toArray, toArray
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
currentConsumerIndex, currentProducerIndex
contains, containsAll, equals, hashCode, parallelStream, remove, removeAll, removeIf, retainAll, spliterator, stream, toArray, toArray
protected long consumerIndex
protected static final long C_INDEX_OFFSET
protected static final long P_INDEX_OFFSET
protected long producerIndex
protected long producerLimit
public static final int MAX_LOOK_AHEAD_STEP
protected final int lookAheadStep
public boolean offer(E e)
Queue.offer(Object)
interface.
This implementation is correct for single producer thread use only.
public E poll()
Queue.poll()
interface.
This implementation is correct for single consumer thread use only.
public E peek()
Queue.peek()
interface.
This implementation is correct for single consumer thread use only.
public final long lvProducerIndex()
lvProducerIndex
in interface IndexedQueueSizeUtil.IndexedQueue
public final long lvConsumerIndex()
lvConsumerIndex
in interface IndexedQueueSizeUtil.IndexedQueue
public boolean relaxedOffer(E message)
MessagePassingQueue
Queue.offer(Object)
this method may return false without the queue being full.relaxedOffer
in interface MessagePassingQueue<E>
message
- not null, will throw NPE if it ispublic E relaxedPoll()
MessagePassingQueue
Queue.poll()
this method may return null without the queue being empty.relaxedPoll
in interface MessagePassingQueue<E>
public E relaxedPeek()
MessagePassingQueue
Queue.peek()
this method may return null without the queue being empty.relaxedPeek
in interface MessagePassingQueue<E>
public int drain(MessagePassingQueue.Consumer<E> c)
MessagePassingQueue
M m;
while((m = relaxedPoll()) != null){
c.accept(m);
}
There's no strong commitment to the queue being empty at the end of a drain. Called from a
consumer thread subject to the restrictions appropriate to the implementation.drain
in interface MessagePassingQueue<E>
public int fill(MessagePassingQueue.Supplier<E> s)
MessagePassingQueue
while(relaxedOffer(s.get());
There's no strong commitment to the queue being full at the end of a fill. Called from a
producer thread subject to the restrictions appropriate to the implementation.fill
in interface MessagePassingQueue<E>
public int drain(MessagePassingQueue.Consumer<E> c, int limit)
MessagePassingQueue
M m;
int i = 0;
for(;i < limit && (m = relaxedPoll()) != null; i++){
c.accept(m);
}
return i;
There's no strong commitment to the queue being empty at the end of a drain. Called from a consumer
thread subject to the restrictions appropriate to the implementation.drain
in interface MessagePassingQueue<E>
public int fill(MessagePassingQueue.Supplier<E> s, int limit)
MessagePassingQueue
for(int i=0; i < limit && relaxedOffer(s.get()); i++);
There's no strong commitment to the queue being full at the end of a fill. Called from a producer
thread subject to the restrictions appropriate to the implementation.fill
in interface MessagePassingQueue<E>
public void drain(MessagePassingQueue.Consumer<E> c, MessagePassingQueue.WaitStrategy w, MessagePassingQueue.ExitCondition exit)
MessagePassingQueue
int idleCounter = 0;
while (exit.keepRunning()) {
E e = relaxedPoll();
if(e==null){
idleCounter = wait.idle(idleCounter);
continue;
}
idleCounter = 0;
c.accept(e);
}
Called from a consumer thread subject to the restrictions appropriate to the implementation.drain
in interface MessagePassingQueue<E>
public void fill(MessagePassingQueue.Supplier<E> s, MessagePassingQueue.WaitStrategy w, MessagePassingQueue.ExitCondition e)
MessagePassingQueue
int idleCounter = 0;
while (exit.keepRunning()) {
E e = s.get();
while (!relaxedOffer(e)) {
idleCounter = wait.idle(idleCounter);
continue;
}
idleCounter = 0;
}
Called from a producer thread subject to the restrictions appropriate to the implementation.fill
in interface MessagePassingQueue<E>
Copyright © 2013–2017. All rights reserved.