Fix event loop scheduler leaks

Author: franz1981

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

@@ -66,19 +66,23 @@ public EventLoopScheduler(IoEventLoopGroup parent, ThreadFactory threadFactory,
       eventLoopIsRunning = new AtomicBoolean(false);
       runQueue = new MpscUnboundedStream<>(resumedContinuationsExpectedCount);
       carrierThread = threadFactory.newThread(this::virtualThreadSchedulerLoop);
-      var rawVTFactory = Thread.ofVirtual().factory();
-      vThreadFactory = runnable ->
-              NettyScheduler.assignUnstarted(rawVTFactory.newThread(
-                      () ->   // this can be inherited by any thread created in the context of this virtual thread
-                              // but only the original virtual thread will have the correct scheduler context
-                              ScopedValue.where(CURRENT_SCHEDULER, new SchedulingContext(Thread.currentThread().threadId(), sharedRef)).run(runnable)
-              ), sharedRef);
+      vThreadFactory = newEventLoopSchedulerFactory(sharedRef);
       eventLoopThread = vThreadFactory.newThread(() -> FastThreadLocalThread.runWithFastThreadLocal(this::nettyEventLoop));
       ioEventLoop = new ManualIoEventLoop(parent, eventLoopThread,
             ioExecutor -> new AwakeAwareIoHandler(eventLoopIsRunning, ioHandlerFactory.newHandler(ioExecutor)));
       carrierThread.start();
    }
 
+   private static ThreadFactory newEventLoopSchedulerFactory(SharedRef sharedRef) {
+       var rawVTFactory = Thread.ofVirtual().factory();
+       return runnable ->
+               NettyScheduler.assignUnstarted(rawVTFactory.newThread(
+                       () ->   // this can be inherited by any thread created in the context of this virtual thread
+                               // but only the original virtual thread will have the correct scheduler context
+                               ScopedValue.where(CURRENT_SCHEDULER, new SchedulingContext(Thread.currentThread().threadId(), sharedRef)).run(runnable)
+               ), sharedRef);
+   }
+
    int externalContinuationsCount() {
       return runQueue.size();
    }

core/src/test/java/io/netty/loom/VirtualMultithreadIoEventLoopGroupTest.java

@@ -3,6 +3,8 @@
 import static java.util.concurrent.StructuredTaskScope.Joiner.allSuccessfulOrThrow;
 import static org.junit.jupiter.api.Assertions.*;
 
+import java.lang.ref.SoftReference;
+import java.lang.ref.WeakReference;
 import java.net.InetSocketAddress;
 import java.net.URI;
 import java.net.http.HttpClient;
@@ -39,6 +41,7 @@
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.internal.ThreadExecutorMap;
+import org.junit.jupiter.api.Timeout;
 
 public class VirtualMultithreadIoEventLoopGroupTest {
 
@@ -366,6 +369,32 @@ void schedulerIsNotInheritedByForkedVT() throws InterruptedException, ExecutionE
         }
     }
 
+    @Test
+    @Timeout(10)
+    void schedulerIsNotLeakingIfItsThreadFactoryOutliveIt() throws InterruptedException, ExecutionException {
+        ThreadFactory vThreadFactory;
+        WeakReference<EventLoopScheduler> schedulerWeakRef;
+        EventLoopScheduler.SharedRef schedulerRef;
+        try (var group = new VirtualMultithreadIoEventLoopGroup(1, LocalIoHandler.newFactory())) {
+            vThreadFactory = group.submit(group::vThreadFactory).get();
+            schedulerRef = group.submit(() -> EventLoopScheduler.currentThreadSchedulerContext().scheduler()).get();
+            schedulerWeakRef = new WeakReference<>(schedulerRef.get());
+        }
+        while (schedulerRef.get() != null) {
+            Thread.yield();
+        }
+        while (schedulerWeakRef.get() != null) {
+            System.gc();
+            System.runFinalization();
+            Thread.sleep(100);
+        }
+        // we can still run virtual threads from the factory without the scheduler
+        var schedulerRefPromise = new CompletableFuture<EventLoopScheduler.SharedRef>();
+        vThreadFactory.newThread(() -> schedulerRefPromise.complete(EventLoopScheduler.currentThreadSchedulerContext().scheduler())).start();
+        assertSame(schedulerRef, schedulerRefPromise.get());
+        assertNull(schedulerRefPromise.get().get());
+    }
+
     @Test
     void virtualThreadCanMakeProgressEvenIfEventLoopIsClosed() throws InterruptedException, ExecutionException, BrokenBarrierException, TimeoutException {
         var group = new VirtualMultithreadIoEventLoopGroup(1, LocalIoHandler.newFactory());