process() — netty Function Reference

Architecture documentation for the process() function in CompletionQueue.java from the netty codebase.

Function java Buffer Telemetry calls 1

Entity Profile

Buffer→ Telemetry→ process() — netty Function Reference

Dependency Diagram

graph TD
  d9b73f6d_bcee_b6c3_41c1_0cc0d26a184d["process()"]
  af6dd3bb_1c78_efaf_ac7c_5f13fef08613["CompletionQueue"]
  d9b73f6d_bcee_b6c3_41c1_0cc0d26a184d -->|defined in| af6dd3bb_1c78_efaf_ac7c_5f13fef08613
  04ea1afd_83d6_fa0b_816d_7f9913fa27ea["cqeIdx()"]
  d9b73f6d_bcee_b6c3_41c1_0cc0d26a184d -->|calls| 04ea1afd_83d6_fa0b_816d_7f9913fa27ea
  style d9b73f6d_bcee_b6c3_41c1_0cc0d26a184d fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

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

    int process(CompletionCallback callback) {
        if (closed) {
            return 0;
        }
        int tail = (int) INT_HANDLE.getVolatile(ktail, 0);
        try {
            int i = 0;
            while (ringHead != tail) {
                int cqeIdx = cqeIdx(ringHead, ringMask);
                int cqePosition = cqeIdx * cqeLength;

                long udata = completionQueueArray.getLong(cqePosition + CQE_USER_DATA_FIELD);
                int res = completionQueueArray.getInt(cqePosition + CQE_RES_FIELD);
                int flags = completionQueueArray.getInt(cqePosition + CQE_FLAGS_FIELD);

                ringHead++;
                final ByteBuffer extraCqeData;
                if ((flags & Native.IORING_CQE_F_32) != 0) {
                    extraCqeData = extraCqeData(cqeIdx + 1);
                    // We used mixed mode and this was a 32 byte CQE, let's increment the head once more.
                    ringHead++;
                } else if (cqeLength == Native.CQE32_SIZE) {
                    extraCqeData = extraCqeData(cqeIdx + 1);
                } else {
                    extraCqeData = null;
                }
                // Check if we should just skip it.
                if ((flags & Native.IORING_CQE_F_SKIP) == 0) {
                    i++;

                    callback.handle(res, flags, udata, extraCqeData);
                }

                if (ringHead == tail) {
                    // Let's fetch the tail one more time as it might have changed because a completion might have
                    // triggered a submission (io_uring_enter). This can happen as we automatically submit once we
                    // run out of space in the submission queue.
                    tail = (int) INT_HANDLE.getVolatile(ktail, 0);
                }
            }
            return i;
        } finally {
            // Ensure that the kernel only sees the new value of the head index after the CQEs have been read.
            INT_HANDLE.setRelease(khead, 0, ringHead);
        }
    }