diff --git a/make/data/hotspot-symbols/symbols-unix b/make/data/hotspot-symbols/symbols-unix index fbb82a11fac..0ce1483f890 100644 --- a/make/data/hotspot-symbols/symbols-unix +++ b/make/data/hotspot-symbols/symbols-unix @@ -224,6 +224,7 @@ JVM_VirtualThreadMount JVM_VirtualThreadUnmount JVM_VirtualThreadHideFrames JVM_VirtualThreadDisableSuspend +JVM_VirtualThreadWaitForPendingList # Scoped values JVM_EnsureMaterializedForStackWalk_func diff --git a/src/hotspot/share/include/jvm.h b/src/hotspot/share/include/jvm.h index 5d6ab27a3a1..b534dc36582 100644 --- a/src/hotspot/share/include/jvm.h +++ b/src/hotspot/share/include/jvm.h @@ -1157,6 +1157,9 @@ JVM_VirtualThreadHideFrames(JNIEnv* env, jobject vthread, jboolean hide); JNIEXPORT void JNICALL JVM_VirtualThreadDisableSuspend(JNIEnv* env, jobject vthread, jboolean enter); +JNIEXPORT jobject JNICALL +JVM_VirtualThreadWaitForPendingList(JNIEnv* env); + /* * Core reflection support. */ diff --git a/src/java.base/share/classes/java/lang/VirtualThread.java b/src/java.base/share/classes/java/lang/VirtualThread.java index 19635db8bbd..1ae08f9ed56 100644 --- a/src/java.base/share/classes/java/lang/VirtualThread.java +++ b/src/java.base/share/classes/java/lang/VirtualThread.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2018, 2023, Oracle and/or its affiliates. All rights reserved. + * Copyright (c) 2018, 2024, 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 @@ -22,13 +22,9 @@ * or visit www.oracle.com if you need additional information or have any * questions. */ -/* - * =========================================================================== - * (c) Copyright IBM Corp. 2022, 2023 All Rights Reserved - * =========================================================================== - */ package java.lang; +import java.nio.charset.StandardCharsets; import java.security.AccessController; import java.security.PrivilegedAction; import java.util.Locale; @@ -46,7 +42,6 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledThreadPoolExecutor; import jdk.internal.event.VirtualThreadEndEvent; -import jdk.internal.event.VirtualThreadPinnedEvent; import jdk.internal.event.VirtualThreadStartEvent; import jdk.internal.event.VirtualThreadSubmitFailedEvent; import jdk.internal.misc.CarrierThread; @@ -74,13 +69,13 @@ 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 UNPARKER = createDelayedTaskScheduler(); - private static final int TRACE_PINNING_MODE = tracePinningMode(); + 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"); private static final long CARRIER_THREAD = U.objectFieldOffset(VirtualThread.class, "carrierThread"); private static final long TERMINATION = U.objectFieldOffset(VirtualThread.class, "termination"); + private static final long ON_WAITING_LIST = U.objectFieldOffset(VirtualThread.class, "onWaitingList"); // scheduler and continuation private final Executor scheduler; @@ -111,6 +106,11 @@ final class VirtualThread extends BaseVirtualThread { * TIMED_PARKED -> UNPARKED // unparked, may be scheduled to continue * TIMED_PINNED -> RUNNING // unparked, continue execution on same carrier * + * RUNNABLE -> BLOCKING // blocking on monitor enter + * BLOCKING -> BLOCKED // blocked on monitor enter + * BLOCKED -> UNBLOCKED // unblocked, may be scheduled to continue + * UNBLOCKED -> RUNNING // continue execution after blocked on monitor enter + * * RUNNING -> YIELDING // Thread.yield * YIELDING -> YIELDED // cont.yield successful, may be scheduled to continue * YIELDING -> RUNNING // cont.yield failed @@ -133,6 +133,11 @@ final class VirtualThread extends BaseVirtualThread { private static final int YIELDING = 10; private static final int YIELDED = 11; // unmounted but runnable + // monitor enter + private static final int BLOCKING = 12; + private static final int BLOCKED = 13; // unmounted + private static final int UNBLOCKED = 14; // unmounted but runnable + private static final int TERMINATED = 99; // final state // can be suspended from scheduling when unmounted @@ -141,12 +146,24 @@ final class VirtualThread extends BaseVirtualThread { // parking permit private volatile boolean parkPermit; + // used to mark thread as ready to be unblocked while it is concurrently blocking + private volatile boolean unblocked; + + // a positive value if "responsible thread" blocked on monitor enter, accessed by VM + private volatile byte recheckInterval; + // carrier thread when mounted, accessed by VM private volatile Thread carrierThread; // termination object when joining, created lazily if needed private volatile CountDownLatch termination; + // has the value 1 when on the list of virtual threads waiting to be unblocked + private volatile byte onWaitingList; + + // next virtual thread on the list of virtual threads waiting to be unblocked + private volatile VirtualThread next; + /** * Returns the continuation scope used for virtual threads. */ @@ -191,22 +208,11 @@ static ContinuationScope continuationScope() { private static class VThreadContinuation extends Continuation { VThreadContinuation(VirtualThread vthread, Runnable task) { super(VTHREAD_SCOPE, wrap(vthread, task)); - this.vthread = vthread; } @Override protected void onPinned(Continuation.Pinned reason) { - if (TRACE_PINNING_MODE > 0) { - boolean printAll = (TRACE_PINNING_MODE == 1); - VirtualThread vthread = (VirtualThread) Thread.currentThread(); - int oldState = vthread.state(); - try { - // avoid printing when in transition states - vthread.setState(RUNNING); - PinnedThreadPrinter.printStackTrace(System.out, reason, printAll); - } finally { - vthread.setState(oldState); - } - } + // emit JFR event + virtualThreadPinnedEvent(reason.value()); } private static Runnable wrap(VirtualThread vthread, Runnable task) { return new Runnable() { @@ -232,7 +238,8 @@ private void runContinuation() { // set state to RUNNING int initialState = state(); - if (initialState == STARTED || initialState == UNPARKED || initialState == YIELDED) { + if (initialState == STARTED || initialState == UNPARKED + || initialState == UNBLOCKED || initialState == YIELDED) { // newly started or continue after parking/blocking/Thread.yield if (!compareAndSetState(initialState, RUNNING)) { return; @@ -274,6 +281,20 @@ private void submitRunContinuation() { } } + /** + * Submits the runContinuation task the scheduler. For the default scheduler, the task + * will be pushed to an external submission queue. + * @throws RejectedExecutionException + */ + private void externalSubmitRunContinuation() { + if (scheduler == DEFAULT_SCHEDULER + && currentCarrierThread() instanceof CarrierThread ct) { + externalSubmitRunContinuation(ct.getPool()); + } else { + submitRunContinuation(); + } + } + /** * Submits the runContinuation task to given scheduler with a lazy submit. * @throws RejectedExecutionException @@ -391,6 +412,8 @@ private void mount() { @ChangesCurrentThread @ReservedStackAccess private void unmount() { + assert !Thread.holdsLock(interruptLock); + // set Thread.currentThread() to return the platform thread Thread carrier = this.carrierThread; carrier.setCurrentThread(carrier); @@ -416,6 +439,7 @@ private void switchToCarrierThread() { assert Thread.currentThread() == this && carrier == Thread.currentCarrierThread(); carrier.setCurrentThread(carrier); + setLockId(this.threadId()); // keep lockid of vthread } /** @@ -466,6 +490,11 @@ private boolean yieldContinuation() { private void afterYield() { assert carrierThread == null; + // re-adjust parallelism if the virtual thread yielded when compensating + if (currentThread() instanceof CarrierThread ct) { + ct.endBlocking(); + } + int s = state(); // LockSupport.park/parkNanos @@ -499,6 +528,33 @@ private void afterYield() { return; } + // blocking on monitorenter + if (s == BLOCKING) { + setState(BLOCKED); + + // may have been unblocked while blocking + if (unblocked && compareAndSetState(BLOCKED, UNBLOCKED)) { + unblocked = false; + submitRunContinuation(); + return; + } + + // if thread is the designated responsible thread for a monitor then schedule + // it to wakeup so that it can check and recover. See objectMonitor.cpp. + int recheckInterval = this.recheckInterval; + if (recheckInterval > 0 && state() == BLOCKED) { + assert recheckInterval >= 1 && recheckInterval <= 6; + // 4 ^ (recheckInterval - 1) = 1, 4, 16, ... 1024 + long delay = 1 << (recheckInterval - 1) << (recheckInterval - 1); + Future unblocker = delayedTaskScheduler().schedule(this::unblock, delay, MILLISECONDS); + // cancel if unblocked while scheduling the unblock + if (state() != BLOCKED) { + unblocker.cancel(false); + } + } + return; + } + assert false; } @@ -562,8 +618,8 @@ void start(ThreadContainer container) { // scoped values may be inherited inheritScopedValueBindings(container); - // submit task to run thread - submitRunContinuation(); + // submit task to run thread, using externalSubmit if possible + externalSubmitRunContinuation(); started = true; } finally { if (!started) { @@ -666,14 +722,6 @@ void parkNanos(long nanos) { private void parkOnCarrierThread(boolean timed, long nanos) { assert state() == RUNNING; - VirtualThreadPinnedEvent event; - try { - event = new VirtualThreadPinnedEvent(); - event.begin(); - } catch (OutOfMemoryError e) { - event = null; - } - setState(timed ? TIMED_PINNED : PINNED); try { if (!parkPermit) { @@ -689,16 +737,15 @@ private void parkOnCarrierThread(boolean timed, long nanos) { // consume parking permit setParkPermit(false); - - if (event != null) { - try { - event.commit(); - } catch (OutOfMemoryError e) { - // ignore - } - } } + /** + * jdk.VirtualThreadPinned is emitted by HotSpot VM when pinned. Call into VM to + * emit event to avoid having a JFR in Java with the same name (but different ID) + * to events emitted by the VM. + */ + private static native void virtualThreadPinnedEvent(int reason); + /** * Schedule this virtual thread to be unparked after a given delay. */ @@ -708,7 +755,7 @@ private Future scheduleUnpark(long nanos) { // need to switch to current carrier thread to avoid nested parking switchToCarrierThread(); try { - return UNPARKER.schedule(this::unpark, nanos, NANOSECONDS); + return delayedTaskScheduler().schedule(this::unpark, nanos, NANOSECONDS); } finally { switchToVirtualThread(this); } @@ -719,6 +766,7 @@ private Future scheduleUnpark(long nanos) { */ @ChangesCurrentThread private void cancel(Future future) { + assert Thread.currentThread() == this; if (!future.isDone()) { // need to switch to current carrier thread to avoid nested parking switchToCarrierThread(); @@ -757,7 +805,7 @@ void unpark() { } } else if ((s == PINNED) || (s == TIMED_PINNED)) { // unpark carrier thread when pinned - notifyJvmtiDisableSuspend(true); + disableSuspendAndPreempt(); try { synchronized (carrierThreadAccessLock()) { Thread carrier = carrierThread; @@ -766,12 +814,25 @@ void unpark() { } } } finally { - notifyJvmtiDisableSuspend(false); + enableSuspendAndPreempt(); } } } } + /** + * Re-enables this virtual thread for scheduling after blocking on monitor enter. + * @throws RejectedExecutionException if the scheduler cannot accept a task + */ + private void unblock() { + assert !Thread.currentThread().isVirtual(); + unblocked = true; + if (state() == BLOCKED && compareAndSetState(BLOCKED, UNBLOCKED)) { + unblocked = false; + submitRunContinuation(); + } + } + /** * Attempts to yield the current virtual thread (Thread.yield). */ @@ -858,18 +919,32 @@ boolean joinNanos(long nanos) throws InterruptedException { return true; } + @Override + void blockedOn(Interruptible b) { + disableSuspendAndPreempt(); + try { + super.blockedOn(b); + } finally { + enableSuspendAndPreempt(); + } + } + @Override @SuppressWarnings("removal") public void interrupt() { if (Thread.currentThread() != this) { checkAccess(); - notifyJvmtiDisableSuspend(true); + + // if current thread is a virtual thread then prevent it from being + // suspended or unmounted when entering or holding interruptLock + Interruptible blocker; + disableSuspendAndPreempt(); try { synchronized (interruptLock) { interrupted = true; - Interruptible b = nioBlocker; - if (b != null) { - b.interrupt(this); + blocker = nioBlocker(); + if (blocker != null) { + blocker.interrupt(this); } // interrupt carrier thread if mounted @@ -877,8 +952,14 @@ public void interrupt() { if (carrier != null) carrier.setInterrupt(); } } finally { - notifyJvmtiDisableSuspend(false); + enableSuspendAndPreempt(); } + + // notify blocker after releasing interruptLock + if (blocker != null) { + blocker.postInterrupt(); + } + } else { interrupted = true; carrierThread.setInterrupt(); @@ -896,14 +977,14 @@ boolean getAndClearInterrupt() { assert Thread.currentThread() == this; boolean oldValue = interrupted; if (oldValue) { - notifyJvmtiDisableSuspend(true); + disableSuspendAndPreempt(); try { synchronized (interruptLock) { interrupted = false; carrierThread.clearInterrupt(); } } finally { - notifyJvmtiDisableSuspend(false); + enableSuspendAndPreempt(); } } return oldValue; @@ -926,18 +1007,31 @@ Thread.State threadState() { case YIELDED: // runnable, not mounted return Thread.State.RUNNABLE; + case UNBLOCKED: + // if designated responsible thread for monitor then thread is blocked + if (isResponsibleForMonitor()) { + return Thread.State.BLOCKED; + } else { + return Thread.State.RUNNABLE; + } case RUNNING: + // if designated responsible thread for monitor then thread is blocked + if (isResponsibleForMonitor()) { + return Thread.State.BLOCKED; + } // if mounted then return state of carrier thread - notifyJvmtiDisableSuspend(true); - try { - synchronized (carrierThreadAccessLock()) { - Thread carrierThread = this.carrierThread; - if (carrierThread != null) { - return carrierThread.threadState(); + if (Thread.currentThread() != this) { + disableSuspendAndPreempt(); + try { + synchronized (carrierThreadAccessLock()) { + Thread carrierThread = this.carrierThread; + if (carrierThread != null) { + return carrierThread.threadState(); + } } + } finally { + enableSuspendAndPreempt(); } - } finally { - notifyJvmtiDisableSuspend(false); } // runnable, mounted return Thread.State.RUNNABLE; @@ -952,6 +1046,9 @@ Thread.State threadState() { case TIMED_PARKED: case TIMED_PINNED: return State.TIMED_WAITING; + case BLOCKING: + case BLOCKED: + return State.BLOCKED; case TERMINATED: return Thread.State.TERMINATED; default: @@ -959,6 +1056,14 @@ Thread.State threadState() { } } + /** + * Returns true if thread is the designated responsible thread for a monitor. + * See objectMonitor.cpp for details. + */ + private boolean isResponsibleForMonitor() { + return (recheckInterval > 0); + } + @Override boolean alive() { int s = state; @@ -997,13 +1102,13 @@ private StackTraceElement[] tryGetStackTrace() { case RUNNING, PINNED, TIMED_PINNED -> { return null; // mounted } - case PARKED, TIMED_PARKED -> { + case PARKED, TIMED_PARKED, BLOCKED -> { // unmounted, not runnable } - case UNPARKED, YIELDED -> { + case UNPARKED, UNBLOCKED, YIELDED -> { // unmounted, runnable } - case PARKING, TIMED_PARKING, YIELDING -> { + case PARKING, TIMED_PARKING, BLOCKING, YIELDING -> { return null; // in transition } default -> throw new InternalError("" + initialState); @@ -1024,7 +1129,7 @@ private StackTraceElement[] tryGetStackTrace() { setState(initialState); } boolean resubmit = switch (initialState) { - case UNPARKED, YIELDED -> { + case UNPARKED, UNBLOCKED, YIELDED -> { // resubmit as task may have run while suspended yield true; } @@ -1032,6 +1137,10 @@ private StackTraceElement[] tryGetStackTrace() { // resubmit if unparked while suspended yield parkPermit && compareAndSetState(initialState, UNPARKED); } + case BLOCKED -> { + // resubmit if unblocked while suspended + yield unblocked && compareAndSetState(BLOCKED, UNBLOCKED); + } default -> throw new InternalError(); }; if (resubmit) { @@ -1050,32 +1159,49 @@ public String toString() { sb.append(name); } sb.append("]/"); - Thread carrier = carrierThread; - if (carrier != null) { - // include the carrier thread state and name when mounted - notifyJvmtiDisableSuspend(true); + + // add the carrier state and thread name when mounted + boolean mounted; + if (Thread.currentThread() == this) { + mounted = appendCarrierInfo(sb); + } else { + disableSuspendAndPreempt(); try { synchronized (carrierThreadAccessLock()) { - carrier = carrierThread; - if (carrier != null) { - String stateAsString = carrier.threadState().toString(); - sb.append(stateAsString.toLowerCase(Locale.ROOT)); - sb.append('@'); - sb.append(carrier.getName()); - } + mounted = appendCarrierInfo(sb); } } finally { - notifyJvmtiDisableSuspend(false); + enableSuspendAndPreempt(); } } - // include virtual thread state when not mounted - if (carrier == null) { + + // add virtual thread state when not mounted + if (!mounted) { String stateAsString = threadState().toString(); sb.append(stateAsString.toLowerCase(Locale.ROOT)); } + return sb.toString(); } + /** + * Appends the carrier state and thread name to the string buffer if mounted. + * @return true if mounted, false if not mounted + */ + private boolean appendCarrierInfo(StringBuilder sb) { + assert Thread.currentThread() == this || Thread.holdsLock(carrierThreadAccessLock()); + Thread carrier = carrierThread; + if (carrier != null) { + String stateAsString = carrier.threadState().toString(); + sb.append(stateAsString.toLowerCase(Locale.ROOT)); + sb.append('@'); + sb.append(carrier.getName()); + return true; + } else { + return false; + } + } + @Override public int hashCode() { return (int) threadId(); @@ -1086,6 +1212,15 @@ public boolean equals(Object obj) { return obj == this; } + /** + * Returns a ScheduledExecutorService to execute a delayed task. + */ + private ScheduledExecutorService delayedTaskScheduler() { + long tid = Thread.currentThread().threadId(); + int index = (int) tid & (DELAYED_TASK_SCHEDULERS.length - 1); + return DELAYED_TASK_SCHEDULERS[index]; + } + /** * Returns the termination object, creating it if needed. */ @@ -1109,6 +1244,22 @@ private Object carrierThreadAccessLock() { return interruptLock; } + /** + * Disallow the current thread be suspended or preempted. + */ + private void disableSuspendAndPreempt() { + notifyJvmtiDisableSuspend(true); + Continuation.pin(); + } + + /** + * Allow the current thread be suspended or preempted. + */ + private void enableSuspendAndPreempt() { + Continuation.unpin(); + notifyJvmtiDisableSuspend(false); + } + // -- wrappers for get/set of state, parking permit, and carrier thread -- private int state() { @@ -1123,6 +1274,10 @@ private boolean compareAndSetState(int expectedValue, int newValue) { return U.compareAndSetInt(this, STATE, expectedValue, newValue); } + private boolean compareAndSetOnWaitingList(byte expectedValue, byte newValue) { + return U.compareAndSetByte(this, ON_WAITING_LIST, expectedValue, newValue); + } + private void setParkPermit(boolean newValue) { if (parkPermit != newValue) { parkPermit = newValue; @@ -1142,6 +1297,9 @@ private void setCarrierThread(Thread carrier) { this.carrierThread = carrier; } + @IntrinsicCandidate + private static native void setLockId(long tid); + // -- JVM TI support -- @IntrinsicCandidate @@ -1162,10 +1320,10 @@ private void setCarrierThread(Thread carrier) { @IntrinsicCandidate @JvmtiMountTransition - private native void notifyJvmtiHideFrames(boolean hide); + private static native void notifyJvmtiHideFrames(boolean hide); @IntrinsicCandidate - private native void notifyJvmtiDisableSuspend(boolean enter); + private static native void notifyJvmtiDisableSuspend(boolean enter); private static native void registerNatives(); static { @@ -1211,37 +1369,75 @@ private static ForkJoinPool createDefaultScheduler() { } /** - * Creates the ScheduledThreadPoolExecutor used for timed unpark. + * Invoked by the VM for the Thread.vthread_scheduler diagnostic command. + */ + private static byte[] printDefaultScheduler() { + return String.format("%s%n", DEFAULT_SCHEDULER.toString()) + .getBytes(StandardCharsets.UTF_8); + } + + /** + * Creates the ScheduledThreadPoolExecutors used to execute delayed tasks. */ - private static ScheduledExecutorService createDelayedTaskScheduler() { - String propValue = GetPropertyAction.privilegedGetProperty("jdk.unparker.maxPoolSize"); - int poolSize; + private static ScheduledExecutorService[] createDelayedTaskSchedulers() { + String propName = "jdk.virtualThreadScheduler.timerQueues"; + String propValue = GetPropertyAction.privilegedGetProperty(propName); + int queueCount; if (propValue != null) { - poolSize = Integer.parseInt(propValue); + queueCount = Integer.parseInt(propValue); + if (queueCount != Integer.highestOneBit(queueCount)) { + throw new RuntimeException("Value of " + propName + " must be power of 2"); + } } else { - poolSize = 1; + int ncpus = Runtime.getRuntime().availableProcessors(); + queueCount = Math.max(Integer.highestOneBit(ncpus / 4), 1); } - ScheduledThreadPoolExecutor stpe = (ScheduledThreadPoolExecutor) - Executors.newScheduledThreadPool(poolSize, task -> { - return InnocuousThread.newThread("VirtualThread-unparker", task); - }); - stpe.setRemoveOnCancelPolicy(true); - return stpe; + 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; } /** - * Reads the value of the jdk.tracePinnedThreads property to determine if stack - * traces should be printed when a carrier thread is pinned when a virtual thread - * attempts to park. + * Schedule virtual threads that are ready to be scheduled after they blocked on + * monitor enter. */ - private static int tracePinningMode() { - String propValue = GetPropertyAction.privilegedGetProperty("jdk.tracePinnedThreads"); - if (propValue != null) { - if (propValue.length() == 0 || "full".equalsIgnoreCase(propValue)) - return 1; - if ("short".equalsIgnoreCase(propValue)) - return 2; + private static void unblockVirtualThreads() { + while (true) { + VirtualThread vthread = takeVirtualThreadListToUnblock(); + while (vthread != null) { + assert vthread.onWaitingList == 1; + VirtualThread nextThread = vthread.next; + + // remove from list and unblock + vthread.next = null; + boolean changed = vthread.compareAndSetOnWaitingList((byte) 1, (byte) 0); + assert changed; + vthread.unblock(); + + vthread = nextThread; + } } - return 0; } -} + + /** + * Retrieves the list of virtual threads that are waiting to be unblocked, waiting + * if necessary until a list of one or more threads becomes available. + */ + private static native VirtualThread takeVirtualThreadListToUnblock(); + + static { + var unblocker = InnocuousThread.newThread("VirtualThread-unblocker", + VirtualThread::unblockVirtualThreads); + unblocker.setDaemon(true); + unblocker.start(); + } +} \ No newline at end of file diff --git a/src/java.base/share/native/libjava/VirtualThread.c b/src/java.base/share/native/libjava/VirtualThread.c index 94dbe0b7e37..1ef29270ddd 100644 --- a/src/java.base/share/native/libjava/VirtualThread.c +++ b/src/java.base/share/native/libjava/VirtualThread.c @@ -38,6 +38,7 @@ static JNINativeMethod methods[] = { { "notifyJvmtiUnmount", "(Z)V", (void *)&JVM_VirtualThreadUnmount }, { "notifyJvmtiHideFrames", "(Z)V", (void *)&JVM_VirtualThreadHideFrames }, { "notifyJvmtiDisableSuspend", "(Z)V", (void *)&JVM_VirtualThreadDisableSuspend }, + { "waitForPendingList", "()" VIRTUAL_THREAD, (void *)&JVM_VirtualThreadWaitForPendingList }, }; JNIEXPORT void JNICALL diff --git a/test/jdk/java/lang/Thread/virtual/MonitorsTest.java b/test/jdk/java/lang/Thread/virtual/MonitorsTest.java new file mode 100644 index 00000000000..43dbd511e97 --- /dev/null +++ b/test/jdk/java/lang/Thread/virtual/MonitorsTest.java @@ -0,0 +1,408 @@ +/* + * Copyright (c) 2023, 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 + * @summary Test virtual threads using synchronized + * @library /test/lib + * @modules java.base/java.lang:+open + * + * @run junit/othervm/timeout=10 -Xint MonitorsTest + * @run junit/othervm/timeout=50 -Xcomp MonitorsTest + * @run junit/othervm/timeout=50 MonitorsTest + * @run junit/othervm/timeout=50 -XX:+FullGCALot -XX:FullGCALotInterval=1000 MonitorsTest + */ + + import java.util.concurrent.atomic.AtomicInteger; + import java.util.concurrent.*; + + import org.junit.jupiter.api.Test; + import static org.junit.jupiter.api.Assertions.*; + + class MonitorsTest { + final int CARRIER_COUNT = 8; + ExecutorService scheduler = Executors.newFixedThreadPool(CARRIER_COUNT); + + static final Object globalLock = new Object(); + static volatile boolean finish = false; + static volatile int counter = 0; + + ///////////////////////////////////////////////////////////////////// + //////////////////////////// BASIC TESTS //////////////////////////// + ///////////////////////////////////////////////////////////////////// + + static final Runnable FOO = () -> { + Object lock = new Object(); + synchronized(lock) { + while(!finish) { + Thread.yield(); + } + } + System.out.println("Exiting FOO from thread " + Thread.currentThread().getName()); + }; + + static final Runnable BAR = () -> { + synchronized(globalLock) { + counter++; + } + System.out.println("Exiting BAR from thread " + Thread.currentThread().getName()); + }; + + /** + * Test yield while holding monitor. + */ + @Test + void testBasic() throws Exception { + final int VT_COUNT = CARRIER_COUNT; + + // Create first batch of VT threads. + Thread firstBatch[] = new Thread[VT_COUNT]; + for (int i = 0; i < VT_COUNT; i++) { + firstBatch[i] = ThreadBuilders.virtualThreadBuilder(scheduler).name("FirstBatchVT-" + i).start(FOO); + } + + // Give time for all threads to reach Thread.yield + Thread.sleep(1000); + + // Create second batch of VT threads. + Thread secondBatch[] = new Thread[VT_COUNT]; + for (int i = 0; i < VT_COUNT; i++) { + secondBatch[i] = ThreadBuilders.virtualThreadBuilder(scheduler).name("SecondBatchVT-" + i).start(BAR); + } + + while(counter != VT_COUNT) {} + + finish = true; + + for (int i = 0; i < VT_COUNT; i++) { + firstBatch[i].join(); + } + for (int i = 0; i < VT_COUNT; i++) { + secondBatch[i].join(); + } + } + + static final Runnable BAR2 = () -> { + synchronized(globalLock) { + counter++; + } + recursive2(10); + System.out.println("Exiting BAR2 from thread " + Thread.currentThread().getName() + "with counter=" + counter); + }; + + static void recursive2(int count) { + synchronized(Thread.currentThread()) { + if (count > 0) { + recursive2(count - 1); + } else { + synchronized(globalLock) { + counter++; + Thread.yield(); + } + } + } + } + + /** + * Test yield while holding monitor with recursive locking. + */ + @Test + void testRecursive() throws Exception { + final int VT_COUNT = CARRIER_COUNT; + counter = 0; + finish = false; + + // Create first batch of VT threads. + Thread firstBatch[] = new Thread[VT_COUNT]; + for (int i = 0; i < VT_COUNT; i++) { + firstBatch[i] = ThreadBuilders.virtualThreadBuilder(scheduler).name("FirstBatchVT-" + i).start(FOO); + } + + // Give time for all threads to reach Thread.yield + Thread.sleep(1000); + + // Create second batch of VT threads. + Thread secondBatch[] = new Thread[VT_COUNT]; + for (int i = 0; i < VT_COUNT; i++) { + secondBatch[i] = ThreadBuilders.virtualThreadBuilder(scheduler).name("SecondBatchVT-" + i).start(BAR2); + } + + while(counter != 2*VT_COUNT) {} + + finish = true; + + for (int i = 0; i < VT_COUNT; i++) { + firstBatch[i].join(); + } + for (int i = 0; i < VT_COUNT; i++) { + secondBatch[i].join(); + } + } + + static final Runnable FOO3 = () -> { + synchronized(globalLock) { + while(!finish) { + Thread.yield(); + } + } + System.out.println("Exiting FOO3 from thread " + Thread.currentThread().getName()); + }; + + /** + * Test contention on monitorenter. + */ + @Test + void testContention() throws Exception { + final int VT_COUNT = CARRIER_COUNT * 8; + counter = 0; + finish = false; + + // Create batch of VT threads. + Thread batch[] = new Thread[VT_COUNT]; + for (int i = 0; i < VT_COUNT; i++) { + batch[i] = ThreadBuilders.virtualThreadBuilder(scheduler).name("BatchVT-" + i).start(FOO3); + } + + // Give time for all threads to reach synchronized(globalLock) + Thread.sleep(2000); + + finish = true; + + for (int i = 0; i < VT_COUNT; i++) { + batch[i].join(); + } + } + + ///////////////////////////////////////////////////////////////////// + //////////////////////////// MAIN TESTS ///////////////////////////// + ///////////////////////////////////////////////////////////////////// + + static final int MONITORS_CNT = 12; + static Object[] globalLockArray; + static AtomicInteger workerCount = new AtomicInteger(0); + + static void recursive4_1(int depth, int lockNumber) { + if (depth > 0) { + recursive4_1(depth - 1, lockNumber); + } else { + if (Math.random() < 0.5) { + Thread.yield(); + } + recursive4_2(lockNumber); + } + } + + static void recursive4_2(int lockNumber) { + if (lockNumber + 2 <= MONITORS_CNT - 1) { + lockNumber += 2; + synchronized(globalLockArray[lockNumber]) { + Thread.yield(); + recursive4_2(lockNumber); + } + } + } + + static final Runnable FOO4 = () -> { + while (!finish) { + int lockNumber = ThreadLocalRandom.current().nextInt(0, MONITORS_CNT - 1); + synchronized(globalLockArray[lockNumber]) { + recursive4_1(lockNumber, lockNumber); + } + } + workerCount.getAndIncrement(); + System.out.println("Exiting FOO4 from thread " + Thread.currentThread().getName()); + }; + + /** + * Test contention on monitorenter with extra monitors on stack shared by all threads. + */ + @Test + void testContentionMultipleMonitors() throws Exception { + final int VT_COUNT = CARRIER_COUNT * 8; + workerCount.getAndSet(0); + finish = false; + + globalLockArray = new Object[MONITORS_CNT]; + for (int i = 0; i < MONITORS_CNT; i++) { + globalLockArray[i] = new Object(); + } + + Thread batch[] = new Thread[VT_COUNT]; + for (int i = 0; i < VT_COUNT; i++) { + batch[i] = ThreadBuilders.virtualThreadBuilder(scheduler).name("BatchVT-" + i).start(FOO4); + } + + Thread.sleep(10000); + finish = true; + + for (int i = 0; i < VT_COUNT; i++) { + batch[i].join(); + } + + if (workerCount.get() != VT_COUNT) { + throw new RuntimeException("testContentionMultipleMonitors2 failed. Expected " + VT_COUNT + "but found " + workerCount.get()); + } + } + + + static void recursive5_1(int depth, int lockNumber, Object[] myLockArray) { + if (depth > 0) { + recursive5_1(depth - 1, lockNumber, myLockArray); + } else { + if (Math.random() < 0.5) { + Thread.yield(); + } + recursive5_2(lockNumber, myLockArray); + } + } + + static void recursive5_2(int lockNumber, Object[] myLockArray) { + if (lockNumber + 2 <= MONITORS_CNT - 1) { + lockNumber += 2; + synchronized (myLockArray[lockNumber]) { + if (Math.random() < 0.5) { + Thread.yield(); + } + synchronized (globalLockArray[lockNumber]) { + Thread.yield(); + recursive5_2(lockNumber, myLockArray); + } + } + } + } + + static final Runnable FOO5 = () -> { + Object[] myLockArray = new Object[MONITORS_CNT]; + for (int i = 0; i < MONITORS_CNT; i++) { + myLockArray[i] = new Object(); + } + + while (!finish) { + int lockNumber = ThreadLocalRandom.current().nextInt(0, MONITORS_CNT - 1); + synchronized (myLockArray[lockNumber]) { + synchronized (globalLockArray[lockNumber]) { + recursive5_1(lockNumber, lockNumber, myLockArray); + } + } + } + workerCount.getAndIncrement(); + System.out.println("Exiting FOO5 from thread " + Thread.currentThread().getName()); + }; + + /** + * Test contention on monitorenter with extra monitors on stack both local only and shared by all threads. + */ + @Test + void testContentionMultipleMonitors2() throws Exception { + final int VT_COUNT = CARRIER_COUNT * 8; + workerCount.getAndSet(0); + finish = false; + + globalLockArray = new Object[MONITORS_CNT]; + for (int i = 0; i < MONITORS_CNT; i++) { + globalLockArray[i] = new Object(); + } + + // Create batch of VT threads. + Thread batch[] = new Thread[VT_COUNT]; + for (int i = 0; i < VT_COUNT; i++) { + //Thread.ofVirtual().name("FirstBatchVT-" + i).start(FOO); + batch[i] = ThreadBuilders.virtualThreadBuilder(scheduler).name("BatchVT-" + i).start(FOO5); + } + + Thread.sleep(10000); + + finish = true; + + for (int i = 0; i < VT_COUNT; i++) { + batch[i].join(); + } + + if (workerCount.get() != VT_COUNT) { + throw new RuntimeException("testContentionMultipleMonitors2 failed. Expected " + VT_COUNT + "but found " + workerCount.get()); + } + } + + static synchronized void recursive6(int depth, Object myLock) { + if (depth > 0) { + recursive6(depth - 1, myLock); + } else { + if (Math.random() < 0.5) { + Thread.yield(); + } else { + synchronized (myLock) { + Thread.yield(); + } + } + } + } + + static final Runnable FOO6 = () -> { + Object myLock = new Object(); + + while (!finish) { + int lockNumber = ThreadLocalRandom.current().nextInt(0, MONITORS_CNT - 1); + synchronized (myLock) { + synchronized (globalLockArray[lockNumber]) { + recursive6(lockNumber, myLock); + } + } + } + workerCount.getAndIncrement(); + System.out.println("Exiting FOO5 from thread " + Thread.currentThread().getName()); + }; + + /** + * Test contention on monitorenter with synchronized methods. + */ + @Test + void testContentionMultipleMonitors3() throws Exception { + final int VT_COUNT = CARRIER_COUNT * 8; + workerCount.getAndSet(0); + finish = false; + + + globalLockArray = new Object[MONITORS_CNT]; + for (int i = 0; i < MONITORS_CNT; i++) { + globalLockArray[i] = new Object(); + } + + // Create batch of VT threads. + Thread batch[] = new Thread[VT_COUNT]; + for (int i = 0; i < VT_COUNT; i++) { + batch[i] = ThreadBuilders.virtualThreadBuilder(scheduler).name("BatchVT-" + i).start(FOO6); + } + + Thread.sleep(10000); + + finish = true; + + for (int i = 0; i < VT_COUNT; i++) { + batch[i].join(); + } + + if (workerCount.get() != VT_COUNT) { + throw new RuntimeException("testContentionMultipleMonitors2 failed. Expected " + VT_COUNT + "but found " + workerCount.get()); + } + } + } \ No newline at end of file