MpscAtomicIntegerArrayQueue Class — netty Architecture

Source Code

common/src/main/java/io/netty/util/concurrent/MpscIntQueue.java lines 96–271

    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));
            if (emptyValue != 0) {
                this.emptyValue = emptyValue;
                int end = length() - 1;
                for (int i = 0; i < end; i++) {
                    lazySet(i, emptyValue);
                }
                getAndSet(end, emptyValue); // 'getAndSet' acts as a full barrier, giving us initialization safety.
            } else {
                this.emptyValue = 0;
            }
            mask = length() - 1;
        }

        @Override
        public boolean offer(int value) {
            if (value == emptyValue) {
                throw new IllegalArgumentException("Cannot offer the \"empty\" value: " + emptyValue);
            }
            // use a cached view on consumer index (potentially updated in loop)
            final int mask = this.mask;
            long producerLimit = this.producerLimit;
            long pIndex;
            do {
                pIndex = producerIndex;
                if (pIndex >= producerLimit) {
                    final long cIndex = consumerIndex;
                    producerLimit = cIndex + mask + 1;
                    if (pIndex >= producerLimit) {
                        // FULL :(
                        return false;
                    } else {
                        // update producer limit to the next index that we must recheck the consumer index
                        // this is racy, but the race is benign
                        PRODUCER_LIMIT.lazySet(this, producerLimit);
                    }
                }
            } while (!PRODUCER_INDEX.compareAndSet(this, pIndex, pIndex + 1));
            /*
             * NOTE: the new producer index value is made visible BEFORE the element in the array. If we relied on
             * the index visibility to poll() we would need to handle the case where the element is not visible.
             */
            // Won CAS, move on to storing
            final int offset = (int) (pIndex & mask);
            lazySet(offset, value);
            // AWESOME :)
            return true;
        }

        @Override
        public int poll() {
            final long cIndex = consumerIndex;
            final int offset = (int) (cIndex & mask);
            // If we can't see the next available element we can't poll
            int value = get(offset);
            if (emptyValue == value) {
                /*
                 * NOTE: Queue may not actually be empty in the case of a producer (P1) being interrupted after
                 * winning the CAS on offer but before storing the element in the queue. Other producers may go on
                 * to fill up the queue after this element.
                 */
                if (cIndex != producerIndex) {
                    do {
                        value = get(offset);
                    } while (emptyValue == value);
                } else {