IoUringStreamUnsafe Class — netty Architecture

Source Code

transport-classes-io_uring/src/main/java/io/netty/channel/uring/AbstractIoUringStreamChannel.java lines 234–623

    protected class IoUringStreamUnsafe extends AbstractUringUnsafe {

        private ByteBuf readBuffer;

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

            int fd = fd().intValue();
            IoRegistration registration = registration();
            IoUringIoHandler handler = registration.attachment();
            IovArray iovArray = handler.iovArray();
            int offset = iovArray.count();

            try {
                in.forEachFlushedMessage(filterWriteMultiple(iovArray));
            } 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;
            // Should not use sendmsg_zc, just use normal writev.
            IoUringIoOps ops = IoUringIoOps.newWritev(fd, (byte) 0, 0, iovArrayAddress, iovArrayLength, nextOpsId());

            byte opCode = ops.opcode();
            writeId = registration.submit(ops);
            writeOpCode = opCode;
            if (writeId == 0) {
                return 0;
            }
            return 1;
        }

        protected ChannelOutboundBuffer.MessageProcessor filterWriteMultiple(IovArray iovArray) {
           return iovArray;
        }

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

            int fd = fd().intValue();
            IoRegistration registration = registration();
            final IoUringIoOps ops;
            if (msg instanceof IoUringFileRegion) {
                IoUringFileRegion fileRegion = (IoUringFileRegion) msg;
                try {
                    fileRegion.open();
                } catch (IOException e) {
                    this.handleWriteError(e);
                    return 0;
                }
                ops = fileRegion.splice(fd);
            } else {
                ByteBuf buf = (ByteBuf) msg;
                long address = IoUring.memoryAddress(buf) + buf.readerIndex();
                int length = buf.readableBytes();
                short opsid = nextOpsId();

                ops = IoUringIoOps.newSend(fd, (byte) 0, 0, address, length, opsid);
            }
            byte opCode = ops.opcode();
            writeId = registration.submit(ops);
            writeOpCode = opCode;
            if (writeId == 0) {
                return 0;
            }
            return 1;
        }

        private int calculateRecvFlags(boolean first) {
            // Depending on if this is the first read or not we will use Native.MSG_DONTWAIT.
            // 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.MSG_DONTWAIT. This allows us to still be
            // able to signal the fireChannelReadComplete() in a timely manner and be consistent with other
            // transports.
            if (first) {
                return 0;
            }