1 /* 2 * Copyright (c) 2018, 2026, Oracle and/or its affiliates. All rights reserved. 3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. 4 * 5 * This code is free software; you can redistribute it and/or modify it 6 * under the terms of the GNU General Public License version 2 only, as 7 * published by the Free Software Foundation. Oracle designates this 8 * particular file as subject to the "Classpath" exception as provided 9 * by Oracle in the LICENSE file that accompanied this code. 10 * 11 * This code is distributed in the hope that it will be useful, but WITHOUT 12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License 14 * version 2 for more details (a copy is included in the LICENSE file that 15 * accompanied this code). 16 * 17 * You should have received a copy of the GNU General Public License version 18 * 2 along with this work; if not, write to the Free Software Foundation, 19 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. 20 * 21 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA 22 * or visit www.oracle.com if you need additional information or have any 23 * questions. 24 */ 25 package java.lang; 26 27 import java.lang.invoke.MethodHandles; 28 import java.lang.invoke.VarHandle; 29 import java.lang.reflect.Constructor; 30 import java.util.Locale; 31 import java.util.Objects; 32 import java.util.concurrent.CountDownLatch; 33 import java.util.concurrent.Executors; 34 import java.util.concurrent.ForkJoinPool; 35 import java.util.concurrent.ForkJoinTask; 36 import java.util.concurrent.Future; 37 import java.util.concurrent.RejectedExecutionException; 38 import java.util.concurrent.ScheduledExecutorService; 39 import java.util.concurrent.ScheduledFuture; 40 import java.util.concurrent.ScheduledThreadPoolExecutor; 41 import java.util.concurrent.TimeUnit; 42 import jdk.internal.event.VirtualThreadEndEvent; 43 import jdk.internal.event.VirtualThreadParkEvent; 44 import jdk.internal.event.VirtualThreadStartEvent; 45 import jdk.internal.event.VirtualThreadSubmitFailedEvent; 46 import jdk.internal.invoke.MhUtil; 47 import jdk.internal.misc.CarrierThread; 48 import jdk.internal.misc.InnocuousThread; 49 import jdk.internal.misc.Unsafe; 50 import jdk.internal.vm.Continuation; 51 import jdk.internal.vm.ContinuationScope; 52 import jdk.internal.vm.StackableScope; 53 import jdk.internal.vm.ThreadContainer; 54 import jdk.internal.vm.ThreadContainers; 55 import jdk.internal.vm.annotation.ChangesCurrentThread; 56 import jdk.internal.vm.annotation.Hidden; 57 import jdk.internal.vm.annotation.IntrinsicCandidate; 58 import jdk.internal.vm.annotation.JvmtiHideEvents; 59 import jdk.internal.vm.annotation.JvmtiMountTransition; 60 import jdk.internal.vm.annotation.ReservedStackAccess; 61 import sun.nio.ch.Interruptible; 62 import static java.util.concurrent.TimeUnit.*; 63 64 /** 65 * A thread that is scheduled by the Java virtual machine rather than the operating system. 66 */ 67 final class VirtualThread extends BaseVirtualThread { 68 private static final Unsafe U = Unsafe.getUnsafe(); 69 private static final ContinuationScope VTHREAD_SCOPE = new ContinuationScope("VirtualThreads"); 70 71 private static final BuiltinScheduler BUILTIN_SCHEDULER; 72 private static final VirtualThreadScheduler DEFAULT_SCHEDULER; 73 private static final VirtualThreadScheduler EXTERNAL_VIEW; 74 static { 75 // experimental 76 String propValue = System.getProperty("jdk.virtualThreadScheduler.implClass"); 77 if (propValue != null) { 78 BuiltinScheduler builtinScheduler = createBuiltinScheduler(true); 79 VirtualThreadScheduler externalView = builtinScheduler.createExternalView(); 80 VirtualThreadScheduler defaultScheduler = loadCustomScheduler(externalView, propValue); 81 BUILTIN_SCHEDULER = builtinScheduler; 82 DEFAULT_SCHEDULER = defaultScheduler; 83 EXTERNAL_VIEW = externalView; 84 } else { 85 var builtinScheduler = createBuiltinScheduler(false); 86 BUILTIN_SCHEDULER = builtinScheduler; 87 DEFAULT_SCHEDULER = builtinScheduler; 88 EXTERNAL_VIEW = builtinScheduler.createExternalView(); 89 } 90 } 91 92 private static final long STATE = U.objectFieldOffset(VirtualThread.class, "state"); 93 private static final long PARK_PERMIT = U.objectFieldOffset(VirtualThread.class, "parkPermit"); 94 private static final long CARRIER_THREAD = U.objectFieldOffset(VirtualThread.class, "carrierThread"); 95 private static final long TERMINATION = U.objectFieldOffset(VirtualThread.class, "termination"); 96 private static final long ON_WAITING_LIST = U.objectFieldOffset(VirtualThread.class, "onWaitingList"); 97 98 // scheduler and continuation 99 private final VirtualThreadScheduler scheduler; 100 private final Continuation cont; 101 private final VirtualThreadTask runContinuation; 102 103 // virtual thread state, accessed by VM 104 private volatile int state; 105 106 /* 107 * Virtual thread state transitions: 108 * 109 * NEW -> STARTED // Thread.start, schedule to run 110 * STARTED -> TERMINATED // failed to start 111 * STARTED -> RUNNING // first run 112 * RUNNING -> TERMINATED // done 113 * 114 * RUNNING -> PARKING // Thread parking with LockSupport.park 115 * PARKING -> PARKED // cont.yield successful, parked indefinitely 116 * PARKED -> UNPARKED // unparked, may be scheduled to continue 117 * UNPARKED -> RUNNING // continue execution after park 118 * 119 * PARKING -> RUNNING // cont.yield failed, need to park on carrier 120 * RUNNING -> PINNED // park on carrier 121 * PINNED -> RUNNING // unparked, continue execution on same carrier 122 * 123 * RUNNING -> TIMED_PARKING // Thread parking with LockSupport.parkNanos 124 * TIMED_PARKING -> TIMED_PARKED // cont.yield successful, timed-parked 125 * TIMED_PARKED -> UNPARKED // unparked, may be scheduled to continue 126 * 127 * TIMED_PARKING -> RUNNING // cont.yield failed, need to park on carrier 128 * RUNNING -> TIMED_PINNED // park on carrier 129 * TIMED_PINNED -> RUNNING // unparked, continue execution on same carrier 130 * 131 * RUNNING -> BLOCKING // blocking on monitor enter 132 * BLOCKING -> BLOCKED // blocked on monitor enter 133 * BLOCKED -> UNBLOCKED // unblocked, may be scheduled to continue 134 * UNBLOCKED -> RUNNING // continue execution after blocked on monitor enter 135 * 136 * RUNNING -> WAITING // transitional state during wait on monitor 137 * WAITING -> WAIT // waiting on monitor 138 * WAIT -> BLOCKED // notified, waiting to be unblocked by monitor owner 139 * WAIT -> UNBLOCKED // interrupted 140 * 141 * RUNNING -> TIMED_WAITING // transition state during timed-waiting on monitor 142 * TIMED_WAITING -> TIMED_WAIT // timed-waiting on monitor 143 * TIMED_WAIT -> BLOCKED // notified, waiting to be unblocked by monitor owner 144 * TIMED_WAIT -> UNBLOCKED // timed-out/interrupted 145 * 146 * RUNNING -> YIELDING // Thread.yield 147 * YIELDING -> YIELDED // cont.yield successful, may be scheduled to continue 148 * YIELDING -> RUNNING // cont.yield failed 149 * YIELDED -> RUNNING // continue execution after Thread.yield 150 */ 151 private static final int NEW = 0; 152 private static final int STARTED = 1; 153 private static final int RUNNING = 2; // runnable-mounted 154 155 // untimed and timed parking 156 private static final int PARKING = 3; 157 private static final int PARKED = 4; // unmounted 158 private static final int PINNED = 5; // mounted 159 private static final int TIMED_PARKING = 6; 160 private static final int TIMED_PARKED = 7; // unmounted 161 private static final int TIMED_PINNED = 8; // mounted 162 private static final int UNPARKED = 9; // unmounted but runnable 163 164 // Thread.yield 165 private static final int YIELDING = 10; 166 private static final int YIELDED = 11; // unmounted but runnable 167 168 // monitor enter 169 private static final int BLOCKING = 12; 170 private static final int BLOCKED = 13; // unmounted 171 private static final int UNBLOCKED = 14; // unmounted but runnable 172 173 // monitor wait/timed-wait 174 private static final int WAITING = 15; 175 private static final int WAIT = 16; // waiting in Object.wait 176 private static final int TIMED_WAITING = 17; 177 private static final int TIMED_WAIT = 18; // waiting in timed-Object.wait 178 179 private static final int TERMINATED = 99; // final state 180 181 // parking permit made available by LockSupport.unpark 182 private volatile boolean parkPermit; 183 184 // blocking permit made available by unblocker thread when another thread exits monitor 185 private volatile boolean blockPermit; 186 187 // true when on the list of virtual threads waiting to be unblocked 188 private volatile boolean onWaitingList; 189 190 // next virtual thread on the list of virtual threads waiting to be unblocked 191 private volatile VirtualThread next; 192 193 // notified by Object.notify/notifyAll while waiting in Object.wait 194 private volatile boolean notified; 195 196 // true when waiting in Object.wait, false for VM internal uninterruptible Object.wait 197 private volatile boolean interruptibleWait; 198 199 // timed-wait support 200 private byte timedWaitSeqNo; 201 202 // timeout for timed-park and timed-wait, only accessed on current/carrier thread 203 private long timeout; 204 205 // timer task for timed-park and timed-wait, only accessed on current/carrier thread 206 private Future<?> timeoutTask; 207 208 // carrier thread when mounted, accessed by VM 209 private volatile Thread carrierThread; 210 211 // termination object when joining, created lazily if needed 212 private volatile CountDownLatch termination; 213 214 /** 215 * Return the built-in scheduler. 216 * @param trusted true if caller is trusted, false if not trusted 217 */ 218 static VirtualThreadScheduler builtinScheduler(boolean trusted) { 219 return trusted ? BUILTIN_SCHEDULER : EXTERNAL_VIEW; 220 } 221 222 /** 223 * Returns the default scheduler, usually the same as the built-in scheduler. 224 */ 225 static VirtualThreadScheduler defaultScheduler() { 226 return DEFAULT_SCHEDULER; 227 } 228 229 /** 230 * Returns the continuation scope used for virtual threads. 231 */ 232 static ContinuationScope continuationScope() { 233 return VTHREAD_SCOPE; 234 } 235 236 /** 237 * Return the scheduler for this thread. 238 * @param trusted true if caller is trusted, false if not trusted 239 */ 240 VirtualThreadScheduler scheduler(boolean trusted) { 241 if (scheduler == BUILTIN_SCHEDULER && !trusted) { 242 return EXTERNAL_VIEW; 243 } else { 244 return scheduler; 245 } 246 } 247 248 /** 249 * Returns the task to start/continue this virtual thread. 250 */ 251 VirtualThreadTask virtualThreadTask() { 252 return runContinuation; 253 } 254 255 /** 256 * Creates a new {@code VirtualThread} to run the given task with the given scheduler. 257 * 258 * @param scheduler the scheduler or null for default scheduler 259 * @param preferredCarrier the preferred carrier or null 260 * @param name thread name 261 * @param characteristics characteristics 262 * @param task the task to execute 263 */ 264 VirtualThread(VirtualThreadScheduler scheduler, 265 Thread preferredCarrier, 266 String name, 267 int characteristics, 268 Runnable task) { 269 super(name, characteristics, /*bound*/ false); 270 Objects.requireNonNull(task); 271 272 // use default scheduler if not provided 273 if (scheduler == null) { 274 scheduler = DEFAULT_SCHEDULER; 275 } else if (scheduler == EXTERNAL_VIEW) { 276 throw new UnsupportedOperationException(); 277 } 278 this.scheduler = scheduler; 279 this.cont = new VThreadContinuation(this, task); 280 281 if (scheduler == BUILTIN_SCHEDULER) { 282 this.runContinuation = new BuiltinSchedulerTask(this); 283 } else { 284 this.runContinuation = new CustomSchedulerTask(this, preferredCarrier); 285 } 286 } 287 288 /** 289 * The task to execute when using the built-in scheduler. 290 */ 291 static final class BuiltinSchedulerTask implements VirtualThreadTask { 292 private final VirtualThread vthread; 293 BuiltinSchedulerTask(VirtualThread vthread) { 294 this.vthread = vthread; 295 } 296 @Override 297 public Thread thread() { 298 return vthread; 299 } 300 @Override 301 public void run() { 302 vthread.runContinuation();; 303 } 304 @Override 305 public Thread preferredCarrier() { 306 throw new UnsupportedOperationException(); 307 } 308 @Override 309 public Object attach(Object att) { 310 throw new UnsupportedOperationException(); 311 } 312 @Override 313 public Object attachment() { 314 throw new UnsupportedOperationException(); 315 } 316 } 317 318 /** 319 * The task to execute when using a custom scheduler. 320 */ 321 static final class CustomSchedulerTask implements VirtualThreadTask { 322 private static final VarHandle ATT = 323 MhUtil.findVarHandle(MethodHandles.lookup(), "att", Object.class); 324 private final VirtualThread vthread; 325 private final Thread preferredCarrier; 326 private volatile Object att; 327 CustomSchedulerTask(VirtualThread vthread, Thread preferredCarrier) { 328 this.vthread = vthread; 329 this.preferredCarrier = preferredCarrier; 330 } 331 @Override 332 public Thread thread() { 333 return vthread; 334 } 335 @Override 336 public void run() { 337 vthread.runContinuation();; 338 } 339 @Override 340 public Thread preferredCarrier() { 341 return preferredCarrier; 342 } 343 @Override 344 public Object attach(Object att) { 345 return ATT.getAndSet(this, att); 346 } 347 @Override 348 public Object attachment() { 349 return att; 350 } 351 } 352 353 /** 354 * The continuation that a virtual thread executes. 355 */ 356 private static class VThreadContinuation extends Continuation { 357 VThreadContinuation(VirtualThread vthread, Runnable task) { 358 super(VTHREAD_SCOPE, wrap(vthread, task)); 359 } 360 @Override 361 protected void onPinned(Continuation.Pinned reason) { 362 } 363 private static Runnable wrap(VirtualThread vthread, Runnable task) { 364 return new Runnable() { 365 @Hidden 366 @JvmtiHideEvents 367 public void run() { 368 vthread.endFirstTransition(); 369 try { 370 vthread.run(task); 371 } finally { 372 vthread.startFinalTransition(); 373 } 374 } 375 }; 376 } 377 } 378 379 /** 380 * Runs or continues execution on the current thread. The virtual thread is mounted 381 * on the current thread before the task runs or continues. It unmounts when the 382 * task completes or yields. 383 */ 384 @ChangesCurrentThread // allow mount/unmount to be inlined 385 private void runContinuation() { 386 // the carrier must be a platform thread 387 if (Thread.currentThread().isVirtual()) { 388 throw new WrongThreadException(); 389 } 390 391 // set state to RUNNING 392 int initialState = state(); 393 if (initialState == STARTED || initialState == UNPARKED 394 || initialState == UNBLOCKED || initialState == YIELDED) { 395 // newly started or continue after parking/blocking/Thread.yield 396 if (!compareAndSetState(initialState, RUNNING)) { 397 return; 398 } 399 // consume permit when continuing after parking or blocking. If continue 400 // after a timed-park or timed-wait then the timeout task is cancelled. 401 if (initialState == UNPARKED) { 402 cancelTimeoutTask(); 403 setParkPermit(false); 404 } else if (initialState == UNBLOCKED) { 405 cancelTimeoutTask(); 406 blockPermit = false; 407 } 408 } else { 409 // not runnable 410 return; 411 } 412 413 mount(); 414 try { 415 cont.run(); 416 } finally { 417 unmount(); 418 if (cont.isDone()) { 419 afterDone(); 420 } else { 421 afterYield(); 422 } 423 } 424 } 425 426 /** 427 * Cancel timeout task when continuing after timed-park or timed-wait. 428 * The timeout task may be executing, or may have already completed. 429 */ 430 private void cancelTimeoutTask() { 431 if (timeoutTask != null) { 432 timeoutTask.cancel(false); 433 timeoutTask = null; 434 } 435 } 436 437 /** 438 * Submits the runContinuation task to the scheduler. For the built-in scheduler, 439 * the task will be pushed to the local queue if possible, otherwise it will be 440 * pushed to an external submission queue. 441 * @param retryOnOOME true to retry indefinitely if OutOfMemoryError is thrown 442 * @throws RejectedExecutionException 443 */ 444 private void submitRunContinuation(boolean retryOnOOME) { 445 boolean done = false; 446 while (!done) { 447 try { 448 // Pin the continuation to prevent the virtual thread from unmounting 449 // when submitting a task. For the default scheduler this ensures that 450 // the carrier doesn't change when pushing a task. For other schedulers 451 // it avoids deadlock that could arise due to carriers and virtual 452 // threads contending for a lock. 453 if (currentThread().isVirtual()) { 454 Continuation.pin(); 455 try { 456 scheduler.onContinue(runContinuation); 457 } finally { 458 Continuation.unpin(); 459 } 460 } else { 461 scheduler.onContinue(runContinuation); 462 } 463 done = true; 464 } catch (RejectedExecutionException ree) { 465 submitFailed(ree); 466 throw ree; 467 } catch (OutOfMemoryError e) { 468 if (retryOnOOME) { 469 U.park(false, 100_000_000); // 100ms 470 } else { 471 throw e; 472 } 473 } 474 } 475 } 476 477 /** 478 * Submits the runContinuation task to the scheduler. For the default scheduler, 479 * and calling it on a worker thread, the task will be pushed to the local queue, 480 * otherwise it will be pushed to an external submission queue. 481 * If OutOfMemoryError is thrown then the submit will be retried until it succeeds. 482 * @throws RejectedExecutionException 483 */ 484 private void submitRunContinuation() { 485 submitRunContinuation(true); 486 } 487 488 /** 489 * Invoked from a carrier thread to lazy submit the runContinuation task to the 490 * carrier's local queue if the queue is empty. If not empty, or invoked by a thread 491 * for a custom scheduler, then it just submits the task to the scheduler. 492 * If OutOfMemoryError is thrown then the submit will be retried until it succeeds. 493 * @throws RejectedExecutionException 494 * @see ForkJoinPool#lazySubmit(ForkJoinTask) 495 */ 496 private void lazySubmitRunContinuation() { 497 assert !currentThread().isVirtual(); 498 if (currentThread() instanceof CarrierThread ct && ct.getQueuedTaskCount() == 0) { 499 try { 500 ct.getPool().lazySubmit(ForkJoinTask.adapt(runContinuation)); 501 } catch (RejectedExecutionException ree) { 502 submitFailed(ree); 503 throw ree; 504 } catch (OutOfMemoryError e) { 505 submitRunContinuation(); 506 } 507 } else { 508 submitRunContinuation(); 509 } 510 } 511 512 /** 513 * Invoked from a carrier thread to externally submit the runContinuation task to the 514 * scheduler. If invoked by a thread for a custom scheduler, then it just submits the 515 * task to the scheduler. 516 * If OutOfMemoryError is thrown then the submit will be retried until it succeeds. 517 * @throws RejectedExecutionException 518 * @see ForkJoinPool#externalSubmit(ForkJoinTask) 519 */ 520 private void externalSubmitRunContinuation() { 521 assert !currentThread().isVirtual(); 522 if (currentThread() instanceof CarrierThread ct) { 523 try { 524 ct.getPool().externalSubmit(ForkJoinTask.adapt(runContinuation)); 525 } catch (RejectedExecutionException ree) { 526 submitFailed(ree); 527 throw ree; 528 } catch (OutOfMemoryError e) { 529 submitRunContinuation(); 530 } 531 } else { 532 submitRunContinuation(); 533 } 534 } 535 536 /** 537 * Invoked from Thread.start to externally submit the runContinuation task to the 538 * scheduler. If this virtual thread is scheduled by the built-in scheduler, 539 * and this method is called from a virtual thread scheduled by the built-in 540 * scheduler, then it uses externalSubmit to ensure that the task is pushed to an 541 * external submission queue rather than the local queue. 542 * @throws RejectedExecutionException 543 * @throws OutOfMemoryError 544 * @see ForkJoinPool#externalSubmit(ForkJoinTask) 545 */ 546 private void externalSubmitRunContinuationOrThrow() { 547 try { 548 if (currentThread().isVirtual()) { 549 // Pin the continuation to prevent the virtual thread from unmounting 550 // when submitting a task. This avoids deadlock that could arise due to 551 // carriers and virtual threads contending for a lock. 552 Continuation.pin(); 553 try { 554 if (scheduler == BUILTIN_SCHEDULER 555 && currentCarrierThread() instanceof CarrierThread ct) { 556 ct.getPool().externalSubmit(ForkJoinTask.adapt(runContinuation)); 557 } else { 558 scheduler.onStart(runContinuation); 559 } 560 } finally { 561 Continuation.unpin(); 562 } 563 } else { 564 scheduler.onStart(runContinuation); 565 } 566 } catch (RejectedExecutionException ree) { 567 submitFailed(ree); 568 throw ree; 569 } 570 } 571 572 /** 573 * If enabled, emits a JFR VirtualThreadSubmitFailedEvent. 574 */ 575 private void submitFailed(RejectedExecutionException ree) { 576 var event = new VirtualThreadSubmitFailedEvent(); 577 if (event.isEnabled()) { 578 event.javaThreadId = threadId(); 579 event.exceptionMessage = ree.getMessage(); 580 event.commit(); 581 } 582 } 583 584 /** 585 * Runs a task in the context of this virtual thread. 586 */ 587 private void run(Runnable task) { 588 assert Thread.currentThread() == this && state == RUNNING; 589 590 // emit JFR event if enabled 591 if (VirtualThreadStartEvent.isTurnedOn()) { 592 var event = new VirtualThreadStartEvent(); 593 event.javaThreadId = threadId(); 594 event.commit(); 595 } 596 597 Object bindings = Thread.scopedValueBindings(); 598 try { 599 runWith(bindings, task); 600 } catch (Throwable exc) { 601 dispatchUncaughtException(exc); 602 } finally { 603 // pop any remaining scopes from the stack, this may block 604 StackableScope.popAll(); 605 606 // emit JFR event if enabled 607 if (VirtualThreadEndEvent.isTurnedOn()) { 608 var event = new VirtualThreadEndEvent(); 609 event.javaThreadId = threadId(); 610 event.commit(); 611 } 612 } 613 } 614 615 /** 616 * Mounts this virtual thread onto the current platform thread. On 617 * return, the current thread is the virtual thread. 618 */ 619 @ChangesCurrentThread 620 @ReservedStackAccess 621 private void mount() { 622 startTransition(/*mount*/true); 623 // We assume following volatile accesses provide equivalent 624 // of acquire ordering, otherwise we need U.loadFence() here. 625 626 // sets the carrier thread 627 Thread carrier = Thread.currentCarrierThread(); 628 setCarrierThread(carrier); 629 630 // sync up carrier thread interrupted status if needed 631 if (interrupted) { 632 carrier.setInterrupt(); 633 } else if (carrier.isInterrupted()) { 634 synchronized (interruptLock) { 635 // need to recheck interrupted status 636 if (!interrupted) { 637 carrier.clearInterrupt(); 638 } 639 } 640 } 641 642 // set Thread.currentThread() to return this virtual thread 643 carrier.setCurrentThread(this); 644 } 645 646 /** 647 * Unmounts this virtual thread from the carrier. On return, the 648 * current thread is the current platform thread. 649 */ 650 @ChangesCurrentThread 651 @ReservedStackAccess 652 private void unmount() { 653 assert !Thread.holdsLock(interruptLock); 654 655 // set Thread.currentThread() to return the platform thread 656 Thread carrier = this.carrierThread; 657 carrier.setCurrentThread(carrier); 658 659 // break connection to carrier thread, synchronized with interrupt 660 synchronized (interruptLock) { 661 setCarrierThread(null); 662 } 663 carrier.clearInterrupt(); 664 665 // We assume previous volatile accesses provide equivalent 666 // of release ordering, otherwise we need U.storeFence() here. 667 endTransition(/*mount*/false); 668 } 669 670 /** 671 * Invokes Continuation.yield, notifying JVMTI (if enabled) to hide frames until 672 * the continuation continues. 673 */ 674 @Hidden 675 private boolean yieldContinuation() { 676 startTransition(/*mount*/false); 677 try { 678 return Continuation.yield(VTHREAD_SCOPE); 679 } finally { 680 endTransition(/*mount*/true); 681 } 682 } 683 684 /** 685 * Invoked in the context of the carrier thread after the Continuation yields when 686 * parking, blocking on monitor enter, Object.wait, or Thread.yield. 687 */ 688 private void afterYield() { 689 assert carrierThread == null; 690 691 // re-adjust parallelism if the virtual thread yielded when compensating 692 if (currentThread() instanceof CarrierThread ct) { 693 ct.endBlocking(); 694 } 695 696 int s = state(); 697 698 // LockSupport.park/parkNanos 699 if (s == PARKING || s == TIMED_PARKING) { 700 int newState; 701 if (s == PARKING) { 702 setState(newState = PARKED); 703 } else { 704 // schedule unpark 705 long timeout = this.timeout; 706 assert timeout > 0; 707 timeoutTask = schedule(this::parkTimeoutExpired, timeout, NANOSECONDS); 708 setState(newState = TIMED_PARKED); 709 } 710 711 // may have been unparked while parking 712 if (parkPermit && compareAndSetState(newState, UNPARKED)) { 713 // lazy submit if local queue is empty 714 lazySubmitRunContinuation(); 715 } 716 return; 717 } 718 719 // Thread.yield 720 if (s == YIELDING) { 721 setState(YIELDED); 722 723 // external submit if there are no tasks in the local task queue 724 if (currentThread() instanceof CarrierThread ct && ct.getQueuedTaskCount() == 0) { 725 externalSubmitRunContinuation(); 726 } else { 727 submitRunContinuation(); 728 } 729 return; 730 } 731 732 // blocking on monitorenter 733 if (s == BLOCKING) { 734 setState(BLOCKED); 735 736 // may have been unblocked while blocking 737 if (blockPermit && compareAndSetState(BLOCKED, UNBLOCKED)) { 738 // lazy submit if local queue is empty 739 lazySubmitRunContinuation(); 740 } 741 return; 742 } 743 744 // Object.wait 745 if (s == WAITING || s == TIMED_WAITING) { 746 int newState; 747 boolean interruptible = interruptibleWait; 748 if (s == WAITING) { 749 setState(newState = WAIT); 750 } else { 751 // For timed-wait, a timeout task is scheduled to execute. The timeout 752 // task will change the thread state to UNBLOCKED and submit the thread 753 // to the scheduler. A sequence number is used to ensure that the timeout 754 // task only unblocks the thread for this timed-wait. We synchronize with 755 // the timeout task to coordinate access to the sequence number and to 756 // ensure the timeout task doesn't execute until the thread has got to 757 // the TIMED_WAIT state. 758 long timeout = this.timeout; 759 assert timeout > 0; 760 synchronized (timedWaitLock()) { 761 byte seqNo = ++timedWaitSeqNo; 762 timeoutTask = schedule(() -> waitTimeoutExpired(seqNo), timeout, MILLISECONDS); 763 setState(newState = TIMED_WAIT); 764 } 765 } 766 767 // may have been notified while in transition to wait state 768 if (notified && compareAndSetState(newState, BLOCKED)) { 769 // may have even been unblocked already 770 if (blockPermit && compareAndSetState(BLOCKED, UNBLOCKED)) { 771 submitRunContinuation(); 772 } 773 return; 774 } 775 776 // may have been interrupted while in transition to wait state 777 if (interruptible && interrupted && compareAndSetState(newState, UNBLOCKED)) { 778 submitRunContinuation(); 779 return; 780 } 781 return; 782 } 783 784 assert false; 785 } 786 787 /** 788 * Invoked after the continuation completes. 789 */ 790 private void afterDone() { 791 afterDone(true); 792 } 793 794 /** 795 * Invoked after the continuation completes (or start failed). Sets the thread 796 * state to TERMINATED and notifies anyone waiting for the thread to terminate. 797 * 798 * @param notifyContainer true if its container should be notified 799 */ 800 private void afterDone(boolean notifyContainer) { 801 assert carrierThread == null; 802 setState(TERMINATED); 803 804 // notify anyone waiting for this virtual thread to terminate 805 CountDownLatch termination = this.termination; 806 if (termination != null) { 807 assert termination.getCount() == 1; 808 termination.countDown(); 809 } 810 811 // notify container 812 if (notifyContainer) { 813 threadContainer().remove(this); 814 } 815 816 // clear references to thread locals 817 clearReferences(); 818 } 819 820 /** 821 * Schedules this {@code VirtualThread} to execute. 822 * 823 * @throws IllegalStateException if the container is shutdown or closed 824 * @throws IllegalThreadStateException if the thread has already been started 825 * @throws RejectedExecutionException if the scheduler cannot accept a task 826 */ 827 @Override 828 void start(ThreadContainer container) { 829 if (!compareAndSetState(NEW, STARTED)) { 830 throw new IllegalThreadStateException("Already started"); 831 } 832 833 // bind thread to container 834 assert threadContainer() == null; 835 setThreadContainer(container); 836 837 // start thread 838 boolean addedToContainer = false; 839 boolean started = false; 840 try { 841 container.add(this); // may throw 842 addedToContainer = true; 843 844 // scoped values may be inherited 845 inheritScopedValueBindings(container); 846 847 // submit task to run thread, using externalSubmit if possible 848 externalSubmitRunContinuationOrThrow(); 849 started = true; 850 } finally { 851 if (!started) { 852 afterDone(addedToContainer); 853 } 854 } 855 } 856 857 @Override 858 public void start() { 859 start(ThreadContainers.root()); 860 } 861 862 @Override 863 public void run() { 864 // do nothing 865 } 866 867 /** 868 * Parks until unparked or interrupted. If already unparked then the parking 869 * permit is consumed and this method completes immediately (meaning it doesn't 870 * yield). It also completes immediately if the interrupted status is set. 871 */ 872 @Override 873 void park() { 874 assert Thread.currentThread() == this; 875 876 // complete immediately if parking permit available or interrupted 877 if (getAndSetParkPermit(false) || interrupted) 878 return; 879 880 // park the thread 881 boolean yielded = false; 882 long eventStartTime = VirtualThreadParkEvent.eventStartTime(); 883 setState(PARKING); 884 try { 885 yielded = yieldContinuation(); 886 } catch (OutOfMemoryError e) { 887 // park on carrier 888 } finally { 889 assert (Thread.currentThread() == this) && (yielded == (state() == RUNNING)); 890 if (yielded) { 891 VirtualThreadParkEvent.offer(eventStartTime, Long.MIN_VALUE); 892 } else { 893 assert state() == PARKING; 894 setState(RUNNING); 895 } 896 } 897 898 // park on the carrier thread when pinned 899 if (!yielded) { 900 parkOnCarrierThread(false, 0); 901 } 902 } 903 904 /** 905 * Parks up to the given waiting time or until unparked or interrupted. 906 * If already unparked then the parking permit is consumed and this method 907 * completes immediately (meaning it doesn't yield). It also completes immediately 908 * if the interrupted status is set or the waiting time is {@code <= 0}. 909 * 910 * @param nanos the maximum number of nanoseconds to wait. 911 */ 912 @Override 913 void parkNanos(long nanos) { 914 assert Thread.currentThread() == this; 915 916 // complete immediately if parking permit available or interrupted 917 if (getAndSetParkPermit(false) || interrupted) 918 return; 919 920 // park the thread for the waiting time 921 if (nanos > 0) { 922 long startTime = System.nanoTime(); 923 924 // park the thread, afterYield will schedule the thread to unpark 925 boolean yielded = false; 926 long eventStartTime = VirtualThreadParkEvent.eventStartTime(); 927 timeout = nanos; 928 setState(TIMED_PARKING); 929 try { 930 yielded = yieldContinuation(); 931 } catch (OutOfMemoryError e) { 932 // park on carrier 933 } finally { 934 assert (Thread.currentThread() == this) && (yielded == (state() == RUNNING)); 935 if (yielded) { 936 VirtualThreadParkEvent.offer(eventStartTime, nanos); 937 } else { 938 assert state() == TIMED_PARKING; 939 setState(RUNNING); 940 } 941 } 942 943 // park on carrier thread for remaining time when pinned (or OOME) 944 if (!yielded) { 945 long remainingNanos = nanos - (System.nanoTime() - startTime); 946 parkOnCarrierThread(true, remainingNanos); 947 } 948 } 949 } 950 951 /** 952 * Parks the current carrier thread up to the given waiting time or until 953 * unparked or interrupted. If the virtual thread is interrupted then the 954 * interrupted status will be propagated to the carrier thread. 955 * @param timed true for a timed park, false for untimed 956 * @param nanos the waiting time in nanoseconds 957 */ 958 private void parkOnCarrierThread(boolean timed, long nanos) { 959 assert state() == RUNNING; 960 961 setState(timed ? TIMED_PINNED : PINNED); 962 try { 963 if (!parkPermit) { 964 if (!timed) { 965 U.park(false, 0); 966 } else if (nanos > 0) { 967 U.park(false, nanos); 968 } 969 } 970 } finally { 971 setState(RUNNING); 972 } 973 974 // consume parking permit 975 setParkPermit(false); 976 977 // JFR jdk.VirtualThreadPinned event 978 postPinnedEvent("LockSupport.park"); 979 } 980 981 /** 982 * Call into VM when pinned to record a JFR jdk.VirtualThreadPinned event. 983 * Recording the event in the VM avoids having JFR event recorded in Java 984 * with the same name, but different ID, to events recorded by the VM. 985 */ 986 @Hidden 987 private static native void postPinnedEvent(String op); 988 989 /** 990 * Re-enables this virtual thread for scheduling. If this virtual thread is parked 991 * then its task is scheduled to continue, otherwise its next call to {@code park} or 992 * {@linkplain #parkNanos(long) parkNanos} is guaranteed not to block. 993 * @param lazySubmit to use lazySubmit if possible 994 * @throws RejectedExecutionException if the scheduler cannot accept a task 995 */ 996 private void unpark(boolean lazySubmit) { 997 if (!getAndSetParkPermit(true) && currentThread() != this) { 998 int s = state(); 999 1000 // unparked while parked 1001 if ((s == PARKED || s == TIMED_PARKED) && compareAndSetState(s, UNPARKED)) { 1002 if (lazySubmit) { 1003 lazySubmitRunContinuation(); 1004 } else { 1005 submitRunContinuation(); 1006 } 1007 return; 1008 } 1009 1010 // unparked while parked when pinned 1011 if (s == PINNED || s == TIMED_PINNED) { 1012 // unpark carrier thread when pinned 1013 disableSuspendAndPreempt(); 1014 try { 1015 synchronized (carrierThreadAccessLock()) { 1016 Thread carrier = carrierThread; 1017 if (carrier != null && ((s = state()) == PINNED || s == TIMED_PINNED)) { 1018 U.unpark(carrier); 1019 } 1020 } 1021 } finally { 1022 enableSuspendAndPreempt(); 1023 } 1024 return; 1025 } 1026 } 1027 } 1028 1029 @Override 1030 void unpark() { 1031 unpark(false); 1032 } 1033 1034 /** 1035 * Invoked by unblocker thread to unblock this virtual thread. 1036 */ 1037 private void unblock() { 1038 assert !Thread.currentThread().isVirtual(); 1039 blockPermit = true; 1040 if (state() == BLOCKED && compareAndSetState(BLOCKED, UNBLOCKED)) { 1041 submitRunContinuation(); 1042 } 1043 } 1044 1045 /** 1046 * Invoked by FJP worker thread or STPE thread when park timeout expires. 1047 */ 1048 private void parkTimeoutExpired() { 1049 assert !VirtualThread.currentThread().isVirtual(); 1050 unpark(true); 1051 } 1052 1053 /** 1054 * Invoked by FJP worker thread or STPE thread when wait timeout expires. 1055 * If the virtual thread is in timed-wait then this method will unblock the thread 1056 * and submit its task so that it continues and attempts to reenter the monitor. 1057 * This method does nothing if the thread has been woken by notify or interrupt. 1058 */ 1059 private void waitTimeoutExpired(byte seqNo) { 1060 assert !Thread.currentThread().isVirtual(); 1061 1062 synchronized (timedWaitLock()) { 1063 if (seqNo != timedWaitSeqNo) { 1064 // this timeout task is for a past timed-wait 1065 return; 1066 } 1067 if (!compareAndSetState(TIMED_WAIT, UNBLOCKED)) { 1068 // already unblocked 1069 return; 1070 } 1071 } 1072 1073 lazySubmitRunContinuation(); 1074 } 1075 1076 /** 1077 * Attempts to yield the current virtual thread (Thread.yield). 1078 */ 1079 void tryYield() { 1080 assert Thread.currentThread() == this; 1081 setState(YIELDING); 1082 boolean yielded = false; 1083 try { 1084 yielded = yieldContinuation(); // may throw 1085 } finally { 1086 assert (Thread.currentThread() == this) && (yielded == (state() == RUNNING)); 1087 if (!yielded) { 1088 assert state() == YIELDING; 1089 setState(RUNNING); 1090 } 1091 } 1092 } 1093 1094 /** 1095 * Sleep the current thread for the given sleep time (in nanoseconds). If 1096 * nanos is 0 then the thread will attempt to yield. 1097 * 1098 * @implNote This implementation parks the thread for the given sleeping time 1099 * and will therefore be observed in PARKED state during the sleep. Parking 1100 * will consume the parking permit so this method makes available the parking 1101 * permit after the sleep. This may be observed as a spurious, but benign, 1102 * wakeup when the thread subsequently attempts to park. 1103 * 1104 * @param nanos the maximum number of nanoseconds to sleep 1105 * @throws InterruptedException if interrupted while sleeping 1106 */ 1107 void sleepNanos(long nanos) throws InterruptedException { 1108 assert Thread.currentThread() == this && nanos >= 0; 1109 if (getAndClearInterrupt()) 1110 throw new InterruptedException(); 1111 if (nanos == 0) { 1112 tryYield(); 1113 } else { 1114 // park for the sleep time 1115 try { 1116 long remainingNanos = nanos; 1117 long startNanos = System.nanoTime(); 1118 while (remainingNanos > 0) { 1119 parkNanos(remainingNanos); 1120 if (getAndClearInterrupt()) { 1121 throw new InterruptedException(); 1122 } 1123 remainingNanos = nanos - (System.nanoTime() - startNanos); 1124 } 1125 } finally { 1126 // may have been unparked while sleeping 1127 setParkPermit(true); 1128 } 1129 } 1130 } 1131 1132 /** 1133 * Waits up to {@code nanos} nanoseconds for this virtual thread to terminate. 1134 * A timeout of {@code 0} means to wait forever. 1135 * 1136 * @throws InterruptedException if interrupted while waiting 1137 * @return true if the thread has terminated 1138 */ 1139 boolean joinNanos(long nanos) throws InterruptedException { 1140 if (state() == TERMINATED) 1141 return true; 1142 1143 // ensure termination object exists, then re-check state 1144 CountDownLatch termination = getTermination(); 1145 if (state() == TERMINATED) 1146 return true; 1147 1148 // wait for virtual thread to terminate 1149 if (nanos == 0) { 1150 termination.await(); 1151 } else { 1152 boolean terminated = termination.await(nanos, NANOSECONDS); 1153 if (!terminated) { 1154 // waiting time elapsed 1155 return false; 1156 } 1157 } 1158 assert state() == TERMINATED; 1159 return true; 1160 } 1161 1162 @Override 1163 void blockedOn(Interruptible b) { 1164 disableSuspendAndPreempt(); 1165 try { 1166 super.blockedOn(b); 1167 } finally { 1168 enableSuspendAndPreempt(); 1169 } 1170 } 1171 1172 @Override 1173 public void interrupt() { 1174 if (Thread.currentThread() != this) { 1175 // if current thread is a virtual thread then prevent it from being 1176 // suspended or unmounted when entering or holding interruptLock 1177 Interruptible blocker; 1178 disableSuspendAndPreempt(); 1179 try { 1180 synchronized (interruptLock) { 1181 interrupted = true; 1182 blocker = nioBlocker(); 1183 if (blocker != null) { 1184 blocker.interrupt(this); 1185 } 1186 1187 // interrupt carrier thread if mounted 1188 Thread carrier = carrierThread; 1189 if (carrier != null) carrier.setInterrupt(); 1190 } 1191 } finally { 1192 enableSuspendAndPreempt(); 1193 } 1194 1195 // notify blocker after releasing interruptLock 1196 if (blocker != null) { 1197 blocker.postInterrupt(); 1198 } 1199 1200 // make available parking permit, unpark thread if parked 1201 unpark(); 1202 1203 // if thread is waiting in Object.wait then schedule to try to reenter 1204 int s = state(); 1205 if ((s == WAIT || s == TIMED_WAIT) && compareAndSetState(s, UNBLOCKED)) { 1206 submitRunContinuation(); 1207 } 1208 1209 } else { 1210 interrupted = true; 1211 carrierThread.setInterrupt(); 1212 setParkPermit(true); 1213 } 1214 } 1215 1216 @Override 1217 public boolean isInterrupted() { 1218 return interrupted; 1219 } 1220 1221 @Override 1222 boolean getAndClearInterrupt() { 1223 assert Thread.currentThread() == this; 1224 boolean oldValue = interrupted; 1225 if (oldValue) { 1226 disableSuspendAndPreempt(); 1227 try { 1228 synchronized (interruptLock) { 1229 interrupted = false; 1230 carrierThread.clearInterrupt(); 1231 } 1232 } finally { 1233 enableSuspendAndPreempt(); 1234 } 1235 } 1236 return oldValue; 1237 } 1238 1239 @Override 1240 Thread.State threadState() { 1241 switch (state()) { 1242 case NEW: 1243 return Thread.State.NEW; 1244 case STARTED: 1245 // return NEW if thread container not yet set 1246 if (threadContainer() == null) { 1247 return Thread.State.NEW; 1248 } else { 1249 return Thread.State.RUNNABLE; 1250 } 1251 case UNPARKED: 1252 case UNBLOCKED: 1253 case YIELDED: 1254 // runnable, not mounted 1255 return Thread.State.RUNNABLE; 1256 case RUNNING: 1257 // if mounted then return state of carrier thread 1258 if (Thread.currentThread() != this) { 1259 disableSuspendAndPreempt(); 1260 try { 1261 synchronized (carrierThreadAccessLock()) { 1262 Thread carrierThread = this.carrierThread; 1263 if (carrierThread != null) { 1264 return carrierThread.threadState(); 1265 } 1266 } 1267 } finally { 1268 enableSuspendAndPreempt(); 1269 } 1270 } 1271 // runnable, mounted 1272 return Thread.State.RUNNABLE; 1273 case PARKING: 1274 case TIMED_PARKING: 1275 case WAITING: 1276 case TIMED_WAITING: 1277 case YIELDING: 1278 // runnable, in transition 1279 return Thread.State.RUNNABLE; 1280 case PARKED: 1281 case PINNED: 1282 case WAIT: 1283 return Thread.State.WAITING; 1284 case TIMED_PARKED: 1285 case TIMED_PINNED: 1286 case TIMED_WAIT: 1287 return Thread.State.TIMED_WAITING; 1288 case BLOCKING: 1289 case BLOCKED: 1290 return Thread.State.BLOCKED; 1291 case TERMINATED: 1292 return Thread.State.TERMINATED; 1293 default: 1294 throw new InternalError(); 1295 } 1296 } 1297 1298 @Override 1299 boolean alive() { 1300 int s = state; 1301 return (s != NEW && s != TERMINATED); 1302 } 1303 1304 @Override 1305 boolean isTerminated() { 1306 return (state == TERMINATED); 1307 } 1308 1309 @Override 1310 public String toString() { 1311 StringBuilder sb = new StringBuilder("VirtualThread[#"); 1312 sb.append(threadId()); 1313 String name = getName(); 1314 if (!name.isEmpty()) { 1315 sb.append(","); 1316 sb.append(name); 1317 } 1318 sb.append("]/"); 1319 1320 // add the carrier state and thread name when mounted 1321 boolean mounted; 1322 if (Thread.currentThread() == this) { 1323 mounted = appendCarrierInfo(sb); 1324 } else { 1325 disableSuspendAndPreempt(); 1326 try { 1327 synchronized (carrierThreadAccessLock()) { 1328 mounted = appendCarrierInfo(sb); 1329 } 1330 } finally { 1331 enableSuspendAndPreempt(); 1332 } 1333 } 1334 1335 // add virtual thread state when not mounted 1336 if (!mounted) { 1337 String stateAsString = threadState().toString(); 1338 sb.append(stateAsString.toLowerCase(Locale.ROOT)); 1339 } 1340 1341 return sb.toString(); 1342 } 1343 1344 /** 1345 * Appends the carrier state and thread name to the string buffer if mounted. 1346 * @return true if mounted, false if not mounted 1347 */ 1348 private boolean appendCarrierInfo(StringBuilder sb) { 1349 assert Thread.currentThread() == this || Thread.holdsLock(carrierThreadAccessLock()); 1350 Thread carrier = carrierThread; 1351 if (carrier != null) { 1352 String stateAsString = carrier.threadState().toString(); 1353 sb.append(stateAsString.toLowerCase(Locale.ROOT)); 1354 sb.append('@'); 1355 sb.append(carrier.getName()); 1356 return true; 1357 } else { 1358 return false; 1359 } 1360 } 1361 1362 @Override 1363 public int hashCode() { 1364 return (int) threadId(); 1365 } 1366 1367 @Override 1368 public boolean equals(Object obj) { 1369 return obj == this; 1370 } 1371 1372 /** 1373 * Returns the termination object, creating it if needed. 1374 */ 1375 private CountDownLatch getTermination() { 1376 CountDownLatch termination = this.termination; 1377 if (termination == null) { 1378 termination = new CountDownLatch(1); 1379 if (!U.compareAndSetReference(this, TERMINATION, null, termination)) { 1380 termination = this.termination; 1381 } 1382 } 1383 return termination; 1384 } 1385 1386 /** 1387 * Returns the lock object to synchronize on when accessing carrierThread. 1388 * The lock prevents carrierThread from being reset to null during unmount. 1389 */ 1390 private Object carrierThreadAccessLock() { 1391 // return interruptLock as unmount has to coordinate with interrupt 1392 return interruptLock; 1393 } 1394 1395 /** 1396 * Returns a lock object for coordinating timed-wait setup and timeout handling. 1397 */ 1398 private Object timedWaitLock() { 1399 // use this object for now to avoid the overhead of introducing another lock 1400 return runContinuation; 1401 } 1402 1403 /** 1404 * Disallow the current thread be suspended or preempted. 1405 */ 1406 private void disableSuspendAndPreempt() { 1407 notifyJvmtiDisableSuspend(true); 1408 Continuation.pin(); 1409 } 1410 1411 /** 1412 * Allow the current thread be suspended or preempted. 1413 */ 1414 private void enableSuspendAndPreempt() { 1415 Continuation.unpin(); 1416 notifyJvmtiDisableSuspend(false); 1417 } 1418 1419 // -- wrappers for get/set of state, parking permit, and carrier thread -- 1420 1421 private int state() { 1422 return state; // volatile read 1423 } 1424 1425 private void setState(int newValue) { 1426 state = newValue; // volatile write 1427 } 1428 1429 private boolean compareAndSetState(int expectedValue, int newValue) { 1430 return U.compareAndSetInt(this, STATE, expectedValue, newValue); 1431 } 1432 1433 private boolean compareAndSetOnWaitingList(boolean expectedValue, boolean newValue) { 1434 return U.compareAndSetBoolean(this, ON_WAITING_LIST, expectedValue, newValue); 1435 } 1436 1437 private void setParkPermit(boolean newValue) { 1438 if (parkPermit != newValue) { 1439 parkPermit = newValue; 1440 } 1441 } 1442 1443 private boolean getAndSetParkPermit(boolean newValue) { 1444 if (parkPermit != newValue) { 1445 return U.getAndSetBoolean(this, PARK_PERMIT, newValue); 1446 } else { 1447 return newValue; 1448 } 1449 } 1450 1451 private void setCarrierThread(Thread carrier) { 1452 // U.putReferenceRelease(this, CARRIER_THREAD, carrier); 1453 this.carrierThread = carrier; 1454 } 1455 1456 // The following four methods notify the VM when a "transition" starts and ends. 1457 // A "mount transition" embodies the steps to transfer control from a platform 1458 // thread to a virtual thread, changing the thread identity, and starting or 1459 // resuming the virtual thread's continuation on the carrier. 1460 // An "unmount transition" embodies the steps to transfer control from a virtual 1461 // thread to its carrier, suspending the virtual thread's continuation, and 1462 // restoring the thread identity to the platform thread. 1463 // The notifications to the VM are necessary in order to coordinate with functions 1464 // (JVMTI mostly) that disable transitions for one or all virtual threads. Starting 1465 // a transition may block if transitions are disabled. Ending a transition may 1466 // notify a thread that is waiting to disable transitions. The notifications are 1467 // also used to post JVMTI events for virtual thread start and end. 1468 1469 @IntrinsicCandidate 1470 @JvmtiMountTransition 1471 private native void endFirstTransition(); 1472 1473 @IntrinsicCandidate 1474 @JvmtiMountTransition 1475 private native void startFinalTransition(); 1476 1477 @IntrinsicCandidate 1478 @JvmtiMountTransition 1479 private native void startTransition(boolean mount); 1480 1481 @IntrinsicCandidate 1482 @JvmtiMountTransition 1483 private native void endTransition(boolean mount); 1484 1485 @IntrinsicCandidate 1486 private static native void notifyJvmtiDisableSuspend(boolean enter); 1487 1488 private static native void registerNatives(); 1489 static { 1490 registerNatives(); 1491 1492 // ensure VTHREAD_GROUP is created, may be accessed by JVMTI 1493 var group = Thread.virtualThreadGroup(); 1494 1495 // ensure event class is initialized 1496 try { 1497 MethodHandles.lookup().ensureInitialized(VirtualThreadParkEvent.class); 1498 } catch (IllegalAccessException e) { 1499 throw new ExceptionInInitializerError(e); 1500 } 1501 } 1502 1503 /** 1504 * Loads a VirtualThreadScheduler with the given class name. The class must be public 1505 * in an exported package, with public one-arg or no-arg constructor, and be visible 1506 * to the system class loader. 1507 * @param delegate the scheduler that the custom scheduler may delegate to 1508 * @param cn the class name of the custom scheduler 1509 */ 1510 private static VirtualThreadScheduler loadCustomScheduler(VirtualThreadScheduler delegate, String cn) { 1511 VirtualThreadScheduler scheduler; 1512 try { 1513 Class<?> clazz = Class.forName(cn, true, ClassLoader.getSystemClassLoader()); 1514 // 1-arg constructor 1515 try { 1516 Constructor<?> ctor = clazz.getConstructor(VirtualThreadScheduler.class); 1517 return (VirtualThreadScheduler) ctor.newInstance(delegate); 1518 } catch (NoSuchMethodException e) { 1519 // 0-arg constructor 1520 Constructor<?> ctor = clazz.getConstructor(); 1521 scheduler = (VirtualThreadScheduler) ctor.newInstance(); 1522 } 1523 } catch (Exception ex) { 1524 throw new Error(ex); 1525 } 1526 System.err.println("WARNING: Using custom default scheduler, this is an experimental feature!"); 1527 return scheduler; 1528 } 1529 1530 /** 1531 * Creates the built-in ForkJoinPool scheduler. 1532 * @param wrapped true if wrapped by a custom default scheduler 1533 */ 1534 private static BuiltinScheduler createBuiltinScheduler(boolean wrapped) { 1535 int parallelism, maxPoolSize, minRunnable; 1536 String parallelismValue = System.getProperty("jdk.virtualThreadScheduler.parallelism"); 1537 String maxPoolSizeValue = System.getProperty("jdk.virtualThreadScheduler.maxPoolSize"); 1538 String minRunnableValue = System.getProperty("jdk.virtualThreadScheduler.minRunnable"); 1539 if (parallelismValue != null) { 1540 parallelism = Integer.parseInt(parallelismValue); 1541 } else { 1542 parallelism = Runtime.getRuntime().availableProcessors(); 1543 } 1544 if (maxPoolSizeValue != null) { 1545 maxPoolSize = Integer.parseInt(maxPoolSizeValue); 1546 parallelism = Integer.min(parallelism, maxPoolSize); 1547 } else { 1548 maxPoolSize = Integer.max(parallelism, 256); 1549 } 1550 if (minRunnableValue != null) { 1551 minRunnable = Integer.parseInt(minRunnableValue); 1552 } else { 1553 minRunnable = Integer.max(parallelism / 2, 1); 1554 } 1555 return new BuiltinScheduler(parallelism, maxPoolSize, minRunnable, wrapped); 1556 } 1557 1558 /** 1559 * The built-in ForkJoinPool scheduler. 1560 */ 1561 private static class BuiltinScheduler 1562 extends ForkJoinPool implements VirtualThreadScheduler { 1563 1564 BuiltinScheduler(int parallelism, int maxPoolSize, int minRunnable, boolean wrapped) { 1565 ForkJoinWorkerThreadFactory factory = wrapped 1566 ? ForkJoinPool.defaultForkJoinWorkerThreadFactory 1567 : CarrierThread::new; 1568 Thread.UncaughtExceptionHandler handler = (t, e) -> { }; 1569 boolean asyncMode = true; // FIFO 1570 super(parallelism, factory, handler, asyncMode, 1571 0, maxPoolSize, minRunnable, pool -> true, 30, SECONDS); 1572 } 1573 1574 private void adaptAndExecute(Runnable task) { 1575 execute(ForkJoinTask.adapt(task)); 1576 } 1577 1578 @Override 1579 public void onStart(VirtualThreadTask task) { 1580 adaptAndExecute(task); 1581 } 1582 1583 @Override 1584 public void onContinue(VirtualThreadTask task) { 1585 adaptAndExecute(task); 1586 } 1587 1588 @Override 1589 public ScheduledFuture<?> schedule(Runnable task, long delay, TimeUnit unit) { 1590 return super.schedule(task, delay, unit); 1591 } 1592 1593 /** 1594 * Wraps the scheduler to avoid leaking a direct reference with 1595 * {@link VirtualThreadScheduler#current()}. 1596 */ 1597 VirtualThreadScheduler createExternalView() { 1598 BuiltinScheduler builtin = this; 1599 return new VirtualThreadScheduler() { 1600 private void execute(VirtualThreadTask task) { 1601 var vthread = (VirtualThread) task.thread(); 1602 VirtualThreadScheduler scheduler = vthread.scheduler; 1603 if (scheduler == this || scheduler == DEFAULT_SCHEDULER) { 1604 builtin.adaptAndExecute(task); 1605 } else { 1606 throw new IllegalArgumentException(); 1607 } 1608 } 1609 @Override 1610 public void onStart(VirtualThreadTask task) { 1611 execute(task); 1612 } 1613 @Override 1614 public void onContinue(VirtualThreadTask task) { 1615 execute(task); 1616 } 1617 @Override 1618 public String toString() { 1619 return builtin.toString(); 1620 } 1621 }; 1622 } 1623 } 1624 1625 /** 1626 * Schedule a runnable task to run after a delay. 1627 */ 1628 private Future<?> schedule(Runnable command, long delay, TimeUnit unit) { 1629 return scheduler.schedule(command, delay, unit); 1630 } 1631 1632 /** 1633 * Supports scheduling a runnable task to run after a delay. It uses a number 1634 * of ScheduledThreadPoolExecutor instances to reduce contention on the delayed 1635 * work queue used. This class is used when using a custom scheduler. 1636 */ 1637 static class DelayedTaskSchedulers { 1638 private static final ScheduledExecutorService[] INSTANCE = createDelayedTaskSchedulers(); 1639 1640 static Future<?> schedule(Runnable command, long delay, TimeUnit unit) { 1641 long tid = Thread.currentThread().threadId(); 1642 int index = (int) tid & (INSTANCE.length - 1); 1643 return INSTANCE[index].schedule(command, delay, unit); 1644 } 1645 1646 private static ScheduledExecutorService[] createDelayedTaskSchedulers() { 1647 String propName = "jdk.virtualThreadScheduler.timerQueues"; 1648 String propValue = System.getProperty(propName); 1649 int queueCount; 1650 if (propValue != null) { 1651 queueCount = Integer.parseInt(propValue); 1652 if (queueCount != Integer.highestOneBit(queueCount)) { 1653 throw new RuntimeException("Value of " + propName + " must be power of 2"); 1654 } 1655 } else { 1656 int ncpus = Runtime.getRuntime().availableProcessors(); 1657 queueCount = Math.max(Integer.highestOneBit(ncpus / 4), 1); 1658 } 1659 var schedulers = new ScheduledExecutorService[queueCount]; 1660 for (int i = 0; i < queueCount; i++) { 1661 ScheduledThreadPoolExecutor stpe = (ScheduledThreadPoolExecutor) 1662 Executors.newScheduledThreadPool(1, task -> { 1663 Thread t = InnocuousThread.newThread("VirtualThread-unparker", task); 1664 t.setDaemon(true); 1665 return t; 1666 }); 1667 stpe.setRemoveOnCancelPolicy(true); 1668 schedulers[i] = stpe; 1669 } 1670 return schedulers; 1671 } 1672 } 1673 1674 /** 1675 * Schedule virtual threads that are ready to be scheduled after they blocked on 1676 * monitor enter. 1677 */ 1678 private static void unblockVirtualThreads() { 1679 while (true) { 1680 VirtualThread vthread = takeVirtualThreadListToUnblock(); 1681 while (vthread != null) { 1682 assert vthread.onWaitingList; 1683 VirtualThread nextThread = vthread.next; 1684 1685 // remove from list and unblock 1686 vthread.next = null; 1687 boolean changed = vthread.compareAndSetOnWaitingList(true, false); 1688 assert changed; 1689 vthread.unblock(); 1690 1691 vthread = nextThread; 1692 } 1693 } 1694 } 1695 1696 /** 1697 * Retrieves the list of virtual threads that are waiting to be unblocked, waiting 1698 * if necessary until a list of one or more threads becomes available. 1699 */ 1700 private static native VirtualThread takeVirtualThreadListToUnblock(); 1701 1702 static { 1703 var unblocker = InnocuousThread.newThread("VirtualThread-unblocker", 1704 VirtualThread::unblockVirtualThreads); 1705 unblocker.setDaemon(true); 1706 unblocker.start(); 1707 } 1708 } --- EOF ---