IoUring: Simplify submission strategy (#15476)

Motivation:

At some point I added some code to submit as soon as a channel becomes
unwritable. This was always kind of a hack and we should better not do
it. With all the recent changes we see the expected performance even
without this.

Modifications:

- Remove the usage of CompletionBuffer as its not neeeded anymore (we
still keep the class for now just in case)
- Don't trigger a submit when the Channel becomes unwritable.

Result:

Cleanup

Author: normanmaurer

transport-classes-io_uring/src/main/java/io/netty/channel/uring/AbstractIoUringStreamChannel.java

@@ -634,16 +634,6 @@ protected boolean socketIsEmpty(int flags) {
         return IoUring.isCqeFSockNonEmptySupported() && (flags & Native.IORING_CQE_F_SOCK_NONEMPTY) == 0;
     }
 
-    @Override
-    protected void submitAndRunNow() {
-        if (writeId != 0) {
-            // Force a submit and processing of the completions to ensure we drain the outbound buffer and
-            // send the data to the remote peer.
-            ((IoUringIoHandler) registration().attachment()).submitAndRunNow(writeId);
-        }
-        super.submitAndRunNow();
-    }
-
     @Override
     boolean isPollInFirst() {
         return bufferRing == null || !bufferRing.isUsable();

transport-classes-io_uring/src/main/java/io/netty/channel/uring/IoUringIoHandler.java

@@ -85,7 +85,6 @@ public final class IoUringIoHandler implements IoHandler {
     private static final int KERNEL_TIMESPEC_TV_SEC_FIELD = 0;
     private static final int KERNEL_TIMESPEC_TV_NSEC_FIELD = 8;
 
-    private final CompletionBuffer completionBuffer;
     private final ThreadAwareExecutor executor;
 
     IoUringIoHandler(ThreadAwareExecutor executor, IoUringIoHandlerConfig config) {
@@ -148,10 +147,6 @@ public final class IoUringIoHandler implements IoHandler {
         timeoutMemoryCleanable = Buffer.allocateDirectBufferWithNativeOrder(KERNEL_TIMESPEC_SIZE);
         timeoutMemory = timeoutMemoryCleanable.buffer();
         timeoutMemoryAddress = Buffer.memoryAddress(timeoutMemory);
-        // We buffer a maximum of 2 * CompletionQueue.ringCapacity completions before we drain them in batches.
-        // Also as we never submit an udata which is 0L we use this as the tombstone marker.
-        completionBuffer = new CompletionBuffer(ringBuffer.ioUringCompletionQueue().ringCapacity * 2, 0);
-
         iovArray = new IovArray(IoUring.NUM_ELEMENTS_IOVEC);
     }
 
@@ -181,19 +176,7 @@ public int run(IoHandlerContext context) {
             // Even if we have some completions already pending we can still try to even fetch more.
             submitAndClearNow(submissionQueue);
         }
-        return drainAndProcessAll(completionQueue, this::handle);
-    }
-
-    void submitAndRunNow(long udata) {
-        if (closeCompleted) {
-            return;
-        }
-        SubmissionQueue submissionQueue = ringBuffer.ioUringSubmissionQueue();
-        CompletionQueue completionQueue = ringBuffer.ioUringCompletionQueue();
-        if (submitAndClearNow(submissionQueue) > 0) {
-            completionBuffer.drain(completionQueue);
-            completionBuffer.processOneNow(this::handle, udata);
-        }
+        return completionQueue.process(this::handle);
     }
 
     private int submitAndClearNow(SubmissionQueue submissionQueue) {
@@ -232,18 +215,6 @@ IoUringBufferRing findBufferRing(short bgId) {
         );
     }
 
-    private int drainAndProcessAll(CompletionQueue completionQueue, CompletionCallback callback) {
-        int processed = 0;
-        for (;;) {
-            boolean drainedAll = completionBuffer.drain(completionQueue);
-            processed += completionBuffer.processNow(callback);
-            if (drainedAll) {
-                break;
-            }
-        }
-        return processed;
-    }
-
     private static void handleLoopException(Throwable throwable) {
         logger.warn("Unexpected exception in the IO event loop.", throwable);
 
@@ -424,11 +395,10 @@ public boolean handle(int res, int flags, long udata) {
                 }
             }
             final DrainFdEventCallback handler = new DrainFdEventCallback();
-            drainAndProcessAll(completionQueue, handler);
             completionQueue.process(handler);
             while (!handler.eventFdDrained) {
                 submissionQueue.submitAndGet();
-                drainAndProcessAll(completionQueue, handler);
+                completionQueue.process(handler);
             }
         }
         // We've consumed any pending eventfd read and `eventfdAsyncNotify` should never