From 68da02c7b536799ccca49e889c22f3e9a2691fb7 Mon Sep 17 00:00:00 2001 From: Alan Bateman Date: Sat, 10 Sep 2022 06:15:48 +0000 Subject: [PATCH] 8292240: CarrierThread.blocking not reset when spare not activated Reviewed-by: dfuchs --- .../classes/jdk/internal/misc/Blocker.java | 12 +- .../Thread/virtual/ActiviateSpareCarrier.java | 148 ++++++++++++++++++ 2 files changed, 154 insertions(+), 6 deletions(-) create mode 100644 test/jdk/java/lang/Thread/virtual/ActiviateSpareCarrier.java diff --git a/src/java.base/share/classes/jdk/internal/misc/Blocker.java b/src/java.base/share/classes/jdk/internal/misc/Blocker.java index 5c74c726067..e427d9520d5 100644 --- a/src/java.base/share/classes/jdk/internal/misc/Blocker.java +++ b/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); diff --git a/test/jdk/java/lang/Thread/virtual/ActiviateSpareCarrier.java b/test/jdk/java/lang/Thread/virtual/ActiviateSpareCarrier.java new file mode 100644 index 00000000000..c97b63bef67 --- /dev/null +++ b/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 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); + } + } + } +}