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