RingConsumer Class — netty Architecture

Source Code

transport-classes-io_uring/src/main/java/io/netty/channel/uring/IoUringBufferRing.java lines 81–158

    private final class RingConsumer implements Consumer<ByteBuf> {
        private int expectedBuffers;
        private short num;
        private short bid;
        private short oldTail;

        short fill(short startBid, int numBuffers) {
            // Fetch the tail once before allocate the batch.
            oldTail = (short) SHORT_HANDLE.get(ioUringBufRing, tailFieldPosition);

            // At the moment we always start with bid 0 and so num and bid is the same. As this is more of an
            // implementation detail it is better to still keep both separated.
            this.num = 0;
            this.bid = startBid;
            this.expectedBuffers = numBuffers;
            try {
                if (batchAllocation) {
                    allocator.allocateBatch(this, numBuffers);
                } else {
                    for (int i = 0; i < numBuffers; i++) {
                        add(oldTail, bid++, num++, allocator.allocate());
                    }
                }
            } catch (Throwable t) {
                corrupted = true;
                for (int i = 0; i < buffers.length; i++) {
                    ByteBuf buffer = buffers[i];
                    if (buffer != null) {
                        buffer.release();
                        buffers[i] = null;
                    }
                }
                throw t;
            }
            // Now advanced the tail by the number of buffers that we just added.
            SHORT_HANDLE.setRelease(ioUringBufRing, tailFieldPosition, (short) (oldTail + num));

            return (short) (bid - 1);
        }

        void fill(short bid) {
            short tail = (short) SHORT_HANDLE.get(ioUringBufRing, tailFieldPosition);
            add(tail, bid, 0, allocator.allocate());
            // Now advanced the tail by one
            SHORT_HANDLE.setRelease(ioUringBufRing, tailFieldPosition, (short) (tail + 1));
        }

        @Override
        public void accept(ByteBuf byteBuf) {
            if (corrupted || closed) {
                byteBuf.release();
                throw new IllegalStateException("Already closed");
            }
            if (expectedBuffers == num) {
                byteBuf.release();
                throw new IllegalStateException("Produced too many buffers");
            }
            add(oldTail, bid++, num++, byteBuf);
        }

        private void add(int tail, short bid, int offset, ByteBuf byteBuf) {
            short ringIndex = (short) ((tail + offset) & mask);
            assert buffers[bid] == null;

            long memoryAddress = IoUring.memoryAddress(byteBuf) + byteBuf.writerIndex();
            int writable = byteBuf.writableBytes();

            //  see:
            //  https://github.com/axboe/liburing/
            //      blob/19134a8fffd406b22595a5813a3e319c19630ac9/src/include/liburing.h#L1561
            int position = Native.SIZEOF_IOURING_BUF * ringIndex;
            ioUringBufRing.putLong(position + Native.IOURING_BUFFER_OFFSETOF_ADDR, memoryAddress);
            ioUringBufRing.putInt(position + Native.IOURING_BUFFER_OFFSETOF_LEN, writable);
            ioUringBufRing.putShort(position + Native.IOURING_BUFFER_OFFSETOF_BID, bid);

            buffers[bid] = byteBuf;
        }
    }