Allow delayed set of owningThread in ManualIoEventLoop (#14976)

yawkat · normanmaurer · web-flow · commit 788071415e0a · 2025-04-01T12:02:30.000+02:00
Motivation: In some scenarios, it is desirable to immediately create a ManualIoEventLoop, without knowing which thread it is assigned to. In particular, to implement `MultiThreadIoEventLoopGroup.newChild` properly, the owning thread is assigned through an `Executor` call, so we don't see it until the `Executor` actually invokes the `Runnable` it receives, but we need to return the `ManualIoEventLoop` from `newChild` immediately. Modification: Allow passing `null` for the `owningThread` constructor parameter, and add a `setOwningThread` setter that can be called at most once. Result: `newChild` can be implemented without hacks. The only use of the `owningThread` field is in `inEventLoop`. That means that the only difference compared to the old constructor approach is that there is a period of time when the future owning thread is running, but inEventLoop still returns `false`. I'm struggling to imagine scenarios where this matters, but @franz1981 had some thoughts, so I'm open to concerns and alternative solutions. --------- Co-authored-by: Norman Maurer <norman_maurer@apple.com>
diff --git a/transport/src/main/java/io/netty/channel/ManualIoEventLoop.java b/transport/src/main/java/io/netty/channel/ManualIoEventLoop.java
@@ -35,6 +35,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * {@link IoEventLoop} implementation that is owned by the user and so needs to be driven by the user manually with the
@@ -75,7 +76,7 @@ public long deadlineNanos() {
     };
     private final BlockingIoHandlerContext blockingContext = new BlockingIoHandlerContext();
     private final IoEventLoopGroup parent;
-    private final Thread owningThread;
+    private final AtomicReference<Thread> owningThread;
     private final IoHandler handler;
 
     private volatile long gracefulShutdownQuietPeriod;
@@ -107,13 +108,14 @@ public ManualIoEventLoop(Thread owningThread, IoHandlerFactory factory) {
      * @param parent            the parent {@link IoEventLoopGroup} or {@code null} if no parent.
      * @param owningThread      the {@link Thread} that executes the IO and tasks for this {@link IoEventLoop}. The
      *                          user will use this {@link Thread} to call {@link #runNow()} or {@link #run(long)} to
-     *                          make progress.
+     *                          make progress. If {@code null}, must be set later using
+     *                          {@link #setOwningThread(Thread)}.
      * @param factory           the {@link IoHandlerFactory} that will be used to create the {@link IoHandler} that is
      *                          used by this {@link IoEventLoop}.
      */
     public ManualIoEventLoop(IoEventLoopGroup parent, Thread owningThread, IoHandlerFactory factory) {
         this.parent = parent;
-        this.owningThread = Objects.requireNonNull(owningThread, "owningThread");
+        this.owningThread = new AtomicReference<>(owningThread);
         this.handler = factory.newHandler(this);
         state = new AtomicInteger(ST_STARTED);
     }
@@ -141,6 +143,9 @@ private int runAllTasks() {
 
     private int run(IoHandlerContext context) {
         if (!initialized) {
+            if (owningThread.get() == null) {
+                throw new IllegalStateException("Owning thread not set");
+            }
             initialized = true;
             handler.initialize();
         }
@@ -295,7 +300,20 @@ public boolean isIoType(Class<? extends IoHandler> handlerType) {
 
     @Override
     public boolean inEventLoop(Thread thread) {
-        return this.owningThread == thread;
+        return this.owningThread.get() == thread;
+    }
+
+    /**
+     * Set the owning thread that will call {@link #run}. May only be called once, and only if the owning thread was
+     * not set in the constructor already.
+     *
+     * @param owningThread The owning thread
+     */
+    public void setOwningThread(Thread owningThread) {
+        Objects.requireNonNull(owningThread, "owningThread");
+        if (!this.owningThread.compareAndSet(null, owningThread)) {
+            throw new IllegalStateException("Owning thread already set");
+        }
     }
 
     private void shutdown0(long quietPeriod, long timeout, int shutdownState) {
diff --git a/transport/src/test/java/io/netty/channel/ManualIoEventLoopTest.java b/transport/src/test/java/io/netty/channel/ManualIoEventLoopTest.java
@@ -26,6 +26,8 @@
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.Semaphore;
@@ -216,6 +218,59 @@ public Boolean call() {
         }
     }
 
+    @Test
+    public void testDelayOwningThread() throws ExecutionException, InterruptedException {
+        Semaphore semaphore = new Semaphore(0);
+        ManualIoEventLoop eventLoop = new ManualIoEventLoop(null, executor ->
+                new TestIoHandler(semaphore));
+        Thread thread = new Thread(() -> {
+            eventLoop.setOwningThread(Thread.currentThread());
+            assertTrue(eventLoop.inEventLoop());
+            while (!eventLoop.isTerminated()) {
+                eventLoop.runNow();
+            }
+        });
+
+        assertFalse(eventLoop.inEventLoop());
+
+        CompletableFuture<Void> cf = new CompletableFuture<>();
+        eventLoop.execute(() -> {
+            assertTrue(eventLoop.inEventLoop());
+            cf.complete(null);
+        });
+
+        thread.start();
+        cf.get();
+
+        eventLoop.shutdownGracefully();
+        thread.join();
+    }
+
+    @Test
+    public void testRunWithoutOwner() throws ExecutionException, InterruptedException {
+        ManualIoEventLoop eventLoop = new ManualIoEventLoop(null, executor ->
+                new TestIoHandler(new Semaphore(0)));
+
+        // prior to setOwningThread, runNow is forbidden
+        assertThrows(IllegalStateException.class, eventLoop::runNow);
+
+        eventLoop.setOwningThread(Thread.currentThread());
+
+        eventLoop.runNow(); // runs fine
+
+        eventLoop.shutdownGracefully();
+    }
+
+    @Test
+    public void testSetOwnerMultipleTimes() {
+        ManualIoEventLoop eventLoop = new ManualIoEventLoop(null, executor ->
+                new TestIoHandler(new Semaphore(0)));
+        eventLoop.setOwningThread(Thread.currentThread());
+        assertThrows(IllegalStateException.class, () -> eventLoop.setOwningThread(Thread.currentThread()));
+
+        eventLoop.shutdownGracefully();
+    }
+
     private static final class TestRunnable implements Runnable {
         private boolean done;
         @Override
