IoUringDomainSocketUnsafe Class — netty Architecture

Source Code

transport-classes-io_uring/src/main/java/io/netty/channel/uring/IoUringDomainSocketChannel.java lines 105–253

    private final class IoUringDomainSocketUnsafe extends IoUringStreamUnsafe {

        private MsgHdrMemory writeMsgHdrMemory;
        private MsgHdrMemory readMsgHdrMemory;

        @Override
        protected int scheduleWriteSingle(Object msg) {
            if (msg instanceof FileDescriptor) {
                // we can reuse the same memory for any fd
                // because we never have more than a single outstanding write.
                if (writeMsgHdrMemory == null) {
                    writeMsgHdrMemory = new MsgHdrMemory();
                }
                IoRegistration registration = registration();
                IoUringIoOps ioUringIoOps = prepSendFdIoOps((FileDescriptor) msg, writeMsgHdrMemory);
                writeId = registration.submit(ioUringIoOps);
                writeOpCode = Native.IORING_OP_SENDMSG;
                if (writeId == 0) {
                    MsgHdrMemory memory = writeMsgHdrMemory;
                    writeMsgHdrMemory = null;
                    memory.release();
                    return 0;
                }
                return 1;
            }
            return super.scheduleWriteSingle(msg);
        }

        @Override
        boolean writeComplete0(byte op, int res, int flags, short data, int outstanding) {
            if (op == Native.IORING_OP_SENDMSG) {
                writeId = 0;
                writeOpCode = 0;
                if (res == Native.ERRNO_ECANCELED_NEGATIVE) {
                    return true;
                }
                try {
                    int nativeCallResult = res >= 0 ? res : Errors.ioResult("io_uring sendmsg", res);
                    if (nativeCallResult >= 0) {
                        ChannelOutboundBuffer channelOutboundBuffer = unsafe().outboundBuffer();
                        channelOutboundBuffer.remove();
                    }
                } catch (Throwable throwable) {
                   handleWriteError(throwable);
                }
                return true;
            }
            return super.writeComplete0(op, res, flags, data, outstanding);
        }

        private IoUringIoOps prepSendFdIoOps(FileDescriptor fileDescriptor, MsgHdrMemory msgHdrMemory) {
            msgHdrMemory.setScmRightsFd(fileDescriptor.intValue());
            return IoUringIoOps.newSendmsg(
                    fd().intValue(), (byte) 0, 0, msgHdrMemory.address(), msgHdrMemory.idx());
        }

        @Override
        protected int scheduleRead0(boolean first, boolean socketIsEmpty) {
            DomainSocketReadMode readMode = config.getReadMode();
            switch (readMode) {
                case FILE_DESCRIPTORS:
                    return scheduleRecvReadFd();
                case BYTES:
                    return super.scheduleRead0(first, socketIsEmpty);
                default:
                    throw new Error("Unexpected read mode: " + readMode);
            }
        }

        private int scheduleRecvReadFd() {
            // we can reuse the same memory for any fd
            // because we only submit one outstanding read
            if (readMsgHdrMemory == null) {
                readMsgHdrMemory = new MsgHdrMemory();
            }
            readMsgHdrMemory.prepRecvReadFd();
            IoRegistration registration = registration();
            IoUringIoOps ioUringIoOps = IoUringIoOps.newRecvmsg(
                    fd().intValue(), (byte) 0, 0, readMsgHdrMemory.address(), readMsgHdrMemory.idx());
            readId = registration.submit(ioUringIoOps);
            readOpCode = Native.IORING_OP_RECVMSG;