Repository: netty/netty
File: transport-classes-io_uring/src/main/java/io/netty/channel/uring/AbstractIoUringStreamChannel.java (Java)

Code before:
                        byteBuf = bufferRing.useBuffer(bid, res, more);
                    } else {
                        attemptedBytesRead = byteBuf.writableBytes();
                        byteBuf.writerIndex(byteBuf.writerIndex() + res);
                    }
                    allocHandle.attemptedBytesRead(attemptedBytesRead);
                    allocHandle.lastBytesRead(res);
                } else {
                    // EOF which we signal with -1.
                    allocHandle.lastBytesRead(-1);
                }
                if (allocHandle.lastBytesRead() <= 0) {
                    // byteBuf might be null if we used a buffer ring.
                    if (byteBuf != null) {
                        // nothing was read, release the buffer.
                        byteBuf.release();
                        byteBuf = null;
                    }
                    allDataRead = allocHandle.lastBytesRead() < 0;
                    if (allDataRead) {
                        // There is nothing left to read as we received an EOF.
                        shutdownInput(true);
                    }
                    allocHandle.readComplete();
                    pipeline.fireChannelReadComplete();
                    return;
                }

                allocHandle.incMessagesRead(1);
                pipeline.fireChannelRead(byteBuf);
                byteBuf = null;
                scheduleNextRead(pipeline, allocHandle, rearm, empty);
            } catch (Throwable t) {
                handleReadException(pipeline, byteBuf, t, allDataRead, allocHandle);
            }
        }

        private void scheduleNextRead(ChannelPipeline pipeline, IoUringRecvByteAllocatorHandle allocHandle,
                                      boolean rearm, boolean empty) {
            if (allocHandle.continueReading() && !empty) {
                if (rearm) {
                    // We only should schedule another read if we need to rearm.
                    // See https://github.com/axboe/liburing/wiki/io_uring-and-networking-in-2023#multi-shot
                    scheduleRead(false);
                }
            } else {
                // We did not fill the whole ByteBuf so we should break the "read loop" and try again later.
                allocHandle.readComplete();
                pipeline.fireChannelReadComplete();
            }
        }

Review comment (suggestion):
I changed this to not use `CompositeByteBuf` and just fire each buffer through the pipeline directly. The reason for this is that otherwise we might hold a lot of memory before the user will be able to do anything with this. This can dramatically increase memory usage as usually the user will do something with the buffer as soon as it is received (which often is releasing it).

Code after:
                    if (useBufferRing) {
                        short bid = (short) (flags >> Native.IORING_CQE_BUFFER_SHIFT);
                        boolean more = (flags & Native.IORING_CQE_F_BUF_MORE) != 0;
                        if (IoUring.isRecvsendBundleSupported()) {
                            // If RECVSEND_BUNDLE is supported 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.
                                    break;
                                }
                                allocHandle.incMessagesRead(1);
                                pipeline.fireChannelRead(byteBuf);
                                byteBuf = null;
                                if (!allocHandle.continueReading()) {
                                    // We should call fireChannelReadComplete() to mimic a normal read loop.
                                    allocHandle.readComplete();
                                    pipeline.fireChannelReadComplete();
                                    allocHandle.reset(config());
                                }
                                bid = bufferRing.nextBid(bid);
                            }
                        } else {
                            int attemptedBytesRead = bufferRing.attemptedBytesRead(bid);
                            byteBuf = bufferRing.useBuffer(bid, res, more);
                            allocHandle.attemptedBytesRead(attemptedBytesRead);
                            allocHandle.lastBytesRead(res);
                        }
                    } else {
                        int attemptedBytesRead = byteBuf.writableBytes();
                        byteBuf.writerIndex(byteBuf.writerIndex() + res);
                        allocHandle.attemptedBytesRead(attemptedBytesRead);
                        allocHandle.lastBytesRead(res);
                    }
                } else {
                    // EOF which we signal with -1.
                    allocHandle.lastBytesRead(-1);
                }
                if (allocHandle.lastBytesRead() <= 0) {
                    // byteBuf might be null if we used a buffer ring.