CompletionQueue Class — netty Architecture

Source Code

transport-classes-io_uring/src/main/java/io/netty/channel/uring/CompletionQueue.java lines 27–201

final class CompletionQueue {
    private static final VarHandle INT_HANDLE =
            MethodHandles.byteBufferViewVarHandle(int[].class, ByteOrder.nativeOrder());

    //these offsets are used to access specific properties
    //CQE (https://github.com/axboe/liburing/blob/master/src/include/liburing/io_uring.h#L162)
    private static final int CQE_USER_DATA_FIELD = 0;
    private static final int CQE_RES_FIELD = 8;
    private static final int CQE_FLAGS_FIELD = 12;

    //these unsigned integer pointers(shared with the kernel) will be changed by the kernel and us
    // using a VarHandle.
    private final ByteBuffer khead;
    private final ByteBuffer ktail;
    private final ByteBuffer kflags;
    private final ByteBuffer completionQueueArray;
    private final ByteBuffer[] extraCqeData;

    final int ringSize;
    final long ringAddress;
    final int ringFd;
    final int ringEntries;
    final int ringCapacity;
    private final int cqeLength;

    private final int ringMask;
    private int ringHead;
    private boolean closed;

    CompletionQueue(ByteBuffer kHead, ByteBuffer kTail, int ringMask, int ringEntries, ByteBuffer kflags,
                    ByteBuffer completionQueueArray, int ringSize, long ringAddress,
                    int ringFd, int ringCapacity, int cqeLength, boolean extraCqeDataNeeded) {
        this.khead = kHead;
        this.ktail = kTail;
        this.completionQueueArray = completionQueueArray;
        this.ringSize = ringSize;
        this.ringAddress = ringAddress;
        this.ringFd = ringFd;
        this.ringCapacity = ringCapacity;
        this.cqeLength = cqeLength;
        this.ringEntries = ringEntries;
        this.kflags = kflags;
        this.ringMask = ringMask;
        ringHead = (int) INT_HANDLE.getVolatile(kHead, 0);

        if (extraCqeDataNeeded) {
            // Let's create the slices up front to reduce GC-pressure and also ensure that the user
            // can not escape the memory range.
            // We slice every Native.CQE_SIZE to support IORING_SETUP_CQE32 and IORING_SETUP_CQE_MIXED.
            this.extraCqeData = new ByteBuffer[ringEntries];
            for (int i = 0; i < ringEntries; i++) {
                int position = i * cqeLength;
                completionQueueArray.position(position).limit(position + Native.CQE_SIZE);
                extraCqeData[i] = completionQueueArray.slice();
                completionQueueArray.clear();
            }
        } else {
            this.extraCqeData = null;
        }
    }

    void close() {
        closed = true;
    }

    int flags() {
        if (closed) {
            return 0;
        }
        // we only need memory_order_relaxed
        return (int) INT_HANDLE.getOpaque(kflags, 0);
    }

    /**
     * Returns {@code true} if any completion event is ready to be processed by
     * {@link #process(CompletionCallback)}, {@code false} otherwise.
     */
    boolean hasCompletions() {
        return !closed && ringHead != (int) INT_HANDLE.getVolatile(ktail, 0);
    }