8351927: Change VirtualThread implementation to use use FJP delayed task handling

Reviewed-by: vklang

Author: Alan Bateman

src/java.base/share/classes/java/lang/System.java

@@ -2290,10 +2290,6 @@ public Executor virtualThreadDefaultScheduler() {
                 return VirtualThread.defaultScheduler();
             }
 
-            public Stream<ScheduledExecutorService> virtualThreadDelayedTaskSchedulers() {
-                return VirtualThread.delayedTaskSchedulers();
-            }
-
             public StackWalker newStackWalkerInstance(Set<StackWalker.Option> options,
                                                       ContinuationScope contScope,
                                                       Continuation continuation) {

src/java.base/share/classes/java/lang/VirtualThread.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2018, 2024, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2018, 2025, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -24,7 +24,6 @@
  */
 package java.lang;
 
-import java.util.Arrays;
 import java.util.Locale;
 import java.util.Objects;
 import java.util.concurrent.CountDownLatch;
@@ -38,7 +37,6 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import java.util.stream.Stream;
 import jdk.internal.event.VirtualThreadEndEvent;
 import jdk.internal.event.VirtualThreadStartEvent;
 import jdk.internal.event.VirtualThreadSubmitFailedEvent;
@@ -66,7 +64,6 @@ final class VirtualThread extends BaseVirtualThread {
     private static final Unsafe U = Unsafe.getUnsafe();
     private static final ContinuationScope VTHREAD_SCOPE = new ContinuationScope("VirtualThreads");
     private static final ForkJoinPool DEFAULT_SCHEDULER = createDefaultScheduler();
-    private static final ScheduledExecutorService[] DELAYED_TASK_SCHEDULERS = createDelayedTaskSchedulers();
 
     private static final long STATE = U.objectFieldOffset(VirtualThread.class, "state");
     private static final long PARK_PERMIT = U.objectFieldOffset(VirtualThread.class, "parkPermit");
@@ -193,13 +190,6 @@ static Executor defaultScheduler() {
         return DEFAULT_SCHEDULER;
     }
 
-    /**
-     * Returns a stream of the delayed task schedulers used to support timed operations.
-     */
-    static Stream<ScheduledExecutorService> delayedTaskSchedulers() {
-        return Arrays.stream(DELAYED_TASK_SCHEDULERS);
-    }
-
     /**
      * Returns the continuation scope used for virtual threads.
      */
@@ -567,8 +557,9 @@ private void afterYield() {
                 setState(newState = PARKED);
             } else {
                 // schedule unpark
+                long timeout = this.timeout;
                 assert timeout > 0;
-                timeoutTask = schedule(this::unpark, timeout, NANOSECONDS);
+                timeoutTask = schedule(this::parkTimeoutExpired, timeout, NANOSECONDS);
                 setState(newState = TIMED_PARKED);
             }
 
@@ -618,6 +609,7 @@ private void afterYield() {
                 // the timeout task to coordinate access to the sequence number and to
                 // ensure the timeout task doesn't execute until the thread has got to
                 // the TIMED_WAIT state.
+                long timeout = this.timeout;
                 assert timeout > 0;
                 synchronized (timedWaitLock()) {
                     byte seqNo = ++timedWaitSeqNo;
@@ -890,7 +882,19 @@ private void unblock() {
     }
 
     /**
-     * Invoked by timer thread when wait timeout for virtual thread has expired.
+     * Invoked by FJP worker thread or STPE thread when park timeout expires.
+     */
+    private void parkTimeoutExpired() {
+        assert !VirtualThread.currentThread().isVirtual();
+        if (!getAndSetParkPermit(true)
+                && (state() == TIMED_PARKED)
+                && compareAndSetState(TIMED_PARKED, UNPARKED)) {
+            lazySubmitRunContinuation();
+        }
+    }
+
+    /**
+     * Invoked by FJP worker thread or STPE thread when wait timeout expires.
      * If the virtual thread is in timed-wait then this method will unblock the thread
      * and submit its task so that it continues and attempts to reenter the monitor.
      * This method does nothing if the thread has been woken by notify or interrupt.
@@ -913,7 +917,7 @@ private void waitTimeoutExpired(byte seqNo) {
                 }
             }
             if (unblocked) {
-                submitRunContinuation();
+                lazySubmitRunContinuation();
                 return;
             }
             // need to retry when thread is suspended in time-wait
@@ -1444,40 +1448,54 @@ private static ForkJoinPool createDefaultScheduler() {
     /**
      * Schedule a runnable task to run after a delay.
      */
-    private static Future<?> schedule(Runnable command, long delay, TimeUnit unit) {
-        long tid = Thread.currentThread().threadId();
-        int index = (int) tid & (DELAYED_TASK_SCHEDULERS.length - 1);
-        return DELAYED_TASK_SCHEDULERS[index].schedule(command, delay, unit);
+    private Future<?> schedule(Runnable command, long delay, TimeUnit unit) {
+        if (scheduler instanceof ForkJoinPool pool) {
+            return pool.schedule(command, delay, unit);
+        } else {
+            return DelayedTaskSchedulers.schedule(command, delay, unit);
+        }
     }
 
     /**
-     * Creates the ScheduledThreadPoolExecutors used to execute delayed tasks.
+     * Supports scheduling a runnable task to run after a delay. It uses a number
+     * of ScheduledThreadPoolExecutor instances to reduce contention on the delayed
+     * work queue used. This class is used when using a custom scheduler.
      */
-    private static ScheduledExecutorService[] createDelayedTaskSchedulers() {
-        String propName = "jdk.virtualThreadScheduler.timerQueues";
-        String propValue = System.getProperty(propName);
-        int queueCount;
-        if (propValue != null) {
-            queueCount = Integer.parseInt(propValue);
-            if (queueCount != Integer.highestOneBit(queueCount)) {
-                throw new RuntimeException("Value of " + propName + " must be power of 2");
-            }
-        } else {
-            int ncpus = Runtime.getRuntime().availableProcessors();
-            queueCount = Math.max(Integer.highestOneBit(ncpus / 4), 1);
+    private static class DelayedTaskSchedulers {
+        private static final ScheduledExecutorService[] INSTANCE = createDelayedTaskSchedulers();
+
+        static Future<?> schedule(Runnable command, long delay, TimeUnit unit) {
+            long tid = Thread.currentThread().threadId();
+            int index = (int) tid & (INSTANCE.length - 1);
+            return INSTANCE[index].schedule(command, delay, unit);
         }
-        var schedulers = new ScheduledExecutorService[queueCount];
-        for (int i = 0; i < queueCount; i++) {
-            ScheduledThreadPoolExecutor stpe = (ScheduledThreadPoolExecutor)
-                Executors.newScheduledThreadPool(1, task -> {
-                    Thread t = InnocuousThread.newThread("VirtualThread-unparker", task);
-                    t.setDaemon(true);
-                    return t;
-                });
-            stpe.setRemoveOnCancelPolicy(true);
-            schedulers[i] = stpe;
+
+        private static ScheduledExecutorService[] createDelayedTaskSchedulers() {
+            String propName = "jdk.virtualThreadScheduler.timerQueues";
+            String propValue = System.getProperty(propName);
+            int queueCount;
+            if (propValue != null) {
+                queueCount = Integer.parseInt(propValue);
+                if (queueCount != Integer.highestOneBit(queueCount)) {
+                    throw new RuntimeException("Value of " + propName + " must be power of 2");
+                }
+            } else {
+                int ncpus = Runtime.getRuntime().availableProcessors();
+                queueCount = Math.max(Integer.highestOneBit(ncpus / 4), 1);
+            }
+            var schedulers = new ScheduledExecutorService[queueCount];
+            for (int i = 0; i < queueCount; i++) {
+                ScheduledThreadPoolExecutor stpe = (ScheduledThreadPoolExecutor)
+                    Executors.newScheduledThreadPool(1, task -> {
+                        Thread t = InnocuousThread.newThread("VirtualThread-unparker", task);
+                        t.setDaemon(true);
+                        return t;
+                    });
+                stpe.setRemoveOnCancelPolicy(true);
+                schedulers[i] = stpe;
+            }
+            return schedulers;
         }
-        return schedulers;
     }
 
     /**
@@ -1514,4 +1532,4 @@ private static void unblockVirtualThreads() {
         unblocker.setDaemon(true);
         unblocker.start();
     }
-}
+}

src/java.base/share/classes/jdk/internal/access/JavaLangAccess.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2003, 2024, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2003, 2025, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -45,7 +45,6 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.stream.Stream;
 
 import jdk.internal.loader.NativeLibraries;
@@ -586,11 +585,6 @@ public interface JavaLangAccess {
      */
     Executor virtualThreadDefaultScheduler();
 
-    /**
-     * Returns a stream of the delayed task schedulers used for virtual threads.
-     */
-    Stream<ScheduledExecutorService> virtualThreadDelayedTaskSchedulers();
-
     /**
      * Creates a new StackWalker
      */

src/java.base/share/classes/jdk/internal/vm/JcmdVThreadCommands.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2023, 2024, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2023, 2025, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -50,19 +50,6 @@ private static byte[] printScheduler() {
         sb.append(JLA.virtualThreadDefaultScheduler())
           .append(System.lineSeparator());
 
-        // break
-        sb.append(System.lineSeparator());
-
-        // delayed task schedulers
-        sb.append("Delayed task schedulers:").append(System.lineSeparator());
-        var delayedTaskSchedulers = JLA.virtualThreadDelayedTaskSchedulers().toList();
-        IntStream.range(0, delayedTaskSchedulers.size())
-                .forEach(i -> sb.append('[')
-                                .append(i)
-                                .append("] ")
-                                .append(delayedTaskSchedulers.get(i))
-                                .append(System.lineSeparator()));
-
         return sb.toString().getBytes(StandardCharsets.UTF_8);
     }
 

test/hotspot/jtreg/serviceability/dcmd/thread/VThreadCommandsTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2023, 2024, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2023, 2025, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -59,13 +59,11 @@ class VThreadCommandsTest {
      */
     @Test
     void testVThreadScheduler() {
-        // ensure default scheduler and timeout schedulers are initialized
+        // ensure default scheduler is initialized
         Thread.startVirtualThread(() -> { });
 
         jcmd("Thread.vthread_scheduler")
-                .shouldContain(Objects.toIdentityString(defaultScheduler()))
-                .shouldContain("Delayed task schedulers:")
-                .shouldContain("[0] " + ScheduledThreadPoolExecutor.class.getName());
+                .shouldContain(Objects.toIdentityString(defaultScheduler()));
     }
 
     /**

test/micro/org/openjdk/bench/java/lang/VirtualThreadParking.java

@@ -0,0 +1,98 @@
+/*
+ * Copyright (c) 2024, 2025, Oracle and/or its affiliates. All rights reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+package org.openjdk.bench.java.lang;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.LockSupport;
+import org.openjdk.jmh.annotations.*;
+
+@BenchmarkMode(Mode.Throughput)
+@State(Scope.Benchmark)
+@Warmup(iterations = 5, time = 10, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 5, time = 10, timeUnit = TimeUnit.SECONDS)
+@Fork(value = 3)
+@OutputTimeUnit(TimeUnit.SECONDS)
+public class VirtualThreadParking {
+
+    @Param({"100", "1000", "10000"})
+    int threadCount;
+
+    /**
+     * Starts N threads that time-park, main thread unparks.
+     */
+    @Benchmark
+    public void timedParkAndUnpark1() throws Exception {
+        var threads = new Thread[threadCount];
+        var unparked = new boolean[threadCount];
+        for (int i = 0; i < threadCount; i++) {
+            threads[i] = Thread.ofVirtual().start(() -> {
+                LockSupport.parkNanos(Long.MAX_VALUE);
+            });
+        }
+        int remaining = threadCount;
+        while (remaining > 0) {
+            for (int i = 0; i < threadCount; i++) {
+                if (!unparked[i]) {
+                    Thread t = threads[i];
+                    if (t.getState() == Thread.State.TIMED_WAITING) {
+                        LockSupport.unpark(t);
+                        unparked[i] = true;
+                        remaining--;
+                    }
+                }
+            }
+            if (remaining > 0) {
+                Thread.yield();
+            }
+        }
+        for (Thread t : threads) {
+            t.join();
+        }
+    }
+
+    /**
+     * Starts N threads that time-park, start another N threads to unpark.
+     */
+    @Benchmark
+    public void timedParkAndUnpark2() throws Exception {
+        var threads = new Thread[threadCount * 2];
+        for (int i = 0; i < threadCount; i++) {
+            threads[i] = Thread.ofVirtual().start(() -> {
+                LockSupport.parkNanos(Long.MAX_VALUE);
+            });
+        }
+        for (int i = 0; i < threadCount; i++) {
+            Thread thread1 = threads[i];
+            Thread thread2 = Thread.ofVirtual().start(() -> {
+                while (thread1.getState() != Thread.State.TIMED_WAITING) {
+                    Thread.yield();
+                }
+                LockSupport.unpark(thread1);
+            });
+            threads[threadCount + i] = thread2;
+        }
+        for (Thread t : threads) {
+            t.join();
+        }
+    }
+}