abstract class MpUnboundedXaddArrayQueue<R extends MpUnboundedXaddChunk<R,E>,E> extends MpUnboundedXaddArrayQueuePad5<R,E> implements MessagePassingQueue<E>, QueueProgressIndicators
MessagePassingQueue.Consumer<T>, MessagePassingQueue.ExitCondition, MessagePassingQueue.Supplier<T>, MessagePassingQueue.WaitStrategy
Modifier and Type | Field and Description |
---|---|
(package private) int |
chunkMask |
(package private) int |
chunkShift |
(package private) SpscArrayQueue<R> |
freeChunksPool |
(package private) int |
maxPooledChunks |
private static long |
ROTATION |
b000, b001, b002, b003, b004, b005, b006, b007, b010, b011, b012, b013, b014, b015, b016, b017, b020, b021, b022, b023, b024, b025, b026, b027, b030, b031, b032, b033, b034, b035, b036, b037, b040, b041, b042, b043, b044, b045, b046, b047, b050, b051, b052, b053, b054, b055, b056, b057, b060, b061, b062, b063, b064, b065, b066, b067, b070, b071, b072, b073, b074, b075, b076, b077, b100, b101, b102, b103, b104, b105, b106, b107, b110, b111, b112, b113, b114, b115, b116, b117, b120, b121, b122, b123, b124, b125, b126, b127, b130, b131, b132, b133, b134, b135, b136, b137, b140, b141, b142, b143, b144, b145, b146, b147, b150, b151, b152, b153, b154, b155, b156, b157, b160, b161, b162, b163, b164, b165, b166, b167
b170, b171, b172, b173, b174, b175, b176, b177
UNBOUNDED_CAPACITY
Constructor and Description |
---|
MpUnboundedXaddArrayQueue(int chunkSize,
int maxPooledChunks) |
Modifier and Type | Method and Description |
---|---|
protected R |
appendNextChunks(R currentChunk,
long currentChunkIndex,
long chunksToAppend) |
int |
capacity() |
int |
chunkSize() |
long |
currentConsumerIndex()
This method has no concurrent visibility semantics.
|
long |
currentProducerIndex()
This method has no concurrent visibility semantics.
|
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 wait,
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.
|
void |
fill(MessagePassingQueue.Supplier<E> s,
MessagePassingQueue.WaitStrategy wait,
MessagePassingQueue.ExitCondition exit)
Stuff the queue with elements from the supplier forever.
|
boolean |
isEmpty()
This method's accuracy is subject to concurrent modifications happening as the observation is carried
out.
|
java.util.Iterator<E> |
iterator() |
int |
maxPooledChunks() |
(package private) void |
moveToNextConsumerChunk(R cChunk,
R next)
Does not null out the first element of `next`, callers must do that
|
(package private) abstract R |
newChunk(long index,
R prev,
int chunkSize,
boolean pooled) |
private R |
newOrPooledChunk(R prevChunk,
long nextChunkIndex) |
(package private) R |
producerChunkForIndex(R initialChunk,
long requiredChunkIndex)
We're here because currentChunk.index doesn't match the expectedChunkIndex.
|
boolean |
relaxedOffer(E e)
Called from a producer thread subject to the restrictions appropriate to the implementation.
|
int |
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.
|
java.lang.String |
toString() |
casConsumerIndex, lpConsumerChunk, lpConsumerIndex, lvConsumerChunk, lvConsumerIndex, soConsumerChunk, soConsumerIndex
casProducerChunkIndex, lvProducerChunk, lvProducerChunkIndex, soProducerChunk, soProducerChunkIndex
getAndAddProducerIndex, getAndIncrementProducerIndex, lvProducerIndex
contains, containsAll, remove, removeAll, retainAll, toArray, toArray
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
clear, fill, offer, peek, poll, relaxedPeek, relaxedPoll
private static final long ROTATION
final int chunkMask
final int chunkShift
final int maxPooledChunks
final SpscArrayQueue<R extends MpUnboundedXaddChunk<R,E>> freeChunksPool
MpUnboundedXaddArrayQueue(int chunkSize, int maxPooledChunks)
chunkSize
- The buffer size to be used in each chunk of this queuemaxPooledChunks
- The maximum number of reused chunks kept around to avoid allocation, chunks are pre-allocatedpublic final int chunkSize()
public final int maxPooledChunks()
public long currentProducerIndex()
QueueProgressIndicators
currentProducerIndex
in interface QueueProgressIndicators
public long currentConsumerIndex()
QueueProgressIndicators
currentConsumerIndex
in interface QueueProgressIndicators
final R producerChunkForIndex(R initialChunk, long requiredChunkIndex)
initialChunk
- the starting point chunk, which does not match the required chunk indexrequiredChunkIndex
- the chunk index we needprotected final R appendNextChunks(R currentChunk, long currentChunkIndex, long chunksToAppend)
final void moveToNextConsumerChunk(R cChunk, R next)
public java.util.Iterator<E> iterator()
public int size()
MessagePassingQueue
size
in interface java.util.Collection<E>
size
in interface MessagePassingQueue<E>
size
in class java.util.AbstractCollection<E>
Integer.MAX_VALUE
but less or equals to
capacity (if bounded).public boolean isEmpty()
MessagePassingQueue
isEmpty
in interface java.util.Collection<E>
isEmpty
in interface MessagePassingQueue<E>
isEmpty
in class java.util.AbstractCollection<E>
public int capacity()
capacity
in interface IndexedQueueSizeUtil.IndexedQueue
capacity
in interface MessagePassingQueue<E>
MessagePassingQueue.UNBOUNDED_CAPACITY
if not boundedpublic boolean relaxedOffer(E e)
MessagePassingQueue
Queue.offer(Object)
this method may return false without the queue being full.relaxedOffer
in interface MessagePassingQueue<E>
e
- not null
, will throw NPE if it ispublic 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.
WARNING: Explicit assumptions are made with regards to MessagePassingQueue.Consumer.accept(T)
make sure you have read
and understood these before using this method.
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.
Unbounded queues will fill up the queue with a fixed amount rather than fill up to oblivion.
WARNING: Explicit assumptions are made with regards to MessagePassingQueue.Supplier.get()
make sure you have read
and understood these before using this method.
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.
WARNING: Explicit assumptions are made with regards to MessagePassingQueue.Consumer.accept(T)
make sure you have read
and understood these before using this method.
drain
in interface MessagePassingQueue<E>
public void drain(MessagePassingQueue.Consumer<E> c, MessagePassingQueue.WaitStrategy wait, 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.
WARNING: Explicit assumptions are made with regards to MessagePassingQueue.Consumer.accept(T)
make sure you have read
and understood these before using this method.
drain
in interface MessagePassingQueue<E>
public void fill(MessagePassingQueue.Supplier<E> s, MessagePassingQueue.WaitStrategy wait, MessagePassingQueue.ExitCondition exit)
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. The main difference
being that implementors MUST assure room in the queue is available BEFORE calling MessagePassingQueue.Supplier.get()
.
WARNING: Explicit assumptions are made with regards to MessagePassingQueue.Supplier.get()
make sure you have read
and understood these before using this method.
fill
in interface MessagePassingQueue<E>
public java.lang.String toString()
toString
in class java.util.AbstractCollection<E>