handleWriteCompleteZeroCopy() — netty Function Reference

Source Code

transport-classes-io_uring/src/main/java/io/netty/channel/uring/IoUringSocketChannel.java lines 188–291

        private boolean handleWriteCompleteZeroCopy(byte op, ChannelOutboundBuffer channelOutboundBuffer,
                                                    int res, int flags) {
            if ((flags & Native.IORING_CQE_F_NOTIF) == 0) {
                // We only want to reset these if IORING_CQE_F_NOTIF is not set.
                // If it's set we know this is only an extra notification for a write but we already handled
                // the write completions before.
                // See https://man7.org/linux/man-pages/man2/io_uring_enter.2.html section: IORING_OP_SEND_ZC
                writeId = 0;
                writeOpCode = 0;

                boolean more = (flags & Native.IORING_CQE_F_MORE) != 0;
                if (more) {
                    // This is the result of send_sz or sendmsg_sc but there will also be another notification
                    // which will let us know that we can release the buffer(s). In this case let's retain the
                    // buffer(s) once and store it in an internal queue. Once we receive the notification we will
                    // call release() on the buffer(s) as it's not used by the kernel anymore.
                    if (zcWriteQueue == null) {
                        zcWriteQueue = new ArrayDeque<>(8);
                    }
                }
                if (res >= 0) {
                    if (more) {

                        // Loop through all the buffers that were part of the operation so we can add them to our
                        // internal queue to release later.
                        do {
                            ByteBuf currentBuffer = (ByteBuf) channelOutboundBuffer.current();
                            assert currentBuffer != null;
                            zcWriteQueue.add(currentBuffer);
                            currentBuffer.retain();
                            int readable = currentBuffer.readableBytes();
                            int skip = Math.min(readable, res);
                            currentBuffer.skipBytes(skip);
                            channelOutboundBuffer.progress(readable);
                            if (readable <= res) {
                                boolean removed = channelOutboundBuffer.remove();
                                assert removed;
                            }
                            res -= readable;
                        } while (res > 0);
                        // Add the marker so we know when we need to stop releasing
                        zcWriteQueue.add(ZC_BATCH_MARKER);
                    } else {
                        // We don't expect any extra notification, just directly let the buffer be released.
                        channelOutboundBuffer.removeBytes(res);
                    }
                    return true;
                } else {
                    if (res == Native.ERRNO_ECANCELED_NEGATIVE) {
                        if (more) {
                            // The send was cancelled but we expect another notification. Just add the marker to the
                            // queue so we don't get into trouble once the final notification for this operation is
                            // received.
                            zcWriteQueue.add(ZC_BATCH_MARKER);
                        }
                        return true;
                    }
                    try {
                        String msg = op == Native.IORING_OP_SEND_ZC ? "io_uring sendzc" : "io_uring sendmsg_zc";
                        int result = ioResult(msg, res);
                        if (more) {
                            try {
                                // We expect another notification so we need to ensure we retain these buffers
                                // so we can release these once we see IORING_CQE_F_NOTIF set.
                                addFlushedToZcWriteQueue(channelOutboundBuffer);
                            } catch (Exception e) {
                                // should never happen but let's handle it anyway.
                                handleWriteError(e);
                            }
                        }
                        if (result == 0) {
                            return false;
                        }
                    } catch (Throwable cause) {
                        if (more) {
                            try {
                                // We expect another notification as handleWriteError(...) will fail all flushed writes
                                // and also release any buffers we need to ensure we retain these buffers
                                // so we can release these once we see IORING_CQE_F_NOTIF set.
                                addFlushedToZcWriteQueue(channelOutboundBuffer);
                            } catch (Exception e) {