IoUringSocketChannel Class — netty Architecture

Source Code

transport-classes-io_uring/src/main/java/io/netty/channel/uring/IoUringSocketChannel.java lines 33–311

public final class IoUringSocketChannel extends AbstractIoUringStreamChannel implements SocketChannel {
    private final IoUringSocketChannelConfig config;

    public IoUringSocketChannel() {
       super(null, LinuxSocket.newSocketStream(), false);
       this.config = new IoUringSocketChannelConfig(this);
    }

    IoUringSocketChannel(Channel parent, LinuxSocket fd) {
        super(parent, fd, true);
        this.config = new IoUringSocketChannelConfig(this);
    }

    IoUringSocketChannel(Channel parent, LinuxSocket fd, SocketAddress remote) {
        super(parent, fd, remote);
        this.config = new IoUringSocketChannelConfig(this);
    }

    @Override
    public ServerSocketChannel parent() {
        return (ServerSocketChannel) super.parent();
    }

    @Override
    public SocketChannelConfig config() {
        return config;
    }

    @Override
    public InetSocketAddress remoteAddress() {
        return (InetSocketAddress) super.remoteAddress();
    }

    @Override
    public InetSocketAddress localAddress() {
        return (InetSocketAddress) super.localAddress();
    }

    @Override
    protected AbstractUringUnsafe newUnsafe() {
        return new IoUringSocketUnsafe();
    }

    // Marker object that is used to mark a batch of buffers that were used with zero-copy write operations.
    private static final Object ZC_BATCH_MARKER = new Object();

    private final class IoUringSocketUnsafe extends IoUringStreamUnsafe {
        /**
         * Queue that holds buffers that we can't release yet as the kernel still holds a reference to these.
         */
        private Queue<Object> zcWriteQueue;

        @Override
        protected int scheduleWriteSingle(Object msg) {
            assert writeId == 0;

            if (IoUring.isSendZcSupported() && msg instanceof ByteBuf) {
                ByteBuf buf = (ByteBuf) msg;
                int length = buf.readableBytes();
                if (((IoUringSocketChannelConfig) config()).shouldWriteZeroCopy(length)) {
                    long address = IoUring.memoryAddress(buf) + buf.readerIndex();
                    IoUringIoOps ops = IoUringIoOps.newSendZc(fd().intValue(), address, length, 0, nextOpsId(), 0);
                    byte opCode = ops.opcode();
                    writeId = registration().submit(ops);
                    writeOpCode = opCode;
                    if (writeId == 0) {
                        return 0;
                    }
                    return 1;
                }
                // Should not use send_zc, just use normal write.
            }
            return super.scheduleWriteSingle(msg);
        }

        @Override
        protected int scheduleWriteMultiple(ChannelOutboundBuffer in) {
            assert writeId == 0;

            IoUringSocketChannelConfig ioUringSocketChannelConfig = (IoUringSocketChannelConfig) config();
            //at least one buffer in the batch exceeds `IO_URING_WRITE_ZERO_COPY_THRESHOLD`.