/*
    * Copyright 2024 The Netty Project
    *
    * The Netty Project licenses this file to you under the Apache License,
    * version 2.0 (the "License"); you may not use this file except in compliance
    * with the License. You may obtain a copy of the License at:
    *
    *   https://www.apache.org/licenses/LICENSE-2.0
    *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   * License for the specific language governing permissions and limitations
   * under the License.
   */
  package io.netty.channel.uring;
  
  import io.netty.buffer.ByteBuf;
  import io.netty.channel.Channel;
  import io.netty.channel.ChannelException;
  import io.netty.channel.ChannelFuture;
  import io.netty.channel.ChannelFutureListener;
  import io.netty.channel.ChannelMetadata;
  import io.netty.channel.ChannelOutboundBuffer;
  import io.netty.channel.ChannelPipeline;
  import io.netty.channel.ChannelPromise;
  import io.netty.channel.DefaultFileRegion;
  import io.netty.channel.EventLoop;
  import io.netty.channel.FileRegion;
  import io.netty.channel.IoRegistration;
  import io.netty.channel.socket.DuplexChannel;
  import io.netty.channel.unix.IovArray;
  import io.netty.util.ReferenceCounted;
  import io.netty.util.internal.SystemPropertyUtil;
  import io.netty.util.internal.logging.InternalLogger;
  import io.netty.util.internal.logging.InternalLoggerFactory;
  
  import java.io.IOException;
  import java.net.SocketAddress;
  import java.nio.ByteBuffer;
  import java.nio.channels.WritableByteChannel;
  
  import static io.netty.channel.unix.Errors.ioResult;
  
  abstract class AbstractIoUringStreamChannel extends AbstractIoUringChannel implements DuplexChannel {
      private static final InternalLogger logger = InternalLoggerFactory.getInstance(AbstractIoUringStreamChannel.class);
      private static final ChannelMetadata METADATA = new ChannelMetadata(false, 16);
  
      /**
       * Maximum bytes per chunk when converting a generic {@link FileRegion} to a {@link ByteBuf}
       * for the io_uring async send path. Overridable via the {@code io.netty.iouring.fileRegionChunkSize}
       * system property; capped at 16 MiB to guard against pathological configurations that would
       * risk direct-memory OOM.
       */
      private static final int FILE_REGION_MAX_CHUNK_SIZE = Math.min(16 * 1024 * 1024,
              Math.max(1, SystemPropertyUtil.getInt("io.netty.iouring.fileRegionChunkSize", 64 * 1024)));
  
      // Store the opCode so we know if we used WRITE or WRITEV.
      byte writeOpCode;
      // Keep track of the ids used for write and read so we can cancel these when needed.
      long writeId;
      byte readOpCode;
      long readId;
  
      // The configured buffer ring if any
      private IoUringBufferRing bufferRing;
  
      AbstractIoUringStreamChannel(Channel parent, LinuxSocket socket, boolean active) {
          super(parent, socket, active);
      }
  
      AbstractIoUringStreamChannel(Channel parent, LinuxSocket socket, SocketAddress remote) {
          super(parent, socket, remote);
      }
  
      @Override
      protected final boolean isStreamSocket() {
          return true;
      }
  
      @Override
      public ChannelMetadata metadata() {
          return METADATA;
      }
  
      @Override
      protected AbstractUringUnsafe newUnsafe() {
          return new IoUringStreamUnsafe();
      }
  
      @Override
      public final ChannelFuture shutdown() {
          return shutdown(newPromise());
      }
  
      @Override
      public final ChannelFuture shutdown(final ChannelPromise promise) {
          ChannelFuture shutdownOutputFuture = shutdownOutput();
          if (shutdownOutputFuture.isDone()) {
             shutdownOutputDone(shutdownOutputFuture, promise);
         } else {
             shutdownOutputFuture.addListener(new ChannelFutureListener() {
                 @Override
                 public void operationComplete(final ChannelFuture shutdownOutputFuture) throws Exception {
                     shutdownOutputDone(shutdownOutputFuture, promise);
                 }
             });
         }
         return promise;
     }
 
     @Override
     protected final void doShutdownOutput0() throws Exception {
         socket.shutdown(false, true);
     }
 
     private void shutdownInput0(final ChannelPromise promise) {
         try {
             socket.shutdown(true, false);
             promise.setSuccess();
         } catch (Throwable cause) {
             promise.setFailure(cause);
         }
     }
 
     @Override
     public final boolean isOutputShutdown() {
         return socket.isOutputShutdown();
     }
 
     @Override
     public final boolean isInputShutdown() {
         return socket.isInputShutdown();
     }
 
     @Override
     public final boolean isShutdown() {
         return socket.isShutdown();
     }
 
     @Override
     public final ChannelFuture shutdownOutput() {
         return shutdownOutput(newPromise());
     }
 
     @Override
     public final ChannelFuture shutdownOutput(final ChannelPromise promise) {
         EventLoop loop = eventLoop();
         if (loop.inEventLoop()) {
             ((AbstractUnsafe) unsafe()).shutdownOutput(promise);
         } else {
             loop.execute(new Runnable() {
                 @Override
                 public void run() {
                     ((AbstractUnsafe) unsafe()).shutdownOutput(promise);
                 }
             });
         }
 
         return promise;
     }
 
     @Override
     public final ChannelFuture shutdownInput() {
         return shutdownInput(newPromise());
     }
 
     @Override
     public final ChannelFuture shutdownInput(final ChannelPromise promise) {
         EventLoop loop = eventLoop();
         if (loop.inEventLoop()) {
             shutdownInput0(promise);
         } else {
             loop.execute(new Runnable() {
                 @Override
                 public void run() {
                     shutdownInput0(promise);
                 }
             });
         }
         return promise;
     }
 
     private void shutdownOutputDone(final ChannelFuture shutdownOutputFuture, final ChannelPromise promise) {
         ChannelFuture shutdownInputFuture = shutdownInput();
         if (shutdownInputFuture.isDone()) {
             shutdownDone(shutdownOutputFuture, shutdownInputFuture, promise);
         } else {
             shutdownInputFuture.addListener(new ChannelFutureListener() {
                 @Override
                 public void operationComplete(ChannelFuture shutdownInputFuture) throws Exception {
                     shutdownDone(shutdownOutputFuture, shutdownInputFuture, promise);
                 }
             });
         }
     }
 
     private static void shutdownDone(ChannelFuture shutdownOutputFuture,
                                      ChannelFuture shutdownInputFuture,
                                      ChannelPromise promise) {
         Throwable shutdownOutputCause = shutdownOutputFuture.cause();
         Throwable shutdownInputCause = shutdownInputFuture.cause();
         if (shutdownOutputCause != null) {
             if (shutdownInputCause != null) {
                 logger.info("Exception suppressed because a previous exception occurred.",
                              shutdownInputCause);
             }
             promise.setFailure(shutdownOutputCause);
         } else if (shutdownInputCause != null) {
             promise.setFailure(shutdownInputCause);
         } else {
             promise.setSuccess();
         }
     }
 
     @Override
     protected final void doRegister(ChannelPromise promise) {
         ChannelPromise registerPromise = this.newPromise();
         // Ensure that the buffer group is properly set before channel::read
         registerPromise.addListener(f -> {
             if (f.isSuccess()) {
                try {
                    short bgid = ((IoUringStreamChannelConfig) config()).getBufferGroupId();
                    if (bgid >= 0) {
                        final IoUringIoHandler ioUringIoHandler = registration().attachment();
                        bufferRing = ioUringIoHandler.findBufferRing(bgid);
                    }
                    if (active) {
                        // Register for POLLRDHUP if this channel is already considered active.
                        schedulePollRdHup();
                    }
                } finally {
                    promise.setSuccess();
                }
             } else {
                 promise.setFailure(f.cause());
             }
         });
 
         super.doRegister(registerPromise);
     }
 
     @Override
     protected Object filterOutboundMessage(Object msg) {
         if (IoUring.isSpliceSupported() && msg instanceof DefaultFileRegion) {
             return new IoUringFileRegion((DefaultFileRegion) msg);
         }
 
         if (msg instanceof FileRegion) {
             // Generic FileRegion -- pass through to the write path for chunked conversion.
             return msg;
         }
 
         return super.filterOutboundMessage(msg);
     }
 
     protected class IoUringStreamUnsafe extends AbstractUringUnsafe {
 
         private ByteBuf readBuffer;
 
         // Chunk buffer for generic FileRegion writes. Non-null while a send is in flight.
         private ByteBuf fileRegionChunkBuf;
 
         @Override
         protected int scheduleWriteMultiple(ChannelOutboundBuffer in) {
             assert writeId == 0;
 
             int fd = fd().intValue();
             IoRegistration registration = registration();
             IoUringIoHandler handler = registration.attachment();
             IovArray iovArray = handler.iovArray();
             int offset = iovArray.count();
 
             IovArrayReferenceCollector collector = handler.iovArrayReferenceCollector();
             try {
                 try {
                     in.forEachFlushedMessage(filterWriteMultiple(collector));
                 } catch (Exception e) {
                     // This should never happen, anyway fallback to single write.
                     return scheduleWriteSingle(in.current());
                 }
                 long iovArrayAddress = iovArray.memoryAddress(offset);
                 int iovArrayLength = iovArray.count() - offset;
                 // Should not use sendmsg_zc, just use normal writev.
                 IoUringIoOps ops = IoUringIoOps.newWritev(
                         fd, (byte) 0, 0, iovArrayAddress, iovArrayLength, nextOpsId());
 
                 byte opCode = ops.opcode();
                 // record(...) copies the collector's references into the slot, so the collector stays reusable.
                 writeTracker.recordStream(opCode, collector.referencesArray(), collector.referencesCount());
                 writeId = registration.submit(ops);
                 writeOpCode = opCode;
                 if (writeId == 0) {
                     writeTracker.abandonStream();
                     return 0;
                 }
                 return 1;
             } finally {
                 // The slot copied the references it needs, and an exception must not leave the event loop's
                 // shared collector holding this write's buffers.
                 collector.reset();
             }
         }
 
         protected ChannelOutboundBuffer.MessageProcessor filterWriteMultiple(IovArrayReferenceCollector collector) {
            return collector;
         }
 
         @Override
         protected int scheduleWriteSingle(Object msg) {
             assert writeId == 0;
 
             int fd = fd().intValue();
             IoRegistration registration = registration();
             final IoUringIoOps ops;
             if (msg instanceof IoUringFileRegion) {
                 IoUringFileRegion fileRegion = (IoUringFileRegion) msg;
                 try {
                     fileRegion.open();
                 } catch (IOException e) {
                     this.handleWriteError(e);
                     return 0;
                 }
                 ops = fileRegion.splice(fd);
             } else if (msg instanceof FileRegion) {
                 return scheduleWriteFileRegion(fd, registration, (FileRegion) msg);
             } else {
                 ByteBuf buf = (ByteBuf) msg;
                 long address = IoUring.memoryAddress(buf) + buf.readerIndex();
                 int length = buf.readableBytes();
                 ops = IoUringIoOps.newSend(fd, (byte) 0, 0, address, length, nextOpsId());
             }
             byte opCode = ops.opcode();
             // A splice picks its own data to tell its two stages apart, so it never enters the channel-level
             // slot array used for zero-copy writes. It still has to occupy this single slot though: the file and
             // pipe descriptors it splices between have to outlive the SQE, and writeTracker.retainAll()
             // only retains what was recorded here.
             writeTracker.recordStream(opCode, (ReferenceCounted) msg);
             writeId = registration.submit(ops);
             writeOpCode = opCode;
             if (writeId == 0) {
                 writeTracker.abandonStream();
                 return 0;
             }
             return 1;
         }
 
         // Read a chunk from a generic FileRegion into a direct ByteBuf and submit it as an
         // io_uring async send. If fileRegionChunkBuf is non-null, re-submits the remaining
         // bytes from a previous partial/failed send.
         private int scheduleWriteFileRegion(int fd, IoRegistration registration, FileRegion region) {
             ByteBuf buf = fileRegionChunkBuf;
             if (buf == null) {
                 long remaining = region.count() - region.transferred();
                 if (remaining > 0) {
                     int chunkSize = (int) Math.min(remaining, FILE_REGION_MAX_CHUNK_SIZE);
                     buf = alloc().directBuffer(chunkSize);
                     try {
                         ByteBufWritableByteChannel ch = new ByteBufWritableByteChannel(buf);
                         // Mirror epoll's writeFileRegion(): stop calling transferTo() once
                         // the region reports it has been fully transferred. The FileRegion
                         // contract permits implementations to assume no further invocations
                         // past transferred() == count().
                         while (buf.writableBytes() > 0 && region.transferred() < region.count()) {
                             long t = region.transferTo(ch, region.transferred());
                             if (t <= 0) {
                                 break;
                             }
                         }
                         if (buf.readableBytes() == 0) {
                             buf.release();
                             handleWriteError(new ChannelException(
                                     "FileRegion.transferTo(...) produced 0 bytes (count="
                                             + region.count() + ", transferred=" + region.transferred() + ')'));
                             return 0;
                         }
                     } catch (Exception e) {
                         buf.release();
                         handleWriteError(e);
                         return 0;
                     }
                 } else {
                     // Empty or fully-transferred region. Submit a 0-byte send so the completion
                     // path removes it from the outbound buffer via the normal async flow.
                     buf = alloc().directBuffer(0);
                 }
                 fileRegionChunkBuf = buf;
             }
             long address = IoUring.memoryAddress(buf) + buf.readerIndex();
             int length = buf.readableBytes();
             IoUringIoOps ops = IoUringIoOps.newSend(fd, (byte) 0, 0, address, length, nextOpsId());
             byte opCode = ops.opcode();
             writeTracker.recordStream(opCode, buf);
             writeId = registration.submit(ops);
             writeOpCode = opCode;
             if (writeId == 0) {
                 writeTracker.abandonStream();
                 // Submission only fails when the registration is no longer valid (channel is
                 // being deregistered). Ending the slot above is the only cleanup needed here:
                 // unregistered() will release fileRegionChunkBuf and the outbound buffer will
                 // release the FileRegion -- mirroring the plain ByteBuf path above.
                 return 0;
             }
             return 1;
         }
 
         private int calculateRecvFlags(boolean first) {
             // Depending on if this is the first read or not we will use Native.MSG_DONTWAIT.
             // The idea is that if the socket is blocking we can do the first read in a blocking fashion
             // and so not need to also register POLLIN. As we can not 100 % sure if reads after the first will
             // be possible directly we schedule these with Native.MSG_DONTWAIT. This allows us to still be
             // able to signal the fireChannelReadComplete() in a timely manner and be consistent with other
             // transports.
             if (first) {
                 return 0;
             }
             return Native.MSG_DONTWAIT;
         }
 
         private short calculateRecvIoPrio(boolean first, boolean socketIsEmpty) {
             // Depending on if socketIsEmpty is true we will arm the poll upfront and skip the initial transfer
             // attempt.
             // See https://github.com/axboe/liburing/wiki/io_uring-and-networking-in-2023#socket-state
             if (first) {
                 // IORING_RECVSEND_POLL_FIRST and IORING_CQE_F_SOCK_NONEMPTY were added in the same release (5.19).
                 // We need to check if it's supported as otherwise providing these would result in an -EINVAL.
                 return socketIsEmpty && IoUring.isCqeFSockNonEmptySupported() ?
                         Native.IORING_RECVSEND_POLL_FIRST : 0;
             }
             return 0;
         }
 
         @Override
         protected int scheduleRead0(boolean first, boolean socketIsEmpty) {
             assert readBuffer == null;
             assert readId == 0 : readId;
             final IoUringRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
 
             if (bufferRing != null && bufferRing.isUsable()) {
                 return scheduleReadProviderBuffer(bufferRing, first, socketIsEmpty);
             }
 
             // We either have no buffer ring configured or we force a recv without using a buffer ring.
             ByteBuf byteBuf = allocHandle.allocate(alloc());
             try {
                 int fd = fd().intValue();
                 IoRegistration registration = registration();
                 short ioPrio = calculateRecvIoPrio(first, socketIsEmpty);
                 int recvFlags = calculateRecvFlags(first);
                 short opsId = nextOpsId();
 
                 IoUringIoOps ops = IoUringIoOps.newRecv(fd, (byte) 0, ioPrio, recvFlags,
                         IoUring.memoryAddress(byteBuf) + byteBuf.writerIndex(), byteBuf.writableBytes(), opsId);
                 readId = registration.submit(ops);
                 readOpCode = Native.IORING_OP_RECV;
                 if (readId == 0) {
                     return 0;
                 }
                 readBuffer = byteBuf;
                 byteBuf = null;
                 return 1;
             } finally {
                 if (byteBuf != null) {
                     byteBuf.release();
                 }
             }
         }
 
         private int scheduleReadProviderBuffer(IoUringBufferRing bufferRing, boolean first, boolean socketIsEmpty) {
             short bgId = bufferRing.bufferGroupId();
             try {
                 boolean multishot = IoUring.isRecvMultishotEnabled();
                 byte flags = (byte) Native.IOSQE_BUFFER_SELECT;
                 short ioPrio;
                 final int recvFlags;
                 if (multishot) {
                     ioPrio = Native.IORING_RECV_MULTISHOT;
                     recvFlags = 0;
                 } else {
                     // We should only use the calculate*() methods if this is not a multishot recv, as otherwise
                     // the would be applied until the multishot will be re-armed.
                     ioPrio = calculateRecvIoPrio(first, socketIsEmpty);
                     recvFlags = calculateRecvFlags(first);
                 }
                 if (IoUring.isRecvsendBundleEnabled()) {
                     // See https://github.com/axboe/liburing/wiki/
                     // What's-new-with-io_uring-in-6.10#add-support-for-sendrecv-bundles
                     ioPrio |= Native.IORING_RECVSEND_BUNDLE;
                 }
                 IoRegistration registration = registration();
                 int fd = fd().intValue();
                 short opsId = nextOpsId();
                 IoUringIoOps ops = IoUringIoOps.newRecv(
                         fd, flags, ioPrio, recvFlags, 0,
                         0, opsId, bgId
                 );
                 readId = registration.submit(ops);
                 readOpCode = Native.IORING_OP_RECV;
                 if (readId == 0) {
                     return 0;
                 }
                 if (multishot) {
                     // Return -1 to signal we used multishot and so expect multiple recvComplete(...) calls.
                     return -1;
                 }
                 return 1;
             } catch (IllegalArgumentException illegalArgumentException) {
                 this.handleReadException(pipeline(), null, illegalArgumentException, false, recvBufAllocHandle());
                 return 0;
             }
         }
 
         @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 completeRead = shouldCompleteReadLoop(flags, isReadMultishot());
             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);
                             // Snapshot the number of allocated buffers before calling useBuffer(...) as that call
                             // might grow the ring (and so change allocatedBuffers) once the last posted buffer is
                             // consumed. The bundle completion we are walking here was produced by the kernel using
                             // the ring size as it was *before* any such growth, so the wrap-around for the next bid
                             // in this bundle must be computed with that same (old) size to stay in sync with the
                             // kernel.
                             int allocatedBuffers = bufferRing.allocatedBuffers();
                             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, allocatedBuffers);
                             if (!allocHandle.continueReading()) {
                                 // We should call fireChannelReadComplete() to mimic a normal read loop.
                                 allocHandle.readComplete();
                                 pipeline.fireChannelReadComplete();
                                 allocHandle.reset(config());
                             }
                         }
                     } 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.
                     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, completeRead);
             } catch (Throwable t) {
                 handleReadException(pipeline, byteBuf, t, allDataRead, allocHandle);
             }
         }
 
         private void scheduleNextRead(ChannelPipeline pipeline, IoUringRecvByteAllocatorHandle allocHandle,
                                       boolean rearm, boolean completeRead) {
             if (allocHandle.continueReading() && !completeRead) {
                 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();
             }
         }
 
         protected final void handleReadException(ChannelPipeline pipeline, ByteBuf byteBuf,
                                          Throwable cause, boolean allDataRead,
                                          IoUringRecvByteAllocatorHandle allocHandle) {
             if (byteBuf != null) {
                 if (byteBuf.isReadable()) {
                     pipeline.fireChannelRead(byteBuf);
                 } else {
                     byteBuf.release();
                 }
             }
             allocHandle.readComplete();
             pipeline.fireChannelReadComplete();
             pipeline.fireExceptionCaught(cause);
             if (allDataRead || cause instanceof IOException) {
                 shutdownInput(true);
             }
         }
 
         private boolean handleWriteCompleteFileRegion(ChannelOutboundBuffer channelOutboundBuffer,
                                                       IoUringFileRegion fileRegion, int res, short data) {
             try {
                 if (res == Native.ERRNO_ECANCELED_NEGATIVE) {
                     return true;
                 }
                 int result = res >= 0 ? res : ioResult("io_uring splice", res);
                 if (result == 0 && fileRegion.count() > 0) {
                     validateFileRegion(fileRegion.fileRegion, fileRegion.transfered());
                     return false;
                 }
                 int progress = fileRegion.handleResult(result, data);
                 if (progress == -1) {
                     // Done with writing
                     channelOutboundBuffer.remove();
                 } else if (progress > 0) {
                     channelOutboundBuffer.progress(progress);
                 }
             } catch (Throwable cause) {
                 handleWriteError(cause);
             }
             return true;
         }
 
         @Override
         boolean writeComplete0(byte op, int res, int flags, long data, int outstanding) {
             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;
                 // A completion that never went through the slot finds it inactive, which makes this a no-op.
                 writeTracker.completeStream(flags);
             }
             ChannelOutboundBuffer channelOutboundBuffer = unsafe().outboundBuffer();
             if (channelOutboundBuffer == null) {
                 // The completion may arrive after close() or shutdownOutput() already dropped the buffer.
                 releaseFileRegionChunkBuf();
                 return true;
             }
             Object current = channelOutboundBuffer.current();
             if (current instanceof IoUringFileRegion) {
                 IoUringFileRegion fileRegion = (IoUringFileRegion) current;
                 // A splice picks its own data to tell its two stages apart, so narrowing here can not drop bits.
                 return handleWriteCompleteFileRegion(channelOutboundBuffer, fileRegion, res, (short) data);
             }
 
             if (current instanceof FileRegion) {
                 return handleWriteCompleteGenericFileRegion(
                         channelOutboundBuffer, (FileRegion) current, res);
             }
 
             if (res >= 0) {
                 channelOutboundBuffer.removeBytes(res);
             } else if (res == Native.ERRNO_ECANCELED_NEGATIVE) {
                 return true;
             } else {
                 try {
                     if (ioResult("io_uring write", res) == 0) {
                         return false;
                     }
                 } catch (Throwable cause) {
                     handleWriteError(cause);
                 }
             }
             return true;
         }
 
         // Returns true when the completion can be treated as "written all" for this SQE
         // (the framework may still schedule further writes from the outbound buffer); returns
         // false to signal the framework that POLLOUT should be armed so the chunk buffer is
         // resubmitted once the socket becomes writable again.
         private boolean handleWriteCompleteGenericFileRegion(
                 ChannelOutboundBuffer channelOutboundBuffer, FileRegion region, int res) {
             try {
                 if (res == Native.ERRNO_ECANCELED_NEGATIVE) {
                     releaseFileRegionChunkBuf();
                     return true;
                 }
                 if (res >= 0) {
                     ByteBuf buf = fileRegionChunkBuf;
                     assert buf != null;
                     buf.skipBytes(res);
                     channelOutboundBuffer.progress(res);
                     if (!buf.isReadable()) {
                         // Chunk fully sent.
                         releaseFileRegionChunkBuf();
                         if (region.transferred() >= region.count()) {
                             channelOutboundBuffer.remove();
                         }
                     } else {
                         // Partial send -- schedule POLLOUT to re-send the remainder.
                         return false;
                     }
                 } else {
                     // Keep the chunk buffer -- on retryable errors (EAGAIN) ioResult returns 0
                     // and scheduleWriteFileRegion() will re-submit the same fileRegionChunkBuf
                     // once POLLOUT fires. On a non-retryable error ioResult throws, and the
                     // outer catch releases the buffer.
                     if (ioResult("io_uring write", res) == 0) {
                         return false;
                     }
                 }
             } catch (Throwable cause) {
                 releaseFileRegionChunkBuf();
                 handleWriteError(cause);
             }
             return true;
         }
 
         private void releaseFileRegionChunkBuf() {
             if (fileRegionChunkBuf != null) {
                 fileRegionChunkBuf.release();
                 fileRegionChunkBuf = null;
             }
         }
 
         @Override
         public void unregistered() {
             // Abandons the single slot through writeTracker.releaseAll() before the chunk buffer is
             // dropped below, so a reference a shutdown retained on that buffer is released first.
             super.unregistered();
             assert readBuffer == null;
             releaseFileRegionChunkBuf();
         }
     }
 
     @Override
     protected final void cancelOutstandingReads(IoRegistration registration, int numOutstandingReads) {
         if (readId != 0) {
             // Let's try to cancel outstanding reads as these might be submitted and waiting for data (via fastpoll).
             assert numOutstandingReads == 1 || numOutstandingReads == -1;
             IoUringIoOps ops = IoUringIoOps.newAsyncCancel((byte) 0, readId, readOpCode);
             long id = registration.submit(ops);
             assert id != 0;
             readId = 0;
         }
     }
 
     @Override
     protected final void cancelOutstandingWrites(IoRegistration registration, int numOutstandingWrites) {
         if (writeId != 0) {
             // Let's try to cancel outstanding writes as these might be submitted and waiting to finish writing
             // (via fastpoll).
             assert numOutstandingWrites == 1;
             assert writeOpCode != 0;
             long id = registration.submit(IoUringIoOps.newAsyncCancel((byte) 0, writeId, writeOpCode));
             assert id != 0;
             writeId = 0;
         }
     }
 
     @Override
     protected boolean socketIsEmpty(int flags) {
         return IoUring.isCqeFSockNonEmptySupported() && (flags & Native.IORING_CQE_F_SOCK_NONEMPTY) == 0;
     }
 
     protected boolean shouldCompleteReadLoop(int flags, boolean multishot) {
         return socketIsEmpty(flags);
     }
 
     @Override
     boolean isPollInFirst() {
         return bufferRing == null || !bufferRing.isUsable();
     }
 
     /**
      * A {@link WritableByteChannel} backed by a {@link ByteBuf}.
      * Writes are capped to {@link ByteBuf#writableBytes()} to prevent overflow
      * when {@link FileRegion#transferTo} writes more than the chunk size.
      */
     private static final class ByteBufWritableByteChannel implements WritableByteChannel {
         private final ByteBuf buf;
 
         ByteBufWritableByteChannel(ByteBuf buf) {
             this.buf = buf;
         }
 
         @Override
         public int write(ByteBuffer src) {
             int toWrite = Math.min(src.remaining(), buf.writableBytes());
             if (toWrite == 0) {
                 return 0;
             }
             if (toWrite < src.remaining()) {
                 int oldLimit = src.limit();
                 src.limit(src.position() + toWrite);
                 buf.writeBytes(src);
                 src.limit(oldLimit);
                 return toWrite;
             }
             buf.writeBytes(src);
             return toWrite;
         }
 
         @Override
         public boolean isOpen() {
             return true;
         }
 
         @Override
         public void close() {
             // NOOP
         }
     }
 }