update global scheduler behaviour

Author: franz1981

core/src/main/java/io/netty/loom/GlobalDelegateThreadNettyScheduler.java

@@ -39,16 +39,39 @@ public void onStart(Thread.VirtualThreadTask virtualThreadTask) {
         // 2. if a vThread will never start, it will leak here forever
         var assignedSchedulerRef = unstartedThreads.remove(virtualThreadTask.thread());
         if (assignedSchedulerRef == null) {
-            // TODO per carrier sub-pollers goes here, but we want them to inherit the scheduler from the caller context
-        }
-        var scheduler = assignedSchedulerRef != null ? assignedSchedulerRef.get() : null;
-        if (scheduler != null) {
-            // attach the assigned scheduler to the task
-            virtualThreadTask.attach(assignedSchedulerRef);
-            if (scheduler.execute(virtualThreadTask)) {
-                return;
+            // Read-Poller threads are special: if we run from a VThread managed by a VirtualThreadNettyScheduler,
+            // we want should continue using that same scheduler for the Read-Poller thread.
+            var currentThread = Thread.currentThread();
+            if (currentThread.isVirtual()) {
+                // TODO https://github.com/openjdk/loom/blob/12ddf39bb59252a8274d8b937bd075b2a6dbc3f8/src/java.base/share/classes/java/lang/VirtualThread.java#L270C18-L270C33
+                //      in theory should be easy to provide a VirtualThreadTask::current method to avoid the ScopedValue lookup
+                var schedulerRef = VirtualThreadNettyScheduler.currentRef();
+                // TODO per carrier sub-pollers goes here, but we want them to inherit the scheduler from the caller context
+                if (schedulerRef != null) {
+                    var scheduler = schedulerRef.get();
+                    // See https://github.com/openjdk/loom/blob/12ddf39bb59252a8274d8b937bd075b2a6dbc3f8/src/java.base/share/classes/sun/nio/ch/Poller.java#L723C48-L723C59
+                    if (scheduler != null) {
+                        if (virtualThreadTask.thread().getName().endsWith("-Read-Poller")) {
+                            // attach the assigned scheduler to the task
+                            virtualThreadTask.attach(schedulerRef);
+                            if (scheduler.execute(virtualThreadTask)) {
+                                return;
+                            }
+                            virtualThreadTask.attach(null);
+                        }
+                    }
+                }
+            }
+        } else {
+            var scheduler = assignedSchedulerRef.get();
+            if (scheduler != null) {
+                // attach the assigned scheduler to the task
+                virtualThreadTask.attach(assignedSchedulerRef);
+                if (scheduler.execute(virtualThreadTask)) {
+                    return;
+                }
             }
-            // the v thread has been rejected by its assigned scheduler, clean it up and fallback to JDK
+            // the v thread has been rejected by its assigned scheduler or its scheduler is gone
             virtualThreadTask.attach(null);
         }
         jdkBuildinScheduler.onStart(virtualThreadTask);

core/src/main/java/io/netty/loom/VirtualThreadNettyScheduler.java

@@ -21,6 +21,7 @@ public class VirtualThreadNettyScheduler {
    private static final long IDLE_YIELD_US = TimeUnit.MICROSECONDS.toNanos(Integer.getInteger("io.netty.loom.idle.yield.us", 1));
    // This is required to allow sub-pollers to run on the correct scheduler
    private static final ScopedValue<AtomicReference<VirtualThreadNettyScheduler>> CURRENT_SCHEDULER = ScopedValue.newInstance();
+   private static final AtomicReference<VirtualThreadNettyScheduler> EMPTY_REFERENCE = new AtomicReference<>();
    private final MpscUnboundedStream<Runnable> runQueue;
    private final ManualIoEventLoop ioEventLoop;
    private final Thread eventLoopThread;
@@ -197,4 +198,9 @@ public boolean execute(Thread.VirtualThreadTask task) {
       }
       return true;
    }
+
+   public static AtomicReference<VirtualThreadNettyScheduler> currentRef() {
+      var ref = CURRENT_SCHEDULER.orElse(EMPTY_REFERENCE);
+      return ref == EMPTY_REFERENCE? null : ref;
+   }
 }