1 /* 2 * Copyright (c) 2018, 2025, 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 * @throws RejectedExecutionException if the scheduler cannot accept a task 989 */ 990 @Override 991 void unpark() { 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 submitRunContinuation(); 998 return; 999 } 1000 1001 // unparked while parked when pinned 1002 if (s == PINNED || s == TIMED_PINNED) { 1003 // unpark carrier thread when pinned 1004 disableSuspendAndPreempt(); 1005 try { 1006 synchronized (carrierThreadAccessLock()) { 1007 Thread carrier = carrierThread; 1008 if (carrier != null && ((s = state()) == PINNED || s == TIMED_PINNED)) { 1009 U.unpark(carrier); 1010 } 1011 } 1012 } finally { 1013 enableSuspendAndPreempt(); 1014 } 1015 return; 1016 } 1017 } 1018 } 1019 1020 /** 1021 * Invoked by unblocker thread to unblock this virtual thread. 1022 */ 1023 private void unblock() { 1024 assert !Thread.currentThread().isVirtual(); 1025 blockPermit = true; 1026 if (state() == BLOCKED && compareAndSetState(BLOCKED, UNBLOCKED)) { 1027 submitRunContinuation(); 1028 } 1029 } 1030 1031 /** 1032 * Invoked by FJP worker thread or STPE thread when park timeout expires. 1033 */ 1034 private void parkTimeoutExpired() { 1035 assert !VirtualThread.currentThread().isVirtual(); 1036 if (!getAndSetParkPermit(true)) { 1037 int s = state(); 1038 if ((s == PARKED || s == TIMED_PARKED) && compareAndSetState(s, UNPARKED)) { 1039 lazySubmitRunContinuation(); 1040 } 1041 } 1042 } 1043 1044 /** 1045 * Invoked by FJP worker thread or STPE thread when wait timeout expires. 1046 * If the virtual thread is in timed-wait then this method will unblock the thread 1047 * and submit its task so that it continues and attempts to reenter the monitor. 1048 * This method does nothing if the thread has been woken by notify or interrupt. 1049 */ 1050 private void waitTimeoutExpired(byte seqNo) { 1051 assert !Thread.currentThread().isVirtual(); 1052 1053 synchronized (timedWaitLock()) { 1054 if (seqNo != timedWaitSeqNo) { 1055 // this timeout task is for a past timed-wait 1056 return; 1057 } 1058 if (!compareAndSetState(TIMED_WAIT, UNBLOCKED)) { 1059 // already unblocked 1060 return; 1061 } 1062 } 1063 1064 lazySubmitRunContinuation(); 1065 } 1066 1067 /** 1068 * Attempts to yield the current virtual thread (Thread.yield). 1069 */ 1070 void tryYield() { 1071 assert Thread.currentThread() == this; 1072 setState(YIELDING); 1073 boolean yielded = false; 1074 try { 1075 yielded = yieldContinuation(); // may throw 1076 } finally { 1077 assert (Thread.currentThread() == this) && (yielded == (state() == RUNNING)); 1078 if (!yielded) { 1079 assert state() == YIELDING; 1080 setState(RUNNING); 1081 } 1082 } 1083 } 1084 1085 /** 1086 * Sleep the current thread for the given sleep time (in nanoseconds). If 1087 * nanos is 0 then the thread will attempt to yield. 1088 * 1089 * @implNote This implementation parks the thread for the given sleeping time 1090 * and will therefore be observed in PARKED state during the sleep. Parking 1091 * will consume the parking permit so this method makes available the parking 1092 * permit after the sleep. This may be observed as a spurious, but benign, 1093 * wakeup when the thread subsequently attempts to park. 1094 * 1095 * @param nanos the maximum number of nanoseconds to sleep 1096 * @throws InterruptedException if interrupted while sleeping 1097 */ 1098 void sleepNanos(long nanos) throws InterruptedException { 1099 assert Thread.currentThread() == this && nanos >= 0; 1100 if (getAndClearInterrupt()) 1101 throw new InterruptedException(); 1102 if (nanos == 0) { 1103 tryYield(); 1104 } else { 1105 // park for the sleep time 1106 try { 1107 long remainingNanos = nanos; 1108 long startNanos = System.nanoTime(); 1109 while (remainingNanos > 0) { 1110 parkNanos(remainingNanos); 1111 if (getAndClearInterrupt()) { 1112 throw new InterruptedException(); 1113 } 1114 remainingNanos = nanos - (System.nanoTime() - startNanos); 1115 } 1116 } finally { 1117 // may have been unparked while sleeping 1118 setParkPermit(true); 1119 } 1120 } 1121 } 1122 1123 /** 1124 * Waits up to {@code nanos} nanoseconds for this virtual thread to terminate. 1125 * A timeout of {@code 0} means to wait forever. 1126 * 1127 * @throws InterruptedException if interrupted while waiting 1128 * @return true if the thread has terminated 1129 */ 1130 boolean joinNanos(long nanos) throws InterruptedException { 1131 if (state() == TERMINATED) 1132 return true; 1133 1134 // ensure termination object exists, then re-check state 1135 CountDownLatch termination = getTermination(); 1136 if (state() == TERMINATED) 1137 return true; 1138 1139 // wait for virtual thread to terminate 1140 if (nanos == 0) { 1141 termination.await(); 1142 } else { 1143 boolean terminated = termination.await(nanos, NANOSECONDS); 1144 if (!terminated) { 1145 // waiting time elapsed 1146 return false; 1147 } 1148 } 1149 assert state() == TERMINATED; 1150 return true; 1151 } 1152 1153 @Override 1154 void blockedOn(Interruptible b) { 1155 disableSuspendAndPreempt(); 1156 try { 1157 super.blockedOn(b); 1158 } finally { 1159 enableSuspendAndPreempt(); 1160 } 1161 } 1162 1163 @Override 1164 public void interrupt() { 1165 if (Thread.currentThread() != this) { 1166 // if current thread is a virtual thread then prevent it from being 1167 // suspended or unmounted when entering or holding interruptLock 1168 Interruptible blocker; 1169 disableSuspendAndPreempt(); 1170 try { 1171 synchronized (interruptLock) { 1172 interrupted = true; 1173 blocker = nioBlocker(); 1174 if (blocker != null) { 1175 blocker.interrupt(this); 1176 } 1177 1178 // interrupt carrier thread if mounted 1179 Thread carrier = carrierThread; 1180 if (carrier != null) carrier.setInterrupt(); 1181 } 1182 } finally { 1183 enableSuspendAndPreempt(); 1184 } 1185 1186 // notify blocker after releasing interruptLock 1187 if (blocker != null) { 1188 blocker.postInterrupt(); 1189 } 1190 1191 // make available parking permit, unpark thread if parked 1192 unpark(); 1193 1194 // if thread is waiting in Object.wait then schedule to try to reenter 1195 int s = state(); 1196 if ((s == WAIT || s == TIMED_WAIT) && compareAndSetState(s, UNBLOCKED)) { 1197 submitRunContinuation(); 1198 } 1199 1200 } else { 1201 interrupted = true; 1202 carrierThread.setInterrupt(); 1203 setParkPermit(true); 1204 } 1205 } 1206 1207 @Override 1208 public boolean isInterrupted() { 1209 return interrupted; 1210 } 1211 1212 @Override 1213 boolean getAndClearInterrupt() { 1214 assert Thread.currentThread() == this; 1215 boolean oldValue = interrupted; 1216 if (oldValue) { 1217 disableSuspendAndPreempt(); 1218 try { 1219 synchronized (interruptLock) { 1220 interrupted = false; 1221 carrierThread.clearInterrupt(); 1222 } 1223 } finally { 1224 enableSuspendAndPreempt(); 1225 } 1226 } 1227 return oldValue; 1228 } 1229 1230 @Override 1231 Thread.State threadState() { 1232 switch (state()) { 1233 case NEW: 1234 return Thread.State.NEW; 1235 case STARTED: 1236 // return NEW if thread container not yet set 1237 if (threadContainer() == null) { 1238 return Thread.State.NEW; 1239 } else { 1240 return Thread.State.RUNNABLE; 1241 } 1242 case UNPARKED: 1243 case UNBLOCKED: 1244 case YIELDED: 1245 // runnable, not mounted 1246 return Thread.State.RUNNABLE; 1247 case RUNNING: 1248 // if mounted then return state of carrier thread 1249 if (Thread.currentThread() != this) { 1250 disableSuspendAndPreempt(); 1251 try { 1252 synchronized (carrierThreadAccessLock()) { 1253 Thread carrierThread = this.carrierThread; 1254 if (carrierThread != null) { 1255 return carrierThread.threadState(); 1256 } 1257 } 1258 } finally { 1259 enableSuspendAndPreempt(); 1260 } 1261 } 1262 // runnable, mounted 1263 return Thread.State.RUNNABLE; 1264 case PARKING: 1265 case TIMED_PARKING: 1266 case WAITING: 1267 case TIMED_WAITING: 1268 case YIELDING: 1269 // runnable, in transition 1270 return Thread.State.RUNNABLE; 1271 case PARKED: 1272 case PINNED: 1273 case WAIT: 1274 return Thread.State.WAITING; 1275 case TIMED_PARKED: 1276 case TIMED_PINNED: 1277 case TIMED_WAIT: 1278 return Thread.State.TIMED_WAITING; 1279 case BLOCKING: 1280 case BLOCKED: 1281 return Thread.State.BLOCKED; 1282 case TERMINATED: 1283 return Thread.State.TERMINATED; 1284 default: 1285 throw new InternalError(); 1286 } 1287 } 1288 1289 @Override 1290 boolean alive() { 1291 int s = state; 1292 return (s != NEW && s != TERMINATED); 1293 } 1294 1295 @Override 1296 boolean isTerminated() { 1297 return (state == TERMINATED); 1298 } 1299 1300 @Override 1301 public String toString() { 1302 StringBuilder sb = new StringBuilder("VirtualThread[#"); 1303 sb.append(threadId()); 1304 String name = getName(); 1305 if (!name.isEmpty()) { 1306 sb.append(","); 1307 sb.append(name); 1308 } 1309 sb.append("]/"); 1310 1311 // add the carrier state and thread name when mounted 1312 boolean mounted; 1313 if (Thread.currentThread() == this) { 1314 mounted = appendCarrierInfo(sb); 1315 } else { 1316 disableSuspendAndPreempt(); 1317 try { 1318 synchronized (carrierThreadAccessLock()) { 1319 mounted = appendCarrierInfo(sb); 1320 } 1321 } finally { 1322 enableSuspendAndPreempt(); 1323 } 1324 } 1325 1326 // add virtual thread state when not mounted 1327 if (!mounted) { 1328 String stateAsString = threadState().toString(); 1329 sb.append(stateAsString.toLowerCase(Locale.ROOT)); 1330 } 1331 1332 return sb.toString(); 1333 } 1334 1335 /** 1336 * Appends the carrier state and thread name to the string buffer if mounted. 1337 * @return true if mounted, false if not mounted 1338 */ 1339 private boolean appendCarrierInfo(StringBuilder sb) { 1340 assert Thread.currentThread() == this || Thread.holdsLock(carrierThreadAccessLock()); 1341 Thread carrier = carrierThread; 1342 if (carrier != null) { 1343 String stateAsString = carrier.threadState().toString(); 1344 sb.append(stateAsString.toLowerCase(Locale.ROOT)); 1345 sb.append('@'); 1346 sb.append(carrier.getName()); 1347 return true; 1348 } else { 1349 return false; 1350 } 1351 } 1352 1353 @Override 1354 public int hashCode() { 1355 return (int) threadId(); 1356 } 1357 1358 @Override 1359 public boolean equals(Object obj) { 1360 return obj == this; 1361 } 1362 1363 /** 1364 * Returns the termination object, creating it if needed. 1365 */ 1366 private CountDownLatch getTermination() { 1367 CountDownLatch termination = this.termination; 1368 if (termination == null) { 1369 termination = new CountDownLatch(1); 1370 if (!U.compareAndSetReference(this, TERMINATION, null, termination)) { 1371 termination = this.termination; 1372 } 1373 } 1374 return termination; 1375 } 1376 1377 /** 1378 * Returns the lock object to synchronize on when accessing carrierThread. 1379 * The lock prevents carrierThread from being reset to null during unmount. 1380 */ 1381 private Object carrierThreadAccessLock() { 1382 // return interruptLock as unmount has to coordinate with interrupt 1383 return interruptLock; 1384 } 1385 1386 /** 1387 * Returns a lock object for coordinating timed-wait setup and timeout handling. 1388 */ 1389 private Object timedWaitLock() { 1390 // use this object for now to avoid the overhead of introducing another lock 1391 return runContinuation; 1392 } 1393 1394 /** 1395 * Disallow the current thread be suspended or preempted. 1396 */ 1397 private void disableSuspendAndPreempt() { 1398 notifyJvmtiDisableSuspend(true); 1399 Continuation.pin(); 1400 } 1401 1402 /** 1403 * Allow the current thread be suspended or preempted. 1404 */ 1405 private void enableSuspendAndPreempt() { 1406 Continuation.unpin(); 1407 notifyJvmtiDisableSuspend(false); 1408 } 1409 1410 // -- wrappers for get/set of state, parking permit, and carrier thread -- 1411 1412 private int state() { 1413 return state; // volatile read 1414 } 1415 1416 private void setState(int newValue) { 1417 state = newValue; // volatile write 1418 } 1419 1420 private boolean compareAndSetState(int expectedValue, int newValue) { 1421 return U.compareAndSetInt(this, STATE, expectedValue, newValue); 1422 } 1423 1424 private boolean compareAndSetOnWaitingList(boolean expectedValue, boolean newValue) { 1425 return U.compareAndSetBoolean(this, ON_WAITING_LIST, expectedValue, newValue); 1426 } 1427 1428 private void setParkPermit(boolean newValue) { 1429 if (parkPermit != newValue) { 1430 parkPermit = newValue; 1431 } 1432 } 1433 1434 private boolean getAndSetParkPermit(boolean newValue) { 1435 if (parkPermit != newValue) { 1436 return U.getAndSetBoolean(this, PARK_PERMIT, newValue); 1437 } else { 1438 return newValue; 1439 } 1440 } 1441 1442 private void setCarrierThread(Thread carrier) { 1443 // U.putReferenceRelease(this, CARRIER_THREAD, carrier); 1444 this.carrierThread = carrier; 1445 } 1446 1447 // The following four methods notify the VM when a "transition" starts and ends. 1448 // A "mount transition" embodies the steps to transfer control from a platform 1449 // thread to a virtual thread, changing the thread identity, and starting or 1450 // resuming the virtual thread's continuation on the carrier. 1451 // An "unmount transition" embodies the steps to transfer control from a virtual 1452 // thread to its carrier, suspending the virtual thread's continuation, and 1453 // restoring the thread identity to the platform thread. 1454 // The notifications to the VM are necessary in order to coordinate with functions 1455 // (JVMTI mostly) that disable transitions for one or all virtual threads. Starting 1456 // a transition may block if transitions are disabled. Ending a transition may 1457 // notify a thread that is waiting to disable transitions. The notifications are 1458 // also used to post JVMTI events for virtual thread start and end. 1459 1460 @IntrinsicCandidate 1461 @JvmtiMountTransition 1462 private native void endFirstTransition(); 1463 1464 @IntrinsicCandidate 1465 @JvmtiMountTransition 1466 private native void startFinalTransition(); 1467 1468 @IntrinsicCandidate 1469 @JvmtiMountTransition 1470 private native void startTransition(boolean mount); 1471 1472 @IntrinsicCandidate 1473 @JvmtiMountTransition 1474 private native void endTransition(boolean mount); 1475 1476 @IntrinsicCandidate 1477 private static native void notifyJvmtiDisableSuspend(boolean enter); 1478 1479 private static native void registerNatives(); 1480 static { 1481 registerNatives(); 1482 1483 // ensure VTHREAD_GROUP is created, may be accessed by JVMTI 1484 var group = Thread.virtualThreadGroup(); 1485 1486 // ensure event class is initialized 1487 try { 1488 MethodHandles.lookup().ensureInitialized(VirtualThreadParkEvent.class); 1489 } catch (IllegalAccessException e) { 1490 throw new ExceptionInInitializerError(e); 1491 } 1492 } 1493 1494 /** 1495 * Loads a VirtualThreadScheduler with the given class name. The class must be public 1496 * in an exported package, with public one-arg or no-arg constructor, and be visible 1497 * to the system class loader. 1498 * @param delegate the scheduler that the custom scheduler may delegate to 1499 * @param cn the class name of the custom scheduler 1500 */ 1501 private static VirtualThreadScheduler loadCustomScheduler(VirtualThreadScheduler delegate, String cn) { 1502 VirtualThreadScheduler scheduler; 1503 try { 1504 Class<?> clazz = Class.forName(cn, true, ClassLoader.getSystemClassLoader()); 1505 // 1-arg constructor 1506 try { 1507 Constructor<?> ctor = clazz.getConstructor(VirtualThreadScheduler.class); 1508 return (VirtualThreadScheduler) ctor.newInstance(delegate); 1509 } catch (NoSuchMethodException e) { 1510 // 0-arg constructor 1511 Constructor<?> ctor = clazz.getConstructor(); 1512 scheduler = (VirtualThreadScheduler) ctor.newInstance(); 1513 } 1514 } catch (Exception ex) { 1515 throw new Error(ex); 1516 } 1517 System.err.println("WARNING: Using custom default scheduler, this is an experimental feature!"); 1518 return scheduler; 1519 } 1520 1521 /** 1522 * Creates the built-in ForkJoinPool scheduler. 1523 * @param wrapped true if wrapped by a custom default scheduler 1524 */ 1525 private static BuiltinScheduler createBuiltinScheduler(boolean wrapped) { 1526 int parallelism, maxPoolSize, minRunnable; 1527 String parallelismValue = System.getProperty("jdk.virtualThreadScheduler.parallelism"); 1528 String maxPoolSizeValue = System.getProperty("jdk.virtualThreadScheduler.maxPoolSize"); 1529 String minRunnableValue = System.getProperty("jdk.virtualThreadScheduler.minRunnable"); 1530 if (parallelismValue != null) { 1531 parallelism = Integer.parseInt(parallelismValue); 1532 } else { 1533 parallelism = Runtime.getRuntime().availableProcessors(); 1534 } 1535 if (maxPoolSizeValue != null) { 1536 maxPoolSize = Integer.parseInt(maxPoolSizeValue); 1537 parallelism = Integer.min(parallelism, maxPoolSize); 1538 } else { 1539 maxPoolSize = Integer.max(parallelism, 256); 1540 } 1541 if (minRunnableValue != null) { 1542 minRunnable = Integer.parseInt(minRunnableValue); 1543 } else { 1544 minRunnable = Integer.max(parallelism / 2, 1); 1545 } 1546 return new BuiltinScheduler(parallelism, maxPoolSize, minRunnable, wrapped); 1547 } 1548 1549 /** 1550 * The built-in ForkJoinPool scheduler. 1551 */ 1552 private static class BuiltinScheduler 1553 extends ForkJoinPool implements VirtualThreadScheduler { 1554 1555 BuiltinScheduler(int parallelism, int maxPoolSize, int minRunnable, boolean wrapped) { 1556 ForkJoinWorkerThreadFactory factory = wrapped 1557 ? ForkJoinPool.defaultForkJoinWorkerThreadFactory 1558 : CarrierThread::new; 1559 Thread.UncaughtExceptionHandler handler = (t, e) -> { }; 1560 boolean asyncMode = true; // FIFO 1561 super(parallelism, factory, handler, asyncMode, 1562 0, maxPoolSize, minRunnable, pool -> true, 30, SECONDS); 1563 } 1564 1565 private void adaptAndExecute(Runnable task) { 1566 execute(ForkJoinTask.adapt(task)); 1567 } 1568 1569 @Override 1570 public void onStart(VirtualThreadTask task) { 1571 adaptAndExecute(task); 1572 } 1573 1574 @Override 1575 public void onContinue(VirtualThreadTask task) { 1576 adaptAndExecute(task); 1577 } 1578 1579 /** 1580 * Wraps the scheduler to avoid leaking a direct reference with 1581 * {@link VirtualThreadScheduler#current()}. 1582 */ 1583 VirtualThreadScheduler createExternalView() { 1584 BuiltinScheduler builtin = this; 1585 return new VirtualThreadScheduler() { 1586 private void execute(VirtualThreadTask task) { 1587 var vthread = (VirtualThread) task.thread(); 1588 VirtualThreadScheduler scheduler = vthread.scheduler; 1589 if (scheduler == this || scheduler == DEFAULT_SCHEDULER) { 1590 builtin.adaptAndExecute(task); 1591 } else { 1592 throw new IllegalArgumentException(); 1593 } 1594 } 1595 @Override 1596 public void onStart(VirtualThreadTask task) { 1597 execute(task); 1598 } 1599 @Override 1600 public void onContinue(VirtualThreadTask task) { 1601 execute(task); 1602 } 1603 @Override 1604 public String toString() { 1605 return builtin.toString(); 1606 } 1607 }; 1608 } 1609 } 1610 1611 /** 1612 * Schedule a runnable task to run after a delay. 1613 */ 1614 private Future<?> schedule(Runnable command, long delay, TimeUnit unit) { 1615 if (scheduler == BUILTIN_SCHEDULER) { 1616 return BUILTIN_SCHEDULER.schedule(command, delay, unit); 1617 } else { 1618 return DelayedTaskSchedulers.schedule(command, delay, unit); 1619 } 1620 } 1621 1622 /** 1623 * Supports scheduling a runnable task to run after a delay. It uses a number 1624 * of ScheduledThreadPoolExecutor instances to reduce contention on the delayed 1625 * work queue used. This class is used when using a custom scheduler. 1626 */ 1627 private static class DelayedTaskSchedulers { 1628 private static final ScheduledExecutorService[] INSTANCE = createDelayedTaskSchedulers(); 1629 1630 static Future<?> schedule(Runnable command, long delay, TimeUnit unit) { 1631 long tid = Thread.currentThread().threadId(); 1632 int index = (int) tid & (INSTANCE.length - 1); 1633 return INSTANCE[index].schedule(command, delay, unit); 1634 } 1635 1636 private static ScheduledExecutorService[] createDelayedTaskSchedulers() { 1637 String propName = "jdk.virtualThreadScheduler.timerQueues"; 1638 String propValue = System.getProperty(propName); 1639 int queueCount; 1640 if (propValue != null) { 1641 queueCount = Integer.parseInt(propValue); 1642 if (queueCount != Integer.highestOneBit(queueCount)) { 1643 throw new RuntimeException("Value of " + propName + " must be power of 2"); 1644 } 1645 } else { 1646 int ncpus = Runtime.getRuntime().availableProcessors(); 1647 queueCount = Math.max(Integer.highestOneBit(ncpus / 4), 1); 1648 } 1649 var schedulers = new ScheduledExecutorService[queueCount]; 1650 for (int i = 0; i < queueCount; i++) { 1651 ScheduledThreadPoolExecutor stpe = (ScheduledThreadPoolExecutor) 1652 Executors.newScheduledThreadPool(1, task -> { 1653 Thread t = InnocuousThread.newThread("VirtualThread-unparker", task); 1654 t.setDaemon(true); 1655 return t; 1656 }); 1657 stpe.setRemoveOnCancelPolicy(true); 1658 schedulers[i] = stpe; 1659 } 1660 return schedulers; 1661 } 1662 } 1663 1664 /** 1665 * Schedule virtual threads that are ready to be scheduled after they blocked on 1666 * monitor enter. 1667 */ 1668 private static void unblockVirtualThreads() { 1669 while (true) { 1670 VirtualThread vthread = takeVirtualThreadListToUnblock(); 1671 while (vthread != null) { 1672 assert vthread.onWaitingList; 1673 VirtualThread nextThread = vthread.next; 1674 1675 // remove from list and unblock 1676 vthread.next = null; 1677 boolean changed = vthread.compareAndSetOnWaitingList(true, false); 1678 assert changed; 1679 vthread.unblock(); 1680 1681 vthread = nextThread; 1682 } 1683 } 1684 } 1685 1686 /** 1687 * Retrieves the list of virtual threads that are waiting to be unblocked, waiting 1688 * if necessary until a list of one or more threads becomes available. 1689 */ 1690 private static native VirtualThread takeVirtualThreadListToUnblock(); 1691 1692 static { 1693 var unblocker = InnocuousThread.newThread("VirtualThread-unblocker", 1694 VirtualThread::unblockVirtualThreads); 1695 unblocker.setDaemon(true); 1696 unblocker.start(); 1697 } 1698 } --- EOF ---