IoUringDatagramChannelUnsafe Class — netty Architecture

Source Code

transport-classes-io_uring/src/main/java/io/netty/channel/uring/IoUringDatagramChannel.java lines 377–649

    private final class IoUringDatagramChannelUnsafe extends AbstractUringUnsafe {
        private final WriteProcessor writeProcessor = new WriteProcessor();

        private ByteBuf readBuffer;

        private final class WriteProcessor implements ChannelOutboundBuffer.MessageProcessor {
            private int written;
            @Override
            public boolean processMessage(Object msg) {
                if (scheduleWrite(msg, written == 0)) {
                    written++;
                    return true;
                }
                return false;
            }

            int write(ChannelOutboundBuffer in) {
                written = 0;
                try {
                    in.forEachFlushedMessage(this);
                } catch (Exception e) {
                    // This should never happen as our processMessage(...) never throws.
                    throw new IllegalStateException(e);
                }
                return written;
            }
        }

        @Override
        protected void readComplete0(byte op, int res, int flags, short data, int outstanding) {
            assert outstanding != -1 : "multi-shot not implemented yet";

            final IoUringRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
            final ChannelPipeline pipeline = pipeline();
            ByteBuf byteBuf = this.readBuffer;
            assert byteBuf != null;
            try {
                recvmsgComplete(pipeline, allocHandle, byteBuf, res, flags, data, outstanding);
            } catch (Throwable t) {
                Throwable e = (connected && t instanceof NativeIoException) ?
                  translateForConnected((NativeIoException) t) : t;
                pipeline.fireExceptionCaught(e);
            }
        }

        private void recvmsgComplete(ChannelPipeline pipeline, IoUringRecvByteAllocatorHandle allocHandle,
                                      ByteBuf byteBuf, int res, int flags, int idx, int outstanding)
                throws IOException {
            MsgHdrMemory hdr = recvmsgHdrs.hdr(idx);
            if (res < 0) {
                if (res != Native.ERRNO_ECANCELED_NEGATIVE) {
                    // If res is negative we should pass it to ioResult(...) which will either throw
                    // or convert it to 0 if we could not read because the socket was not readable.
                    allocHandle.lastBytesRead(ioResult("io_uring recvmsg", res));
                }
            } else {
                allocHandle.lastBytesRead(res);
                if (hdr.hasPort(IoUringDatagramChannel.this)) {
                    allocHandle.incMessagesRead(1);
                    DatagramPacket packet = hdr.get(
                            IoUringDatagramChannel.this, registration().attachment(), byteBuf, res);
                    pipeline.fireChannelRead(packet);
                }
            }

            // Reset the id as this read was completed and so don't need to be cancelled later.
            recvmsgHdrs.setId(idx, MsgHdrMemoryArray.NO_ID);
            if (outstanding == 0) {
                // There are no outstanding completion events, release the readBuffer and see if we need to schedule
                // another one or if the user will do it.
                this.readBuffer.release();
                this.readBuffer = null;
                recvmsgHdrs.clear();

                if (res != Native.ERRNO_ECANCELED_NEGATIVE) {
                    if (allocHandle.lastBytesRead() > 0 &&
                            allocHandle.continueReading(UncheckedBooleanSupplier.TRUE_SUPPLIER) &&
                            // If IORING_CQE_F_SOCK_NONEMPTY is supported we should check for it first before
                            // trying to schedule a read. If it's supported and not part of the flags we know for sure
                            // that the next read (which would be using Native.MSG_DONTWAIT) will complete without
                            // be able to read any data. This is useless work and we can skip it.