forked from openjdk/jdk
-
Notifications
You must be signed in to change notification settings - Fork 20
Expand file tree
/
Copy pathVirtualThread.java
More file actions
1476 lines (1322 loc) · 53.1 KB
/
VirtualThread.java
File metadata and controls
1476 lines (1322 loc) · 53.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*
* Copyright (c) 2018, 2026, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation. Oracle designates this
* particular file as subject to the "Classpath" exception as provided
* by Oracle in the LICENSE file that accompanied this code.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
package java.lang;
import java.util.Locale;
import java.util.Objects;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinPool.ForkJoinWorkerThreadFactory;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import jdk.internal.event.VirtualThreadEndEvent;
import jdk.internal.event.VirtualThreadStartEvent;
import jdk.internal.event.VirtualThreadSubmitFailedEvent;
import jdk.internal.misc.CarrierThread;
import jdk.internal.misc.InnocuousThread;
import jdk.internal.misc.Unsafe;
import jdk.internal.vm.Continuation;
import jdk.internal.vm.ContinuationScope;
import jdk.internal.vm.StackableScope;
import jdk.internal.vm.ThreadContainer;
import jdk.internal.vm.ThreadContainers;
import jdk.internal.vm.annotation.ChangesCurrentThread;
import jdk.internal.vm.annotation.Hidden;
import jdk.internal.vm.annotation.IntrinsicCandidate;
import jdk.internal.vm.annotation.JvmtiHideEvents;
import jdk.internal.vm.annotation.JvmtiMountTransition;
import jdk.internal.vm.annotation.ReservedStackAccess;
import sun.nio.ch.Interruptible;
import static java.util.concurrent.TimeUnit.*;
/**
* A thread that is scheduled by the Java virtual machine rather than the operating system.
*/
final class VirtualThread extends BaseVirtualThread {
private static final Unsafe U = Unsafe.getUnsafe();
private static final ContinuationScope VTHREAD_SCOPE = new ContinuationScope("VirtualThreads");
private static final ForkJoinPool DEFAULT_SCHEDULER = createDefaultScheduler();
private static final long STATE = U.objectFieldOffset(VirtualThread.class, "state");
private static final long PARK_PERMIT = U.objectFieldOffset(VirtualThread.class, "parkPermit");
private static final long CARRIER_THREAD = U.objectFieldOffset(VirtualThread.class, "carrierThread");
private static final long ON_WAITING_LIST = U.objectFieldOffset(VirtualThread.class, "onWaitingList");
// scheduler and continuation
private final Executor scheduler;
private final Continuation cont;
private final Runnable runContinuation;
// virtual thread state, accessed by VM
private volatile int state;
/*
* Virtual thread state transitions:
*
* NEW -> STARTED // Thread.start, schedule to run
* STARTED -> TERMINATED // failed to start
* STARTED -> RUNNING // first run
* RUNNING -> TERMINATED // done
*
* RUNNING -> PARKING // Thread parking with LockSupport.park
* PARKING -> PARKED // cont.yield successful, parked indefinitely
* PARKED -> UNPARKED // unparked, may be scheduled to continue
* UNPARKED -> RUNNING // continue execution after park
*
* PARKING -> RUNNING // cont.yield failed, need to park on carrier
* RUNNING -> PINNED // park on carrier
* PINNED -> RUNNING // unparked, continue execution on same carrier
*
* RUNNING -> TIMED_PARKING // Thread parking with LockSupport.parkNanos
* TIMED_PARKING -> TIMED_PARKED // cont.yield successful, timed-parked
* TIMED_PARKED -> UNPARKED // unparked, may be scheduled to continue
*
* TIMED_PARKING -> RUNNING // cont.yield failed, need to park on carrier
* RUNNING -> TIMED_PINNED // park on carrier
* TIMED_PINNED -> RUNNING // unparked, continue execution on same carrier
*
* RUNNING -> BLOCKING // blocking on monitor enter
* BLOCKING -> BLOCKED // blocked on monitor enter
* BLOCKED -> UNBLOCKED // unblocked, may be scheduled to continue
* UNBLOCKED -> RUNNING // continue execution after blocked on monitor enter
*
* RUNNING -> WAITING // transitional state during wait on monitor
* WAITING -> WAIT // waiting on monitor
* WAIT -> BLOCKED // notified, waiting to be unblocked by monitor owner
* WAIT -> UNBLOCKED // interrupted
*
* RUNNING -> TIMED_WAITING // transition state during timed-waiting on monitor
* TIMED_WAITING -> TIMED_WAIT // timed-waiting on monitor
* TIMED_WAIT -> BLOCKED // notified, waiting to be unblocked by monitor owner
* TIMED_WAIT -> UNBLOCKED // timed-out/interrupted
*
* RUNNING -> YIELDING // Thread.yield
* YIELDING -> YIELDED // cont.yield successful, may be scheduled to continue
* YIELDING -> RUNNING // cont.yield failed
* YIELDED -> RUNNING // continue execution after Thread.yield
*/
private static final int NEW = 0;
private static final int STARTED = 1;
private static final int RUNNING = 2; // runnable-mounted
// untimed and timed parking
private static final int PARKING = 3;
private static final int PARKED = 4; // unmounted
private static final int PINNED = 5; // mounted
private static final int TIMED_PARKING = 6;
private static final int TIMED_PARKED = 7; // unmounted
private static final int TIMED_PINNED = 8; // mounted
private static final int UNPARKED = 9; // unmounted but runnable
// Thread.yield
private static final int YIELDING = 10;
private static final int YIELDED = 11; // unmounted but runnable
// monitor enter
private static final int BLOCKING = 12;
private static final int BLOCKED = 13; // unmounted
private static final int UNBLOCKED = 14; // unmounted but runnable
// monitor wait/timed-wait
private static final int WAITING = 15;
private static final int WAIT = 16; // waiting in Object.wait
private static final int TIMED_WAITING = 17;
private static final int TIMED_WAIT = 18; // waiting in timed-Object.wait
private static final int TERMINATED = 99; // final state
// parking permit made available by LockSupport.unpark
private volatile boolean parkPermit;
// blocking permit made available by unblocker thread when another thread exits monitor
private volatile boolean blockPermit;
// true when on the list of virtual threads waiting to be unblocked
private volatile boolean onWaitingList;
// next virtual thread on the list of virtual threads waiting to be unblocked
private volatile VirtualThread next;
// notified by Object.notify/notifyAll while waiting in Object.wait
private volatile boolean notified;
// true when waiting in Object.wait, false for VM internal uninterruptible Object.wait
private volatile boolean interruptibleWait;
// timed-wait support
private byte timedWaitSeqNo;
// timeout for timed-park and timed-wait, only accessed on current/carrier thread
private long timeout;
// timer task for timed-park and timed-wait, only accessed on current/carrier thread
private Future<?> timeoutTask;
// carrier thread when mounted, accessed by VM
private volatile Thread carrierThread;
// true to notifyAll after this virtual thread terminates
private volatile boolean notifyAllAfterTerminate;
/**
* Returns the default scheduler.
*/
static Executor defaultScheduler() {
return DEFAULT_SCHEDULER;
}
/**
* Returns the continuation scope used for virtual threads.
*/
static ContinuationScope continuationScope() {
return VTHREAD_SCOPE;
}
/**
* Creates a new {@code VirtualThread} to run the given task with the given
* scheduler. If the given scheduler is {@code null} and the current thread
* is a platform thread then the newly created virtual thread will use the
* default scheduler. If given scheduler is {@code null} and the current
* thread is a virtual thread then the current thread's scheduler is used.
*
* @param scheduler the scheduler or null
* @param name thread name
* @param characteristics characteristics
* @param task the task to execute
*/
VirtualThread(Executor scheduler, String name, int characteristics, Runnable task) {
super(name, characteristics, /*bound*/ false);
Objects.requireNonNull(task);
// choose scheduler if not specified
if (scheduler == null) {
Thread parent = Thread.currentThread();
if (parent instanceof VirtualThread vparent) {
scheduler = vparent.scheduler;
} else {
scheduler = DEFAULT_SCHEDULER;
}
}
this.scheduler = scheduler;
this.cont = new VThreadContinuation(this, task);
this.runContinuation = this::runContinuation;
}
/**
* The continuation that a virtual thread executes.
*/
private static class VThreadContinuation extends Continuation {
VThreadContinuation(VirtualThread vthread, Runnable task) {
super(VTHREAD_SCOPE, wrap(vthread, task));
}
@Override
protected void onPinned(Continuation.Pinned reason) {
}
private static Runnable wrap(VirtualThread vthread, Runnable task) {
return new Runnable() {
@Hidden
@JvmtiHideEvents
public void run() {
vthread.endFirstTransition();
try {
vthread.run(task);
} finally {
vthread.startFinalTransition();
}
}
};
}
}
/**
* Runs or continues execution on the current thread. The virtual thread is mounted
* on the current thread before the task runs or continues. It unmounts when the
* task completes or yields.
*/
@ChangesCurrentThread // allow mount/unmount to be inlined
private void runContinuation() {
// the carrier must be a platform thread
if (Thread.currentThread().isVirtual()) {
throw new WrongThreadException();
}
// set state to RUNNING
int initialState = state();
if (initialState == STARTED || initialState == UNPARKED
|| initialState == UNBLOCKED || initialState == YIELDED) {
// newly started or continue after parking/blocking/Thread.yield
if (!compareAndSetState(initialState, RUNNING)) {
return;
}
// consume permit when continuing after parking or blocking. If continue
// after a timed-park or timed-wait then the timeout task is cancelled.
if (initialState == UNPARKED) {
cancelTimeoutTask();
setParkPermit(false);
} else if (initialState == UNBLOCKED) {
cancelTimeoutTask();
blockPermit = false;
}
} else {
// not runnable
return;
}
mount();
try {
cont.run();
} finally {
unmount();
if (cont.isDone()) {
afterDone();
} else {
afterYield();
}
}
}
/**
* Cancel timeout task when continuing after timed-park or timed-wait.
* The timeout task may be executing, or may have already completed.
*/
private void cancelTimeoutTask() {
if (timeoutTask != null) {
timeoutTask.cancel(false);
timeoutTask = null;
}
}
/**
* Submits the given task to the given executor. If the scheduler is a
* ForkJoinPool then the task is first adapted to a ForkJoinTask.
*/
private void submit(Executor executor, Runnable task) {
if (executor instanceof ForkJoinPool pool) {
pool.submit(ForkJoinTask.adapt(task));
} else {
executor.execute(task);
}
}
/**
* Submits the runContinuation task to the scheduler. For the default scheduler,
* and calling it on a worker thread, the task will be pushed to the local queue,
* otherwise it will be pushed to an external submission queue.
* @param scheduler the scheduler
* @param retryOnOOME true to retry indefinitely if OutOfMemoryError is thrown
* @throws RejectedExecutionException
*/
private void submitRunContinuation(Executor scheduler, boolean retryOnOOME) {
boolean done = false;
while (!done) {
try {
// Pin the continuation to prevent the virtual thread from unmounting
// when submitting a task. For the default scheduler this ensures that
// the carrier doesn't change when pushing a task. For other schedulers
// it avoids deadlock that could arise due to carriers and virtual
// threads contending for a lock.
if (currentThread().isVirtual()) {
Continuation.pin();
try {
submit(scheduler, runContinuation);
} finally {
Continuation.unpin();
}
} else {
submit(scheduler, runContinuation);
}
done = true;
} catch (RejectedExecutionException ree) {
submitFailed(ree);
throw ree;
} catch (OutOfMemoryError e) {
if (retryOnOOME) {
U.park(false, 100_000_000); // 100ms
} else {
throw e;
}
}
}
}
/**
* Submits the runContinuation task to the given scheduler as an external submit.
* If OutOfMemoryError is thrown then the submit will be retried until it succeeds.
* @throws RejectedExecutionException
* @see ForkJoinPool#externalSubmit(ForkJoinTask)
*/
private void externalSubmitRunContinuation(ForkJoinPool pool) {
assert Thread.currentThread() instanceof CarrierThread;
try {
pool.externalSubmit(ForkJoinTask.adapt(runContinuation));
} catch (RejectedExecutionException ree) {
submitFailed(ree);
throw ree;
} catch (OutOfMemoryError e) {
submitRunContinuation(pool, true);
}
}
/**
* Submits the runContinuation task to the scheduler. For the default scheduler,
* and calling it on a worker thread, the task will be pushed to the local queue,
* otherwise it will be pushed to an external submission queue.
* If OutOfMemoryError is thrown then the submit will be retried until it succeeds.
* @throws RejectedExecutionException
*/
private void submitRunContinuation() {
submitRunContinuation(scheduler, true);
}
/**
* Lazy submit the runContinuation task if invoked on a carrier thread and its local
* queue is empty. If not empty, or invoked by another thread, then this method works
* like submitRunContinuation and just submits the task to the scheduler.
* If OutOfMemoryError is thrown then the submit will be retried until it succeeds.
* @throws RejectedExecutionException
* @see ForkJoinPool#lazySubmit(ForkJoinTask)
*/
private void lazySubmitRunContinuation() {
if (currentThread() instanceof CarrierThread ct && ct.getQueuedTaskCount() == 0) {
ForkJoinPool pool = ct.getPool();
try {
pool.lazySubmit(ForkJoinTask.adapt(runContinuation));
} catch (RejectedExecutionException ree) {
submitFailed(ree);
throw ree;
} catch (OutOfMemoryError e) {
submitRunContinuation();
}
} else {
submitRunContinuation();
}
}
/**
* Submits the runContinuation task to the scheduler. For the default scheduler, and
* calling it a virtual thread that uses the default scheduler, the task will be
* pushed to an external submission queue. This method may throw OutOfMemoryError.
* @throws RejectedExecutionException
* @throws OutOfMemoryError
*/
private void externalSubmitRunContinuationOrThrow() {
if (scheduler == DEFAULT_SCHEDULER && currentCarrierThread() instanceof CarrierThread ct) {
try {
ct.getPool().externalSubmit(ForkJoinTask.adapt(runContinuation));
} catch (RejectedExecutionException ree) {
submitFailed(ree);
throw ree;
}
} else {
submitRunContinuation(scheduler, false);
}
}
/**
* If enabled, emits a JFR VirtualThreadSubmitFailedEvent.
*/
private void submitFailed(RejectedExecutionException ree) {
var event = new VirtualThreadSubmitFailedEvent();
if (event.isEnabled()) {
event.javaThreadId = threadId();
event.exceptionMessage = ree.getMessage();
event.commit();
}
}
/**
* Runs a task in the context of this virtual thread.
*/
private void run(Runnable task) {
assert Thread.currentThread() == this && state == RUNNING;
// emit JFR event if enabled
if (VirtualThreadStartEvent.isTurnedOn()) {
var event = new VirtualThreadStartEvent();
event.javaThreadId = threadId();
event.commit();
}
Object bindings = Thread.scopedValueBindings();
try {
runWith(bindings, task);
} catch (Throwable exc) {
dispatchUncaughtException(exc);
} finally {
// pop any remaining scopes from the stack, this may block
StackableScope.popAll();
// emit JFR event if enabled
if (VirtualThreadEndEvent.isTurnedOn()) {
var event = new VirtualThreadEndEvent();
event.javaThreadId = threadId();
event.commit();
}
}
}
/**
* Mounts this virtual thread onto the current platform thread. On
* return, the current thread is the virtual thread.
*/
@ChangesCurrentThread
@ReservedStackAccess
private void mount() {
startTransition(/*mount*/true);
// We assume following volatile accesses provide equivalent
// of acquire ordering, otherwise we need U.loadFence() here.
// sets the carrier thread
Thread carrier = Thread.currentCarrierThread();
setCarrierThread(carrier);
// sync up carrier thread interrupted status if needed
if (interrupted) {
carrier.setInterrupt();
} else if (carrier.isInterrupted()) {
synchronized (interruptLock) {
// need to recheck interrupted status
if (!interrupted) {
carrier.clearInterrupt();
}
}
}
// set Thread.currentThread() to return this virtual thread
carrier.setCurrentThread(this);
}
/**
* Unmounts this virtual thread from the carrier. On return, the
* current thread is the current platform thread.
*/
@ChangesCurrentThread
@ReservedStackAccess
private void unmount() {
assert !Thread.holdsLock(interruptLock);
// set Thread.currentThread() to return the platform thread
Thread carrier = this.carrierThread;
carrier.setCurrentThread(carrier);
// break connection to carrier thread, synchronized with interrupt
synchronized (interruptLock) {
setCarrierThread(null);
}
carrier.clearInterrupt();
// We assume previous volatile accesses provide equivalent
// of release ordering, otherwise we need U.storeFence() here.
endTransition(/*mount*/false);
}
/**
* Invokes Continuation.yield, notifying JVMTI (if enabled) to hide frames until
* the continuation continues.
*/
@Hidden
private boolean yieldContinuation() {
startTransition(/*mount*/false);
try {
return Continuation.yield(VTHREAD_SCOPE);
} finally {
endTransition(/*mount*/true);
}
}
/**
* Invoked in the context of the carrier thread after the Continuation yields when
* parking, blocking on monitor enter, Object.wait, or Thread.yield.
*/
private void afterYield() {
assert carrierThread == null;
// re-adjust parallelism if the virtual thread yielded when compensating
if (currentThread() instanceof CarrierThread ct) {
ct.endBlocking();
}
int s = state();
// LockSupport.park/parkNanos
if (s == PARKING || s == TIMED_PARKING) {
int newState;
if (s == PARKING) {
setState(newState = PARKED);
} else {
// schedule unpark
long timeout = this.timeout;
assert timeout > 0;
timeoutTask = schedule(this::parkTimeoutExpired, timeout, NANOSECONDS);
setState(newState = TIMED_PARKED);
}
// Full fence (StoreLoad) to ensure the PARKED/TIMED_PARKED state
// is visible before reading parkPermit (Dekker pattern with
// unpark which writes parkPermit then reads state).
// Note: storeFence is insufficient — on ARM64 it only emits
// LoadStore+StoreStore (dmb ishst), not StoreLoad (dmb ish).
U.fullFence();
// may have been unparked while parking
if (parkPermit && compareAndSetState(newState, UNPARKED)) {
// lazy submit if local queue is empty
lazySubmitRunContinuation();
}
return;
}
// Thread.yield
if (s == YIELDING) {
setState(YIELDED);
// external submit if there are no tasks in the local task queue
if (currentThread() instanceof CarrierThread ct && ct.getQueuedTaskCount() == 0) {
externalSubmitRunContinuation(ct.getPool());
} else {
submitRunContinuation();
}
return;
}
// blocking on monitorenter
if (s == BLOCKING) {
setState(BLOCKED);
// Full fence (StoreLoad) for Dekker pattern with unblock
// which writes blockPermit then reads state.
U.fullFence();
// may have been unblocked while blocking
if (blockPermit && compareAndSetState(BLOCKED, UNBLOCKED)) {
// lazy submit if local queue is empty
lazySubmitRunContinuation();
}
return;
}
// Object.wait
if (s == WAITING || s == TIMED_WAITING) {
int newState;
boolean blocked;
boolean interruptible = interruptibleWait;
if (s == WAITING) {
setState(newState = WAIT);
// Full fence (StoreLoad) for Dekker pattern with notify
// which writes notified then reads state.
U.fullFence();
// may have been notified while in transition
blocked = notified && compareAndSetState(WAIT, BLOCKED);
} else {
// For timed-wait, a timeout task is scheduled to execute. The timeout
// task will change the thread state to UNBLOCKED and submit the thread
// to the scheduler. A sequence number is used to ensure that the timeout
// task only unblocks the thread for this timed-wait. We synchronize with
// the timeout task to coordinate access to the sequence number and to
// ensure the timeout task doesn't execute until the thread has got to
// the TIMED_WAIT state.
long timeout = this.timeout;
assert timeout > 0;
synchronized (timedWaitLock()) {
byte seqNo = ++timedWaitSeqNo;
timeoutTask = schedule(() -> waitTimeoutExpired(seqNo), timeout, MILLISECONDS);
setState(newState = TIMED_WAIT);
// Full fence (StoreLoad) for Dekker pattern with notify
// which writes notified then reads state.
U.fullFence();
// May have been notified while in transition. This must be done while
// holding the monitor to avoid changing the state of a new timed wait call.
blocked = notified && compareAndSetState(TIMED_WAIT, BLOCKED);
}
}
if (blocked) {
// may have been unblocked already
if (blockPermit && compareAndSetState(BLOCKED, UNBLOCKED)) {
lazySubmitRunContinuation();
}
} else {
// may have been interrupted while in transition to wait state
if (interruptible && interrupted && compareAndSetState(newState, UNBLOCKED)) {
lazySubmitRunContinuation();
}
}
return;
}
assert false;
}
/**
* Invoked after the continuation completes.
*/
private void afterDone() {
afterDone(true);
}
/**
* Invoked after the continuation completes (or start failed). Sets the thread
* state to TERMINATED and notifies anyone waiting for the thread to terminate.
*
* @param notifyContainer true if its container should be notified
*/
private void afterDone(boolean notifyContainer) {
assert carrierThread == null;
setState(TERMINATED);
// Full fence (StoreLoad) to ensure the TERMINATED state is
// visible before reading notifyAllAfterTerminate (Dekker pattern
// with beforeJoin which writes notifyAllAfterTerminate then
// reads state). Without this, on ARM64 the volatile write of
// state and the subsequent volatile read can be reordered,
// causing a missed-wakeup where both sides miss each other's
// store.
U.fullFence();
// notifyAll to wakeup any threads waiting for this thread to terminate
if (notifyAllAfterTerminate) {
synchronized (this) {
notifyAll();
}
}
// notify container
if (notifyContainer) {
threadContainer().remove(this);
}
// clear references to thread locals
clearReferences();
}
/**
* Schedules this {@code VirtualThread} to execute.
*
* @throws IllegalStateException if the container is shutdown or closed
* @throws IllegalThreadStateException if the thread has already been started
* @throws RejectedExecutionException if the scheduler cannot accept a task
*/
@Override
void start(ThreadContainer container) {
if (!compareAndSetState(NEW, STARTED)) {
throw new IllegalThreadStateException("Already started");
}
// bind thread to container
assert threadContainer() == null;
setThreadContainer(container);
// start thread
boolean addedToContainer = false;
boolean started = false;
try {
container.add(this); // may throw
addedToContainer = true;
// scoped values may be inherited
inheritScopedValueBindings(container);
// submit task to run thread, using externalSubmit if possible
externalSubmitRunContinuationOrThrow();
started = true;
} finally {
if (!started) {
afterDone(addedToContainer);
}
}
}
@Override
public void start() {
start(ThreadContainers.root());
}
@Override
public void run() {
// do nothing
}
/**
* Invoked by Thread.join before a thread waits for this virtual thread to terminate.
*/
void beforeJoin() {
notifyAllAfterTerminate = true;
}
/**
* Parks until unparked or interrupted. If already unparked then the parking
* permit is consumed and this method completes immediately (meaning it doesn't
* yield). It also completes immediately if the interrupted status is set.
*/
@Override
void park() {
assert Thread.currentThread() == this;
// complete immediately if parking permit available or interrupted
if (getAndSetParkPermit(false) || interrupted)
return;
// park the thread
boolean yielded = false;
setState(PARKING);
try {
yielded = yieldContinuation();
} catch (OutOfMemoryError e) {
// park on carrier
} finally {
assert (Thread.currentThread() == this) && (yielded == (state() == RUNNING));
if (!yielded) {
assert state() == PARKING;
setState(RUNNING);
}
}
// park on the carrier thread when pinned
if (!yielded) {
parkOnCarrierThread(false, 0);
}
}
/**
* Parks up to the given waiting time or until unparked or interrupted.
* If already unparked then the parking permit is consumed and this method
* completes immediately (meaning it doesn't yield). It also completes immediately
* if the interrupted status is set or the waiting time is {@code <= 0}.
*
* @param nanos the maximum number of nanoseconds to wait.
*/
@Override
void parkNanos(long nanos) {
assert Thread.currentThread() == this;
// complete immediately if parking permit available or interrupted
if (getAndSetParkPermit(false) || interrupted)
return;
// park the thread for the waiting time
if (nanos > 0) {
long startTime = System.nanoTime();
// park the thread, afterYield will schedule the thread to unpark
boolean yielded = false;
timeout = nanos;
setState(TIMED_PARKING);
try {
yielded = yieldContinuation();
} catch (OutOfMemoryError e) {
// park on carrier
} finally {
assert (Thread.currentThread() == this) && (yielded == (state() == RUNNING));
if (!yielded) {
assert state() == TIMED_PARKING;
setState(RUNNING);
}
}
// park on carrier thread for remaining time when pinned (or OOME)
if (!yielded) {
long remainingNanos = nanos - (System.nanoTime() - startTime);
parkOnCarrierThread(true, remainingNanos);
}
}
}
/**
* Parks the current carrier thread up to the given waiting time or until
* unparked or interrupted. If the virtual thread is interrupted then the
* interrupted status will be propagated to the carrier thread.
* @param timed true for a timed park, false for untimed
* @param nanos the waiting time in nanoseconds
*/
private void parkOnCarrierThread(boolean timed, long nanos) {
assert state() == RUNNING;
setState(timed ? TIMED_PINNED : PINNED);
try {
if (!parkPermit) {
if (!timed) {
U.park(false, 0);
} else if (nanos > 0) {
U.park(false, nanos);
}
}
} finally {
setState(RUNNING);
}
// consume parking permit
setParkPermit(false);
// JFR jdk.VirtualThreadPinned event
postPinnedEvent("LockSupport.park");
}
/**
* Call into VM when pinned to record a JFR jdk.VirtualThreadPinned event.
* Recording the event in the VM avoids having JFR event recorded in Java
* with the same name, but different ID, to events recorded by the VM.
*/
@Hidden
private static native void postPinnedEvent(String op);
/**
* Re-enables this virtual thread for scheduling. If this virtual thread is parked
* then its task is scheduled to continue, otherwise its next call to {@code park} or
* {@linkplain #parkNanos(long) parkNanos} is guaranteed not to block.
* @param lazySubmit to use lazySubmit if possible
* @throws RejectedExecutionException if the scheduler cannot accept a task
*/
private void unpark(boolean lazySubmit) {
if (!getAndSetParkPermit(true) && currentThread() != this) {
// Full fence (StoreLoad) to ensure parkPermit=true is visible
// before reading state (Dekker pattern with afterYield PARKING
// path which writes state then reads parkPermit).
U.fullFence();
int s = state();
// unparked while parked
if ((s == PARKED || s == TIMED_PARKED) && compareAndSetState(s, UNPARKED)) {
if (lazySubmit) {
lazySubmitRunContinuation();
} else {
submitRunContinuation();
}
return;
}
// unparked while parked when pinned
if (s == PINNED || s == TIMED_PINNED) {
// unpark carrier thread when pinned
disableSuspendAndPreempt();
try {
synchronized (carrierThreadAccessLock()) {
Thread carrier = carrierThread;
if (carrier != null && ((s = state()) == PINNED || s == TIMED_PINNED)) {
U.unpark(carrier);
}
}
} finally {
enableSuspendAndPreempt();
}
return;
}
}
}
@Override
void unpark() {
unpark(false);
}
/**
* Invoked by unblocker thread to unblock this virtual thread.
*/
private void unblock() {
assert !Thread.currentThread().isVirtual();
blockPermit = true;
U.fullFence(); // Full fence (StoreLoad) for Dekker with afterYield BLOCKING path
if (state() == BLOCKED && compareAndSetState(BLOCKED, UNBLOCKED)) {
submitRunContinuation();
}
}
/**
* Invoked by FJP worker thread or STPE thread when park timeout expires.
*/
private void parkTimeoutExpired() {
assert !VirtualThread.currentThread().isVirtual();
unpark(true);
}
/**
* Invoked by FJP worker thread or STPE thread when wait timeout expires.
* If the virtual thread is in timed-wait then this method will unblock the thread
* and submit its task so that it continues and attempts to reenter the monitor.
* This method does nothing if the thread has been woken by notify or interrupt.
*/
private void waitTimeoutExpired(byte seqNo) {
assert !Thread.currentThread().isVirtual();
synchronized (timedWaitLock()) {
if (seqNo != timedWaitSeqNo) {
// this timeout task is for a past timed-wait
return;
}
if (!compareAndSetState(TIMED_WAIT, UNBLOCKED)) {
// already notified (or interrupted)
return;
}
}
lazySubmitRunContinuation();
}
/**
* Attempts to yield the current virtual thread (Thread.yield).
*/
void tryYield() {
assert Thread.currentThread() == this;
setState(YIELDING);
boolean yielded = false;
try {
yielded = yieldContinuation(); // may throw
} finally {
assert (Thread.currentThread() == this) && (yielded == (state() == RUNNING));
if (!yielded) {
assert state() == YIELDING;
setState(RUNNING);
}
}
}
/**