MpscIntQueue Type — netty Architecture

Source Code

common/src/main/java/io/netty/util/concurrent/MpscIntQueue.java lines 31–272

public interface MpscIntQueue {
    /**
     * Create a new queue instance of the given size.
     * <p>
     * Note: the size of the queue may be rounded up to nearest power-of-2.
     *
     * @param size The required fixed size of the queue.
     * @param emptyValue The special value that the queue should use to signal the "empty" case.
     * This value will be returned from {@link #poll()} when the queue is empty,
     * and giving this value to {@link #offer(int)} will cause an exception to be thrown.
     * @return The queue instance.
     */
    static MpscIntQueue create(int size, int emptyValue) {
        return new MpscAtomicIntegerArrayQueue(size, emptyValue);
    }

    /**
     * Offer the given value to the queue. This will throw an exception if the given value is the "empty" value.
     * @param value The value to add to the queue.
     * @return {@code true} if the value was added to the queue,
     * or {@code false} if the value could not be added because the queue is full.
     */
    boolean offer(int value);

    /**
     * Remove and return the next value from the queue, or return the "empty" value if the queue is empty.
     * @return The next value or the "empty" value.
     */
    int poll();

    /**
     * Remove up to the given limit of elements from the queue, and pass them to the consumer in order.
     * @param limit The maximum number of elements to dequeue.
     * @param consumer The consumer to pass the removed elements to.
     * @return The actual number of elements removed.
     */
    int drain(int limit, IntConsumer consumer);

    /**
     * Add up to the given limit of elements to this queue, from the given supplier.
     * @param limit The maximum number of elements to enqueue.
     * @param supplier The supplier to obtain the elements from.
     * @return The actual number of elements added.
     */
    int fill(int limit, IntSupplier supplier);

    /**
     * Query if the queue is empty or not.
     * <p>
     * This method is inherently racy and the result may be out of date by the time the method returns.
     * @return {@code true} if the queue was observed to be empty, otherwise {@code false.
     */
    boolean isEmpty();

    /**
     * Query the number of elements currently in the queue.
     * <p>
     * This method is inherently racy and the result may be out of date by the time the method returns.
     * @return An estimate of the number of elements observed in the queue.
     */
    int size();

    /**
     * This implementation is based on MpscAtomicUnpaddedArrayQueue from JCTools.
     */
    final class MpscAtomicIntegerArrayQueue extends AtomicIntegerArray implements MpscIntQueue {
        private static final long serialVersionUID = 8740338425124821455L;
        private static final AtomicLongFieldUpdater<MpscAtomicIntegerArrayQueue> PRODUCER_INDEX =
                AtomicLongFieldUpdater.newUpdater(MpscAtomicIntegerArrayQueue.class, "producerIndex");
        private static final AtomicLongFieldUpdater<MpscAtomicIntegerArrayQueue> PRODUCER_LIMIT =
                AtomicLongFieldUpdater.newUpdater(MpscAtomicIntegerArrayQueue.class, "producerLimit");
        private static final AtomicLongFieldUpdater<MpscAtomicIntegerArrayQueue> CONSUMER_INDEX =
                AtomicLongFieldUpdater.newUpdater(MpscAtomicIntegerArrayQueue.class, "consumerIndex");
        private final int mask;
        private final int emptyValue;
        private volatile long producerIndex;
        private volatile long producerLimit;
        private volatile long consumerIndex;

        public MpscAtomicIntegerArrayQueue(int capacity, int emptyValue) {
            super(MathUtil.safeFindNextPositivePowerOfTwo(capacity));