IoUringSocketUnsafe Class — netty Architecture

Source Code

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

    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`.
            if (IoUring.isSendmsgZcSupported()
                    && (ioUringSocketChannelConfig.shouldWriteZeroCopy(((ByteBuf) in.current()).readableBytes()))) {
                IoUringIoHandler handler = registration().attachment();

                IovArray iovArray = handler.iovArray();
                int offset = iovArray.count();
                // Limit to the maximum number of fragments to ensure we don't get an error when we have too many
                // buffers.
                iovArray.maxCount(Native.MAX_SKB_FRAGS);
                try {
                    in.forEachFlushedMessage(new ChannelOutboundBuffer.MessageProcessor() {
                        @Override
                        public boolean processMessage(Object msg) throws Exception {
                            if (msg instanceof ByteBuf) {
                                ByteBuf buf = (ByteBuf) msg;
                                int length = buf.readableBytes();
                                if (ioUringSocketChannelConfig.shouldWriteZeroCopy(length)) {
                                    return iovArray.processMessage(msg);
                                }
                            }
                            return false;
                        }
                    });
                } catch (Exception e) {
                    // This should never happen, anyway fallback to single write.
                    return scheduleWriteSingle(in.current());
                }
                long iovArrayAddress = iovArray.memoryAddress(offset);
                int iovArrayLength = iovArray.count() - offset;

                MsgHdrMemoryArray msgHdrArray = handler.msgHdrMemoryArray();
                MsgHdrMemory hdr = msgHdrArray.nextHdr();
                assert hdr != null;
                hdr.set(iovArrayAddress, iovArrayLength);
                IoUringIoOps ops = IoUringIoOps.newSendmsgZc(fd().intValue(), (byte) 0, 0, hdr.address(), nextOpsId());
                byte opCode = ops.opcode();
                writeId = registration().submit(ops);
                writeOpCode = opCode;
                if (writeId == 0) {
                    return 0;
                }
                return 1;
            }
            // Should not use sendmsg_zc, just use normal writev.
            return super.scheduleWriteMultiple(in);
        }