Change names (again, sorry!) and disable tests to fix

Author: franz1981

benchmarks/src/main/java/io/netty/loom/benchmark/NettySchedulerBenchmark.java

@@ -2,7 +2,7 @@
 
 import io.netty.channel.nio.NioIoHandler;
 import io.netty.loom.LoomSupport;
-import io.netty.loom.MultithreadVirtualEventExecutorGroup;
+import io.netty.loom.VirtualMultithreadIoEventLoopGroup;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.openjdk.jmh.annotations.BenchmarkMode;
 import org.openjdk.jmh.annotations.Fork;
@@ -33,13 +33,13 @@ public class NettySchedulerBenchmark {
     @Param({"1000", "100000"})
     private int tasks;
 
-    private MultithreadVirtualEventExecutorGroup executorGroup;
+    private VirtualMultithreadIoEventLoopGroup executorGroup;
 
     private ThreadFactory vtFactory;
 
     @Setup
     public void setup() throws ExecutionException, InterruptedException {
-        executorGroup = new MultithreadVirtualEventExecutorGroup(1, NioIoHandler.newFactory());
+        executorGroup = new VirtualMultithreadIoEventLoopGroup(1, NioIoHandler.newFactory());
         vtFactory = executorGroup.submit(executorGroup::vThreadFactory).get();
     }
 

benchmarks/src/main/java/io/netty/loom/benchmark/SchedulerBenchmark.java

@@ -31,7 +31,7 @@
 import io.netty.channel.local.LocalIoHandler;
 import io.netty.channel.nio.NioIoHandler;
 import io.netty.loom.LoomSupport;
-import io.netty.loom.MultithreadVirtualEventExecutorGroup;
+import io.netty.loom.VirtualMultithreadIoEventLoopGroup;
 
 @BenchmarkMode(Mode.Throughput)
 @OutputTimeUnit(TimeUnit.MILLISECONDS)
@@ -82,7 +82,7 @@ public void setup() throws ExecutionException, InterruptedException {
       durationNs = TimeUnit.MICROSECONDS.toNanos(durationUs);
       switch (scheduler) {
          case Netty:
-            var virtualGroup = new MultithreadVirtualEventExecutorGroup(1, factory);
+            var virtualGroup = new VirtualMultithreadIoEventLoopGroup(1, factory);
             executorGroup = virtualGroup;
             LoomSupport.checkSupported();
             vtFactory = executorGroup.submit(virtualGroup::vThreadFactory).get();

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

@@ -180,7 +180,7 @@ private int runExternalContinuations(long deadlineNs) {
       return runContinuations;
    }
 
-   private boolean eventLoopContinuation(Thread.VirtualThreadTask task) {
+   private boolean rescheduleEventLoop(Thread.VirtualThreadTask task) {
       if (eventLoopContinuatioToRun != null) {
          assert task.thread() != eventLoopThread;
          return false;
@@ -193,22 +193,25 @@ private boolean eventLoopContinuation(Thread.VirtualThreadTask task) {
    }
 
    public boolean execute(Thread.VirtualThreadTask task) {
-      boolean isEventLoopContinuation = eventLoopContinuation(task);
-      if (!isEventLoopContinuation) {
+      boolean eventLoopTask = rescheduleEventLoop(task);
+      if (!eventLoopTask) {
         if (!runQueue.offer(task)) {
             return false;
         }
       }
-      if (!ioEventLoop.inEventLoop(Thread.currentThread())) {
-          // this is checking for "local" submissions: it assumes that
-          // currentThreadEventLoopScheduler() is the currently assigned one, and up to date
-          // WARNING!!!!!
-          // work-stealing could break this assumption if we don't update the CURRENT_SCHEDULER scoped value accordingly
-          if (currentThreadEventLoopScheduler() != publishedReference) {
-              ioEventLoop.wakeup();
-              LockSupport.unpark(parkedCarrierThread);
+      var currentThread = Thread.currentThread();
+      if (currentThread != eventLoopThread) {
+          if (currentThread != carrierThread) {
+              // this is checking for "local" submissions: it assumes that
+              // currentThreadEventLoopScheduler() is the currently assigned one, and up to date
+              // WARNING!!!!!
+              // work-stealing could break this assumption if we don't update the CURRENT_SCHEDULER scoped value accordingly
+              if (currentThreadEventLoopScheduler() != publishedReference) {
+                  ioEventLoop.wakeup();
+                  LockSupport.unpark(parkedCarrierThread);
+              }
           }
-      } else if (!isEventLoopContinuation && !eventLoopIsRunning.get()) {
+      } else if (!eventLoopTask && !eventLoopIsRunning.get()) {
           // the event loop thread is allowed to give up cycles to consume external continuations
           // whilst is just woken up for I/O
           assert eventLoopContinuatioToRun == null;

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

@@ -4,7 +4,6 @@
 
 import java.lang.invoke.VarHandle;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicReference;
 
 /**
  * Global scheduler — provides execution for virtual threads that do not specify their own scheduler.
@@ -34,6 +33,10 @@ public NettyScheduler(Thread.VirtualThreadScheduler jdkBuildinScheduler) {
         VarHandle.storeStoreFence();
     }
 
+    Thread.VirtualThreadScheduler jdkBuildinScheduler() {
+        return jdkBuildinScheduler;
+    }
+
     @Override
     public void onStart(Thread.VirtualThreadTask virtualThreadTask) {
         // TODO this is not great for 2 reasons:
@@ -85,10 +88,7 @@ public void onStart(Thread.VirtualThreadTask virtualThreadTask) {
     @Override
     public void onContinue(Thread.VirtualThreadTask virtualThreadTask) {
         var attachment = virtualThreadTask.attachment();
-        if (attachment instanceof AtomicReference<?>) {
-            // TODO create a EventLoopSchedulerRef type
-            @SuppressWarnings("unchecked")
-            var ref = (AtomicReference<EventLoopScheduler>) attachment;
+        if (attachment instanceof SchedulerRef ref) {
             var scheduler = ref.get();
             if (scheduler != null) {
                 if (scheduler.execute(virtualThreadTask)) {
@@ -101,14 +101,13 @@ public void onContinue(Thread.VirtualThreadTask virtualThreadTask) {
         jdkBuildinScheduler.onContinue(virtualThreadTask);
     }
 
-    // just for benchmark
-    public Thread.VirtualThreadScheduler getJdkBuildinScheduler() {
-        return jdkBuildinScheduler;
-    }
-
     static Thread assignUnstarted(Thread unstarted, SchedulerRef ref) {
         INSTANCE.unstartedThreads.put(unstarted, ref);
         return unstarted;
     }
+
+    public static boolean isAvailable() {
+        return INSTANCE != null;
+    }
 }
 

…ultithreadVirtualEventExecutorGroup.java …/VirtualMultithreadIoEventLoopGroup.javacore/src/main/java/io/netty/loom/MultithreadVirtualEventExecutorGroup.java renamed to core/src/main/java/io/netty/loom/VirtualMultithreadIoEventLoopGroup.java core/src/main/java/io/netty/loom/MultithreadVirtualEventExecutorGroup.java renamed to core/src/main/java/io/netty/loom/VirtualMultithreadIoEventLoopGroup.java

@@ -9,7 +9,7 @@
 import io.netty.channel.MultiThreadIoEventLoopGroup;
 import io.netty.util.concurrent.FastThreadLocal;
 
-public class MultithreadVirtualEventExecutorGroup extends MultiThreadIoEventLoopGroup {
+public class VirtualMultithreadIoEventLoopGroup extends MultiThreadIoEventLoopGroup {
 
    private static final int RESUMED_CONTINUATIONS_EXPECTED_COUNT = Integer.getInteger("io.netty.loom.resumed.continuations", 1024);
    private ThreadFactory threadFactory;
@@ -21,10 +21,13 @@ protected EventLoopScheduler initialValue() {
       }
    };
 
-   public MultithreadVirtualEventExecutorGroup(int nThreads, IoHandlerFactory ioHandlerFactory) {
+   public VirtualMultithreadIoEventLoopGroup(int nThreads, IoHandlerFactory ioHandlerFactory) {
       super(nThreads, (Executor) command -> {
          throw new UnsupportedOperationException("this executor is not supposed to be used");
       }, ioHandlerFactory);
+      if (!NettyScheduler.isAvailable()) {
+         throw new IllegalStateException("-Djdk.virtualThreadScheduler.implClass=io.netty.loom.NettyScheduler is required to use VirtualMultithreadIoEventLoopGroup");
+      }
    }
 
    public ThreadFactory vThreadFactory() {