UringServerChannelUnsafe Class — netty Architecture

Source Code

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

    private final class UringServerChannelUnsafe extends AbstractIoUringChannel.AbstractUringUnsafe {

        @Override
        protected int scheduleWriteMultiple(ChannelOutboundBuffer in) {
            throw new UnsupportedOperationException();
        }

        @Override
        protected int scheduleWriteSingle(Object msg) {
            throw new UnsupportedOperationException();
        }

        @Override
        boolean writeComplete0(byte op, int res, int flags, short data, int outstanding) {
            throw new UnsupportedOperationException();
        }

        @Override
        protected int scheduleRead0(boolean first, boolean socketIsEmpty) {
            assert acceptId == 0;
            final IoUringRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
            allocHandle.attemptedBytesRead(1);

            int fd = fd().intValue();
            IoRegistration registration = registration();

            final short ioPrio;

            final long acceptedAddressMemoryAddress;
            final long acceptedAddressLengthMemoryAddress;
            if (IoUring.isAcceptMultishotEnabled()) {
                // Let's use multi-shot accept to reduce overhead.
                ioPrio = Native.IORING_ACCEPT_MULTISHOT;
                acceptedAddressMemoryAddress = 0;
                acceptedAddressLengthMemoryAddress = 0;
            } else {
                // Depending on if socketIsEmpty is true we will arm the poll upfront and skip the initial transfer
                // attempt.
                // See https://github.com/axboe/liburing/wiki/io_uring-and-networking-in-2023#socket-state
                //
                // Depending on if this is the first read or not we will use Native.IORING_ACCEPT_DONT_WAIT.
                // The idea is that if the socket is blocking we can do the first read in a blocking fashion
                // and so not need to also register POLLIN. As we can not 100 % sure if reads after the first will
                // be possible directly we schedule these with Native.IORING_ACCEPT_DONT_WAIT. This allows us to still
                // be able to signal the fireChannelReadComplete() in a timely manner and be consistent with other
                // transports.
                //
                // IORING_ACCEPT_POLL_FIRST and IORING_ACCEPT_DONTWAIT were added in the same release.
                // We need to check if its supported as otherwise providing these would result in an -EINVAL.
                if (IoUring.isAcceptNoWaitSupported()) {
                    if (first) {
                        ioPrio = socketIsEmpty ? Native.IORING_ACCEPT_POLL_FIRST : 0;
                    } else {
                        ioPrio = Native.IORING_ACCEPT_DONTWAIT;
                    }
                } else {
                    ioPrio = 0;
                }

                assert acceptedAddressMemory != null;
                acceptedAddressMemoryAddress = acceptedAddressMemory.acceptedAddressMemoryAddress;
                acceptedAddressLengthMemoryAddress = acceptedAddressMemory.acceptedAddressLengthMemoryAddress;
            }

            // See https://github.com/axboe/liburing/wiki/What's-new-with-io_uring-in-6.10#improvements-for-accept
            IoUringIoOps ops = IoUringIoOps.newAccept(fd, (byte) 0, 0, ioPrio,
                    acceptedAddressMemoryAddress, acceptedAddressLengthMemoryAddress, nextOpsId());
            acceptId = registration.submit(ops);
            if (acceptId == 0) {
                return 0;
            }
            if ((ioPrio & Native.IORING_ACCEPT_MULTISHOT) != 0) {
                // Let's return -1 to signal that we used multi-shot.
                return -1;
            }
            return 1;
        }

        @Override
        protected void readComplete0(byte op, int res, int flags, short data, int outstanding) {
            if (res == Native.ERRNO_ECANCELED_NEGATIVE) {