< prev index next >

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

Print this page
@@ -22,12 +22,14 @@
   * or visit www.oracle.com if you need additional information or have any
   * questions.
   */
  package java.lang;
  
+ import java.lang.reflect.Constructor;
  import java.security.AccessController;
  import java.security.PrivilegedAction;
+ import java.util.Arrays;
  import java.util.Locale;
  import java.util.Objects;
  import java.util.concurrent.Callable;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.Executor;

@@ -38,12 +40,13 @@
  import java.util.concurrent.ForkJoinWorkerThread;
  import java.util.concurrent.Future;
  import java.util.concurrent.RejectedExecutionException;
  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.VirtualThreadPinnedEvent;
  import jdk.internal.event.VirtualThreadStartEvent;
  import jdk.internal.event.VirtualThreadSubmitFailedEvent;
  import jdk.internal.misc.CarrierThread;
  import jdk.internal.misc.InnocuousThread;
  import jdk.internal.misc.Unsafe;

@@ -60,24 +63,23 @@
  import sun.nio.ch.Interruptible;
  import sun.security.action.GetPropertyAction;
  import static java.util.concurrent.TimeUnit.*;
  
  /**
-  * A thread that is scheduled by the Java virtual machine rather than the operating
-  * system.
+  * A thread that is scheduled by the Java virtual machine rather than the operating system.
   */
  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 Executor 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");
      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;
      private final Continuation cont;
      private final Runnable runContinuation;

@@ -104,10 +106,25 @@
       * TIMED_PARKING -> TIMED_PARKED    // cont.yield successful, timed-parked
       * TIMED_PARKING -> TIMED_PINNED    // cont.yield failed, timed-parked on carrier
       *  TIMED_PARKED -> UNPARKED        // unparked, may be scheduled to continue
       *  TIMED_PINNED -> RUNNING         // unparked, continue execution on same carrier
       *
+      *   RUNNING -> 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 -> WAITING        // transitional state during wait on monitor
+      *   WAITING -> WAITED         // waiting on monitor
+      *    WAITED -> BLOCKED        // notified, waiting to be unblocked by monitor owner
+      *    WAITED -> UNBLOCKED      // timed-out/interrupted
+      *
+      *       RUNNING -> TIMED_WAITING   // transition state during timed-waiting on monitor
+      * TIMED_WAITING -> TIMED_WAITED    // timed-waiting on monitor
+      *  TIMED_WAITED -> BLOCKED         // notified, waiting to be unblocked by monitor owner
+      *  TIMED_WAITED -> UNBLOCKED       // timed-out/interrupted
+      *
       *  RUNNING -> YIELDING        // Thread.yield
       * YIELDING -> YIELDED         // cont.yield successful, may be scheduled to continue
       * YIELDING -> RUNNING         // cont.yield failed
       *  YIELDED -> RUNNING         // continue execution after Thread.yield
       */

@@ -126,24 +143,69 @@
  
      // Thread.yield
      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
+ 
+     // monitor wait/timed-wait
+     private static final int WAITING       = 15;
+     private static final int WAIT          = 16;    // waiting in Object.wait
+     private static final int TIMED_WAITING = 17;
+     private static final int TIMED_WAIT    = 18;    // waiting in timed-Object.wait
+ 
      private static final int TERMINATED = 99;  // final state
  
      // can be suspended from scheduling when unmounted
      private static final int SUSPENDED = 1 << 8;
  
      // parking permit
      private volatile boolean parkPermit;
  
+     // used to mark thread as ready to be unblocked
+     private volatile boolean unblocked;
+ 
+     // notified by Object.notify/notifyAll while waiting in Object.wait
+     private volatile boolean notified;
+ 
+     // timed-wait support
+     private long waitTimeout;
+     private byte timedWaitNonce;
+     private volatile Future<?> waitTimeoutTask;
+ 
+     // 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 default scheduler.
+      */
+     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.
       */
      static ContinuationScope continuationScope() {
          return VTHREAD_SCOPE;

@@ -187,22 +249,12 @@
          VThreadContinuation(VirtualThread vthread, Runnable task) {
              super(VTHREAD_SCOPE, wrap(vthread, task));
          }
          @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.reasonCode(), reason.reasonString());
          }
          private static Runnable wrap(VirtualThread vthread, Runnable task) {
              return new Runnable() {
                  @Hidden
                  public void run() {

@@ -215,20 +267,21 @@
      /**
       * Runs or continues execution on the current thread. The virtual thread is mounted
       * on the current thread before the task runs or continues. It unmounts when the
       * task completes or yields.
       */
-     @ChangesCurrentThread
+     @ChangesCurrentThread // allow mount/unmount to be inlined
      private void runContinuation() {
          // the carrier must be a platform thread
          if (Thread.currentThread().isVirtual()) {
              throw new WrongThreadException();
          }
  
          // 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;
              }
              // consume parking permit when continuing after parking

@@ -255,46 +308,114 @@
  
      /**
       * Submits the runContinuation task to the scheduler. For the default scheduler,
       * and calling it on a worker thread, the task will be pushed to the local queue,
       * otherwise it will be pushed to an external submission queue.
+      * If OutOfMemoryError is thrown then the submit will be retried until it succeeds.
+      * @param retryOnOOME to retry indefinitely if OutOfMemoryError is thrown
       * @throws RejectedExecutionException
       */
-     private void submitRunContinuation() {
-         try {
-             scheduler.execute(runContinuation);
-         } catch (RejectedExecutionException ree) {
-             submitFailed(ree);
-             throw ree;
+     @ChangesCurrentThread
+     private void submitRunContinuation(boolean retryOnOOME) {
+         boolean done = false;
+         while (!done) {
+             try {
+                 // The scheduler's execute method is invoked in the context of the
+                 // carrier thread. For the default scheduler this ensures that the
+                 // current thread is a ForkJoinWorkerThread so the task will be pushed
+                 // to the local queue. For other schedulers, it avoids deadlock that
+                 // would arise due to platform and virtual threads contenting for a
+                 // lock on the scheduler's submission queue.
+                 if (currentThread() instanceof VirtualThread vthread) {
+                     vthread.switchToCarrierThread();
+                     try {
+                         scheduler.execute(runContinuation);
+                     } finally {
+                         switchToVirtualThread(vthread);
+                     }
+                 } else {
+                     scheduler.execute(runContinuation);
+                 }
+                 done = true;
+             } catch (RejectedExecutionException ree) {
+                 submitFailed(ree);
+                 throw ree;
+             } catch (OutOfMemoryError e) {
+                 if (retryOnOOME) {
+                     U.park(false, 100_000_000); // 100ms
+                 } else {
+                     throw e;
+                 }
+             }
          }
      }
  
      /**
       * Submits the runContinuation task to given scheduler with a lazy submit.
+      * If OutOfMemoryError is thrown then the submit will be retried until it succeeds.
       * @throws RejectedExecutionException
       * @see ForkJoinPool#lazySubmit(ForkJoinTask)
       */
      private void lazySubmitRunContinuation(ForkJoinPool pool) {
+         assert Thread.currentThread() instanceof CarrierThread;
          try {
              pool.lazySubmit(ForkJoinTask.adapt(runContinuation));
          } catch (RejectedExecutionException ree) {
              submitFailed(ree);
              throw ree;
+         } catch (OutOfMemoryError e) {
+             submitRunContinuation();
          }
      }
  
      /**
       * Submits the runContinuation task to the given scheduler as an external submit.
+      * If OutOfMemoryError is thrown then the submit will be retried until it succeeds.
       * @throws RejectedExecutionException
       * @see ForkJoinPool#externalSubmit(ForkJoinTask)
       */
      private void externalSubmitRunContinuation(ForkJoinPool pool) {
+         assert Thread.currentThread() instanceof CarrierThread;
          try {
              pool.externalSubmit(ForkJoinTask.adapt(runContinuation));
          } catch (RejectedExecutionException ree) {
              submitFailed(ree);
              throw ree;
+         } catch (OutOfMemoryError e) {
+             submitRunContinuation();
+         }
+     }
+ 
+     /**
+      * Submits the runContinuation task to the scheduler. For the default scheduler,
+      * and calling it on a worker thread, the task will be pushed to the local queue,
+      * otherwise it will be pushed to an external submission queue.
+      * If OutOfMemoryError is thrown then the submit will be retried until it succeeds.
+      * @throws RejectedExecutionException
+      */
+     private void submitRunContinuation() {
+         submitRunContinuation(true);
+     }
+ 
+     /**
+      * Submits the runContinuation task the scheduler. For the default scheduler, and
+      * calling it a virtual thread that uses the default scheduler, the task will be
+      * pushed to an external submission queue. This method may throw OutOfMemoryError.
+      * @throws RejectedExecutionException
+      * @throws OutOfMemoryError
+      */
+     private void externalSubmitRunContinuationOrThrow() {
+         if (scheduler == DEFAULT_SCHEDULER
+                 && currentCarrierThread() instanceof CarrierThread ct) {
+             try {
+                 ct.getPool().externalSubmit(ForkJoinTask.adapt(runContinuation));
+             } catch (RejectedExecutionException ree) {
+                 submitFailed(ree);
+                 throw ree;
+             }
+         } else {
+             submitRunContinuation(false);
          }
      }
  
      /**
       * If enabled, emits a JFR VirtualThreadSubmitFailedEvent.

@@ -383,10 +504,12 @@
       * current thread is the current platform thread.
       */
      @ChangesCurrentThread
      @ReservedStackAccess
      private void unmount() {
+         assert !Thread.holdsLock(interruptLock);
+ 
          // set Thread.currentThread() to return the platform thread
          Thread carrier = this.carrierThread;
          carrier.setCurrentThread(carrier);
  
          // break connection to carrier thread, synchronized with interrupt

@@ -408,18 +531,19 @@
          notifyJvmtiHideFrames(true);
          Thread carrier = this.carrierThread;
          assert Thread.currentThread() == this
                  && carrier == Thread.currentCarrierThread();
          carrier.setCurrentThread(carrier);
+         setLockId(this.threadId()); // keep lockid of vthread
      }
  
      /**
       * Sets the current thread to the given virtual thread.
       */
      @ChangesCurrentThread
      @JvmtiMountTransition
-     private void switchToVirtualThread(VirtualThread vthread) {
+     private static void switchToVirtualThread(VirtualThread vthread) {
          Thread carrier = vthread.carrierThread;
          assert carrier == Thread.currentCarrierThread();
          carrier.setCurrentThread(vthread);
          notifyJvmtiHideFrames(false);
      }

@@ -478,11 +602,10 @@
                  if (currentThread() instanceof CarrierThread ct) {
                      lazySubmitRunContinuation(ct.getPool());
                  } else {
                      submitRunContinuation();
                  }
- 
              }
              return;
          }
  
          // Thread.yield

@@ -496,10 +619,72 @@
                  submitRunContinuation();
              }
              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) {
+                 assert recheckInterval >= 1 && recheckInterval <= 6;
+                 // 4 ^ (recheckInterval - 1) = 1, 4, 16, ... 1024
+                 long delay = 1 << (recheckInterval - 1) << (recheckInterval - 1);
+                 schedule(this::unblock, delay, MILLISECONDS);
+             }
+             return;
+         }
+ 
+         // Object.wait
+         if (s == WAITING || s == TIMED_WAITING) {
+             byte nonce;
+             int newState;
+             if (s == WAITING) {
+                 nonce = 0;  // not used
+                 setState(newState = WAIT);
+             } else {
+                 // synchronize with timeout task (previous timed-wait may be running)
+                 synchronized (timedWaitLock()) {
+                     nonce = ++timedWaitNonce;
+                     setState(newState = TIMED_WAIT);
+                 }
+             }
+ 
+             // may have been notified while in transition to wait state
+             if (notified && compareAndSetState(newState, BLOCKED)) {
+                 // may have even been unblocked already
+                 if (unblocked && compareAndSetState(BLOCKED, UNBLOCKED)) {
+                     unblocked = false;
+                     submitRunContinuation();
+                 }
+                 return;
+             }
+ 
+             // may have been interrupted while in transition to wait state
+             if (interrupted && compareAndSetState(newState, UNBLOCKED)) {
+                 submitRunContinuation();
+                 return;
+             }
+ 
+             // schedule wakeup
+             if (newState == TIMED_WAIT) {
+                 assert waitTimeout > 0;
+                 waitTimeoutTask = schedule(() -> waitTimeoutExpired(nonce), waitTimeout, MILLISECONDS);
+             }
+             return;
+         }
+ 
          assert false;
      }
  
      /**
       * Invoked after the continuation completes.

@@ -559,12 +744,12 @@
              addedToContainer = true;
  
              // scoped values may be inherited
              inheritScopedValueBindings(container);
  
-             // submit task to run thread
-             submitRunContinuation();
+             // submit task to run thread, using externalSubmit if possible
+             externalSubmitRunContinuationOrThrow();
              started = true;
          } finally {
              if (!started) {
                  afterDone(addedToContainer);
              }

@@ -580,11 +765,11 @@
      public void run() {
          // do nothing
      }
  
      /**
-      * Parks until unparked or interrupted. If already unparked then the parking
+      * Parks unless unparked or interrupted. If already unparked then the parking
       * permit is consumed and this method completes immediately (meaning it doesn't
       * yield). It also completes immediately if the interrupt status is set.
       */
      @Override
      void park() {

@@ -663,18 +848,10 @@
       * @param nanos the waiting time in nanoseconds
       */
      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) {
                  if (!timed) {
                      U.park(false, 0);

@@ -686,40 +863,40 @@
              setState(RUNNING);
          }
  
          // 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 event in Java with the same name (but different ID)
+      * to events emitted by the VM.
+      */
+     private static native void virtualThreadPinnedEvent(int reason, String reasonString);
+ 
      /**
       * Schedule this virtual thread to be unparked after a given delay.
       */
      @ChangesCurrentThread
      private Future<?> scheduleUnpark(long nanos) {
          assert Thread.currentThread() == this;
          // need to switch to current carrier thread to avoid nested parking
          switchToCarrierThread();
          try {
-             return UNPARKER.schedule(this::unpark, nanos, NANOSECONDS);
+             return schedule(this::unpark, nanos, NANOSECONDS);
          } finally {
              switchToVirtualThread(this);
          }
      }
  
      /**
       * Cancels a task if it has not completed.
       */
      @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();
              try {
                  future.cancel(false);

@@ -728,48 +905,97 @@
              }
          }
      }
  
      /**
-      * Re-enables this virtual thread for scheduling. If the virtual thread was
-      * {@link #park() parked} then it will be unblocked, otherwise its next call
-      * to {@code park} or {@linkplain #parkNanos(long) parkNanos} is guaranteed
-      * not to block.
+      * Re-enables this virtual thread for scheduling. If this virtual thread is parked
+      * then its task is scheduled to continue, otherwise its next call to {@code park} or
+      * {@linkplain #parkNanos(long) parkNanos} is guaranteed not to block.
       * @throws RejectedExecutionException if the scheduler cannot accept a task
       */
      @Override
-     @ChangesCurrentThread
      void unpark() {
-         Thread currentThread = Thread.currentThread();
-         if (!getAndSetParkPermit(true) && currentThread != this) {
+         if (!getAndSetParkPermit(true) && currentThread() != this) {
              int s = state();
-             boolean parked = (s == PARKED) || (s == TIMED_PARKED);
-             if (parked && compareAndSetState(s, UNPARKED)) {
-                 if (currentThread instanceof VirtualThread vthread) {
-                     vthread.switchToCarrierThread();
-                     try {
-                         submitRunContinuation();
-                     } finally {
-                         switchToVirtualThread(vthread);
-                     }
-                 } else {
-                     submitRunContinuation();
-                 }
-             } else if ((s == PINNED) || (s == TIMED_PINNED)) {
+ 
+             // unparked while parked
+             if ((s == PARKED || s == TIMED_PARKED) && compareAndSetState(s, UNPARKED)) {
+                 submitRunContinuation();
+                 return;
+             }
+ 
+             // unparked while parked when pinned
+             if (s == PINNED || s == TIMED_PINNED) {
                  // unpark carrier thread when pinned
-                 notifyJvmtiDisableSuspend(true);
+                 disableSuspendAndPreempt();
                  try {
                      synchronized (carrierThreadAccessLock()) {
                          Thread carrier = carrierThread;
                          if (carrier != null && ((s = state()) == PINNED || s == TIMED_PINNED)) {
                              U.unpark(carrier);
                          }
                      }
                  } finally {
-                     notifyJvmtiDisableSuspend(false);
+                     enableSuspendAndPreempt();
+                 }
+                 return;
+             }
+         }
+     }
+ 
+     /**
+      * Invoked by unblocker thread to unblock this virtual thread.
+      */
+     private void unblock() {
+         assert !Thread.currentThread().isVirtual();
+         unblocked = true;
+         if (state() == BLOCKED && compareAndSetState(BLOCKED, UNBLOCKED)) {
+             unblocked = false;
+             submitRunContinuation();
+         }
+     }
+ 
+     /**
+      * Invoked by timer thread when wait timeout for virtual thread has expired.
+      * 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.
+      */
+     private void waitTimeoutExpired(byte nounce) {
+         assert !Thread.currentThread().isVirtual();
+         for (;;) {
+             boolean unblocked = false;
+             synchronized (timedWaitLock()) {
+                 if (nounce != timedWaitNonce) {
+                     // this timeout task is for a past timed-wait
+                     return;
+                 }
+                 int s = state();
+                 if (s == TIMED_WAIT) {
+                     unblocked = compareAndSetState(TIMED_WAIT, UNBLOCKED);
+                 } else if (s != (TIMED_WAIT | SUSPENDED)) {
+                     // notified or interrupted, no longer waiting
+                     return;
                  }
              }
+             if (unblocked) {
+                 submitRunContinuation();
+                 return;
+             }
+             // need to retry when thread is suspended in time-wait
+             Thread.yield();
+         }
+     }
+ 
+     /**
+      * Invoked by Object.wait to cancel wait timer.
+      */
+     void cancelWaitTimeout() {
+         assert Thread.currentThread() == this;
+         Future<?> timeoutTask = this.waitTimeoutTask;
+         if (timeoutTask != null) {
+             timeoutTask.cancel(false);
          }
      }
  
      /**
       * Attempts to yield the current virtual thread (Thread.yield).

@@ -857,28 +1083,28 @@
          return true;
      }
  
      @Override
      void blockedOn(Interruptible b) {
-         notifyJvmtiDisableSuspend(true);
+         disableSuspendAndPreempt();
          try {
              super.blockedOn(b);
          } finally {
-             notifyJvmtiDisableSuspend(false);
+             enableSuspendAndPreempt();
          }
      }
  
      @Override
      @SuppressWarnings("removal")
      public void interrupt() {
          if (Thread.currentThread() != this) {
              checkAccess();
  
              // if current thread is a virtual thread then prevent it from being
-             // suspended when entering or holding interruptLock
+             // suspended or unmounted when entering or holding interruptLock
              Interruptible blocker;
-             notifyJvmtiDisableSuspend(true);
+             disableSuspendAndPreempt();
              try {
                  synchronized (interruptLock) {
                      interrupted = true;
                      blocker = nioBlocker();
                      if (blocker != null) {

@@ -888,22 +1114,32 @@
                      // interrupt carrier thread if mounted
                      Thread carrier = carrierThread;
                      if (carrier != null) carrier.setInterrupt();
                  }
              } finally {
-                 notifyJvmtiDisableSuspend(false);
+                 enableSuspendAndPreempt();
              }
  
              // notify blocker after releasing interruptLock
              if (blocker != null) {
                  blocker.postInterrupt();
              }
+ 
+             // make available parking permit, unpark thread if parked
+             unpark();
+ 
+             // if thread is waiting in Object.wait then schedule to try to reenter
+             int s = state();
+             if ((s == WAIT || s == TIMED_WAIT) && compareAndSetState(s, UNBLOCKED)) {
+                 submitRunContinuation();
+             }
+ 
          } else {
              interrupted = true;
              carrierThread.setInterrupt();
+             setParkPermit(true);
          }
-         unpark();
      }
  
      @Override
      public boolean isInterrupted() {
          return interrupted;

@@ -912,18 +1148,18 @@
      @Override
      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;
      }
  

@@ -942,43 +1178,71 @@
                  }
              case UNPARKED:
              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;
              case PARKING:
              case TIMED_PARKING:
              case YIELDING:
+             case WAITING:
+             case TIMED_WAITING:
                  // runnable, in transition
                  return Thread.State.RUNNABLE;
              case PARKED:
              case PINNED:
+             case WAIT:
                  return State.WAITING;
              case TIMED_PARKED:
              case TIMED_PINNED:
+             case TIMED_WAIT:
                  return State.TIMED_WAITING;
+             case BLOCKING:
+             case BLOCKED:
+                 return State.BLOCKED;
              case TERMINATED:
                  return Thread.State.TERMINATED;
              default:
                  throw new InternalError();
          }
      }
  
+     /**
+      * 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;
          return (s != NEW && s != TERMINATED);
      }

@@ -1013,17 +1277,17 @@
                  return new StackTraceElement[0];  // unmounted, empty stack
              }
              case RUNNING, PINNED, TIMED_PINNED -> {
                  return null;   // mounted
              }
-             case PARKED, TIMED_PARKED -> {
+             case PARKED, TIMED_PARKED, BLOCKED, WAIT, TIMED_WAIT -> {
                  // unmounted, not runnable
              }
-             case UNPARKED, YIELDED -> {
+             case UNPARKED, UNBLOCKED, YIELDED -> {
                  // unmounted, runnable
              }
-             case PARKING, TIMED_PARKING, YIELDING -> {
+             case PARKING, TIMED_PARKING, BLOCKING, YIELDING, WAITING, TIMED_WAITING -> {
                  return null;  // in transition
              }
              default -> throw new InternalError("" + initialState);
          }
  

@@ -1040,18 +1304,27 @@
          } finally {
              assert state == suspendedState;
              setState(initialState);
          }
          boolean resubmit = switch (initialState) {
-             case UNPARKED, YIELDED -> {
+             case UNPARKED, UNBLOCKED, YIELDED -> {
                  // resubmit as task may have run while suspended
                  yield true;
              }
              case PARKED, TIMED_PARKED -> {
                  // resubmit if unparked while suspended
                  yield parkPermit && compareAndSetState(initialState, UNPARKED);
              }
+             case BLOCKED -> {
+                 // resubmit if unblocked while suspended
+                 yield unblocked && compareAndSetState(BLOCKED, UNBLOCKED);
+             }
+             case WAIT, TIMED_WAIT -> {
+                 // resubmit if notified or interrupted while waiting (Object.wait)
+                 // waitTimeoutExpired will retry if the timed expired when suspended
+                 yield (notified || interrupted) && compareAndSetState(initialState, UNBLOCKED);
+             }
              default -> throw new InternalError();
          };
          if (resubmit) {
              submitRunContinuation();
          }

@@ -1066,36 +1339,53 @@
          if (!name.isEmpty()) {
              sb.append(",");
              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();
      }
  

@@ -1125,10 +1415,34 @@
      private Object carrierThreadAccessLock() {
          // return interruptLock as unmount has to coordinate with interrupt
          return interruptLock;
      }
  
+     /**
+      * Returns a lock object to coordinating timed-wait setup and timeout handling.
+      */
+     private Object timedWaitLock() {
+         // use this object for now to avoid the overhead of introducing another lock
+         return runContinuation;
+     }
+ 
+     /**
+      * 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() {
          return state;  // volatile read
      }

@@ -1139,10 +1453,14 @@
  
      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;
          }
      }

@@ -1158,10 +1476,13 @@
      private void setCarrierThread(Thread carrier) {
          // U.putReferenceRelease(this, CARRIER_THREAD, carrier);
          this.carrierThread = carrier;
      }
  
+     @IntrinsicCandidate
+     private static native void setLockId(long tid);
+ 
      // -- JVM TI support --
  
      @IntrinsicCandidate
      @JvmtiMountTransition
      private native void notifyJvmtiStart();

@@ -1190,13 +1511,39 @@
          registerNatives();
      }
  
      /**
       * Creates the default scheduler.
+      * If the system property {@code jdk.virtualThreadScheduler.implClass} is set then
+      * its value is name of a class that implements java.util.concurrent.Executor.
+      * The class is public in an exported package, has a public no-arg constructor,
+      * and is visible to the system class loader.
+      * If the system property is not set then the default scheduler will be a
+      * ForkJoinPool instance.
+      */
+     private static Executor createDefaultScheduler() {
+         String propName = "jdk.virtualThreadScheduler.implClass";
+         String propValue = GetPropertyAction.privilegedGetProperty(propName);
+         if (propValue != null) {
+             try {
+                 Class<?> clazz = Class.forName(propValue, true,
+                         ClassLoader.getSystemClassLoader());
+                 Constructor<?> ctor = clazz.getConstructor();
+                 return (Executor) ctor.newInstance();
+             } catch (Exception ex) {
+                 throw new Error(ex);
+             }
+         } else {
+             return createDefaultForkJoinPoolScheduler();
+         }
+     }
+ 
+     /**
+      * Creates the default ForkJoinPool scheduler.
       */
      @SuppressWarnings("removal")
-     private static ForkJoinPool createDefaultScheduler() {
+     private static ForkJoinPool createDefaultForkJoinPoolScheduler() {
          ForkJoinWorkerThreadFactory factory = pool -> {
              PrivilegedAction<ForkJoinWorkerThread> pa = () -> new CarrierThread(pool);
              return AccessController.doPrivileged(pa);
          };
          PrivilegedAction<ForkJoinPool> pa = () -> {

@@ -1227,39 +1574,78 @@
          };
          return AccessController.doPrivileged(pa);
      }
  
      /**
-      * Creates the ScheduledThreadPoolExecutor used for timed unpark.
+      * Schedule a runnable task to run after a delay.
       */
-     private static ScheduledExecutorService createDelayedTaskScheduler() {
-         String propValue = GetPropertyAction.privilegedGetProperty("jdk.unparker.maxPoolSize");
-         int poolSize;
+     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);
+     }
+ 
+     /**
+      * Creates the ScheduledThreadPoolExecutors used to execute delayed tasks.
+      */
+     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);
+         }
+         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;
          }
-         ScheduledThreadPoolExecutor stpe = (ScheduledThreadPoolExecutor)
-             Executors.newScheduledThreadPool(poolSize, task -> {
-                 return InnocuousThread.newThread("VirtualThread-unparker", task);
-             });
-         stpe.setRemoveOnCancelPolicy(true);
-         return 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();
      }
  }
< prev index next >