IoUringBufferRing Class — netty Architecture

Source Code

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

final class IoUringBufferRing {
    private static final VarHandle SHORT_HANDLE =
            MethodHandles.byteBufferViewVarHandle(short[].class, ByteOrder.nativeOrder());
    private final ByteBuffer ioUringBufRing;
    private final int tailFieldPosition;
    private final short entries;
    private final short mask;
    private final short bufferGroupId;
    private final int ringFd;
    private final ByteBuf[] buffers;
    private final IoUringBufferRingAllocator allocator;
    private final boolean batchAllocation;
    private final IoUringBufferRingExhaustedEvent exhaustedEvent;
    private final RingConsumer ringConsumer;
    private final boolean incremental;
    private final int batchSize;
    private boolean corrupted;
    private boolean closed;
    private int usableBuffers;
    private int allocatedBuffers;
    private boolean needExpand;
    private short lastGeneratedBid;

    IoUringBufferRing(int ringFd, ByteBuffer ioUringBufRing,
                      short entries, int batchSize, short bufferGroupId, boolean incremental,
                      IoUringBufferRingAllocator allocator, boolean batchAllocation) {
        assert entries % 2 == 0;
        assert batchSize % 2 == 0;
        this.batchSize = batchSize;
        this.ioUringBufRing = ioUringBufRing;
        this.tailFieldPosition = Native.IO_URING_BUFFER_RING_TAIL;
        this.entries = entries;
        this.mask = (short) (entries - 1);
        this.bufferGroupId = bufferGroupId;
        this.ringFd = ringFd;
        this.buffers = new ByteBuf[entries];
        this.incremental = incremental;
        this.allocator = allocator;
        this.batchAllocation = batchAllocation;
        this.ringConsumer  = new RingConsumer();
        this.exhaustedEvent = new IoUringBufferRingExhaustedEvent(bufferGroupId);
    }

    boolean isUsable() {
        return !closed && !corrupted;
    }

    void initialize() {
        // We already validated that batchSize is <= ring length.
        fill((short) 0, batchSize);
        allocatedBuffers = batchSize;
    }

    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) {