8292240: CarrierThread.blocking not reset when spare not activated

Reviewed-by: dfuchs

src/java.base/share/classes/jdk/internal/misc/Blocker.java

@@ -63,7 +63,7 @@ public class Blocker {
 
     /**
      * Marks the beginning of a possibly blocking operation.
-     * @return the return value from the attempt to compensate
+     * @return the return value from the attempt to compensate or -1 if not attempted
      */
     public static long begin() {
         if (VM.isBooted()
@@ -81,17 +81,17 @@ public class Blocker {
                 }
             }
         }
-        return 0;
+        return -1;
     }
 
     /**
      * Marks the beginning of a possibly blocking operation.
      * @param blocking true if the operation may block, otherwise false
-     * @return the return value from the attempt to compensate when blocking is true,
-     * another value when blocking is false
+     * @return the return value from the attempt to compensate, -1 if not attempted
+     * or blocking is false
      */
     public static long begin(boolean blocking) {
-        return (blocking) ? begin() : 0;
+        return (blocking) ? begin() : -1;
     }
 
     /**
@@ -99,7 +99,7 @@ public class Blocker {
      * @param compensateReturn the value returned by the begin method
      */
     public static void end(long compensateReturn) {
-        if (compensateReturn > 0) {
+        if (compensateReturn >= 0) {
             assert currentCarrierThread() instanceof CarrierThread ct && ct.inBlocking();
             CarrierThread ct = (CarrierThread) currentCarrierThread();
             ForkJoinPools.endCompensatedBlock(ct.getPool(), compensateReturn);

test/jdk/java/lang/Thread/virtual/ActiviateSpareCarrier.java

@@ -0,0 +1,148 @@
+/*
+ * Copyright (c) 2022, 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.
+ */
+
+/**
+ * @test
+ * @bug 8292240
+ * @summary Test the scenario where a blocking operation pins a virtual thread to its
+ *   carrier thread (cT1) and doesn't activate a spare. Subsequent blocking operations
+ *   that pin a virtual thread to cT1 should attempt to activate a spare.
+ * @requires vm.continuations
+ * @enablePreview
+ * @run main/othervm
+ *     -Djdk.virtualThreadScheduler.parallelism=1
+ *     -Djdk.virtualThreadScheduler.maxPoolSize=2 ActiviateSpareCarrier 100
+ */
+
+import java.time.Duration;
+import java.util.Comparator;
+import java.util.List;
+import java.util.concurrent.ForkJoinWorkerThread;
+import java.util.stream.Collectors;
+
+public class ActiviateSpareCarrier {
+
+    private static final int DEFAULT_ITERTAIONS = 10_000;
+
+    private static final Object LOCK = new Object();
+
+    public static void main(String[] args) throws Exception {
+        int iterations;
+        if (args.length == 0) {
+            iterations = DEFAULT_ITERTAIONS;
+        } else {
+            iterations = Integer.parseInt(args[0]);
+        }
+        for (int i = 0; i < iterations; i++) {
+            test(i);
+        }
+    }
+
+    /**
+     * This method creates 3 virtual threads:
+     * - thread1 blocks in Object.wait, activating a spare carrier thread
+     * - thread2 is started and runs on the spare carrier thread
+     * - thread1 is notified causing it to re-adjust the release count and terminate
+     * - thread3 is started and should run on the one active thread
+     *
+     * This method need invoked at least twice in the same VM.
+     */
+    private static void test(int i) throws Exception {
+        System.out.printf("---- %d ----%n", i);
+
+        // thread1 blocks in wait, this triggers a tryCompensate to activate a spare thread
+        Thread thread1 = Thread.ofVirtual().unstarted(() -> {
+            System.out.println(Thread.currentThread());
+            synchronized (LOCK) {
+                try {
+                    LOCK.wait();
+                } catch (InterruptedException e) { }
+            }
+        });
+        System.out.printf("starting waiter thread #%d%n", thread1.threadId());
+        thread1.start();
+
+        // wait for thread1 to block in Object.wait
+        while (thread1.getState() != Thread.State.WAITING) {
+            Thread.sleep(10);
+        }
+
+        // start another virtual thread, it should run on the spare carrier thread
+        startAndJoinVirtualThread();
+
+        // notify thread1, this releases the blocker
+        synchronized (LOCK) {
+            LOCK.notifyAll();
+        }
+        joinThread(thread1);
+
+        // start another virtual thread after counts have been re-adjusted
+        startAndJoinVirtualThread();
+    }
+
+    /**
+     * Start a virtual thread and wait for it to terminate.
+     */
+    private static void startAndJoinVirtualThread() throws InterruptedException {
+        Thread thread = Thread.ofVirtual().unstarted(() -> {
+            System.out.println(Thread.currentThread());
+        });
+        System.out.format("starting #%d%n", thread.threadId());
+        thread.start();
+        joinThread(thread);
+    }
+
+    /**
+     * Wait for the give thread to terminate with diagnostic output if the thread does
+     * not terminate quickly.
+     */
+    private static void joinThread(Thread thread) throws InterruptedException {
+        long tid = thread.threadId();
+        System.out.printf("Waiting for #%d to terminate%n", tid);
+        boolean terminated = thread.join(Duration.ofSeconds(2));
+        if (!terminated) {
+            System.out.printf("#%d did not terminate quickly, continue to wait...%n", tid);
+            printForkJoinWorkerThreads();
+            thread.join();
+        }
+        System.out.printf("#%d terminated%n", tid);
+    }
+
+    /**
+     * Print the list of ForkJoinWorkerThreads and their stack traces.
+     */
+    private static void printForkJoinWorkerThreads() {
+        List<Thread> threads = Thread.getAllStackTraces().keySet().stream()
+                .filter(t -> t instanceof ForkJoinWorkerThread)
+                .sorted(Comparator.comparingLong(Thread::threadId))
+                .collect(Collectors.toList());
+        System.out.println("ForkJoinWorkerThreads:");
+        for (Thread t : threads) {
+            System.out.printf("    %s%n", t);
+            StackTraceElement[] stack = t.getStackTrace();
+            for (StackTraceElement e : stack) {
+                System.out.printf("      %s%n", e);
+            }
+        }
+    }
+}