1 /*
2 * Copyright (c) 2018, 2021, 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.MethodHandle;
28 import java.lang.invoke.MethodHandles;
29 import java.lang.invoke.MethodType;
30 import java.lang.invoke.VarHandle;
31 import java.security.AccessControlContext;
32 import java.security.AccessController;
33 import java.security.PrivilegedAction;
34 import java.security.PrivilegedExceptionAction;
35 import java.security.ProtectionDomain;
36 import java.util.Locale;
37 import java.util.Objects;
38 import java.util.concurrent.CountDownLatch;
39 import java.util.concurrent.Executor;
40 import java.util.concurrent.Executors;
41 import java.util.concurrent.ForkJoinPool;
42 import java.util.concurrent.ForkJoinPool.ForkJoinWorkerThreadFactory;
43 import java.util.concurrent.ForkJoinWorkerThread;
44 import java.util.concurrent.Future;
45 import java.util.concurrent.RejectedExecutionException;
46 import java.util.concurrent.ScheduledExecutorService;
47 import java.util.concurrent.ScheduledThreadPoolExecutor;
48 import jdk.internal.event.VirtualThreadEndEvent;
49 import jdk.internal.event.VirtualThreadPinnedEvent;
50 import jdk.internal.event.VirtualThreadStartEvent;
51 import jdk.internal.event.VirtualThreadSubmitFailedEvent;
52 import jdk.internal.misc.InnocuousThread;
53 import jdk.internal.misc.Unsafe;
54 import jdk.internal.vm.Continuation;
55 import jdk.internal.vm.ContinuationScope;
56 import jdk.internal.vm.ThreadContainer;
57 import jdk.internal.vm.ThreadContainers;
58 import jdk.internal.vm.annotation.ChangesCurrentThread;
59 import jdk.internal.vm.annotation.JvmtiMountTransition;
60 import sun.nio.ch.Interruptible;
61 import sun.security.action.GetPropertyAction;
62 import static java.util.concurrent.TimeUnit.*;
63
64 /**
65 * A thread that is scheduled by the Java virtual machine rather than the operating
66 * system.
67 */
68 class VirtualThread extends Thread {
69 private static final Unsafe U = Unsafe.getUnsafe();
70 private static final ContinuationScope VTHREAD_SCOPE = new ContinuationScope("VirtualThreads");
71 private static final ForkJoinPool DEFAULT_SCHEDULER = createDefaultScheduler();
72 private static final ScheduledExecutorService UNPARKER = createDelayedTaskScheduler();
73 private static final int TRACE_PINNING_MODE = tracePinningMode();
74
75 private static final VarHandle STATE;
76 private static final VarHandle PARK_PERMIT;
77 private static final VarHandle CARRIER_THREAD;
78 private static final VarHandle TERMINATION;
79 private static final VarHandle NEXT_UNPARKER_ID;
80 static {
81 try {
82 MethodHandles.Lookup l = MethodHandles.lookup();
83 STATE = l.findVarHandle(VirtualThread.class, "state", int.class);
84 PARK_PERMIT = l.findVarHandle(VirtualThread.class, "parkPermit", boolean.class);
85 CARRIER_THREAD = l.findVarHandle(VirtualThread.class, "carrierThread", Thread.class);
86 TERMINATION = l.findVarHandle(VirtualThread.class, "termination", CountDownLatch.class);
87 NEXT_UNPARKER_ID = l.findStaticVarHandle(VirtualThread.class, "nextUnparkerId", int.class);
88 } catch (Exception e) {
89 throw new InternalError(e);
90 }
91 }
92
93 // scheduler and continuation
94 private final Executor scheduler;
95 private final Continuation cont;
96 private final Runnable runContinuation;
97
98 // virtual thread state, accessed by VM so need to coordinate changes
99 private volatile int state;
100
101 private static final int NEW = 0;
102 private static final int STARTED = 1;
103 private static final int RUNNABLE = 2; // runnable-unmounted
104 private static final int RUNNING = 3; // runnable-mounted
105 private static final int PARKING = 4;
106 private static final int PARKED = 5; // unmounted
107 private static final int PINNED = 6; // mounted
108 private static final int YIELDING = 7; // Thread.yield
109 private static final int TERMINATED = 99; // final state
110
111 // can be suspended from scheduling when parked
112 private static final int SUSPENDED = 1 << 8;
113 private static final int PARKED_SUSPENDED = (PARKED | SUSPENDED);
114
115 // parking permit, may eventually be merged into state
116 private volatile boolean parkPermit;
117
118 // carrier thread when mounted
119 private volatile Thread carrierThread;
120
121 // termination object when joining, created lazily if needed
122 private volatile CountDownLatch termination;
123
124 /**
125 * Returns the continuation scope used for virtual threads.
126 */
127 static ContinuationScope continuationScope() {
128 return VTHREAD_SCOPE;
129 }
130
131 /**
132 * Creates a new {@code VirtualThread} to run the given task with the given
133 * scheduler. If the given scheduler is {@code null} and the current thread
134 * is a platform thread then the newly created virtual thread will use the
135 * default scheduler. If given scheduler is {@code null} and the current
136 * thread is a virtual thread then the current thread's scheduler is used.
137 *
138 * @param scheduler the scheduler or null.
139 * @param name thread name
140 * @param characteristics characteristics
141 * @param task the task to execute
142 */
143 VirtualThread(Executor scheduler, String name, int characteristics, Runnable task) {
144 super(name, characteristics);
145 Objects.requireNonNull(task);
146
147 // choose scheduler if not specified
148 if (scheduler == null) {
149 Thread parent = Thread.currentThread();
150 if (parent instanceof VirtualThread vparent) {
151 scheduler = vparent.scheduler;
152 } else {
153 scheduler = DEFAULT_SCHEDULER;
154 }
155 }
156
157 this.scheduler = scheduler;
158 this.cont = new VThreadContinuation(this, task);
159 this.runContinuation = this::runContinuation;
160 }
161
162 /**
163 * The continuation that a virtual thread executes.
164 */
165 private static class VThreadContinuation extends Continuation {
166 VThreadContinuation(VirtualThread vthread, Runnable task) {
167 super(VTHREAD_SCOPE, () -> vthread.run(task));
168 }
169 @Override
170 protected void onPinned(Continuation.Pinned reason) {
171 if (TRACE_PINNING_MODE > 0) {
172 boolean printAll = (TRACE_PINNING_MODE == 1);
173 PinnedThreadPrinter.printStackTrace(System.out, printAll);
174 }
175 }
176 }
177
178 /**
179 * Runs or continues execution of the continuation on the current thread.
180 */
181 private void runContinuation() {
182 // the carrier thread should be a platform thread
183 if (Thread.currentThread().isVirtual()) {
184 throw new IllegalCallerException();
185 }
186
187 // set state to RUNNING
188 boolean firstRun;
189 int initialState = state();
190 if (initialState == STARTED && compareAndSetState(STARTED, RUNNING)) {
191 // first run
192 firstRun = true;
193 } else if (initialState == RUNNABLE && compareAndSetState(RUNNABLE, RUNNING)) {
194 // consume parking permit
195 setParkPermit(false);
196 firstRun = false;
197 } else {
198 throw new IllegalStateException();
199 }
200
201 // notify JVMTI before mount
202 if (notifyJvmtiEvents) notifyJvmtiMountBegin(firstRun);
203
204 try {
205 cont.run();
206 } finally {
207 if (cont.isDone()) {
208 afterTerminate(/*executed*/ true);
209 } else {
210 afterYield();
211 }
212 }
213 }
214
215 /**
216 * Submits the runContinuation task to the scheduler.
217 * In the case of the default scheduler, and the current carrier thread is one
218 * of FJP worker threads, then the task is queued to the current thread's queue.
219 * In that case, the parameter {@code signalOnEmpty} indicates if workers are
220 * signalled when its queue is empty.
221 * @throws RejectedExecutionException
222 */
223 private void submitRunContinuationAndSignal(boolean signalOnEmpty) {
224 try {
225 if (scheduler == DEFAULT_SCHEDULER) {
226 ForkJoinPools.externalExecuteTask(DEFAULT_SCHEDULER,runContinuation, signalOnEmpty);
227 } else {
228 scheduler.execute(runContinuation);
229 }
230 } catch (RejectedExecutionException ree) {
231 // record event
232 var event = new VirtualThreadSubmitFailedEvent();
233 if (event.isEnabled()) {
234 event.javaThreadId = getId();
235 event.exceptionMessage = ree.getMessage();
236 event.commit();
237 }
238 throw ree;
239 }
240 }
241
242 /**
243 * Submits the runContinuation task to the scheduler.
244 * @throws RejectedExecutionException
245 */
246 private void submitRunContinuation() {
247 submitRunContinuationAndSignal(true);
248 }
249
250 /**
251 * Runs a task in the context of this virtual thread. The virtual thread is
252 * mounted on the current (carrier) thread before the task runs. It unmounts
253 * from its carrier thread when the task completes.
254 *
255 * When enabled, JVMTI must be notified from this method.
256 */
257 @ChangesCurrentThread
258 private void run(Runnable task) {
259 assert state == RUNNING;
260 boolean notifyJvmti = notifyJvmtiEvents;
261
262 // first mount
263 mount();
264 if (notifyJvmti) notifyJvmtiMountEnd(true);
265
266 try {
267 task.run();
268 } catch (Throwable exc) {
269 dispatchUncaughtException(exc);
270 } finally {
271
272 // last unmount
273 if (notifyJvmti) notifyJvmtiUnmountBegin(true);
274 unmount();
275 if (notifyJvmti) notifyJvmtiUnmountEnd(true);
276
277 // final state
278 setState(TERMINATED);
279 }
280 }
281
282 /**
283 * Mounts this virtual thread onto the current carrier thread.
284 */
285 @ChangesCurrentThread
286 private void mount() {
287 //assert this.carrierThread == null;
288
289 // sets the carrier thread
290 Thread carrier = Thread.currentCarrierThread();
291 CARRIER_THREAD.setRelease(this, carrier);
292
293 // sync up carrier thread interrupt status if needed
294 if (interrupted) {
295 carrier.setInterrupt();
296 } else if (carrier.isInterrupted()) {
297 synchronized (interruptLock) {
298 if (!interrupted) {
299 carrier.clearInterrupt();
300 }
301 }
302 }
303
304 // set Thread.currentThread() to return this virtual thread
305 carrier.setCurrentThread(this);
306 }
307
308 /**
309 * Unmounts this virtual thread from the current carrier thread.
310 */
311 @ChangesCurrentThread
312 private void unmount() {
313 //assert this.carrierThread == Thread.currentCarrierThread();
314
315 // set Thread.currentThread() to return the carrier thread
316 Thread carrier = this.carrierThread;
317 carrier.setCurrentThread(carrier);
318
319 // break connection to carrier thread
320 synchronized (interruptLock) { // synchronize with interrupt
321 CARRIER_THREAD.setRelease(this, null);
322 }
323 carrier.clearInterrupt();
324 }
325
326 /**
327 * Unmounts this virtual thread, invokes Continuation.yield, and re-mounts the
328 * thread when continued. When enabled, JVMTI must be notified from this method.
329 * @return true if the yield was successful
330 */
331 @ChangesCurrentThread
332 private boolean yieldContinuation() {
333 boolean notifyJvmti = notifyJvmtiEvents;
334
335 // unmount
336 if (notifyJvmti) notifyJvmtiUnmountBegin(false);
337 unmount();
338 try {
339 return Continuation.yield(VTHREAD_SCOPE);
340 } finally {
341 // re-mount
342 mount();
343 if (notifyJvmti) notifyJvmtiMountEnd(false);
344 }
345 }
346
347 /**
348 * Invoked after the continuation yields. If parking then it sets the state
349 * and also re-submits the task to continue if unparked while parking.
350 * If yielding due to Thread.yield then it just submits the task to continue.
351 */
352 private void afterYield() {
353 int s = state();
354 assert (s == PARKING || s == YIELDING) && (carrierThread == null);
355
356 if (s == PARKING) {
357 setState(PARKED);
358
359 // notify JVMTI that unmount has completed, thread is parked
360 if (notifyJvmtiEvents) notifyJvmtiUnmountEnd(false);
361
362 // may have been unparked while parking
363 if (parkPermit && compareAndSetState(PARKED, RUNNABLE)) {
364 // re-submit to continue on this carrier thread if possible
365 submitRunContinuationAndSignal(false);
366 }
367 } else if (s == YIELDING) { // Thread.yield
368 setState(RUNNABLE);
369
370 // notify JVMTI that unmount has completed, thread is runnable
371 if (notifyJvmtiEvents) notifyJvmtiUnmountEnd(false);
372
373 // re-submit to continue on this carrier thread if possible
374 submitRunContinuationAndSignal(false);
375 }
376 }
377
378 /**
379 * Invoked after the thread terminates (or start failed). This method
380 * notifies anyone waiting for the thread to terminate.
381 *
382 * @param executed true if the thread executed, false if it failed to start
383 */
384 private void afterTerminate(boolean executed) {
385 assert (state() == TERMINATED) && (carrierThread == null);
386
387 // notify anyone waiting for this virtual thread to terminate
388 CountDownLatch termination = this.termination;
389 if (termination != null) {
390 assert termination.getCount() == 1;
391 termination.countDown();
392 }
393
394 if (VirtualThreadEndEvent.isTurnedOn()) {
395 var event = new VirtualThreadEndEvent();
396 event.javaThreadId = getId();
397 event.commit();
398 }
399
400 if (executed) {
401 // notify container if thread executed
402 threadContainer().onExit(this);
403
404 // clear references to thread locals
405 clearReferences();
406 }
407 }
408
409 /**
410 * Parks on the carrier thread up to the given waiting time or until unparked
411 * or interrupted. If the virtual thread is interrupted then the interrupt
412 * status will be propagated to the carrier thread so it will wakeup.
413 * @param timed true for a timed park, false for untimed
414 * @param nanos the waiting time in nanoseconds
415 */
416 private void parkOnCarrierThread(boolean timed, long nanos) {
417 assert state() == PARKING;
418
419 var pinnedEvent = new VirtualThreadPinnedEvent();
420 pinnedEvent.begin();
421
422 setState(PINNED);
423 try {
424 if (!parkPermit) {
425 if (!timed) {
426 U.park(false, 0);
427 } else if (nanos > 0) {
428 U.park(false, nanos);
429 }
430 }
431 } finally {
432 setState(RUNNING);
433 }
434
435 // consume parking permit
436 setParkPermit(false);
437
438 // commit event if enabled
439 if (pinnedEvent.isEnabled()) {
440 pinnedEvent.commit();
441 }
442 }
443
444 /**
445 * Schedules this {@code VirtualThread} to execute.
446 *
447 * @throws IllegalStateException if the container is shutdown or closed
448 * @throws IllegalThreadStateException if the thread has already been started
449 * @throws RejectedExecutionException if the scheduler cannot accept a task
450 */
451 @Override
452 void start(ThreadContainer container) {
453 if (!compareAndSetState(NEW, STARTED)) {
454 throw new IllegalThreadStateException("Already started");
455 }
456
457 boolean started = false;
458 container.onStart(this); // may throw
459 try {
460 // inherit scope locals from structured container
461 Object bindings = container.scopeLocalBindings();
462 if (bindings != null) {
463 if (Thread.currentThread().scopeLocalBindings != bindings) {
464 throw new IllegalStateException("Scope local bindings have changed");
465 }
466 this.scopeLocalBindings = (ScopeLocal.Snapshot) bindings;
467 }
468
469 // bind thread to container
470 setThreadContainer(container);
471 if (VirtualThreadStartEvent.isTurnedOn()) {
472 var event = new VirtualThreadStartEvent();
473 event.javaThreadId = getId();
474 event.commit();
475 }
476
477 // submit task to run thread
478 submitRunContinuation();
479 started = true;
480 } finally {
481 if (!started) {
482 setState(TERMINATED);
483 container.onExit(this);
484 afterTerminate(/*executed*/ false);
485 }
486 }
487 }
488
489 @Override
490 public void start() {
491 start(ThreadContainers.root());
492 }
493
494 /**
495 * Parks the current virtual thread until it is unparked or interrupted.
496 * If already unparked then the parking permit is consumed and this method
497 * completes immediately (meaning it doesn't yield). It also completes
498 * immediately if the interrupt status is set.
499 */
500 void park() {
501 assert Thread.currentThread() == this;
502
503 // complete immediately if parking permit available or interrupted
504 if (getAndSetParkPermit(false) || interrupted)
505 return;
506
507 // park the thread
508 setState(PARKING);
509 try {
510 if (!yieldContinuation()) {
511 // park on the carrier thread when pinned
512 parkOnCarrierThread(false, 0);
513 }
514 } finally {
515 assert (Thread.currentThread() == this) && (state() == RUNNING);
516 }
517 }
518
519 /**
520 * Parks the current virtual thread up to the given waiting time or until it
521 * is unparked or interrupted. If already unparked then the parking permit is
522 * consumed and this method completes immediately (meaning it doesn't yield).
523 * It also completes immediately if the interrupt status is set or the waiting
524 * time is {@code <= 0}.
525 *
526 * @param nanos the maximum number of nanoseconds to wait.
527 */
528 void parkNanos(long nanos) {
529 assert Thread.currentThread() == this;
530
531 // complete immediately if parking permit available or interrupted
532 if (getAndSetParkPermit(false) || interrupted)
533 return;
534
535 // park the thread for the waiting time
536 if (nanos > 0) {
537 long startTime = System.nanoTime();
538
539 boolean yielded;
540 Future<?> unparker = scheduleUnpark(nanos);
541 setState(PARKING);
542 try {
543 yielded = yieldContinuation();
544 } finally {
545 assert (Thread.currentThread() == this)
546 && (state() == RUNNING || state() == PARKING);
547 cancel(unparker);
548 }
549
550 // park on the carrier thread for remaining time when pinned
551 if (!yielded) {
552 long deadline = startTime + nanos;
553 if (deadline < 0L)
554 deadline = Long.MAX_VALUE;
555 parkOnCarrierThread(true, deadline - System.nanoTime());
556 }
557 }
558 }
559
560 /**
561 * Schedules this thread to be unparked after the given delay.
562 */
563 @ChangesCurrentThread
564 private Future<?> scheduleUnpark(long nanos) {
565 //assert Thread.currentThread() == this;
566 Thread carrier = this.carrierThread;
567 // need to switch to carrier thread to avoid nested parking
568 carrier.setCurrentThread(carrier);
569 try {
570 return UNPARKER.schedule(() -> unpark(), nanos, NANOSECONDS);
571 } finally {
572 carrier.setCurrentThread(this);
573 }
574 }
575
576 /**
577 * Cancels a task if it has not completed.
578 */
579 @ChangesCurrentThread
580 private void cancel(Future<?> future) {
581 //assert Thread.currentThread() == this;
582 if (!future.isDone()) {
583 Thread carrier = this.carrierThread;
584 // need to switch to carrier thread to avoid nested parking
585 carrier.setCurrentThread(carrier);
586 try {
587 future.cancel(false);
588 } finally {
589 carrier.setCurrentThread(this);
590 }
591 }
592 }
593
594 /**
595 * Re-enables this virtual thread for scheduling. If the virtual thread was
596 * {@link #park() parked} then it will be unblocked, otherwise its next call
597 * to {@code park} or {@linkplain #parkNanos(long) parkNanos} is guaranteed
598 * not to block.
599 * @throws RejectedExecutionException if the scheduler cannot accept a task
600 */
601 @ChangesCurrentThread
602 void unpark() {
603 Thread currentThread = Thread.currentThread();
604 if (!getAndSetParkPermit(true) && currentThread != this) {
605 int s = state();
606 if (s == PARKED && compareAndSetState(PARKED, RUNNABLE)) {
607 if (currentThread instanceof VirtualThread vthread) {
608 Thread carrier = vthread.carrierThread;
609 carrier.setCurrentThread(carrier);
610 try {
611 submitRunContinuation();
612 } finally {
613 carrier.setCurrentThread(vthread);
614 }
615 } else {
616 submitRunContinuation();
617 }
618 } else if (s == PINNED) {
619 // unpark carrier thread when pinned.
620 synchronized (interruptLock) {
621 Thread carrier = carrierThread;
622 if (carrier != null && state() == PINNED) {
623 U.unpark(carrier);
624 }
625 }
626 }
627 }
628 }
629
630 /**
631 * Attempts to yield (Thread.yield).
632 */
633 void tryYield() {
634 assert Thread.currentThread() == this;
635 setState(YIELDING);
636 try {
637 yieldContinuation();
638 } finally {
639 assert Thread.currentThread() == this;
640 if (state() != RUNNING) {
641 assert state() == YIELDING;
642 setState(RUNNING);
643 }
644 }
645 }
646
647 /**
648 * Waits up to {@code nanos} nanoseconds for this virtual thread to terminate.
649 * A timeout of {@code 0} means to wait forever.
650 *
651 * @throws InterruptedException if interrupted while waiting
652 * @return true if the thread has terminated
653 */
654 boolean joinNanos(long nanos) throws InterruptedException {
655 if (state() == TERMINATED)
656 return true;
657
658 // ensure termination object exists, then re-check state
659 CountDownLatch termination = getTermination();
660 if (state() == TERMINATED)
661 return true;
662
663 // wait for virtual thread to terminate
664 if (nanos == 0) {
665 termination.await();
666 } else {
667 boolean terminated = termination.await(nanos, NANOSECONDS);
668 if (!terminated) {
669 // waiting time elapsed
670 return false;
671 }
672 }
673 assert state() == TERMINATED;
674 return true;
675 }
676
677 @Override
678 @SuppressWarnings("removal")
679 public void interrupt() {
680 if (Thread.currentThread() != this) {
681 checkAccess();
682 synchronized (interruptLock) {
683 interrupted = true;
684 Interruptible b = nioBlocker;
685 if (b != null) {
686 b.interrupt(this);
687 }
688
689 // interrupt carrier thread
690 Thread carrier = carrierThread;
691 if (carrier != null) carrier.setInterrupt();
692 }
693 } else {
694 interrupted = true;
695 carrierThread.setInterrupt();
696 }
697 unpark();
698 }
699
700 @Override
701 public boolean isInterrupted() {
702 return interrupted;
703 }
704
705 @Override
706 boolean getAndClearInterrupt() {
707 assert Thread.currentThread() == this;
708 synchronized (interruptLock) {
709 boolean oldValue = interrupted;
710 if (oldValue)
711 interrupted = false;
712 carrierThread.clearInterrupt();
713 return oldValue;
714 }
715 }
716
717 /**
718 * Sleep the current thread for the given sleep time (in nanoseconds)
719 *
720 * @implNote This implementation parks the thread for the given sleeping time
721 * and will therefore be observed in PARKED state during the sleep. Parking
722 * will consume the parking permit so this method makes available the parking
723 * permit after the sleep. This will observed as a spurious, but benign, wakeup
724 * when the thread subsequently attempts to park.
725 *
726 * @throws InterruptedException if interrupted while sleeping
727 */
728 void sleepNanos(long nanos) throws InterruptedException {
729 assert Thread.currentThread() == this;
730 if (nanos >= 0) {
731 if (getAndClearInterrupt())
732 throw new InterruptedException();
733 if (nanos == 0) {
734 tryYield();
735 } else {
736 // park for the sleep time
737 try {
738 long remainingNanos = nanos;
739 long startNanos = System.nanoTime();
740 while (remainingNanos > 0) {
741 parkNanos(remainingNanos);
742 if (getAndClearInterrupt()) {
743 throw new InterruptedException();
744 }
745 remainingNanos = nanos - (System.nanoTime() - startNanos);
746 }
747 } finally {
748 // may have been unparked while sleeping
749 setParkPermit(true);
750 }
751 }
752 }
753 }
754
755 @Override
756 public Thread.State getState() {
757 switch (state()) {
758 case NEW:
759 return Thread.State.NEW;
760 case STARTED:
761 case RUNNABLE:
762 // runnable, not mounted
763 return Thread.State.RUNNABLE;
764 case RUNNING:
765 // if mounted then return state of carrier thread
766 synchronized (interruptLock) {
767 Thread carrierThread = this.carrierThread;
768 if (carrierThread != null) {
769 return carrierThread.threadState();
770 }
771 }
772 // runnable, mounted
773 return Thread.State.RUNNABLE;
774 case PARKING:
775 case YIELDING:
776 // runnable, mounted, not yet waiting
777 return Thread.State.RUNNABLE;
778 case PARKED:
779 case PARKED_SUSPENDED:
780 case PINNED:
781 return Thread.State.WAITING;
782 case TERMINATED:
783 return Thread.State.TERMINATED;
784 default:
785 throw new InternalError();
786 }
787 }
788
789 @Override
790 StackTraceElement[] asyncGetStackTrace() {
791 StackTraceElement[] stackTrace;
792 do {
793 stackTrace = (carrierThread != null)
794 ? super.asyncGetStackTrace() // mounted
795 : tryGetStackTrace(); // unmounted
796 if (stackTrace == null) {
797 Thread.yield();
798 }
799 } while (stackTrace == null);
800 return stackTrace;
801 }
802
803 /**
804 * Returns the stack trace for this virtual thread if it newly created,
805 * started, parked, or terminated. Returns null if the thread is in
806 * another state.
807 */
808 private StackTraceElement[] tryGetStackTrace() {
809 if (compareAndSetState(PARKED, PARKED_SUSPENDED)) {
810 try {
811 return cont.getStackTrace();
812 } finally {
813 assert state() == PARKED_SUSPENDED;
814 setState(PARKED);
815
816 // may have been unparked while suspended
817 if (parkPermit && compareAndSetState(PARKED, RUNNABLE)) {
818 try {
819 submitRunContinuation();
820 } catch (RejectedExecutionException ignore) { }
821 }
822 }
823 } else {
824 int s = state();
825 if (s == NEW || s == STARTED || s == TERMINATED) {
826 return new StackTraceElement[0]; // empty stack
827 } else {
828 return null;
829 }
830 }
831 }
832
833 @Override
834 public String toString() {
835 StringBuilder sb = new StringBuilder("VirtualThread[#");
836 sb.append(getId());
837 String name = getName();
838 if (!name.isEmpty() && !name.equals("<unnamed>")) {
839 sb.append(",");
840 sb.append(name);
841 }
842 sb.append("]/");
843 Thread carrier = carrierThread;
844 if (carrier != null) {
845 // include the carrier thread state and name when mounted
846 synchronized (interruptLock) {
847 carrier = carrierThread;
848 if (carrier != null) {
849 String stateAsString = carrier.threadState().toString();
850 sb.append(stateAsString.toLowerCase(Locale.ROOT));
851 sb.append('@');
852 sb.append(carrier.getName());
853 }
854 }
855 }
856 // include virtual thread state when not mounted
857 if (carrier == null) {
858 String stateAsString = getState().toString();
859 sb.append(stateAsString.toLowerCase(Locale.ROOT));
860 }
861 return sb.toString();
862 }
863
864 @Override
865 public int hashCode() {
866 return (int) getId();
867 }
868
869 @Override
870 public boolean equals(Object obj) {
871 return obj == this;
872 }
873
874 /**
875 * Returns the termination object, creating it if needed.
876 */
877 private CountDownLatch getTermination() {
878 CountDownLatch termination = this.termination;
879 if (termination == null) {
880 termination = new CountDownLatch(1);
881 if (!TERMINATION.compareAndSet(this, null, termination)) {
882 termination = this.termination;
883 }
884 }
885 return termination;
886 }
887
888 // -- wrappers for VarHandle methods --
889
890 private int state() {
891 return state; // volatile read
892 }
893
894 private void setState(int newValue) {
895 state = newValue; // volatile write
896 }
897
898 private void setReleaseState(int newValue) {
899 STATE.setRelease(this, newValue);
900 }
901
902 private boolean compareAndSetState(int expectedValue, int newValue) {
903 return STATE.compareAndSet(this, expectedValue, newValue);
904 }
905
906 private void setParkPermit(boolean newValue) {
907 if (parkPermit != newValue) {
908 parkPermit = newValue;
909 }
910 }
911
912 private boolean getAndSetParkPermit(boolean newValue) {
913 if (parkPermit != newValue) {
914 return (boolean) PARK_PERMIT.getAndSet(this, newValue);
915 } else {
916 return newValue;
917 }
918 }
919
920 // -- JVM TI support --
921
922 private static volatile boolean notifyJvmtiEvents; // set by VM
923
924 @JvmtiMountTransition
925 private native void notifyJvmtiMountBegin(boolean firstMount);
926
927 @JvmtiMountTransition
928 private native void notifyJvmtiMountEnd(boolean firstMount);
929
930 @JvmtiMountTransition
931 private native void notifyJvmtiUnmountBegin(boolean lastUnmount);
932
933 @JvmtiMountTransition
934 private native void notifyJvmtiUnmountEnd(boolean lastUnmount);
935
936 private static native void registerNatives();
937 static {
938 registerNatives();
939 }
940
941 /**
942 * Creates the default scheduler.
943 */
944 @SuppressWarnings("removal")
945 private static ForkJoinPool createDefaultScheduler() {
946 ForkJoinWorkerThreadFactory factory = pool -> {
947 PrivilegedAction<ForkJoinWorkerThread> pa = () -> new CarrierThread(pool);
948 return AccessController.doPrivileged(pa);
949 };
950 PrivilegedAction<ForkJoinPool> pa = () -> {
951 int parallelism, maxPoolSize, minRunnable;
952 String parallelismValue = System.getProperty("jdk.defaultScheduler.parallelism");
953 String maxPoolSizeValue = System.getProperty("jdk.defaultScheduler.maxPoolSize");
954 String minRunnableValue = System.getProperty("jdk.defaultScheduler.minRunnable");
955 if (parallelismValue != null) {
956 parallelism = Integer.parseInt(parallelismValue);
957 } else {
958 parallelism = Runtime.getRuntime().availableProcessors();
959 }
960 if (maxPoolSizeValue != null) {
961 maxPoolSize = Integer.parseInt(maxPoolSizeValue);
962 parallelism = Integer.min(parallelism, maxPoolSize);
963 } else {
964 maxPoolSize = Integer.max(parallelism, 256);
965 }
966 if (minRunnableValue != null) {
967 minRunnable = Integer.parseInt(minRunnableValue);
968 } else {
969 minRunnable = Integer.max(parallelism / 2, 1);
970 }
971 Thread.UncaughtExceptionHandler handler = (t, e) -> { };
972 boolean asyncMode = true; // FIFO
973 return new ForkJoinPool(parallelism, factory, handler, asyncMode,
974 0, maxPoolSize, minRunnable, pool -> true, 30, SECONDS);
975 };
976 return AccessController.doPrivileged(pa);
977 }
978
979 /**
980 * Defines static methods to invoke non-public ForkJoinPool methods.
981 */
982 private static class ForkJoinPools {
983 static final MethodHandle externalExecuteTask;
984 static {
985 try {
986 PrivilegedExceptionAction<MethodHandles.Lookup> pa = () ->
987 MethodHandles.privateLookupIn(ForkJoinPool.class, MethodHandles.lookup());
988 @SuppressWarnings("removal")
989 MethodHandles.Lookup l = AccessController.doPrivileged(pa);
990 MethodType methodType = MethodType.methodType(void.class, Runnable.class, boolean.class);
991 externalExecuteTask = l.findVirtual(ForkJoinPool.class, "externalExecuteTask", methodType);
992 } catch (Exception e) {
993 throw new InternalError(e);
994 }
995 }
996 /**
997 * Invokes the non-public ForkJoinPool.externalExecuteTask method to
998 * submit the task to the current carrier thread's work queue.
999 */
1000 static void externalExecuteTask(ForkJoinPool pool, Runnable task, boolean signalOnEmpty) {
1001 try {
1002 externalExecuteTask.invoke(pool, task, signalOnEmpty);
1003 } catch (RuntimeException | Error e) {
1004 throw e;
1005 } catch (Throwable e) {
1006 throw new InternalError(e);
1007 }
1008 }
1009 }
1010
1011 /**
1012 * A thread in the ForkJoinPool created by the default scheduler.
1013 */
1014 private static class CarrierThread extends ForkJoinWorkerThread {
1015 private static final ThreadGroup CARRIER_THREADGROUP = carrierThreadGroup();
1016 @SuppressWarnings("removal")
1017 private static final AccessControlContext INNOCUOUS_ACC = innocuousACC();
1018
1019 private static final long CONTEXTCLASSLOADER;
1020 private static final long INHERITABLETHREADLOCALS;
1021 private static final long INHERITEDACCESSCONTROLCONTEXT;
1022
1023 CarrierThread(ForkJoinPool pool) {
1024 super(CARRIER_THREADGROUP, pool);
1025 U.putReference(this, CONTEXTCLASSLOADER, ClassLoader.getSystemClassLoader());
1026 U.putReference(this, INHERITABLETHREADLOCALS, null);
1027 U.putReferenceRelease(this, INHERITEDACCESSCONTROLCONTEXT, INNOCUOUS_ACC);
1028 }
1029
1030 @Override
1031 public void setUncaughtExceptionHandler(UncaughtExceptionHandler ueh) { }
1032
1033 @Override
1034 public void setContextClassLoader(ClassLoader cl) {
1035 throw new SecurityException("setContextClassLoader");
1036 }
1037
1038 /**
1039 * The thread group for the carrier threads.
1040 */
1041 @SuppressWarnings("removal")
1042 private static final ThreadGroup carrierThreadGroup() {
1043 return AccessController.doPrivileged(new PrivilegedAction<ThreadGroup>() {
1044 public ThreadGroup run() {
1045 ThreadGroup group = Thread.currentCarrierThread().getThreadGroup();
1046 for (ThreadGroup p; (p = group.getParent()) != null; )
1047 group = p;
1048 @SuppressWarnings("deprecation")
1049 var carrierThreadsGroup = new ThreadGroup(group, "CarrierThreads");
1050 return carrierThreadsGroup;
1051 }
1052 });
1053 }
1054
1055 /**
1056 * Return an AccessControlContext that doesn't support any permissions.
1057 */
1058 @SuppressWarnings("removal")
1059 private static AccessControlContext innocuousACC() {
1060 return new AccessControlContext(new ProtectionDomain[] {
1061 new ProtectionDomain(null, null)
1062 });
1063 }
1064
1065 static {
1066 CONTEXTCLASSLOADER = U.objectFieldOffset(Thread.class,
1067 "contextClassLoader");
1068 INHERITABLETHREADLOCALS = U.objectFieldOffset(Thread.class,
1069 "inheritableThreadLocals");
1070 INHERITEDACCESSCONTROLCONTEXT = U.objectFieldOffset(Thread.class,
1071 "inheritedAccessControlContext");
1072 }
1073 }
1074
1075 /**
1076 * Creates the ScheduledThreadPoolExecutor used for timed unpark.
1077 */
1078 private static ScheduledExecutorService createDelayedTaskScheduler() {
1079 String propValue = GetPropertyAction.privilegedGetProperty("jdk.unparker.maxPoolSize");
1080 int poolSize;
1081 if (propValue != null) {
1082 poolSize = Integer.parseInt(propValue);
1083 } else {
1084 poolSize = 1;
1085 }
1086 ScheduledThreadPoolExecutor stpe = (ScheduledThreadPoolExecutor)
1087 Executors.newScheduledThreadPool(poolSize, task -> {
1088 String name = "VirtualThread-unparker";
1089 if (poolSize > 1)
1090 name += "-" + nextUnparkerId();
1091 var thread = InnocuousThread.newThread(name, task);
1092 return thread;
1093 });
1094 stpe.setRemoveOnCancelPolicy(true);
1095 return stpe;
1096 }
1097
1098 private static volatile int nextUnparkerId;
1099 private static int nextUnparkerId() {
1100 return (int) NEXT_UNPARKER_ID.getAndAdd(1) + 1;
1101 }
1102
1103 /**
1104 * Reads the value of the jdk.tracePinnedThreads property to determine if stack
1105 * traces should be printed when a carrier thread is pinned when a virtual thread
1106 * attempts to park.
1107 */
1108 private static int tracePinningMode() {
1109 String propValue = GetPropertyAction.privilegedGetProperty("jdk.tracePinnedThreads");
1110 if (propValue != null) {
1111 if (propValue.length() == 0 || "full".equalsIgnoreCase(propValue))
1112 return 1;
1113 if ("short".equalsIgnoreCase(propValue))
1114 return 2;
1115 }
1116 return 0;
1117 }
1118 }