This repository was archived by the owner on Sep 2, 2022. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 157
Expand file tree
/
Copy pathForkJoinPool.java
More file actions
3515 lines (3354 loc) · 150 KB
/
ForkJoinPool.java
File metadata and controls
3515 lines (3354 loc) · 150 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
/*
* 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.
*/
/*
* This file is available under and governed by the GNU General Public
* License version 2 only, as published by the Free Software Foundation.
* However, the following notice accompanied the original version of this
* file:
*
* Written by Doug Lea with assistance from members of JCP JSR-166
* Expert Group and released to the public domain, as explained at
* http://creativecommons.org/publicdomain/zero/1.0/
*/
package java.util.concurrent;
import java.lang.Thread.UncaughtExceptionHandler;
import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.security.AccessController;
import java.security.AccessControlContext;
import java.security.Permission;
import java.security.Permissions;
import java.security.PrivilegedAction;
import java.security.ProtectionDomain;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.function.Predicate;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.Condition;
/**
* An {@link ExecutorService} for running {@link ForkJoinTask}s.
* A {@code ForkJoinPool} provides the entry point for submissions
* from non-{@code ForkJoinTask} clients, as well as management and
* monitoring operations.
*
* <p>A {@code ForkJoinPool} differs from other kinds of {@link
* ExecutorService} mainly by virtue of employing
* <em>work-stealing</em>: all threads in the pool attempt to find and
* execute tasks submitted to the pool and/or created by other active
* tasks (eventually blocking waiting for work if none exist). This
* enables efficient processing when most tasks spawn other subtasks
* (as do most {@code ForkJoinTask}s), as well as when many small
* tasks are submitted to the pool from external clients. Especially
* when setting <em>asyncMode</em> to true in constructors, {@code
* ForkJoinPool}s may also be appropriate for use with event-style
* tasks that are never joined. All worker threads are initialized
* with {@link Thread#isDaemon} set {@code true}.
*
* <p>A static {@link #commonPool()} is available and appropriate for
* most applications. The common pool is used by any ForkJoinTask that
* is not explicitly submitted to a specified pool. Using the common
* pool normally reduces resource usage (its threads are slowly
* reclaimed during periods of non-use, and reinstated upon subsequent
* use).
*
* <p>For applications that require separate or custom pools, a {@code
* ForkJoinPool} may be constructed with a given target parallelism
* level; by default, equal to the number of available processors.
* The pool attempts to maintain enough active (or available) threads
* by dynamically adding, suspending, or resuming internal worker
* threads, even if some tasks are stalled waiting to join others.
* However, no such adjustments are guaranteed in the face of blocked
* I/O or other unmanaged synchronization. The nested {@link
* ManagedBlocker} interface enables extension of the kinds of
* synchronization accommodated. The default policies may be
* overridden using a constructor with parameters corresponding to
* those documented in class {@link ThreadPoolExecutor}.
*
* <p>In addition to execution and lifecycle control methods, this
* class provides status check methods (for example
* {@link #getStealCount}) that are intended to aid in developing,
* tuning, and monitoring fork/join applications. Also, method
* {@link #toString} returns indications of pool state in a
* convenient form for informal monitoring.
*
* <p>As is the case with other ExecutorServices, there are three
* main task execution methods summarized in the following table.
* These are designed to be used primarily by clients not already
* engaged in fork/join computations in the current pool. The main
* forms of these methods accept instances of {@code ForkJoinTask},
* but overloaded forms also allow mixed execution of plain {@code
* Runnable}- or {@code Callable}- based activities as well. However,
* tasks that are already executing in a pool should normally instead
* use the within-computation forms listed in the table unless using
* async event-style tasks that are not usually joined, in which case
* there is little difference among choice of methods.
*
* <table class="plain">
* <caption>Summary of task execution methods</caption>
* <tr>
* <td></td>
* <th scope="col"> Call from non-fork/join clients</th>
* <th scope="col"> Call from within fork/join computations</th>
* </tr>
* <tr>
* <th scope="row" style="text-align:left"> Arrange async execution</th>
* <td> {@link #execute(ForkJoinTask)}</td>
* <td> {@link ForkJoinTask#fork}</td>
* </tr>
* <tr>
* <th scope="row" style="text-align:left"> Await and obtain result</th>
* <td> {@link #invoke(ForkJoinTask)}</td>
* <td> {@link ForkJoinTask#invoke}</td>
* </tr>
* <tr>
* <th scope="row" style="text-align:left"> Arrange exec and obtain Future</th>
* <td> {@link #submit(ForkJoinTask)}</td>
* <td> {@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td>
* </tr>
* </table>
*
* <p>The parameters used to construct the common pool may be controlled by
* setting the following {@linkplain System#getProperty system properties}:
* <ul>
* <li>{@systemProperty java.util.concurrent.ForkJoinPool.common.parallelism}
* - the parallelism level, a non-negative integer
* <li>{@systemProperty java.util.concurrent.ForkJoinPool.common.threadFactory}
* - the class name of a {@link ForkJoinWorkerThreadFactory}.
* The {@linkplain ClassLoader#getSystemClassLoader() system class loader}
* is used to load this class.
* <li>{@systemProperty java.util.concurrent.ForkJoinPool.common.exceptionHandler}
* - the class name of a {@link UncaughtExceptionHandler}.
* The {@linkplain ClassLoader#getSystemClassLoader() system class loader}
* is used to load this class.
* <li>{@systemProperty java.util.concurrent.ForkJoinPool.common.maximumSpares}
* - the maximum number of allowed extra threads to maintain target
* parallelism (default 256).
* </ul>
* If no thread factory is supplied via a system property, then the
* common pool uses a factory that uses the system class loader as the
* {@linkplain Thread#getContextClassLoader() thread context class loader}.
* In addition, if a {@link SecurityManager} is present, then
* the common pool uses a factory supplying threads that have no
* {@link Permissions} enabled.
*
* Upon any error in establishing these settings, default parameters
* are used. It is possible to disable or limit the use of threads in
* the common pool by setting the parallelism property to zero, and/or
* using a factory that may return {@code null}. However doing so may
* cause unjoined tasks to never be executed.
*
* <p><b>Implementation notes:</b> This implementation restricts the
* maximum number of running threads to 32767. Attempts to create
* pools with greater than the maximum number result in
* {@code IllegalArgumentException}.
*
* <p>This implementation rejects submitted tasks (that is, by throwing
* {@link RejectedExecutionException}) only when the pool is shut down
* or internal resources have been exhausted.
*
* @since 1.7
* @author Doug Lea
*/
public class ForkJoinPool extends AbstractExecutorService {
/*
* Implementation Overview
*
* This class and its nested classes provide the main
* functionality and control for a set of worker threads:
* Submissions from non-FJ threads enter into submission queues.
* Workers take these tasks and typically split them into subtasks
* that may be stolen by other workers. Work-stealing based on
* randomized scans generally leads to better throughput than
* "work dealing" in which producers assign tasks to idle threads,
* in part because threads that have finished other tasks before
* the signalled thread wakes up (which can be a long time) can
* take the task instead. Preference rules give first priority to
* processing tasks from their own queues (LIFO or FIFO, depending
* on mode), then to randomized FIFO steals of tasks in other
* queues. This framework began as vehicle for supporting
* tree-structured parallelism using work-stealing. Over time,
* its scalability advantages led to extensions and changes to
* better support more diverse usage contexts. Because most
* internal methods and nested classes are interrelated, their
* main rationale and descriptions are presented here; individual
* methods and nested classes contain only brief comments about
* details.
*
* WorkQueues
* ==========
*
* Most operations occur within work-stealing queues (in nested
* class WorkQueue). These are special forms of Deques that
* support only three of the four possible end-operations -- push,
* pop, and poll (aka steal), under the further constraints that
* push and pop are called only from the owning thread (or, as
* extended here, under a lock), while poll may be called from
* other threads. (If you are unfamiliar with them, you probably
* want to read Herlihy and Shavit's book "The Art of
* Multiprocessor programming", chapter 16 describing these in
* more detail before proceeding.) The main work-stealing queue
* design is roughly similar to those in the papers "Dynamic
* Circular Work-Stealing Deque" by Chase and Lev, SPAA 2005
* (http://research.sun.com/scalable/pubs/index.html) and
* "Idempotent work stealing" by Michael, Saraswat, and Vechev,
* PPoPP 2009 (http://portal.acm.org/citation.cfm?id=1504186).
* The main differences ultimately stem from GC requirements that
* we null out taken slots as soon as we can, to maintain as small
* a footprint as possible even in programs generating huge
* numbers of tasks. To accomplish this, we shift the CAS
* arbitrating pop vs poll (steal) from being on the indices
* ("base" and "top") to the slots themselves.
*
* Adding tasks then takes the form of a classic array push(task)
* in a circular buffer:
* q.array[q.top++ % length] = task;
*
* The actual code needs to null-check and size-check the array,
* uses masking, not mod, for indexing a power-of-two-sized array,
* enforces memory ordering, supports resizing, and possibly
* signals waiting workers to start scanning -- see below.
*
* The pop operation (always performed by owner) is of the form:
* if ((task = getAndSet(q.array, (q.top-1) % length, null)) != null)
* decrement top and return task;
* If this fails, the queue is empty.
*
* The poll operation by another stealer thread is, basically:
* if (CAS nonnull task at q.array[q.base % length] to null)
* increment base and return task;
*
* This may fail due to contention, and may be retried.
* Implementations must ensure a consistent snapshot of the base
* index and the task (by looping or trying elsewhere) before
* trying CAS. There isn't actually a method of this form,
* because failure due to inconsistency or contention is handled
* in different ways in different contexts, normally by first
* trying other queues. (For the most straightforward example, see
* method pollScan.) There are further variants for cases
* requiring inspection of elements before extracting them, so
* must interleave these with variants of this code. Also, a more
* efficient version (nextLocalTask) is used for polls by owners.
* It avoids some overhead because the queue cannot be growing
* during call.
*
* Memory ordering. See "Correct and Efficient Work-Stealing for
* Weak Memory Models" by Le, Pop, Cohen, and Nardelli, PPoPP 2013
* (http://www.di.ens.fr/~zappa/readings/ppopp13.pdf) for an
* analysis of memory ordering requirements in work-stealing
* algorithms similar to the one used here. Inserting and
* extracting tasks in array slots via volatile or atomic accesses
* or explicit fences provides primary synchronization.
*
* Operations on deque elements require reads and writes of both
* indices and slots. When possible, we allow these to occur in
* any order. Because the base and top indices (along with other
* pool or array fields accessed in many methods) only imprecisely
* guide where to extract from, we let accesses other than the
* element getAndSet/CAS/setVolatile appear in any order, using
* plain mode. But we must still preface some methods (mainly
* those that may be accessed externally) with an acquireFence to
* avoid unbounded staleness. This is equivalent to acting as if
* callers use an acquiring read of the reference to the pool or
* queue when invoking the method, even when they do not. We use
* explicit acquiring reads (getSlot) rather than plain array
* access when acquire mode is required but not otherwise ensured
* by context. To reduce stalls by other stealers, we encourage
* timely writes to the base index by immediately following
* updates with a write of a volatile field that must be updated
* anyway, or an Opaque-mode write if there is no such
* opportunity.
*
* Because indices and slot contents cannot always be consistent,
* the emptiness check base == top is only quiescently accurate
* (and so used where this suffices). Otherwise, it may err on the
* side of possibly making the queue appear nonempty when a push,
* pop, or poll have not fully committed, or making it appear
* empty when an update of top or base has not yet been seen.
* Similarly, the check in push for the queue array being full may
* trigger when not completely full, causing a resize earlier than
* required.
*
* Mainly because of these potential inconsistencies among slots
* vs indices, the poll operation, considered individually, is not
* wait-free. One thief cannot successfully continue until another
* in-progress one (or, if previously empty, a push) visibly
* completes. This can stall threads when required to consume
* from a given queue (which may spin). However, in the
* aggregate, we ensure probabilistic non-blockingness at least
* until checking quiescence (which is intrinsically blocking):
* If an attempted steal fails, a scanning thief chooses a
* different victim target to try next. So, in order for one thief
* to progress, it suffices for any in-progress poll or new push
* on any empty queue to complete. The worst cases occur when many
* threads are looking for tasks being produced by a stalled
* producer.
*
* This approach also enables support of a user mode in which
* local task processing is in FIFO, not LIFO order, simply by
* using poll rather than pop. This can be useful in
* message-passing frameworks in which tasks are never joined,
* although with increased contention among task producers and
* consumers.
*
* WorkQueues are also used in a similar way for tasks submitted
* to the pool. We cannot mix these tasks in the same queues used
* by workers. Instead, we randomly associate submission queues
* with submitting threads, using a form of hashing. The
* ThreadLocalRandom probe value serves as a hash code for
* choosing existing queues, and may be randomly repositioned upon
* contention with other submitters. In essence, submitters act
* like workers except that they are restricted to executing local
* tasks that they submitted (or when known, subtasks thereof).
* Insertion of tasks in shared mode requires a lock. We use only
* a simple spinlock (using field "source"), because submitters
* encountering a busy queue move to a different position to use
* or create other queues. They block only when registering new
* queues.
*
* Management
* ==========
*
* The main throughput advantages of work-stealing stem from
* decentralized control -- workers mostly take tasks from
* themselves or each other, at rates that can exceed a billion
* per second. Most non-atomic control is performed by some form
* of scanning across or within queues. The pool itself creates,
* activates (enables scanning for and running tasks),
* deactivates, blocks, and terminates threads, all with minimal
* central information. There are only a few properties that we
* can globally track or maintain, so we pack them into a small
* number of variables, often maintaining atomicity without
* blocking or locking. Nearly all essentially atomic control
* state is held in a few volatile variables that are by far most
* often read (not written) as status and consistency checks. We
* pack as much information into them as we can.
*
* Field "ctl" contains 64 bits holding information needed to
* atomically decide to add, enqueue (on an event queue), and
* dequeue and release workers. To enable this packing, we
* restrict maximum parallelism to (1<<15)-1 (which is far in
* excess of normal operating range) to allow ids, counts, and
* their negations (used for thresholding) to fit into 16bit
* subfields.
*
* Field "mode" holds configuration parameters as well as lifetime
* status, atomically and monotonically setting SHUTDOWN, STOP,
* and finally TERMINATED bits. It is updated only via bitwise
* atomics (getAndBitwiseOr).
*
* Array "queues" holds references to WorkQueues. It is updated
* (only during worker creation and termination) under the
* registrationLock, but is otherwise concurrently readable, and
* accessed directly (although always prefaced by acquireFences or
* other acquiring reads). To simplify index-based operations, the
* array size is always a power of two, and all readers must
* tolerate null slots. Worker queues are at odd indices. Worker
* ids masked with SMASK match their index. Shared (submission)
* queues are at even indices. Grouping them together in this way
* simplifies and speeds up task scanning.
*
* All worker thread creation is on-demand, triggered by task
* submissions, replacement of terminated workers, and/or
* compensation for blocked workers. However, all other support
* code is set up to work with other policies. To ensure that we
* do not hold on to worker or task references that would prevent
* GC, all accesses to workQueues are via indices into the
* queues array (which is one source of some of the messy code
* constructions here). In essence, the queues array serves as
* a weak reference mechanism. Thus for example the stack top
* subfield of ctl stores indices, not references.
*
* Queuing Idle Workers. Unlike HPC work-stealing frameworks, we
* cannot let workers spin indefinitely scanning for tasks when
* none can be found immediately, and we cannot start/resume
* workers unless there appear to be tasks available. On the
* other hand, we must quickly prod them into action when new
* tasks are submitted or generated. These latencies are mainly a
* function of JVM park/unpark (and underlying OS) performance,
* which can be slow and variable. In many usages, ramp-up time
* is the main limiting factor in overall performance, which is
* compounded at program start-up by JIT compilation and
* allocation. On the other hand, throughput degrades when too
* many threads poll for too few tasks.
*
* The "ctl" field atomically maintains total and "released"
* worker counts, plus the head of the available worker queue
* (actually stack, represented by the lower 32bit subfield of
* ctl). Released workers are those known to be scanning for
* and/or running tasks. Unreleased ("available") workers are
* recorded in the ctl stack. These workers are made available for
* signalling by enqueuing in ctl (see method awaitWork). The
* "queue" is a form of Treiber stack. This is ideal for
* activating threads in most-recently used order, and improves
* performance and locality, outweighing the disadvantages of
* being prone to contention and inability to release a worker
* unless it is topmost on stack. The top stack state holds the
* value of the "phase" field of the worker: its index and status,
* plus a version counter that, in addition to the count subfields
* (also serving as version stamps) provide protection against
* Treiber stack ABA effects.
*
* Creating workers. To create a worker, we pre-increment counts
* (serving as a reservation), and attempt to construct a
* ForkJoinWorkerThread via its factory. On starting, the new
* thread first invokes registerWorker, where it constructs a
* WorkQueue and is assigned an index in the queues array
* (expanding the array if necessary). Upon any exception across
* these steps, or null return from factory, deregisterWorker
* adjusts counts and records accordingly. If a null return, the
* pool continues running with fewer than the target number
* workers. If exceptional, the exception is propagated, generally
* to some external caller.
*
* WorkQueue field "phase" is used by both workers and the pool to
* manage and track whether a worker is UNSIGNALLED (possibly
* blocked waiting for a signal). When a worker is enqueued its
* phase field is set negative. Note that phase field updates lag
* queue CAS releases; seeing a negative phase does not guarantee
* that the worker is available. When queued, the lower 16 bits of
* its phase must hold its pool index. So we place the index there
* upon initialization and never modify these bits.
*
* The ctl field also serves as the basis for memory
* synchronization surrounding activation. This uses a more
* efficient version of a Dekker-like rule that task producers and
* consumers sync with each other by both writing/CASing ctl (even
* if to its current value). However, rather than CASing ctl to
* its current value in the common case where no action is
* required, we reduce write contention by ensuring that
* signalWork invocations are prefaced with a full-volatile memory
* access (which is usually needed anyway).
*
* Signalling. Signals (in signalWork) cause new or reactivated
* workers to scan for tasks. Method signalWork and its callers
* try to approximate the unattainable goal of having the right
* number of workers activated for the tasks at hand, but must err
* on the side of too many workers vs too few to avoid stalls. If
* computations are purely tree structured, it suffices for every
* worker to activate another when it pushes a task into an empty
* queue, resulting in O(log(#threads)) steps to full activation.
* If instead, tasks come in serially from only a single producer,
* each worker taking its first (since the last quiescence) task
* from a queue should signal another if there are more tasks in
* that queue. This is equivalent to, but generally faster than,
* arranging the stealer take two tasks, re-pushing one on its own
* queue, and signalling (because its queue is empty), also
* resulting in logarithmic full activation time. Because we don't
* know about usage patterns (or most commonly, mixtures), we use
* both approaches. We approximate the second rule by arranging
* that workers in scan() do not repeat signals when repeatedly
* taking tasks from any given queue, by remembering the previous
* one. There are narrow windows in which both rules may apply,
* leading to duplicate or unnecessary signals. Despite such
* limitations, these rules usually avoid slowdowns that otherwise
* occur when too many workers contend to take too few tasks, or
* when producers waste most of their time resignalling. However,
* contention and overhead effects may still occur during ramp-up,
* ramp-down, and small computations involving only a few workers.
*
* Scanning. Method scan performs top-level scanning for (and
* execution of) tasks. Scans by different workers and/or at
* different times are unlikely to poll queues in the same
* order. Each scan traverses and tries to poll from each queue in
* a pseudorandom permutation order by starting at a random index,
* and using a constant cyclically exhaustive stride; restarting
* upon contention. (Non-top-level scans; for example in
* helpJoin, use simpler linear probes because they do not
* systematically contend with top-level scans.) The pseudorandom
* generator need not have high-quality statistical properties in
* the long term. We use Marsaglia XorShifts, seeded with the Weyl
* sequence from ThreadLocalRandom probes, which are cheap and
* suffice. Scans do not otherwise explicitly take into account
* core affinities, loads, cache localities, etc, However, they do
* exploit temporal locality (which usually approximates these) by
* preferring to re-poll from the same queue after a successful
* poll before trying others (see method topLevelExec). This
* reduces fairness, which is partially counteracted by using a
* one-shot form of poll (tryPoll) that may lose to other workers.
*
* Deactivation. Method scan returns a sentinel when no tasks are
* found, leading to deactivation (see awaitWork). The count
* fields in ctl allow accurate discovery of quiescent states
* (i.e., when all workers are idle) after deactivation. However,
* this may also race with new (external) submissions, so a
* recheck is also needed to determine quiescence. Upon apparently
* triggering quiescence, awaitWork re-scans and self-signals if
* it may have missed a signal. In other cases, a missed signal
* may transiently lower parallelism because deactivation does not
* necessarily mean that there is no more work, only that that
* there were no tasks not taken by other workers. But more
* signals are generated (see above) to eventually reactivate if
* needed.
*
* Trimming workers. To release resources after periods of lack of
* use, a worker starting to wait when the pool is quiescent will
* time out and terminate if the pool has remained quiescent for
* period given by field keepAlive.
*
* Shutdown and Termination. A call to shutdownNow invokes
* tryTerminate to atomically set a mode bit. The calling thread,
* as well as every other worker thereafter terminating, helps
* terminate others by cancelling their unprocessed tasks, and
* waking them up. Calls to non-abrupt shutdown() preface this by
* checking isQuiescent before triggering the "STOP" phase of
* termination. To conform to ExecutorService invoke, invokeAll,
* and invokeAny specs, we must track pool status while waiting,
* and interrupt interruptible callers on termination (see
* ForkJoinTask.joinForPoolInvoke etc).
*
* Joining Tasks
* =============
*
* Normally, the first option when joining a task that is not done
* is to try to unfork it from local queue and run it. Otherwise,
* any of several actions may be taken when one worker is waiting
* to join a task stolen (or always held) by another. Because we
* are multiplexing many tasks on to a pool of workers, we can't
* always just let them block (as in Thread.join). We also cannot
* just reassign the joiner's run-time stack with another and
* replace it later, which would be a form of "continuation", that
* even if possible is not necessarily a good idea since we may
* need both an unblocked task and its continuation to progress.
* Instead we combine two tactics:
*
* Helping: Arranging for the joiner to execute some task that it
* could be running if the steal had not occurred.
*
* Compensating: Unless there are already enough live threads,
* method tryCompensate() may create or re-activate a spare
* thread to compensate for blocked joiners until they unblock.
*
* A third form (implemented via tryRemove) amounts to helping a
* hypothetical compensator: If we can readily tell that a
* possible action of a compensator is to steal and execute the
* task being joined, the joining thread can do so directly,
* without the need for a compensation thread; although with a
* (rare) possibility of reduced parallelism because of a
* transient gap in the queue array.
*
* Other intermediate forms available for specific task types (for
* example helpAsyncBlocker) often avoid or postpone the need for
* blocking or compensation.
*
* The ManagedBlocker extension API can't use helping so relies
* only on compensation in method awaitBlocker.
*
* The algorithm in helpJoin entails a form of "linear helping".
* Each worker records (in field "source") the id of the queue
* from which it last stole a task. The scan in method helpJoin
* uses these markers to try to find a worker to help (i.e., steal
* back a task from and execute it) that could hasten completion
* of the actively joined task. Thus, the joiner executes a task
* that would be on its own local deque if the to-be-joined task
* had not been stolen. This is a conservative variant of the
* approach described in Wagner & Calder "Leapfrogging: a portable
* technique for implementing efficient futures" SIGPLAN Notices,
* 1993 (http://portal.acm.org/citation.cfm?id=155354). It differs
* mainly in that we only record queue ids, not full dependency
* links. This requires a linear scan of the queues array to
* locate stealers, but isolates cost to when it is needed, rather
* than adding to per-task overhead. Also, searches are limited to
* direct and at most two levels of indirect stealers, after which
* there are rapidly diminishing returns on increased overhead.
* Searches can fail to locate stealers when stalls delay
* recording sources. Further, even when accurately identified,
* stealers might not ever produce a task that the joiner can in
* turn help with. So, compensation is tried upon failure to find
* tasks to run.
*
* Joining CountedCompleters (see helpComplete) differs from (and
* is generally more efficient than) other cases because task
* eligibility is determined by checking completion chains rather
* than tracking stealers.
*
* Joining under timeouts (ForkJoinTask timed get) uses a
* constrained mixture of helping and compensating in part because
* pools (actually, only the common pool) may not have any
* available threads: If the pool is saturated (all available
* workers are busy), the caller tries to remove and otherwise
* help; else it blocks under compensation so that it may time out
* independently of any tasks.
*
* Compensation does not by default aim to keep exactly the target
* parallelism number of unblocked threads running at any given
* time. Some previous versions of this class employed immediate
* compensations for any blocked join. However, in practice, the
* vast majority of blockages are transient byproducts of GC and
* other JVM or OS activities that are made worse by replacement
* when they cause longer-term oversubscription. Rather than
* impose arbitrary policies, we allow users to override the
* default of only adding threads upon apparent starvation. The
* compensation mechanism may also be bounded. Bounds for the
* commonPool (see COMMON_MAX_SPARES) better enable JVMs to cope
* with programming errors and abuse before running out of
* resources to do so.
*
* Common Pool
* ===========
*
* The static common pool always exists after static
* initialization. Since it (or any other created pool) need
* never be used, we minimize initial construction overhead and
* footprint to the setup of about a dozen fields.
*
* When external threads submit to the common pool, they can
* perform subtask processing (see helpComplete and related
* methods) upon joins. This caller-helps policy makes it
* sensible to set common pool parallelism level to one (or more)
* less than the total number of available cores, or even zero for
* pure caller-runs. We do not need to record whether external
* submissions are to the common pool -- if not, external help
* methods return quickly. These submitters would otherwise be
* blocked waiting for completion, so the extra effort (with
* liberally sprinkled task status checks) in inapplicable cases
* amounts to an odd form of limited spin-wait before blocking in
* ForkJoinTask.join.
*
* Guarantees for common pool parallelism zero are limited to
* tasks that are joined by their callers in a tree-structured
* fashion or use CountedCompleters (as is true for jdk
* parallelStreams). Support infiltrates several methods,
* including those that retry helping steps until we are sure that
* none apply if there are no workers.
*
* As a more appropriate default in managed environments, unless
* overridden by system properties, we use workers of subclass
* InnocuousForkJoinWorkerThread when there is a SecurityManager
* present. These workers have no permissions set, do not belong
* to any user-defined ThreadGroup, and erase all ThreadLocals
* after executing any top-level task. The associated mechanics
* may be JVM-dependent and must access particular Thread class
* fields to achieve this effect.
*
* Interrupt handling
* ==================
*
* The framework is designed to manage task cancellation
* (ForkJoinTask.cancel) independently from the interrupt status
* of threads running tasks. (See the public ForkJoinTask
* documentation for rationale.) Interrupts are issued only in
* tryTerminate, when workers should be terminating and tasks
* should be cancelled anyway. Interrupts are cleared only when
* necessary to ensure that calls to LockSupport.park do not loop
* indefinitely (park returns immediately if the current thread is
* interrupted). If so, interruption is reinstated after blocking
* if status could be visible during the scope of any task. For
* cases in which task bodies are specified or desired to
* interrupt upon cancellation, ForkJoinTask.cancel can be
* overridden to do so (as is done for invoke{Any,All}).
*
* Memory placement
* ================
*
* Performance can be very sensitive to placement of instances of
* ForkJoinPool and WorkQueues and their queue arrays. To reduce
* false-sharing impact, the @Contended annotation isolates the
* ForkJoinPool.ctl field as well as the most heavily written
* WorkQueue fields. These mainly reduce cache traffic by scanners.
* WorkQueue arrays are presized large enough to avoid resizing
* (which transiently reduces throughput) in most tree-like
* computations, although not in some streaming usages. Initial
* sizes are not large enough to avoid secondary contention
* effects (especially for GC cardmarks) when queues are placed
* near each other in memory. This is common, but has different
* impact in different collectors and remains incompletely
* addressed.
*
* Style notes
* ===========
*
* Memory ordering relies mainly on atomic operations (CAS,
* getAndSet, getAndAdd) along with explicit fences. This can be
* awkward and ugly, but also reflects the need to control
* outcomes across the unusual cases that arise in very racy code
* with very few invariants. All fields are read into locals
* before use, and null-checked if they are references, even if
* they can never be null under current usages. Array accesses
* using masked indices include checks (that are always true) that
* the array length is non-zero to avoid compilers inserting more
* expensive traps. This is usually done in a "C"-like style of
* listing declarations at the heads of methods or blocks, and
* using inline assignments on first encounter. Nearly all
* explicit checks lead to bypass/return, not exception throws,
* because they may legitimately arise during shutdown.
*
* There is a lot of representation-level coupling among classes
* ForkJoinPool, ForkJoinWorkerThread, and ForkJoinTask. The
* fields of WorkQueue maintain data structures managed by
* ForkJoinPool, so are directly accessed. There is little point
* trying to reduce this, since any associated future changes in
* representations will need to be accompanied by algorithmic
* changes anyway. Several methods intrinsically sprawl because
* they must accumulate sets of consistent reads of fields held in
* local variables. Some others are artificially broken up to
* reduce producer/consumer imbalances due to dynamic compilation.
* There are also other coding oddities (including several
* unnecessary-looking hoisted null checks) that help some methods
* perform reasonably even when interpreted (not compiled).
*
* The order of declarations in this file is (with a few exceptions):
* (1) Static utility functions
* (2) Nested (static) classes
* (3) Static fields
* (4) Fields, along with constants used when unpacking some of them
* (5) Internal control methods
* (6) Callbacks and other support for ForkJoinTask methods
* (7) Exported methods
* (8) Static block initializing statics in minimally dependent order
*
* Revision notes
* ==============
*
* The main sources of differences of January 2020 ForkJoin
* classes from previous version are:
*
* * ForkJoinTask now uses field "aux" to support blocking joins
* and/or record exceptions, replacing reliance on builtin
* monitors and side tables.
* * Scans probe slots (vs compare indices), along with related
* changes that reduce performance differences across most
* garbage collectors, and reduce contention.
* * Refactoring for better integration of special task types and
* other capabilities that had been incrementally tacked on. Plus
* many minor reworkings to improve consistency.
*/
// Static utilities
/**
* If there is a security manager, makes sure caller has
* permission to modify threads.
*/
private static void checkPermission() {
@SuppressWarnings("removal")
SecurityManager security = System.getSecurityManager();
if (security != null)
security.checkPermission(modifyThreadPermission);
}
@SuppressWarnings("removal")
static AccessControlContext contextWithPermissions(Permission ... perms) {
Permissions permissions = new Permissions();
for (Permission perm : perms)
permissions.add(perm);
return new AccessControlContext(
new ProtectionDomain[] { new ProtectionDomain(null, permissions) });
}
// Nested classes
/**
* Factory for creating new {@link ForkJoinWorkerThread}s.
* A {@code ForkJoinWorkerThreadFactory} must be defined and used
* for {@code ForkJoinWorkerThread} subclasses that extend base
* functionality or initialize threads with different contexts.
*/
public static interface ForkJoinWorkerThreadFactory {
/**
* Returns a new worker thread operating in the given pool.
* Returning null or throwing an exception may result in tasks
* never being executed. If this method throws an exception,
* it is relayed to the caller of the method (for example
* {@code execute}) causing attempted thread creation. If this
* method returns null or throws an exception, it is not
* retried until the next attempted creation (for example
* another call to {@code execute}).
*
* @param pool the pool this thread works in
* @return the new worker thread, or {@code null} if the request
* to create a thread is rejected
* @throws NullPointerException if the pool is null
*/
public ForkJoinWorkerThread newThread(ForkJoinPool pool);
}
/**
* Default ForkJoinWorkerThreadFactory implementation; creates a
* new ForkJoinWorkerThread using the system class loader as the
* thread context class loader.
*/
static final class DefaultForkJoinWorkerThreadFactory
implements ForkJoinWorkerThreadFactory {
// ACC for access to the factory
@SuppressWarnings("removal")
private static final AccessControlContext ACC = contextWithPermissions(
new RuntimePermission("getClassLoader"),
new RuntimePermission("setContextClassLoader"));
@SuppressWarnings("removal")
public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
return AccessController.doPrivileged(
new PrivilegedAction<>() {
public ForkJoinWorkerThread run() {
return new ForkJoinWorkerThread(null, pool, true, false);
}},
ACC);
}
}
/**
* Factory for CommonPool unless overridden by System property.
* Creates InnocuousForkJoinWorkerThreads if a security manager is
* present at time of invocation. Support requires that we break
* quite a lot of encapsulation (some via helper methods in
* ThreadLocalRandom) to access and set Thread fields.
*/
static final class DefaultCommonPoolForkJoinWorkerThreadFactory
implements ForkJoinWorkerThreadFactory {
@SuppressWarnings("removal")
private static final AccessControlContext ACC = contextWithPermissions(
modifyThreadPermission,
new RuntimePermission("enableContextClassLoaderOverride"),
new RuntimePermission("modifyThreadGroup"),
new RuntimePermission("getClassLoader"),
new RuntimePermission("setContextClassLoader"));
@SuppressWarnings("removal")
public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
return AccessController.doPrivileged(
new PrivilegedAction<>() {
public ForkJoinWorkerThread run() {
return System.getSecurityManager() == null ?
new ForkJoinWorkerThread(null, pool, true, true):
new ForkJoinWorkerThread.
InnocuousForkJoinWorkerThread(pool); }},
ACC);
}
}
// Constants shared across ForkJoinPool and WorkQueue
// Bounds
static final int SWIDTH = 16; // width of short
static final int SMASK = 0xffff; // short bits == max index
static final int MAX_CAP = 0x7fff; // max #workers - 1
// Masks and units for WorkQueue.phase and ctl sp subfield
static final int UNSIGNALLED = 1 << 31; // must be negative
static final int SS_SEQ = 1 << 16; // version count
// Mode bits and sentinels, some also used in WorkQueue fields
static final int FIFO = 1 << 16; // fifo queue or access mode
static final int SRC = 1 << 17; // set for valid queue ids
static final int INNOCUOUS = 1 << 18; // set for Innocuous workers
static final int QUIET = 1 << 19; // quiescing phase or source
static final int SHUTDOWN = 1 << 24;
static final int TERMINATED = 1 << 25;
static final int STOP = 1 << 31; // must be negative
static final int UNCOMPENSATE = 1 << 16; // tryCompensate return
/**
* Initial capacity of work-stealing queue array. Must be a power
* of two, at least 2. See above.
*/
static final int INITIAL_QUEUE_CAPACITY = 1 << 8;
/**
* Queues supporting work-stealing as well as external task
* submission. See above for descriptions and algorithms.
*/
static final class WorkQueue {
volatile int phase; // versioned, negative if inactive
int stackPred; // pool stack (ctl) predecessor link
int config; // index, mode, ORed with SRC after init
int base; // index of next slot for poll
ForkJoinTask<?>[] array; // the queued tasks; power of 2 size
final ForkJoinWorkerThread owner; // owning thread or null if shared
// segregate fields frequently updated but not read by scans or steals
@jdk.internal.vm.annotation.Contended("w")
int top; // index of next slot for push
@jdk.internal.vm.annotation.Contended("w")
volatile int source; // source queue id, lock, or sentinel
@jdk.internal.vm.annotation.Contended("w")
int nsteals; // number of steals from other queues
// Support for atomic operations
private static final VarHandle QA; // for array slots
private static final VarHandle SOURCE;
private static final VarHandle BASE;
static final ForkJoinTask<?> getSlot(ForkJoinTask<?>[] a, int i) {
return (ForkJoinTask<?>)QA.getAcquire(a, i);
}
static final ForkJoinTask<?> getAndClearSlot(ForkJoinTask<?>[] a,
int i) {
return (ForkJoinTask<?>)QA.getAndSet(a, i, null);
}
static final void setSlotVolatile(ForkJoinTask<?>[] a, int i,
ForkJoinTask<?> v) {
QA.setVolatile(a, i, v);
}
static final boolean casSlotToNull(ForkJoinTask<?>[] a, int i,
ForkJoinTask<?> c) {
return QA.compareAndSet(a, i, c, null);
}
final boolean tryLock() {
return SOURCE.compareAndSet(this, 0, 1);
}
final void setBaseOpaque(int b) {
BASE.setOpaque(this, b);
}
/**
* Constructor used by ForkJoinWorkerThreads. Most fields
* are initialized upon thread start, in pool.registerWorker.
*/
WorkQueue(ForkJoinWorkerThread owner, boolean isInnocuous) {
this.config = (isInnocuous) ? INNOCUOUS : 0;
this.owner = owner;
}
/**
* Constructor used for external queues.
*/
WorkQueue(int config) {
array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY];
this.config = config;
owner = null;
phase = -1;
}
/**
* Returns an exportable index (used by ForkJoinWorkerThread).
*/
final int getPoolIndex() {
return (config & 0xffff) >>> 1; // ignore odd/even tag bit
}
/**
* Returns the approximate number of tasks in the queue.
*/
final int queueSize() {
VarHandle.acquireFence(); // ensure fresh reads by external callers
int n = top - base;
return (n < 0) ? 0 : n; // ignore transient negative
}
/**
* Provides a more conservative estimate of whether this queue
* has any tasks than does queueSize.
*/
final boolean isEmpty() {
return !((source != 0 && owner == null) || top - base > 0);
}
/**
* Pushes a task. Call only by owner in unshared queues.
*
* @param task the task. Caller must ensure non-null.
* @param pool (no-op if null)
* @throws RejectedExecutionException if array cannot be resized
*/
final void push(ForkJoinTask<?> task, ForkJoinPool pool) {
ForkJoinTask<?>[] a = array;
int s = top++, d = s - base, cap, m; // skip insert if disabled
if (a != null && pool != null && (cap = a.length) > 0) {
setSlotVolatile(a, (m = cap - 1) & s, task);
if (d == m)
growArray();
if (d == m || a[m & (s - 1)] == null)
pool.signalWork(); // signal if was empty or resized
}
}
/**
* Pushes task to a shared queue with lock already held, and unlocks.
*
* @return true if caller should signal work
*/
final boolean lockedPush(ForkJoinTask<?> task) {
ForkJoinTask<?>[] a = array;
int s = top++, d = s - base, cap, m;
if (a != null && (cap = a.length) > 0) {
a[(m = cap - 1) & s] = task;
if (d == m)
growArray();
source = 0; // unlock
if (d == m || a[m & (s - 1)] == null)
return true;
}
return false;
}