readComplete0() — netty Function Reference

Source Code

transport-classes-io_uring/src/main/java/io/netty/channel/uring/AbstractIoUringServerChannel.java lines 210–271

        @Override
        protected void readComplete0(byte op, int res, int flags, short data, int outstanding) {
            if (res == Native.ERRNO_ECANCELED_NEGATIVE) {
                acceptId = 0;
                return;
            }
            boolean rearm = (flags & Native.IORING_CQE_F_MORE) == 0;
            if (rearm) {
                // Only reset if we don't use multi-shot or we need to re-arm because the multi-shot was cancelled.
                acceptId = 0;
            }
            final IoUringRecvByteAllocatorHandle allocHandle =
                    (IoUringRecvByteAllocatorHandle) unsafe()
                            .recvBufAllocHandle();
            final ChannelPipeline pipeline = pipeline();
            allocHandle.lastBytesRead(res);

            if (res >= 0) {
                allocHandle.incMessagesRead(1);
                final ByteBuffer acceptedAddressBuffer;
                final long acceptedAddressLengthMemoryAddress;
                if (acceptedAddressMemory == null) {
                    acceptedAddressBuffer = null;
                } else {
                    acceptedAddressBuffer = acceptedAddressMemory.acceptedAddressMemory;
                }
                try {
                    Channel channel = newChildChannel(res, acceptedAddressBuffer);
                    pipeline.fireChannelRead(channel);

                    if (allocHandle.continueReading() &&
                            // If IORING_CQE_F_SOCK_NONEMPTY is supported we should check for it first before
                            // trying to schedule a read. If it's supported and not part of the flags we know for sure
                            // that the next read (which would be using Native.IORING_ACCEPT_DONTWAIT) will complete
                            // without be able to read any data. This is useless work and we can skip it.
                            //
                            // See https://github.com/axboe/liburing/wiki/What's-new-with-io_uring-in-6.10
                            !socketIsEmpty(flags)) {
                        if (rearm) {
                            // We only should schedule another read if we need to rearm.
                            // See https://github.com/axboe/liburing/wiki/io_uring-and-networking-in-2023#multi-shot
                            scheduleRead(false);
                        }
                    } else {
                        allocHandle.readComplete();
                        pipeline.fireChannelReadComplete();
                    }
                } catch (Throwable cause) {
                    allocHandle.readComplete();
                    pipeline.fireChannelReadComplete();
                    pipeline.fireExceptionCaught(cause);
                }
            } else {
                allocHandle.readComplete();
                pipeline.fireChannelReadComplete();
                // Check if we did fail because there was nothing to accept atm.
                if (res != ERRNO_EAGAIN_NEGATIVE && res != ERRNO_EWOULDBLOCK_NEGATIVE) {
                    // Something bad happened. Convert to an exception.
                    pipeline.fireExceptionCaught(Errors.newIOException("io_uring accept", res));
                }
            }
        }