readComplete0() — netty Function Reference

Source Code

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

        @Override
        protected void readComplete0(byte op, int res, int flags, short data, int outstanding) {
            ByteBuf byteBuf = readBuffer;
            readBuffer = null;
            if (res == Native.ERRNO_ECANCELED_NEGATIVE) {
                readId = 0;
                // In case of cancellation we should reset the last used buffer ring to null as we will select a new one
                // when calling scheduleRead(..)
                if (byteBuf != null) {
                    //recv without buffer ring
                    byteBuf.release();
                }
                return;
            }
            boolean rearm = (flags & Native.IORING_CQE_F_MORE) == 0;
            boolean useBufferRing = (flags & Native.IORING_CQE_F_BUFFER) != 0;
            short bid = (short) (flags >> Native.IORING_CQE_BUFFER_SHIFT);
            boolean more = (flags & Native.IORING_CQE_F_BUF_MORE) != 0;

            boolean empty = socketIsEmpty(flags);
            if (rearm) {
                // Only reset if we don't use multi-shot or we need to re-arm because the multi-shot was cancelled.
                readId = 0;
            }

            boolean allDataRead = false;

            final IoUringRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
            final ChannelPipeline pipeline = pipeline();

            try {
                if (res < 0) {
                    if (res == Native.ERRNO_NOBUFS_NEGATIVE) {
                        // try to expand the buffer ring by adding more buffers to it if there is any space left.
                        if (!bufferRing.expand()) {
                            // We couldn't expand the ring anymore so notify the user that we did run out of buffers
                            // without the ability to expand it.
                            // If this happens to often the user should most likely increase the buffer ring size.
                            pipeline.fireUserEventTriggered(bufferRing.getExhaustedEvent());
                        }

                        // Let's trigger a read again without consulting the RecvByteBufAllocator.Handle as
                        // we can't count this as a "real" read operation.
                        // Because of how our BufferRing works we should have it filled again.
                        scheduleRead(allocHandle.isFirstRead());
                        return;
                    }

                    // 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 read", res));
                } else if (res > 0) {
                    if (useBufferRing) {
                        // If RECVSEND_BUNDLE is used we need to do a bit more work here.
                        // In this case we might need to obtain multiple buffers out of the buffer ring as
                        // multiple of them might have been filled for one recv operation.
                        // See https://github.com/axboe/liburing/wiki/
                        // What's-new-with-io_uring-in-6.10#add-support-for-sendrecv-bundles
                        int read = res;
                        for (;;) {
                            int attemptedBytesRead = bufferRing.attemptedBytesRead(bid);
                            byteBuf = bufferRing.useBuffer(bid, read, more);
                            read -= byteBuf.readableBytes();
                            allocHandle.attemptedBytesRead(attemptedBytesRead);
                            allocHandle.lastBytesRead(byteBuf.readableBytes());

                            assert read >= 0;
                            if (read == 0) {
                                // Just break here, we will handle the byteBuf below and also fill the bufferRing
                                // if needed later.
                                break;
                            }
                            allocHandle.incMessagesRead(1);
                            pipeline.fireChannelRead(byteBuf);
                            byteBuf = null;
                            bid = bufferRing.nextBid(bid);
                            if (!allocHandle.continueReading()) {
                                // We should call fireChannelReadComplete() to mimic a normal read loop.
                                allocHandle.readComplete();
                                pipeline.fireChannelReadComplete();
                                allocHandle.reset(config());