IoUring: Allow to create IoHandlerFactory that supports changing the (#15608)

Thread and so supports AutoScalingEventExecutorChooserFactory.

Modifications:

By default io_uring does not support
AutoScalingEventExecutorChooserFactory as it uses
IORING_SETUP_SINGLE_ISSUER for performance reasons. While this is a good
default sometimes the user might want to tradeoff performance for
flexibility / scalability.

Modifications:

Allow to disable the use of IORING_SETUP_SINGLE_ISSUER and so support
AutoScalingEventExecutorChooserFactory even with io_uring

Result:

Follow up of #15603

Author: normanmaurer

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

@@ -93,7 +93,7 @@ public final class IoUringIoHandler implements IoHandler {
         IoUring.ensureAvailability();
         this.executor = requireNonNull(executor, "executor");
         requireNonNull(config, "config");
-        int setupFlags = Native.setupFlags();
+        int setupFlags = Native.setupFlags(config.singleIssuer());
 
         //The default cq size is always twice the ringSize.
         // It only makes sense when the user actually specifies the cq ring size.
@@ -696,6 +696,16 @@ public static IoHandlerFactory newFactory(int ringSize) {
     public static IoHandlerFactory newFactory(IoUringIoHandlerConfig config) {
         IoUring.ensureAvailability();
         ObjectUtil.checkNotNull(config, "config");
-        return eventLoop -> new IoUringIoHandler(eventLoop, config);
+        return new IoHandlerFactory() {
+            @Override
+            public IoHandler newHandler(ThreadAwareExecutor eventLoop) {
+                return new IoUringIoHandler(eventLoop, config);
+            }
+
+            @Override
+            public boolean isChangingThreadSupported() {
+                return !config.singleIssuer();
+            }
+        };
     }
 }

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

@@ -92,6 +92,12 @@
  *         It will be used to register the buffer ring for the io_uring instance.
  *       </td>
  *     </tr>
+ *     <tr>
+ *       <td>{@link IoUringIoHandlerConfig#setSingleIssuer}</td>
+ *       <td>
+ *         Enable or disable the use of {@code IORING_SETUP_SINGLE_ISSUER}.
+ *       </td>
+ *     </tr>
  *   </tbody>
  * </table>
  */
@@ -100,12 +106,10 @@ public final class IoUringIoHandlerConfig {
 
     private int ringSize = IoUring.DEFAULT_RING_SIZE;
     private int cqSize = IoUring.DEFAULT_CQ_SIZE;
-
     private int maxBoundedWorker;
-
     private int maxUnboundedWorker;
-
     private Set<IoUringBufferRingConfig> bufferRingConfigs;
+    private boolean singleIssuer = true;
 
     /**
      * Return the ring size of the io_uring instance.
@@ -213,6 +217,22 @@ public IoUringIoHandlerConfig setBufferRingConfig(IoUringBufferRingConfig... rin
         return this;
     }
 
+    /**
+     * Set if {@code IORING_SETUP_SINGLE_ISSUER} should be used when setup the ring. This is {@code true} by default
+     * for performance reasons but also means that the {@link Thread} that is used to drive the
+     * {@link io.netty.channel.IoHandler} can never change. If you want the flexibility to change the {@link Thread},
+     * to for example make use of the {@link io.netty.util.concurrent.AutoScalingEventExecutorChooserFactory} it's
+     * possible to not use {@code IORING_SETUP_SINGLE_ISSUER}. This will trade scalibility / flexibility
+     * and performance.
+     *
+     * @param singleIssuer  {@code true} if {@code IORING_SETUP_SINGLE_ISSUER} should be used, {@code false} otherwise
+     * @return reference to this, so the API can be used fluently
+     */
+    public IoUringIoHandlerConfig setSingleIssuer(boolean singleIssuer) {
+        this.singleIssuer = singleIssuer;
+        return this;
+    }
+
     /**
      * Get the list of buffer ring configurations.
      * @return the copy of buffer ring configurations.
@@ -232,4 +252,8 @@ boolean needSetupCqeSize() {
     Set<IoUringBufferRingConfig> getInternBufferRingConfigs() {
         return bufferRingConfigs;
     }
+
+    boolean singleIssuer() {
+        return singleIssuer;
+    }
 }

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

@@ -355,18 +355,21 @@ static String opToStr(byte op) {
             IORING_OP_SEND
     };
 
-    static int setupFlags() {
+    static int setupFlags(boolean useSingleIssuer) {
         int flags = Native.IORING_SETUP_R_DISABLED | Native.IORING_SETUP_CLAMP;
         if (IoUring.isSetupSubmitAllSupported()) {
             flags |= Native.IORING_SETUP_SUBMIT_ALL;
         }
 
-        // See https://github.com/axboe/liburing/wiki/io_uring-and-networking-in-2023#task-work
-        if (IoUring.isSetupSingleIssuerSupported()) {
-            flags |= Native.IORING_SETUP_SINGLE_ISSUER;
-        }
-        if (IoUring.isSetupDeferTaskrunSupported()) {
-            flags |= Native.IORING_SETUP_DEFER_TASKRUN;
+        if (useSingleIssuer) {
+            // See https://github.com/axboe/liburing/wiki/io_uring-and-networking-in-2023#task-work
+            if (IoUring.isSetupSingleIssuerSupported()) {
+                flags |= Native.IORING_SETUP_SINGLE_ISSUER;
+            }
+            // IORING_SETUP_DEFER_TASKRUN also requires IORING_SETUP_SINGLE_ISSUER.
+            if (IoUring.isSetupDeferTaskrunSupported()) {
+                flags |= Native.IORING_SETUP_DEFER_TASKRUN;
+            }
         }
         // liburing uses IORING_SETUP_NO_SQARRAY by default these days, we should do the same by default if possible.
         // See https://github.com/axboe/liburing/releases/tag/liburing-2.6

transport-native-io_uring/src/test/java/io/netty/channel/uring/IoUringEventLoopTest.java

@@ -27,6 +27,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -53,7 +54,8 @@ protected EventLoopGroup newEventLoopGroup() {
 
     @Override
     protected EventLoopGroup newAutoScalingEventLoopGroup() {
-        return null;
+        return new MultiThreadIoEventLoopGroup(SCALING_MAX_THREADS, (Executor) null, AUTO_SCALING_CHOOSER_FACTORY,
+                IoUringIoHandler.newFactory(new IoUringIoHandlerConfig().setSingleIssuer(false)));
     }
 
     @Override