1 /*
2  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
3  *
4  * This code is free software; you can redistribute it and/or modify it
5  * under the terms of the GNU General Public License version 2 only, as
6  * published by the Free Software Foundation.  Oracle designates this
7  * particular file as subject to the "Classpath" exception as provided
8  * by Oracle in the LICENSE file that accompanied this code.
9  *
10  * This code is distributed in the hope that it will be useful, but WITHOUT
11  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
13  * version 2 for more details (a copy is included in the LICENSE file that
14  * accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License version
17  * 2 along with this work; if not, write to the Free Software Foundation,
18  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
19  *
20  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
21  * or visit www.oracle.com if you need additional information or have any
22  * questions.
23  */
24 
25 /*
26  * This file is available under and governed by the GNU General Public
27  * License version 2 only, as published by the Free Software Foundation.
28  * However, the following notice accompanied the original version of this
29  * file:
30  *
31  * Written by Doug Lea with assistance from members of JCP JSR-166
32  * Expert Group and released to the public domain, as explained at
33  * http://creativecommons.org/publicdomain/zero/1.0/
34  */
35 
36 package java.util.concurrent;
37 
38 import java.lang.Thread.UncaughtExceptionHandler;
39 import java.lang.invoke.MethodHandles;
40 import java.lang.invoke.VarHandle;
41 import java.security.AccessController;
42 import java.security.AccessControlContext;
43 import java.security.Permission;
44 import java.security.Permissions;
45 import java.security.PrivilegedAction;
46 import java.security.ProtectionDomain;
47 import java.util.ArrayList;
48 import java.util.Collection;
49 import java.util.Collections;
50 import java.util.List;
51 import java.util.function.Predicate;
52 import java.util.concurrent.atomic.AtomicInteger;
53 import java.util.concurrent.locks.LockSupport;
54 import java.util.concurrent.locks.ReentrantLock;
55 import java.util.concurrent.locks.Condition;
56 
57 // Android-changed: Substituted @systemProperty tag with @code.
58 /**
59  * An {@link ExecutorService} for running {@link ForkJoinTask}s.
60  * A {@code ForkJoinPool} provides the entry point for submissions
61  * from non-{@code ForkJoinTask} clients, as well as management and
62  * monitoring operations.
63  *
64  * <p>A {@code ForkJoinPool} differs from other kinds of {@link
65  * ExecutorService} mainly by virtue of employing
66  * <em>work-stealing</em>: all threads in the pool attempt to find and
67  * execute tasks submitted to the pool and/or created by other active
68  * tasks (eventually blocking waiting for work if none exist). This
69  * enables efficient processing when most tasks spawn other subtasks
70  * (as do most {@code ForkJoinTask}s), as well as when many small
71  * tasks are submitted to the pool from external clients.  Especially
72  * when setting <em>asyncMode</em> to true in constructors, {@code
73  * ForkJoinPool}s may also be appropriate for use with event-style
74  * tasks that are never joined. All worker threads are initialized
75  * with {@link Thread#isDaemon} set {@code true}.
76  *
77  * <p>A static {@link #commonPool()} is available and appropriate for
78  * most applications. The common pool is used by any ForkJoinTask that
79  * is not explicitly submitted to a specified pool. Using the common
80  * pool normally reduces resource usage (its threads are slowly
81  * reclaimed during periods of non-use, and reinstated upon subsequent
82  * use).
83  *
84  * <p>For applications that require separate or custom pools, a {@code
85  * ForkJoinPool} may be constructed with a given target parallelism
86  * level; by default, equal to the number of available processors.
87  * The pool attempts to maintain enough active (or available) threads
88  * by dynamically adding, suspending, or resuming internal worker
89  * threads, even if some tasks are stalled waiting to join others.
90  * However, no such adjustments are guaranteed in the face of blocked
91  * I/O or other unmanaged synchronization. The nested {@link
92  * ManagedBlocker} interface enables extension of the kinds of
93  * synchronization accommodated. The default policies may be
94  * overridden using a constructor with parameters corresponding to
95  * those documented in class {@link ThreadPoolExecutor}.
96  *
97  * <p>In addition to execution and lifecycle control methods, this
98  * class provides status check methods (for example
99  * {@link #getStealCount}) that are intended to aid in developing,
100  * tuning, and monitoring fork/join applications. Also, method
101  * {@link #toString} returns indications of pool state in a
102  * convenient form for informal monitoring.
103  *
104  * <p>As is the case with other ExecutorServices, there are three
105  * main task execution methods summarized in the following table.
106  * These are designed to be used primarily by clients not already
107  * engaged in fork/join computations in the current pool.  The main
108  * forms of these methods accept instances of {@code ForkJoinTask},
109  * but overloaded forms also allow mixed execution of plain {@code
110  * Runnable}- or {@code Callable}- based activities as well.  However,
111  * tasks that are already executing in a pool should normally instead
112  * use the within-computation forms listed in the table unless using
113  * async event-style tasks that are not usually joined, in which case
114  * there is little difference among choice of methods.
115  *
116  * <table class="plain">
117  * <caption>Summary of task execution methods</caption>
118  *  <tr>
119  *    <td></td>
120  *    <th scope="col"> Call from non-fork/join clients</th>
121  *    <th scope="col"> Call from within fork/join computations</th>
122  *  </tr>
123  *  <tr>
124  *    <th scope="row" style="text-align:left"> Arrange async execution</th>
125  *    <td> {@link #execute(ForkJoinTask)}</td>
126  *    <td> {@link ForkJoinTask#fork}</td>
127  *  </tr>
128  *  <tr>
129  *    <th scope="row" style="text-align:left"> Await and obtain result</th>
130  *    <td> {@link #invoke(ForkJoinTask)}</td>
131  *    <td> {@link ForkJoinTask#invoke}</td>
132  *  </tr>
133  *  <tr>
134  *    <th scope="row" style="text-align:left"> Arrange exec and obtain Future</th>
135  *    <td> {@link #submit(ForkJoinTask)}</td>
136  *    <td> {@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td>
137  *  </tr>
138  * </table>
139  *
140  * <p>The parameters used to construct the common pool may be controlled by
141  * setting the following {@linkplain System#getProperty system properties}:
142  * <ul>
143  * <li>{@code java.util.concurrent.ForkJoinPool.common.parallelism}
144  * - the parallelism level, a non-negative integer
145  * <li>{@code java.util.concurrent.ForkJoinPool.common.threadFactory}
146  * - the class name of a {@link ForkJoinWorkerThreadFactory}.
147  * The {@linkplain ClassLoader#getSystemClassLoader() system class loader}
148  * is used to load this class.
149  * <li>{@code java.util.concurrent.ForkJoinPool.common.exceptionHandler}
150  * - the class name of a {@link UncaughtExceptionHandler}.
151  * The {@linkplain ClassLoader#getSystemClassLoader() system class loader}
152  * is used to load this class.
153  * <li>{@code java.util.concurrent.ForkJoinPool.common.maximumSpares}
154  * - the maximum number of allowed extra threads to maintain target
155  * parallelism (default 256).
156  * </ul>
157  * If no thread factory is supplied via a system property, then the
158  * common pool uses a factory that uses the system class loader as the
159  * {@linkplain Thread#getContextClassLoader() thread context class loader}.
160  * In addition, if a {@link SecurityManager} is present, then
161  * the common pool uses a factory supplying threads that have no
162  * {@link Permissions} enabled.
163  *
164  * Upon any error in establishing these settings, default parameters
165  * are used. It is possible to disable or limit the use of threads in
166  * the common pool by setting the parallelism property to zero, and/or
167  * using a factory that may return {@code null}. However doing so may
168  * cause unjoined tasks to never be executed.
169  *
170  * <p><b>Implementation notes:</b> This implementation restricts the
171  * maximum number of running threads to 32767. Attempts to create
172  * pools with greater than the maximum number result in
173  * {@code IllegalArgumentException}.
174  *
175  * <p>This implementation rejects submitted tasks (that is, by throwing
176  * {@link RejectedExecutionException}) only when the pool is shut down
177  * or internal resources have been exhausted.
178  *
179  * @since 1.7
180  * @author Doug Lea
181  */
182 @jdk.internal.vm.annotation.Contended
183 public class ForkJoinPool extends AbstractExecutorService {
184 
185     // Android-changed: Substituted time reference with Android API version reference.
186     /*
187      * Implementation Overview
188      *
189      * This class and its nested classes provide the main
190      * functionality and control for a set of worker threads:
191      * Submissions from non-FJ threads enter into submission queues.
192      * Workers take these tasks and typically split them into subtasks
193      * that may be stolen by other workers. Work-stealing based on
194      * randomized scans generally leads to better throughput than
195      * "work dealing" in which producers assign tasks to idle threads,
196      * in part because threads that have finished other tasks before
197      * the signalled thread wakes up (which can be a long time) can
198      * take the task instead.  Preference rules give first priority to
199      * processing tasks from their own queues (LIFO or FIFO, depending
200      * on mode), then to randomized FIFO steals of tasks in other
201      * queues.  This framework began as vehicle for supporting
202      * tree-structured parallelism using work-stealing.  Over time,
203      * its scalability advantages led to extensions and changes to
204      * better support more diverse usage contexts.  Because most
205      * internal methods and nested classes are interrelated, their
206      * main rationale and descriptions are presented here; individual
207      * methods and nested classes contain only brief comments about
208      * details.
209      *
210      * WorkQueues
211      * ==========
212      *
213      * Most operations occur within work-stealing queues (in nested
214      * class WorkQueue).  These are special forms of Deques that
215      * support only three of the four possible end-operations -- push,
216      * pop, and poll (aka steal), under the further constraints that
217      * push and pop are called only from the owning thread (or, as
218      * extended here, under a lock), while poll may be called from
219      * other threads.  (If you are unfamiliar with them, you probably
220      * want to read Herlihy and Shavit's book "The Art of
221      * Multiprocessor programming", chapter 16 describing these in
222      * more detail before proceeding.)  The main work-stealing queue
223      * design is roughly similar to those in the papers "Dynamic
224      * Circular Work-Stealing Deque" by Chase and Lev, SPAA 2005
225      * (http://research.sun.com/scalable/pubs/index.html) and
226      * "Idempotent work stealing" by Michael, Saraswat, and Vechev,
227      * PPoPP 2009 (http://portal.acm.org/citation.cfm?id=1504186).
228      * The main differences ultimately stem from GC requirements that
229      * we null out taken slots as soon as we can, to maintain as small
230      * a footprint as possible even in programs generating huge
231      * numbers of tasks. To accomplish this, we shift the CAS
232      * arbitrating pop vs poll (steal) from being on the indices
233      * ("base" and "top") to the slots themselves.
234      *
235      * Adding tasks then takes the form of a classic array push(task)
236      * in a circular buffer:
237      *    q.array[q.top++ % length] = task;
238      *
239      * The actual code needs to null-check and size-check the array,
240      * uses masking, not mod, for indexing a power-of-two-sized array,
241      * enforces memory ordering, supports resizing, and possibly
242      * signals waiting workers to start scanning -- see below.
243      *
244      * The pop operation (always performed by owner) is of the form:
245      *   if ((task = getAndSet(q.array, (q.top-1) % length, null)) != null)
246      *        decrement top and return task;
247      * If this fails, the queue is empty.
248      *
249      * The poll operation by another stealer thread is, basically:
250      *   if (CAS nonnull task at q.array[q.base % length] to null)
251      *       increment base and return task;
252      *
253      * This may fail due to contention, and may be retried.
254      * Implementations must ensure a consistent snapshot of the base
255      * index and the task (by looping or trying elsewhere) before
256      * trying CAS.  There isn't actually a method of this form,
257      * because failure due to inconsistency or contention is handled
258      * in different ways in different contexts, normally by first
259      * trying other queues. (For the most straightforward example, see
260      * method pollScan.) There are further variants for cases
261      * requiring inspection of elements before extracting them, so
262      * must interleave these with variants of this code.  Also, a more
263      * efficient version (nextLocalTask) is used for polls by owners.
264      * It avoids some overhead because the queue cannot be growing
265      * during call.
266      *
267      * Memory ordering.  See "Correct and Efficient Work-Stealing for
268      * Weak Memory Models" by Le, Pop, Cohen, and Nardelli, PPoPP 2013
269      * (http://www.di.ens.fr/~zappa/readings/ppopp13.pdf) for an
270      * analysis of memory ordering requirements in work-stealing
271      * algorithms similar to the one used here.  Inserting and
272      * extracting tasks in array slots via volatile or atomic accesses
273      * or explicit fences provides primary synchronization.
274      *
275      * Operations on deque elements require reads and writes of both
276      * indices and slots. When possible, we allow these to occur in
277      * any order.  Because the base and top indices (along with other
278      * pool or array fields accessed in many methods) only imprecisely
279      * guide where to extract from, we let accesses other than the
280      * element getAndSet/CAS/setVolatile appear in any order, using
281      * plain mode. But we must still preface some methods (mainly
282      * those that may be accessed externally) with an acquireFence to
283      * avoid unbounded staleness. This is equivalent to acting as if
284      * callers use an acquiring read of the reference to the pool or
285      * queue when invoking the method, even when they do not. We use
286      * explicit acquiring reads (getSlot) rather than plain array
287      * access when acquire mode is required but not otherwise ensured
288      * by context. To reduce stalls by other stealers, we encourage
289      * timely writes to the base index by immediately following
290      * updates with a write of a volatile field that must be updated
291      * anyway, or an Opaque-mode write if there is no such
292      * opportunity.
293      *
294      * Because indices and slot contents cannot always be consistent,
295      * the emptiness check base == top is only quiescently accurate
296      * (and so used where this suffices). Otherwise, it may err on the
297      * side of possibly making the queue appear nonempty when a push,
298      * pop, or poll have not fully committed, or making it appear
299      * empty when an update of top or base has not yet been seen.
300      * Similarly, the check in push for the queue array being full may
301      * trigger when not completely full, causing a resize earlier than
302      * required.
303      *
304      * Mainly because of these potential inconsistencies among slots
305      * vs indices, the poll operation, considered individually, is not
306      * wait-free. One thief cannot successfully continue until another
307      * in-progress one (or, if previously empty, a push) visibly
308      * completes.  This can stall threads when required to consume
309      * from a given queue (which may spin).  However, in the
310      * aggregate, we ensure probabilistic non-blockingness at least
311      * until checking quiescence (which is intrinsically blocking):
312      * If an attempted steal fails, a scanning thief chooses a
313      * different victim target to try next. So, in order for one thief
314      * to progress, it suffices for any in-progress poll or new push
315      * on any empty queue to complete. The worst cases occur when many
316      * threads are looking for tasks being produced by a stalled
317      * producer.
318      *
319      * This approach also enables support of a user mode in which
320      * local task processing is in FIFO, not LIFO order, simply by
321      * using poll rather than pop.  This can be useful in
322      * message-passing frameworks in which tasks are never joined,
323      * although with increased contention among task producers and
324      * consumers.
325      *
326      * WorkQueues are also used in a similar way for tasks submitted
327      * to the pool. We cannot mix these tasks in the same queues used
328      * by workers. Instead, we randomly associate submission queues
329      * with submitting threads, using a form of hashing.  The
330      * ThreadLocalRandom probe value serves as a hash code for
331      * choosing existing queues, and may be randomly repositioned upon
332      * contention with other submitters.  In essence, submitters act
333      * like workers except that they are restricted to executing local
334      * tasks that they submitted (or when known, subtasks thereof).
335      * Insertion of tasks in shared mode requires a lock. We use only
336      * a simple spinlock (using field "source"), because submitters
337      * encountering a busy queue move to a different position to use
338      * or create other queues. They block only when registering new
339      * queues.
340      *
341      * Management
342      * ==========
343      *
344      * The main throughput advantages of work-stealing stem from
345      * decentralized control -- workers mostly take tasks from
346      * themselves or each other, at rates that can exceed a billion
347      * per second.  Most non-atomic control is performed by some form
348      * of scanning across or within queues.  The pool itself creates,
349      * activates (enables scanning for and running tasks),
350      * deactivates, blocks, and terminates threads, all with minimal
351      * central information.  There are only a few properties that we
352      * can globally track or maintain, so we pack them into a small
353      * number of variables, often maintaining atomicity without
354      * blocking or locking.  Nearly all essentially atomic control
355      * state is held in a few volatile variables that are by far most
356      * often read (not written) as status and consistency checks. We
357      * pack as much information into them as we can.
358      *
359      * Field "ctl" contains 64 bits holding information needed to
360      * atomically decide to add, enqueue (on an event queue), and
361      * dequeue and release workers.  To enable this packing, we
362      * restrict maximum parallelism to (1<<15)-1 (which is far in
363      * excess of normal operating range) to allow ids, counts, and
364      * their negations (used for thresholding) to fit into 16bit
365      * subfields.
366      *
367      * Field "mode" holds configuration parameters as well as lifetime
368      * status, atomically and monotonically setting SHUTDOWN, STOP,
369      * and finally TERMINATED bits. It is updated only via bitwise
370      * atomics (getAndBitwiseOr).
371      *
372      * Array "queues" holds references to WorkQueues.  It is updated
373      * (only during worker creation and termination) under the
374      * registrationLock, but is otherwise concurrently readable, and
375      * accessed directly (although always prefaced by acquireFences or
376      * other acquiring reads). To simplify index-based operations, the
377      * array size is always a power of two, and all readers must
378      * tolerate null slots.  Worker queues are at odd indices. Worker
379      * ids masked with SMASK match their index. Shared (submission)
380      * queues are at even indices. Grouping them together in this way
381      * simplifies and speeds up task scanning.
382      *
383      * All worker thread creation is on-demand, triggered by task
384      * submissions, replacement of terminated workers, and/or
385      * compensation for blocked workers. However, all other support
386      * code is set up to work with other policies.  To ensure that we
387      * do not hold on to worker or task references that would prevent
388      * GC, all accesses to workQueues are via indices into the
389      * queues array (which is one source of some of the messy code
390      * constructions here). In essence, the queues array serves as
391      * a weak reference mechanism. Thus for example the stack top
392      * subfield of ctl stores indices, not references.
393      *
394      * Queuing Idle Workers. Unlike HPC work-stealing frameworks, we
395      * cannot let workers spin indefinitely scanning for tasks when
396      * none can be found immediately, and we cannot start/resume
397      * workers unless there appear to be tasks available.  On the
398      * other hand, we must quickly prod them into action when new
399      * tasks are submitted or generated. These latencies are mainly a
400      * function of JVM park/unpark (and underlying OS) performance,
401      * which can be slow and variable.  In many usages, ramp-up time
402      * is the main limiting factor in overall performance, which is
403      * compounded at program start-up by JIT compilation and
404      * allocation. On the other hand, throughput degrades when too
405      * many threads poll for too few tasks.
406      *
407      * The "ctl" field atomically maintains total and "released"
408      * worker counts, plus the head of the available worker queue
409      * (actually stack, represented by the lower 32bit subfield of
410      * ctl).  Released workers are those known to be scanning for
411      * and/or running tasks. Unreleased ("available") workers are
412      * recorded in the ctl stack. These workers are made available for
413      * signalling by enqueuing in ctl (see method awaitWork).  The
414      * "queue" is a form of Treiber stack. This is ideal for
415      * activating threads in most-recently used order, and improves
416      * performance and locality, outweighing the disadvantages of
417      * being prone to contention and inability to release a worker
418      * unless it is topmost on stack. The top stack state holds the
419      * value of the "phase" field of the worker: its index and status,
420      * plus a version counter that, in addition to the count subfields
421      * (also serving as version stamps) provide protection against
422      * Treiber stack ABA effects.
423      *
424      * Creating workers. To create a worker, we pre-increment counts
425      * (serving as a reservation), and attempt to construct a
426      * ForkJoinWorkerThread via its factory. On starting, the new
427      * thread first invokes registerWorker, where it constructs a
428      * WorkQueue and is assigned an index in the queues array
429      * (expanding the array if necessary).  Upon any exception across
430      * these steps, or null return from factory, deregisterWorker
431      * adjusts counts and records accordingly.  If a null return, the
432      * pool continues running with fewer than the target number
433      * workers. If exceptional, the exception is propagated, generally
434      * to some external caller.
435      *
436      * WorkQueue field "phase" is used by both workers and the pool to
437      * manage and track whether a worker is UNSIGNALLED (possibly
438      * blocked waiting for a signal).  When a worker is enqueued its
439      * phase field is set negative. Note that phase field updates lag
440      * queue CAS releases; seeing a negative phase does not guarantee
441      * that the worker is available. When queued, the lower 16 bits of
442      * its phase must hold its pool index. So we place the index there
443      * upon initialization and never modify these bits.
444      *
445      * The ctl field also serves as the basis for memory
446      * synchronization surrounding activation. This uses a more
447      * efficient version of a Dekker-like rule that task producers and
448      * consumers sync with each other by both writing/CASing ctl (even
449      * if to its current value).  However, rather than CASing ctl to
450      * its current value in the common case where no action is
451      * required, we reduce write contention by ensuring that
452      * signalWork invocations are prefaced with a full-volatile memory
453      * access (which is usually needed anyway).
454      *
455      * Signalling. Signals (in signalWork) cause new or reactivated
456      * workers to scan for tasks.  Method signalWork and its callers
457      * try to approximate the unattainable goal of having the right
458      * number of workers activated for the tasks at hand, but must err
459      * on the side of too many workers vs too few to avoid stalls.  If
460      * computations are purely tree structured, it suffices for every
461      * worker to activate another when it pushes a task into an empty
462      * queue, resulting in O(log(#threads)) steps to full activation.
463      * If instead, tasks come in serially from only a single producer,
464      * each worker taking its first (since the last quiescence) task
465      * from a queue should signal another if there are more tasks in
466      * that queue. This is equivalent to, but generally faster than,
467      * arranging the stealer take two tasks, re-pushing one on its own
468      * queue, and signalling (because its queue is empty), also
469      * resulting in logarithmic full activation time. Because we don't
470      * know about usage patterns (or most commonly, mixtures), we use
471      * both approaches.  We approximate the second rule by arranging
472      * that workers in scan() do not repeat signals when repeatedly
473      * taking tasks from any given queue, by remembering the previous
474      * one. There are narrow windows in which both rules may apply,
475      * leading to duplicate or unnecessary signals. Despite such
476      * limitations, these rules usually avoid slowdowns that otherwise
477      * occur when too many workers contend to take too few tasks, or
478      * when producers waste most of their time resignalling.  However,
479      * contention and overhead effects may still occur during ramp-up,
480      * ramp-down, and small computations involving only a few workers.
481      *
482      * Scanning. Method scan performs top-level scanning for (and
483      * execution of) tasks.  Scans by different workers and/or at
484      * different times are unlikely to poll queues in the same
485      * order. Each scan traverses and tries to poll from each queue in
486      * a pseudorandom permutation order by starting at a random index,
487      * and using a constant cyclically exhaustive stride; restarting
488      * upon contention.  (Non-top-level scans; for example in
489      * helpJoin, use simpler linear probes because they do not
490      * systematically contend with top-level scans.)  The pseudorandom
491      * generator need not have high-quality statistical properties in
492      * the long term. We use Marsaglia XorShifts, seeded with the Weyl
493      * sequence from ThreadLocalRandom probes, which are cheap and
494      * suffice. Scans do not otherwise explicitly take into account
495      * core affinities, loads, cache localities, etc, However, they do
496      * exploit temporal locality (which usually approximates these) by
497      * preferring to re-poll from the same queue after a successful
498      * poll before trying others (see method topLevelExec).  This
499      * reduces fairness, which is partially counteracted by using a
500      * one-shot form of poll (tryPoll) that may lose to other workers.
501      *
502      * Deactivation. Method scan returns a sentinel when no tasks are
503      * found, leading to deactivation (see awaitWork). The count
504      * fields in ctl allow accurate discovery of quiescent states
505      * (i.e., when all workers are idle) after deactivation. However,
506      * this may also race with new (external) submissions, so a
507      * recheck is also needed to determine quiescence. Upon apparently
508      * triggering quiescence, awaitWork re-scans and self-signals if
509      * it may have missed a signal. In other cases, a missed signal
510      * may transiently lower parallelism because deactivation does not
511      * necessarily mean that there is no more work, only that that
512      * there were no tasks not taken by other workers.  But more
513      * signals are generated (see above) to eventually reactivate if
514      * needed.
515      *
516      * Trimming workers. To release resources after periods of lack of
517      * use, a worker starting to wait when the pool is quiescent will
518      * time out and terminate if the pool has remained quiescent for
519      * period given by field keepAlive.
520      *
521      * Shutdown and Termination. A call to shutdownNow invokes
522      * tryTerminate to atomically set a mode bit. The calling thread,
523      * as well as every other worker thereafter terminating, helps
524      * terminate others by cancelling their unprocessed tasks, and
525      * waking them up. Calls to non-abrupt shutdown() preface this by
526      * checking isQuiescent before triggering the "STOP" phase of
527      * termination. To conform to ExecutorService invoke, invokeAll,
528      * and invokeAny specs, we must track pool status while waiting,
529      * and interrupt interruptible callers on termination (see
530      * ForkJoinTask.joinForPoolInvoke etc).
531      *
532      * Joining Tasks
533      * =============
534      *
535      * Normally, the first option when joining a task that is not done
536      * is to try to unfork it from local queue and run it.  Otherwise,
537      * any of several actions may be taken when one worker is waiting
538      * to join a task stolen (or always held) by another.  Because we
539      * are multiplexing many tasks on to a pool of workers, we can't
540      * always just let them block (as in Thread.join).  We also cannot
541      * just reassign the joiner's run-time stack with another and
542      * replace it later, which would be a form of "continuation", that
543      * even if possible is not necessarily a good idea since we may
544      * need both an unblocked task and its continuation to progress.
545      * Instead we combine two tactics:
546      *
547      *   Helping: Arranging for the joiner to execute some task that it
548      *      could be running if the steal had not occurred.
549      *
550      *   Compensating: Unless there are already enough live threads,
551      *      method tryCompensate() may create or re-activate a spare
552      *      thread to compensate for blocked joiners until they unblock.
553      *
554      * A third form (implemented via tryRemove) amounts to helping a
555      * hypothetical compensator: If we can readily tell that a
556      * possible action of a compensator is to steal and execute the
557      * task being joined, the joining thread can do so directly,
558      * without the need for a compensation thread; although with a
559      * (rare) possibility of reduced parallelism because of a
560      * transient gap in the queue array.
561      *
562      * Other intermediate forms available for specific task types (for
563      * example helpAsyncBlocker) often avoid or postpone the need for
564      * blocking or compensation.
565      *
566      * The ManagedBlocker extension API can't use helping so relies
567      * only on compensation in method awaitBlocker.
568      *
569      * The algorithm in helpJoin entails a form of "linear helping".
570      * Each worker records (in field "source") the id of the queue
571      * from which it last stole a task.  The scan in method helpJoin
572      * uses these markers to try to find a worker to help (i.e., steal
573      * back a task from and execute it) that could hasten completion
574      * of the actively joined task.  Thus, the joiner executes a task
575      * that would be on its own local deque if the to-be-joined task
576      * had not been stolen. This is a conservative variant of the
577      * approach described in Wagner & Calder "Leapfrogging: a portable
578      * technique for implementing efficient futures" SIGPLAN Notices,
579      * 1993 (http://portal.acm.org/citation.cfm?id=155354). It differs
580      * mainly in that we only record queue ids, not full dependency
581      * links.  This requires a linear scan of the queues array to
582      * locate stealers, but isolates cost to when it is needed, rather
583      * than adding to per-task overhead. Also, searches are limited to
584      * direct and at most two levels of indirect stealers, after which
585      * there are rapidly diminishing returns on increased overhead.
586      * Searches can fail to locate stealers when stalls delay
587      * recording sources.  Further, even when accurately identified,
588      * stealers might not ever produce a task that the joiner can in
589      * turn help with. So, compensation is tried upon failure to find
590      * tasks to run.
591      *
592      * Joining CountedCompleters (see helpComplete) differs from (and
593      * is generally more efficient than) other cases because task
594      * eligibility is determined by checking completion chains rather
595      * than tracking stealers.
596      *
597      * Joining under timeouts (ForkJoinTask timed get) uses a
598      * constrained mixture of helping and compensating in part because
599      * pools (actually, only the common pool) may not have any
600      * available threads: If the pool is saturated (all available
601      * workers are busy), the caller tries to remove and otherwise
602      * help; else it blocks under compensation so that it may time out
603      * independently of any tasks.
604      *
605      * Compensation does not by default aim to keep exactly the target
606      * parallelism number of unblocked threads running at any given
607      * time. Some previous versions of this class employed immediate
608      * compensations for any blocked join. However, in practice, the
609      * vast majority of blockages are transient byproducts of GC and
610      * other JVM or OS activities that are made worse by replacement
611      * when they cause longer-term oversubscription.  Rather than
612      * impose arbitrary policies, we allow users to override the
613      * default of only adding threads upon apparent starvation.  The
614      * compensation mechanism may also be bounded.  Bounds for the
615      * commonPool (see COMMON_MAX_SPARES) better enable JVMs to cope
616      * with programming errors and abuse before running out of
617      * resources to do so.
618      *
619      * Common Pool
620      * ===========
621      *
622      * The static common pool always exists after static
623      * initialization.  Since it (or any other created pool) need
624      * never be used, we minimize initial construction overhead and
625      * footprint to the setup of about a dozen fields.
626      *
627      * When external threads submit to the common pool, they can
628      * perform subtask processing (see helpComplete and related
629      * methods) upon joins.  This caller-helps policy makes it
630      * sensible to set common pool parallelism level to one (or more)
631      * less than the total number of available cores, or even zero for
632      * pure caller-runs.  We do not need to record whether external
633      * submissions are to the common pool -- if not, external help
634      * methods return quickly. These submitters would otherwise be
635      * blocked waiting for completion, so the extra effort (with
636      * liberally sprinkled task status checks) in inapplicable cases
637      * amounts to an odd form of limited spin-wait before blocking in
638      * ForkJoinTask.join.
639      *
640      * Guarantees for common pool parallelism zero are limited to
641      * tasks that are joined by their callers in a tree-structured
642      * fashion or use CountedCompleters (as is true for jdk
643      * parallelStreams). Support infiltrates several methods,
644      * including those that retry helping steps until we are sure that
645      * none apply if there are no workers.
646      *
647      * As a more appropriate default in managed environments, unless
648      * overridden by system properties, we use workers of subclass
649      * InnocuousForkJoinWorkerThread when there is a SecurityManager
650      * present. These workers have no permissions set, do not belong
651      * to any user-defined ThreadGroup, and erase all ThreadLocals
652      * after executing any top-level task.  The associated mechanics
653      * may be JVM-dependent and must access particular Thread class
654      * fields to achieve this effect.
655      *
656      * Interrupt handling
657      * ==================
658      *
659      * The framework is designed to manage task cancellation
660      * (ForkJoinTask.cancel) independently from the interrupt status
661      * of threads running tasks. (See the public ForkJoinTask
662      * documentation for rationale.)  Interrupts are issued only in
663      * tryTerminate, when workers should be terminating and tasks
664      * should be cancelled anyway. Interrupts are cleared only when
665      * necessary to ensure that calls to LockSupport.park do not loop
666      * indefinitely (park returns immediately if the current thread is
667      * interrupted). If so, interruption is reinstated after blocking
668      * if status could be visible during the scope of any task.  For
669      * cases in which task bodies are specified or desired to
670      * interrupt upon cancellation, ForkJoinTask.cancel can be
671      * overridden to do so (as is done for invoke{Any,All}).
672      *
673      * Memory placement
674      * ================
675      *
676      * Performance can be very sensitive to placement of instances of
677      * ForkJoinPool and WorkQueues and their queue arrays. To reduce
678      * false-sharing impact, the @Contended annotation isolates the
679      * ForkJoinPool.ctl field as well as the most heavily written
680      * WorkQueue fields. These mainly reduce cache traffic by scanners.
681      * WorkQueue arrays are presized large enough to avoid resizing
682      * (which transiently reduces throughput) in most tree-like
683      * computations, although not in some streaming usages. Initial
684      * sizes are not large enough to avoid secondary contention
685      * effects (especially for GC cardmarks) when queues are placed
686      * near each other in memory. This is common, but has different
687      * impact in different collectors and remains incompletely
688      * addressed.
689      *
690      * Style notes
691      * ===========
692      *
693      * Memory ordering relies mainly on atomic operations (CAS,
694      * getAndSet, getAndAdd) along with explicit fences.  This can be
695      * awkward and ugly, but also reflects the need to control
696      * outcomes across the unusual cases that arise in very racy code
697      * with very few invariants. All fields are read into locals
698      * before use, and null-checked if they are references, even if
699      * they can never be null under current usages.  Array accesses
700      * using masked indices include checks (that are always true) that
701      * the array length is non-zero to avoid compilers inserting more
702      * expensive traps.  This is usually done in a "C"-like style of
703      * listing declarations at the heads of methods or blocks, and
704      * using inline assignments on first encounter.  Nearly all
705      * explicit checks lead to bypass/return, not exception throws,
706      * because they may legitimately arise during shutdown.
707      *
708      * There is a lot of representation-level coupling among classes
709      * ForkJoinPool, ForkJoinWorkerThread, and ForkJoinTask.  The
710      * fields of WorkQueue maintain data structures managed by
711      * ForkJoinPool, so are directly accessed.  There is little point
712      * trying to reduce this, since any associated future changes in
713      * representations will need to be accompanied by algorithmic
714      * changes anyway. Several methods intrinsically sprawl because
715      * they must accumulate sets of consistent reads of fields held in
716      * local variables. Some others are artificially broken up to
717      * reduce producer/consumer imbalances due to dynamic compilation.
718      * There are also other coding oddities (including several
719      * unnecessary-looking hoisted null checks) that help some methods
720      * perform reasonably even when interpreted (not compiled).
721      *
722      * The order of declarations in this file is (with a few exceptions):
723      * (1) Static utility functions
724      * (2) Nested (static) classes
725      * (3) Static fields
726      * (4) Fields, along with constants used when unpacking some of them
727      * (5) Internal control methods
728      * (6) Callbacks and other support for ForkJoinTask methods
729      * (7) Exported methods
730      * (8) Static block initializing statics in minimally dependent order
731      *
732      * Revision notes
733      * ==============
734      *
735      * The main sources of differences of ForkJoin classes from previous
736      * versions, up to Android API level 33, are:
737      *
738      * * ForkJoinTask now uses field "aux" to support blocking joins
739      *   and/or record exceptions, replacing reliance on builtin
740      *   monitors and side tables.
741      * * Scans probe slots (vs compare indices), along with related
742      *   changes that reduce performance differences across most
743      *   garbage collectors, and reduce contention.
744      * * Refactoring for better integration of special task types and
745      *   other capabilities that had been incrementally tacked on. Plus
746      *   many minor reworkings to improve consistency.
747      */
748 
749     // Static utilities
750 
751     /**
752      * If there is a security manager, makes sure caller has
753      * permission to modify threads.
754      */
checkPermission()755     private static void checkPermission() {
756         @SuppressWarnings("removal")
757         SecurityManager security = System.getSecurityManager();
758         if (security != null)
759             security.checkPermission(modifyThreadPermission);
760     }
761 
762     @SuppressWarnings("removal")
contextWithPermissions(Permission .... perms)763     static AccessControlContext contextWithPermissions(Permission ... perms) {
764         Permissions permissions = new Permissions();
765         for (Permission perm : perms)
766             permissions.add(perm);
767         return new AccessControlContext(
768             new ProtectionDomain[] { new ProtectionDomain(null, permissions) });
769     }
770 
771     // Nested classes
772 
773     /**
774      * Factory for creating new {@link ForkJoinWorkerThread}s.
775      * A {@code ForkJoinWorkerThreadFactory} must be defined and used
776      * for {@code ForkJoinWorkerThread} subclasses that extend base
777      * functionality or initialize threads with different contexts.
778      */
779     public static interface ForkJoinWorkerThreadFactory {
780         /**
781          * Returns a new worker thread operating in the given pool.
782          * Returning null or throwing an exception may result in tasks
783          * never being executed.  If this method throws an exception,
784          * it is relayed to the caller of the method (for example
785          * {@code execute}) causing attempted thread creation. If this
786          * method returns null or throws an exception, it is not
787          * retried until the next attempted creation (for example
788          * another call to {@code execute}).
789          *
790          * @param pool the pool this thread works in
791          * @return the new worker thread, or {@code null} if the request
792          *         to create a thread is rejected
793          * @throws NullPointerException if the pool is null
794          */
newThread(ForkJoinPool pool)795         public ForkJoinWorkerThread newThread(ForkJoinPool pool);
796     }
797 
798     /**
799      * Default ForkJoinWorkerThreadFactory implementation; creates a
800      * new ForkJoinWorkerThread using the system class loader as the
801      * thread context class loader.
802      */
803     static final class DefaultForkJoinWorkerThreadFactory
804         implements ForkJoinWorkerThreadFactory {
805         // ACC for access to the factory
806         @SuppressWarnings("removal")
807         private static final AccessControlContext ACC = contextWithPermissions(
808             new RuntimePermission("getClassLoader"),
809             new RuntimePermission("setContextClassLoader"));
810         @SuppressWarnings("removal")
newThread(ForkJoinPool pool)811         public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
812             return AccessController.doPrivileged(
813                 new PrivilegedAction<>() {
814                     public ForkJoinWorkerThread run() {
815                         return new ForkJoinWorkerThread(null, pool, true, false);
816                     }},
817                 ACC);
818         }
819     }
820 
821     /**
822      * Factory for CommonPool unless overridden by System property.
823      * Creates InnocuousForkJoinWorkerThreads if a security manager is
824      * present at time of invocation.  Support requires that we break
825      * quite a lot of encapsulation (some via helper methods in
826      * ThreadLocalRandom) to access and set Thread fields.
827      */
828     static final class DefaultCommonPoolForkJoinWorkerThreadFactory
829         implements ForkJoinWorkerThreadFactory {
830         @SuppressWarnings("removal")
831         private static final AccessControlContext ACC = contextWithPermissions(
832             modifyThreadPermission,
833             new RuntimePermission("enableContextClassLoaderOverride"),
834             new RuntimePermission("modifyThreadGroup"),
835             new RuntimePermission("getClassLoader"),
836             new RuntimePermission("setContextClassLoader"));
837 
838         @SuppressWarnings("removal")
839         public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
840             return AccessController.doPrivileged(
841                  new PrivilegedAction<>() {
842                      public ForkJoinWorkerThread run() {
843                          return System.getSecurityManager() == null ?
844                              new ForkJoinWorkerThread(null, pool, true, true):
845                              new ForkJoinWorkerThread.
846                              InnocuousForkJoinWorkerThread(pool); }},
847                  ACC);
848         }
849     }
850 
851     // Constants shared across ForkJoinPool and WorkQueue
852 
853     // Bounds
854     static final int SWIDTH       = 16;            // width of short
855     static final int SMASK        = 0xffff;        // short bits == max index
856     static final int MAX_CAP      = 0x7fff;        // max #workers - 1
857 
858     // Masks and units for WorkQueue.phase and ctl sp subfield
859     static final int UNSIGNALLED  = 1 << 31;       // must be negative
860     static final int SS_SEQ       = 1 << 16;       // version count
861 
862     // Mode bits and sentinels, some also used in WorkQueue fields
863     static final int FIFO         = 1 << 16;       // fifo queue or access mode
864     static final int SRC          = 1 << 17;       // set for valid queue ids
865     static final int INNOCUOUS    = 1 << 18;       // set for Innocuous workers
866     static final int QUIET        = 1 << 19;       // quiescing phase or source
867     static final int SHUTDOWN     = 1 << 24;
868     static final int TERMINATED   = 1 << 25;
869     static final int STOP         = 1 << 31;       // must be negative
870     static final int UNCOMPENSATE = 1 << 16;       // tryCompensate return
871 
872     /**
873      * Initial capacity of work-stealing queue array.  Must be a power
874      * of two, at least 2. See above.
875      */
876     static final int INITIAL_QUEUE_CAPACITY = 1 << 8;
877 
878     /**
879      * Queues supporting work-stealing as well as external task
880      * submission. See above for descriptions and algorithms.
881      */
882     static final class WorkQueue {
883         volatile int phase;        // versioned, negative if inactive
884         int stackPred;             // pool stack (ctl) predecessor link
885         int config;                // index, mode, ORed with SRC after init
886         int base;                  // index of next slot for poll
887         ForkJoinTask<?>[] array;   // the queued tasks; power of 2 size
888         final ForkJoinWorkerThread owner; // owning thread or null if shared
889 
890         // segregate fields frequently updated but not read by scans or steals
891         @jdk.internal.vm.annotation.Contended("w")
892         int top;                   // index of next slot for push
893         @jdk.internal.vm.annotation.Contended("w")
894         volatile int source;       // source queue id, lock, or sentinel
895         @jdk.internal.vm.annotation.Contended("w")
896         int nsteals;               // number of steals from other queues
897 
898         // Support for atomic operations
899         private static final VarHandle QA; // for array slots
900         private static final VarHandle SOURCE;
901         private static final VarHandle BASE;
902         static final ForkJoinTask<?> getSlot(ForkJoinTask<?>[] a, int i) {
903             return (ForkJoinTask<?>)QA.getAcquire(a, i);
904         }
905         static final ForkJoinTask<?> getAndClearSlot(ForkJoinTask<?>[] a,
906                                                      int i) {
907             return (ForkJoinTask<?>)QA.getAndSet(a, i, null);
908         }
909         static final void setSlotVolatile(ForkJoinTask<?>[] a, int i,
910                                           ForkJoinTask<?> v) {
911             QA.setVolatile(a, i, v);
912         }
913         static final boolean casSlotToNull(ForkJoinTask<?>[] a, int i,
914                                           ForkJoinTask<?> c) {
915             return QA.compareAndSet(a, i, c, null);
916         }
917         final boolean tryLock() {
918             return SOURCE.compareAndSet(this, 0, 1);
919         }
920         final void setBaseOpaque(int b) {
921             BASE.setOpaque(this, b);
922         }
923 
924         /**
925          * Constructor used by ForkJoinWorkerThreads. Most fields
926          * are initialized upon thread start, in pool.registerWorker.
927          */
928         WorkQueue(ForkJoinWorkerThread owner, boolean isInnocuous) {
929             this.config = (isInnocuous) ? INNOCUOUS : 0;
930             this.owner = owner;
931         }
932 
933         /**
934          * Constructor used for external queues.
935          */
936         WorkQueue(int config) {
937             array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY];
938             this.config = config;
939             owner = null;
940             phase = -1;
941         }
942 
943         /**
944          * Returns an exportable index (used by ForkJoinWorkerThread).
945          */
946         final int getPoolIndex() {
947             return (config & 0xffff) >>> 1; // ignore odd/even tag bit
948         }
949 
950         /**
951          * Returns the approximate number of tasks in the queue.
952          */
953         final int queueSize() {
954             VarHandle.acquireFence(); // ensure fresh reads by external callers
955             int n = top - base;
956             return (n < 0) ? 0 : n;   // ignore transient negative
957         }
958 
959         /**
960          * Provides a more conservative estimate of whether this queue
961          * has any tasks than does queueSize.
962          */
963         final boolean isEmpty() {
964             return !((source != 0 && owner == null) || top - base > 0);
965         }
966 
967         /**
968          * Pushes a task. Call only by owner in unshared queues.
969          *
970          * @param task the task. Caller must ensure non-null.
971          * @param pool (no-op if null)
972          * @throws RejectedExecutionException if array cannot be resized
973          */
974         final void push(ForkJoinTask<?> task, ForkJoinPool pool) {
975             ForkJoinTask<?>[] a = array;
976             int s = top++, d = s - base, cap, m; // skip insert if disabled
977             if (a != null && pool != null && (cap = a.length) > 0) {
978                 setSlotVolatile(a, (m = cap - 1) & s, task);
979                 if (d == m)
980                     growArray();
981                 if (d == m || a[m & (s - 1)] == null)
982                     pool.signalWork(); // signal if was empty or resized
983             }
984         }
985 
986         /**
987          * Pushes task to a shared queue with lock already held, and unlocks.
988          *
989          * @return true if caller should signal work
990          */
991         final boolean lockedPush(ForkJoinTask<?> task) {
992             ForkJoinTask<?>[] a = array;
993             int s = top++, d = s - base, cap, m;
994             if (a != null && (cap = a.length) > 0) {
995                 a[(m = cap - 1) & s] = task;
996                 if (d == m)
997                     growArray();
998                 source = 0; // unlock
999                 if (d == m || a[m & (s - 1)] == null)
1000                     return true;
1001             }
1002             return false;
1003         }
1004 
1005         /**
1006          * Doubles the capacity of array. Called by owner or with lock
1007          * held after pre-incrementing top, which is reverted on
1008          * allocation failure.
1009          */
1010         final void growArray() {
1011             ForkJoinTask<?>[] oldArray = array, newArray;
1012             int s = top - 1, oldCap, newCap;
1013             if (oldArray != null && (oldCap = oldArray.length) > 0 &&
1014                 (newCap = oldCap << 1) > 0) { // skip if disabled
1015                 try {
1016                     newArray = new ForkJoinTask<?>[newCap];
1017                 } catch (Throwable ex) {
1018                     top = s;
1019                     if (owner == null)
1020                         source = 0; // unlock
1021                     throw new RejectedExecutionException(
1022                         "Queue capacity exceeded");
1023                 }
1024                 int newMask = newCap - 1, oldMask = oldCap - 1;
1025                 for (int k = oldCap; k > 0; --k, --s) {
1026                     ForkJoinTask<?> x;        // poll old, push to new
1027                     if ((x = getAndClearSlot(oldArray, s & oldMask)) == null)
1028                         break;                // others already taken
1029                     newArray[s & newMask] = x;
1030                 }
1031                 VarHandle.releaseFence();     // fill before publish
1032                 array = newArray;
1033             }
1034         }
1035 
1036         // Variants of pop
1037 
1038         /**
1039          * Pops and returns task, or null if empty. Called only by owner.
1040          */
1041         private ForkJoinTask<?> pop() {
1042             ForkJoinTask<?> t = null;
1043             int s = top, cap; ForkJoinTask<?>[] a;
1044             if ((a = array) != null && (cap = a.length) > 0 && base != s-- &&
1045                 (t = getAndClearSlot(a, (cap - 1) & s)) != null)
1046                 top = s;
1047             return t;
1048         }
1049 
1050         /**
1051          * Pops the given task for owner only if it is at the current top.
1052          */
1053         final boolean tryUnpush(ForkJoinTask<?> task) {
1054             int s = top, cap; ForkJoinTask<?>[] a;
1055             if ((a = array) != null && (cap = a.length) > 0 && base != s-- &&
1056                 casSlotToNull(a, (cap - 1) & s, task)) {
1057                 top = s;
1058                 return true;
1059             }
1060             return false;
1061         }
1062 
1063         /**
1064          * Locking version of tryUnpush.
1065          */
1066         final boolean externalTryUnpush(ForkJoinTask<?> task) {
1067             boolean taken = false;
1068             for (;;) {
1069                 int s = top, cap, k; ForkJoinTask<?>[] a;
1070                 if ((a = array) == null || (cap = a.length) <= 0 ||
1071                     a[k = (cap - 1) & (s - 1)] != task)
1072                     break;
1073                 if (tryLock()) {
1074                     if (top == s && array == a) {
1075                         if (taken = casSlotToNull(a, k, task)) {
1076                             top = s - 1;
1077                             source = 0;
1078                             break;
1079                         }
1080                     }
1081                     source = 0; // release lock for retry
1082                 }
1083                 Thread.yield(); // trylock failure
1084             }
1085             return taken;
1086         }
1087 
1088         /**
1089          * Deep form of tryUnpush: Traverses from top and removes task if
1090          * present, shifting others to fill gap.
1091          */
1092         final boolean tryRemove(ForkJoinTask<?> task, boolean owned) {
1093             boolean taken = false;
1094             int p = top, cap; ForkJoinTask<?>[] a; ForkJoinTask<?> t;
1095             if ((a = array) != null && task != null && (cap = a.length) > 0) {
1096                 int m = cap - 1, s = p - 1, d = p - base;
1097                 for (int i = s, k; d > 0; --i, --d) {
1098                     if ((t = a[k = i & m]) == task) {
1099                         if (owned || tryLock()) {
1100                             if ((owned || (array == a && top == p)) &&
1101                                 (taken = casSlotToNull(a, k, t))) {
1102                                 for (int j = i; j != s; ) // shift down
1103                                     a[j & m] = getAndClearSlot(a, ++j & m);
1104                                 top = s;
1105                             }
1106                             if (!owned)
1107                                 source = 0;
1108                         }
1109                         break;
1110                     }
1111                 }
1112             }
1113             return taken;
1114         }
1115 
1116         // variants of poll
1117 
1118         /**
1119          * Tries once to poll next task in FIFO order, failing on
1120          * inconsistency or contention.
1121          */
1122         final ForkJoinTask<?> tryPoll() {
1123             int cap, b, k; ForkJoinTask<?>[] a;
1124             if ((a = array) != null && (cap = a.length) > 0) {
1125                 ForkJoinTask<?> t = getSlot(a, k = (cap - 1) & (b = base));
1126                 if (base == b++ && t != null && casSlotToNull(a, k, t)) {
1127                     setBaseOpaque(b);
1128                     return t;
1129                 }
1130             }
1131             return null;
1132         }
1133 
1134         /**
1135          * Takes next task, if one exists, in order specified by mode.
1136          */
1137         final ForkJoinTask<?> nextLocalTask(int cfg) {
1138             ForkJoinTask<?> t = null;
1139             int s = top, cap; ForkJoinTask<?>[] a;
1140             if ((a = array) != null && (cap = a.length) > 0) {
1141                 for (int b, d;;) {
1142                     if ((d = s - (b = base)) <= 0)
1143                         break;
1144                     if (d == 1 || (cfg & FIFO) == 0) {
1145                         if ((t = getAndClearSlot(a, --s & (cap - 1))) != null)
1146                             top = s;
1147                         break;
1148                     }
1149                     if ((t = getAndClearSlot(a, b++ & (cap - 1))) != null) {
1150                         setBaseOpaque(b);
1151                         break;
1152                     }
1153                 }
1154             }
1155             return t;
1156         }
1157 
1158         /**
1159          * Takes next task, if one exists, using configured mode.
1160          */
1161         final ForkJoinTask<?> nextLocalTask() {
1162             return nextLocalTask(config);
1163         }
1164 
1165         /**
1166          * Returns next task, if one exists, in order specified by mode.
1167          */
1168         final ForkJoinTask<?> peek() {
1169             VarHandle.acquireFence();
1170             int cap; ForkJoinTask<?>[] a;
1171             return ((a = array) != null && (cap = a.length) > 0) ?
1172                 a[(cap - 1) & ((config & FIFO) != 0 ? base : top - 1)] : null;
1173         }
1174 
1175         // specialized execution methods
1176 
1177         /**
1178          * Runs the given (stolen) task if nonnull, as well as
1179          * remaining local tasks and/or others available from the
1180          * given queue.
1181          */
1182         final void topLevelExec(ForkJoinTask<?> task, WorkQueue q) {
1183             int cfg = config, nstolen = 1;
1184             while (task != null) {
1185                 task.doExec();
1186                 if ((task = nextLocalTask(cfg)) == null &&
1187                     q != null && (task = q.tryPoll()) != null)
1188                     ++nstolen;
1189             }
1190             nsteals += nstolen;
1191             source = 0;
1192             if ((cfg & INNOCUOUS) != 0)
1193                 ThreadLocalRandom.eraseThreadLocals(Thread.currentThread());
1194         }
1195 
1196         /**
1197          * Tries to pop and run tasks within the target's computation
1198          * until done, not found, or limit exceeded.
1199          *
1200          * @param task root of CountedCompleter computation
1201          * @param owned true if owned by a ForkJoinWorkerThread
1202          * @param limit max runs, or zero for no limit
1203          * @return task status on exit
1204          */
1205         final int helpComplete(ForkJoinTask<?> task, boolean owned, int limit) {
1206             int status = 0, cap, k, p, s; ForkJoinTask<?>[] a; ForkJoinTask<?> t;
1207             while (task != null && (status = task.status) >= 0 &&
1208                    (a = array) != null && (cap = a.length) > 0 &&
1209                    (t = a[k = (cap - 1) & (s = (p = top) - 1)])
1210                    instanceof CountedCompleter) {
1211                 CountedCompleter<?> f = (CountedCompleter<?>)t;
1212                 boolean taken = false;
1213                 for (;;) {     // exec if root task is a completer of t
1214                     if (f == task) {
1215                         if (owned) {
1216                             if ((taken = casSlotToNull(a, k, t)))
1217                                 top = s;
1218                         }
1219                         else if (tryLock()) {
1220                             if (top == p && array == a &&
1221                                 (taken = casSlotToNull(a, k, t)))
1222                                 top = s;
1223                             source = 0;
1224                         }
1225                         if (taken)
1226                             t.doExec();
1227                         else if (!owned)
1228                             Thread.yield(); // tryLock failure
1229                         break;
1230                     }
1231                     else if ((f = f.completer) == null)
1232                         break;
1233                 }
1234                 if (taken && limit != 0 && --limit == 0)
1235                     break;
1236             }
1237             return status;
1238         }
1239 
1240         /**
1241          * Tries to poll and run AsynchronousCompletionTasks until
1242          * none found or blocker is released.
1243          *
1244          * @param blocker the blocker
1245          */
1246         final void helpAsyncBlocker(ManagedBlocker blocker) {
1247             int cap, b, d, k; ForkJoinTask<?>[] a; ForkJoinTask<?> t;
1248             while (blocker != null && (d = top - (b = base)) > 0 &&
1249                    (a = array) != null && (cap = a.length) > 0 &&
1250                    (((t = getSlot(a, k = (cap - 1) & b)) == null && d > 1) ||
1251                     t instanceof
1252                     CompletableFuture.AsynchronousCompletionTask) &&
1253                    !blocker.isReleasable()) {
1254                 if (t != null && base == b++ && casSlotToNull(a, k, t)) {
1255                     setBaseOpaque(b);
1256                     t.doExec();
1257                 }
1258             }
1259         }
1260 
1261         // misc
1262 
1263         /** AccessControlContext for innocuous workers, created on 1st use. */
1264         @SuppressWarnings("removal")
1265         private static AccessControlContext INNOCUOUS_ACC;
1266 
1267         /**
1268          * Initializes (upon registration) InnocuousForkJoinWorkerThreads.
1269          */
1270         @SuppressWarnings("removal")
1271         final void initializeInnocuousWorker() {
1272             AccessControlContext acc; // racy construction OK
1273             if ((acc = INNOCUOUS_ACC) == null)
1274                 INNOCUOUS_ACC = acc = new AccessControlContext(
1275                     new ProtectionDomain[] { new ProtectionDomain(null, null) });
1276             Thread t = Thread.currentThread();
1277             ThreadLocalRandom.setInheritedAccessControlContext(t, acc);
1278             ThreadLocalRandom.eraseThreadLocals(t);
1279         }
1280 
1281         /**
1282          * Returns true if owned by a worker thread and not known to be blocked.
1283          */
1284         final boolean isApparentlyUnblocked() {
1285             Thread wt; Thread.State s;
1286             return ((wt = owner) != null &&
1287                     (s = wt.getState()) != Thread.State.BLOCKED &&
1288                     s != Thread.State.WAITING &&
1289                     s != Thread.State.TIMED_WAITING);
1290         }
1291 
1292         static {
1293             try {
1294                 QA = MethodHandles.arrayElementVarHandle(ForkJoinTask[].class);
1295                 MethodHandles.Lookup l = MethodHandles.lookup();
1296                 SOURCE = l.findVarHandle(WorkQueue.class, "source", int.class);
1297                 BASE = l.findVarHandle(WorkQueue.class, "base", int.class);
1298             } catch (ReflectiveOperationException e) {
1299                 throw new ExceptionInInitializerError(e);
1300             }
1301         }
1302     }
1303 
1304     // static fields (initialized in static initializer below)
1305 
1306     /**
1307      * Creates a new ForkJoinWorkerThread. This factory is used unless
1308      * overridden in ForkJoinPool constructors.
1309      */
1310     public static final ForkJoinWorkerThreadFactory
1311         defaultForkJoinWorkerThreadFactory;
1312 
1313     /**
1314      * Permission required for callers of methods that may start or
1315      * kill threads.
1316      */
1317     static final RuntimePermission modifyThreadPermission;
1318 
1319     /**
1320      * Common (static) pool. Non-null for public use unless a static
1321      * construction exception, but internal usages null-check on use
1322      * to paranoically avoid potential initialization circularities
1323      * as well as to simplify generated code.
1324      */
1325     static final ForkJoinPool common;
1326 
1327     /**
1328      * Common pool parallelism. To allow simpler use and management
1329      * when common pool threads are disabled, we allow the underlying
1330      * common.parallelism field to be zero, but in that case still report
1331      * parallelism as 1 to reflect resulting caller-runs mechanics.
1332      */
1333     static final int COMMON_PARALLELISM;
1334 
1335     /**
1336      * Limit on spare thread construction in tryCompensate.
1337      */
1338     private static final int COMMON_MAX_SPARES;
1339 
1340     /**
1341      * Sequence number for creating worker names
1342      */
1343     private static volatile int poolIds;
1344 
1345     // static configuration constants
1346 
1347     /**
1348      * Default idle timeout value (in milliseconds) for the thread
1349      * triggering quiescence to park waiting for new work
1350      */
1351     private static final long DEFAULT_KEEPALIVE = 60_000L;
1352 
1353     /**
1354      * Undershoot tolerance for idle timeouts
1355      */
1356     private static final long TIMEOUT_SLOP = 20L;
1357 
1358     /**
1359      * The default value for COMMON_MAX_SPARES.  Overridable using the
1360      * "java.util.concurrent.ForkJoinPool.common.maximumSpares" system
1361      * property.  The default value is far in excess of normal
1362      * requirements, but also far short of MAX_CAP and typical OS
1363      * thread limits, so allows JVMs to catch misuse/abuse before
1364      * running out of resources needed to do so.
1365      */
1366     private static final int DEFAULT_COMMON_MAX_SPARES = 256;
1367 
1368     /*
1369      * Bits and masks for field ctl, packed with 4 16 bit subfields:
1370      * RC: Number of released (unqueued) workers minus target parallelism
1371      * TC: Number of total workers minus target parallelism
1372      * SS: version count and status of top waiting thread
1373      * ID: poolIndex of top of Treiber stack of waiters
1374      *
1375      * When convenient, we can extract the lower 32 stack top bits
1376      * (including version bits) as sp=(int)ctl.  The offsets of counts
1377      * by the target parallelism and the positionings of fields makes
1378      * it possible to perform the most common checks via sign tests of
1379      * fields: When ac is negative, there are not enough unqueued
1380      * workers, when tc is negative, there are not enough total
1381      * workers.  When sp is non-zero, there are waiting workers.  To
1382      * deal with possibly negative fields, we use casts in and out of
1383      * "short" and/or signed shifts to maintain signedness.
1384      *
1385      * Because it occupies uppermost bits, we can add one release
1386      * count using getAndAdd of RC_UNIT, rather than CAS, when
1387      * returning from a blocked join.  Other updates entail multiple
1388      * subfields and masking, requiring CAS.
1389      *
1390      * The limits packed in field "bounds" are also offset by the
1391      * parallelism level to make them comparable to the ctl rc and tc
1392      * fields.
1393      */
1394 
1395     // Lower and upper word masks
1396     private static final long SP_MASK    = 0xffffffffL;
1397     private static final long UC_MASK    = ~SP_MASK;
1398 
1399     // Release counts
1400     private static final int  RC_SHIFT   = 48;
1401     private static final long RC_UNIT    = 0x0001L << RC_SHIFT;
1402     private static final long RC_MASK    = 0xffffL << RC_SHIFT;
1403 
1404     // Total counts
1405     private static final int  TC_SHIFT   = 32;
1406     private static final long TC_UNIT    = 0x0001L << TC_SHIFT;
1407     private static final long TC_MASK    = 0xffffL << TC_SHIFT;
1408     private static final long ADD_WORKER = 0x0001L << (TC_SHIFT + 15); // sign
1409 
1410     // Instance fields
1411 
1412     final long keepAlive;                // milliseconds before dropping if idle
1413     volatile long stealCount;            // collects worker nsteals
1414     int scanRover;                       // advances across pollScan calls
1415     volatile int threadIds;              // for worker thread names
1416     final int bounds;                    // min, max threads packed as shorts
1417     volatile int mode;                   // parallelism, runstate, queue mode
1418     WorkQueue[] queues;                  // main registry
1419     final ReentrantLock registrationLock;
1420     Condition termination;               // lazily constructed
1421     final String workerNamePrefix;       // null for common pool
1422     final ForkJoinWorkerThreadFactory factory;
1423     final UncaughtExceptionHandler ueh;  // per-worker UEH
1424     final Predicate<? super ForkJoinPool> saturate;
1425 
1426     @jdk.internal.vm.annotation.Contended("fjpctl") // segregate
1427     volatile long ctl;                   // main pool control
1428 
1429     // Support for atomic operations
1430     private static final VarHandle CTL;
1431     private static final VarHandle MODE;
1432     private static final VarHandle THREADIDS;
1433     private static final VarHandle POOLIDS;
1434     private boolean compareAndSetCtl(long c, long v) {
1435         return CTL.compareAndSet(this, c, v);
1436     }
1437     private long compareAndExchangeCtl(long c, long v) {
1438         return (long)CTL.compareAndExchange(this, c, v);
1439     }
1440     private long getAndAddCtl(long v) {
1441         return (long)CTL.getAndAdd(this, v);
1442     }
1443     private int getAndBitwiseOrMode(int v) {
1444         return (int)MODE.getAndBitwiseOr(this, v);
1445     }
1446     private int getAndAddThreadIds(int x) {
1447         return (int)THREADIDS.getAndAdd(this, x);
1448     }
1449     private static int getAndAddPoolIds(int x) {
1450         return (int)POOLIDS.getAndAdd(x);
1451     }
1452 
1453     // Creating, registering and deregistering workers
1454 
1455     /**
1456      * Tries to construct and start one worker. Assumes that total
1457      * count has already been incremented as a reservation.  Invokes
1458      * deregisterWorker on any failure.
1459      *
1460      * @return true if successful
1461      */
1462     private boolean createWorker() {
1463         ForkJoinWorkerThreadFactory fac = factory;
1464         Throwable ex = null;
1465         ForkJoinWorkerThread wt = null;
1466         try {
1467             if (fac != null && (wt = fac.newThread(this)) != null) {
1468                 wt.start();
1469                 return true;
1470             }
1471         } catch (Throwable rex) {
1472             ex = rex;
1473         }
1474         deregisterWorker(wt, ex);
1475         return false;
1476     }
1477 
1478     /**
1479      * Provides a name for ForkJoinWorkerThread constructor.
1480      */
1481     final String nextWorkerThreadName() {
1482         String prefix = workerNamePrefix;
1483         int tid = getAndAddThreadIds(1) + 1;
1484         if (prefix == null) // commonPool has no prefix
1485             prefix = "ForkJoinPool.commonPool-worker-";
1486         return prefix.concat(Integer.toString(tid));
1487     }
1488 
1489     /**
1490      * Finishes initializing and records owned queue.
1491      *
1492      * @param w caller's WorkQueue
1493      */
1494     final void registerWorker(WorkQueue w) {
1495         ReentrantLock lock = registrationLock;
1496         ThreadLocalRandom.localInit();
1497         int seed = ThreadLocalRandom.getProbe();
1498         if (w != null && lock != null) {
1499             int modebits = (mode & FIFO) | w.config;
1500             w.array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY];
1501             w.stackPred = seed;                         // stash for runWorker
1502             if ((modebits & INNOCUOUS) != 0)
1503                 w.initializeInnocuousWorker();
1504             int id = (seed << 1) | 1;                   // initial index guess
1505             lock.lock();
1506             try {
1507                 WorkQueue[] qs; int n;                  // find queue index
1508                 if ((qs = queues) != null && (n = qs.length) > 0) {
1509                     int k = n, m = n - 1;
1510                     for (; qs[id &= m] != null && k > 0; id -= 2, k -= 2);
1511                     if (k == 0)
1512                         id = n | 1;                     // resize below
1513                     w.phase = w.config = id | modebits; // now publishable
1514 
1515                     if (id < n)
1516                         qs[id] = w;
1517                     else {                              // expand array
1518                         int an = n << 1, am = an - 1;
1519                         WorkQueue[] as = new WorkQueue[an];
1520                         as[id & am] = w;
1521                         for (int j = 1; j < n; j += 2)
1522                             as[j] = qs[j];
1523                         for (int j = 0; j < n; j += 2) {
1524                             WorkQueue q;
1525                             if ((q = qs[j]) != null)    // shared queues may move
1526                                 as[q.config & am] = q;
1527                         }
1528                         VarHandle.releaseFence();       // fill before publish
1529                         queues = as;
1530                     }
1531                 }
1532             } finally {
1533                 lock.unlock();
1534             }
1535         }
1536     }
1537 
1538     /**
1539      * Final callback from terminating worker, as well as upon failure
1540      * to construct or start a worker.  Removes record of worker from
1541      * array, and adjusts counts. If pool is shutting down, tries to
1542      * complete termination.
1543      *
1544      * @param wt the worker thread, or null if construction failed
1545      * @param ex the exception causing failure, or null if none
1546      */
1547     final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) {
1548         ReentrantLock lock = registrationLock;
1549         WorkQueue w = null;
1550         int cfg = 0;
1551         if (wt != null && (w = wt.workQueue) != null && lock != null) {
1552             WorkQueue[] qs; int n, i;
1553             cfg = w.config;
1554             long ns = w.nsteals & 0xffffffffL;
1555             lock.lock();                             // remove index from array
1556             if ((qs = queues) != null && (n = qs.length) > 0 &&
1557                 qs[i = cfg & (n - 1)] == w)
1558                 qs[i] = null;
1559             stealCount += ns;                        // accumulate steals
1560             lock.unlock();
1561             long c = ctl;
1562             if ((cfg & QUIET) == 0) // unless self-signalled, decrement counts
1563                 do {} while (c != (c = compareAndExchangeCtl(
1564                                        c, ((RC_MASK & (c - RC_UNIT)) |
1565                                            (TC_MASK & (c - TC_UNIT)) |
1566                                            (SP_MASK & c)))));
1567             else if ((int)c == 0)                    // was dropped on timeout
1568                 cfg = 0;                             // suppress signal if last
1569             for (ForkJoinTask<?> t; (t = w.pop()) != null; )
1570                 ForkJoinTask.cancelIgnoringExceptions(t); // cancel tasks
1571         }
1572 
1573         if (!tryTerminate(false, false) && w != null && (cfg & SRC) != 0)
1574             signalWork();                            // possibly replace worker
1575         if (ex != null)
1576             ForkJoinTask.rethrow(ex);
1577     }
1578 
1579     /*
1580      * Tries to create or release a worker if too few are running.
1581      */
1582     final void signalWork() {
1583         for (long c = ctl; c < 0L;) {
1584             int sp, i; WorkQueue[] qs; WorkQueue v;
1585             if ((sp = (int)c & ~UNSIGNALLED) == 0) {  // no idle workers
1586                 if ((c & ADD_WORKER) == 0L)           // enough total workers
1587                     break;
1588                 if (c == (c = compareAndExchangeCtl(
1589                               c, ((RC_MASK & (c + RC_UNIT)) |
1590                                   (TC_MASK & (c + TC_UNIT)))))) {
1591                     createWorker();
1592                     break;
1593                 }
1594             }
1595             else if ((qs = queues) == null)
1596                 break;                                // unstarted/terminated
1597             else if (qs.length <= (i = sp & SMASK))
1598                 break;                                // terminated
1599             else if ((v = qs[i]) == null)
1600                 break;                                // terminating
1601             else {
1602                 long nc = (v.stackPred & SP_MASK) | (UC_MASK & (c + RC_UNIT));
1603                 Thread vt = v.owner;
1604                 if (c == (c = compareAndExchangeCtl(c, nc))) {
1605                     v.phase = sp;
1606                     LockSupport.unpark(vt);           // release idle worker
1607                     break;
1608                 }
1609             }
1610         }
1611     }
1612 
1613     /**
1614      * Top-level runloop for workers, called by ForkJoinWorkerThread.run.
1615      * See above for explanation.
1616      *
1617      * @param w caller's WorkQueue (may be null on failed initialization)
1618      */
1619     final void runWorker(WorkQueue w) {
1620         if (mode >= 0 && w != null) {           // skip on failed init
1621             w.config |= SRC;                    // mark as valid source
1622             int r = w.stackPred, src = 0;       // use seed from registerWorker
1623             do {
1624                 r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift
1625             } while ((src = scan(w, src, r)) >= 0 ||
1626                      (src = awaitWork(w)) == 0);
1627         }
1628     }
1629 
1630     /**
1631      * Scans for and if found executes top-level tasks: Tries to poll
1632      * each queue starting at a random index with random stride,
1633      * returning source id or retry indicator if contended or
1634      * inconsistent.
1635      *
1636      * @param w caller's WorkQueue
1637      * @param prevSrc the previous queue stolen from in current phase, or 0
1638      * @param r random seed
1639      * @return id of queue if taken, negative if none found, prevSrc for retry
1640      */
1641     private int scan(WorkQueue w, int prevSrc, int r) {
1642         WorkQueue[] qs = queues;
1643         int n = (w == null || qs == null) ? 0 : qs.length;
1644         for (int step = (r >>> 16) | 1, i = n; i > 0; --i, r += step) {
1645             int j, cap, b; WorkQueue q; ForkJoinTask<?>[] a;
1646             if ((q = qs[j = r & (n - 1)]) != null && // poll at qs[j].array[k]
1647                 (a = q.array) != null && (cap = a.length) > 0) {
1648                 int k = (cap - 1) & (b = q.base), nextBase = b + 1;
1649                 int nextIndex = (cap - 1) & nextBase, src = j | SRC;
1650                 ForkJoinTask<?> t = WorkQueue.getSlot(a, k);
1651                 if (q.base != b)                // inconsistent
1652                     return prevSrc;
1653                 else if (t != null && WorkQueue.casSlotToNull(a, k, t)) {
1654                     q.base = nextBase;
1655                     ForkJoinTask<?> next = a[nextIndex];
1656                     if ((w.source = src) != prevSrc && next != null)
1657                         signalWork();           // propagate
1658                     w.topLevelExec(t, q);
1659                     return src;
1660                 }
1661                 else if (a[nextIndex] != null)  // revisit
1662                     return prevSrc;
1663             }
1664         }
1665         return (queues != qs) ? prevSrc: -1;    // possibly resized
1666     }
1667 
1668     /**
1669      * Advances worker phase, pushes onto ctl stack, and awaits signal
1670      * or reports termination.
1671      *
1672      * @return negative if terminated, else 0
1673      */
1674     private int awaitWork(WorkQueue w) {
1675         if (w == null)
1676             return -1;                       // already terminated
1677         int phase = (w.phase + SS_SEQ) & ~UNSIGNALLED;
1678         w.phase = phase | UNSIGNALLED;       // advance phase
1679         long prevCtl = ctl, c;               // enqueue
1680         do {
1681             w.stackPred = (int)prevCtl;
1682             c = ((prevCtl - RC_UNIT) & UC_MASK) | (phase & SP_MASK);
1683         } while (prevCtl != (prevCtl = compareAndExchangeCtl(prevCtl, c)));
1684 
1685         Thread.interrupted();                // clear status
1686         LockSupport.setCurrentBlocker(this); // prepare to block (exit also OK)
1687         long deadline = 0L;                  // nonzero if possibly quiescent
1688         int ac = (int)(c >> RC_SHIFT), md;
1689         if ((md = mode) < 0)                 // pool is terminating
1690             return -1;
1691         else if ((md & SMASK) + ac <= 0) {
1692             boolean checkTermination = (md & SHUTDOWN) != 0;
1693             if ((deadline = System.currentTimeMillis() + keepAlive) == 0L)
1694                 deadline = 1L;               // avoid zero
1695             WorkQueue[] qs = queues;         // check for racing submission
1696             int n = (qs == null) ? 0 : qs.length;
1697             for (int i = 0; i < n; i += 2) {
1698                 WorkQueue q; ForkJoinTask<?>[] a; int cap, b;
1699                 if (ctl != c) {              // already signalled
1700                     checkTermination = false;
1701                     break;
1702                 }
1703                 else if ((q = qs[i]) != null &&
1704                          (a = q.array) != null && (cap = a.length) > 0 &&
1705                          ((b = q.base) != q.top || a[(cap - 1) & b] != null ||
1706                           q.source != 0)) {
1707                     if (compareAndSetCtl(c, prevCtl))
1708                         w.phase = phase;     // self-signal
1709                     checkTermination = false;
1710                     break;
1711                 }
1712             }
1713             if (checkTermination && tryTerminate(false, false))
1714                 return -1;                   // trigger quiescent termination
1715         }
1716 
1717         for (boolean alt = false;;) {        // await activation or termination
1718             if (w.phase >= 0)
1719                 break;
1720             else if (mode < 0)
1721                 return -1;
1722             else if ((c = ctl) == prevCtl)
1723                 Thread.onSpinWait();         // signal in progress
1724             else if (!(alt = !alt))          // check between park calls
1725                 Thread.interrupted();
1726             else if (deadline == 0L)
1727                 LockSupport.park();
1728             else if (deadline - System.currentTimeMillis() > TIMEOUT_SLOP)
1729                 LockSupport.parkUntil(deadline);
1730             else if (((int)c & SMASK) == (w.config & SMASK) &&
1731                      compareAndSetCtl(c, ((UC_MASK & (c - TC_UNIT)) |
1732                                           (prevCtl & SP_MASK)))) {
1733                 w.config |= QUIET;           // sentinel for deregisterWorker
1734                 return -1;                   // drop on timeout
1735             }
1736             else if ((deadline += keepAlive) == 0L)
1737                 deadline = 1L;               // not at head; restart timer
1738         }
1739         return 0;
1740     }
1741 
1742     // Utilities used by ForkJoinTask
1743 
1744     /**
1745      * Returns true if can start terminating if enabled, or already terminated
1746      */
1747     final boolean canStop() {
1748         outer: for (long oldSum = 0L;;) { // repeat until stable
1749             int md; WorkQueue[] qs;  long c;
1750             if ((qs = queues) == null || ((md = mode) & STOP) != 0)
1751                 return true;
1752             if ((md & SMASK) + (int)((c = ctl) >> RC_SHIFT) > 0)
1753                 break;
1754             long checkSum = c;
1755             for (int i = 1; i < qs.length; i += 2) { // scan submitters
1756                 WorkQueue q; ForkJoinTask<?>[] a; int s = 0, cap;
1757                 if ((q = qs[i]) != null && (a = q.array) != null &&
1758                     (cap = a.length) > 0 &&
1759                     ((s = q.top) != q.base || a[(cap - 1) & s] != null ||
1760                      q.source != 0))
1761                     break outer;
1762                 checkSum += (((long)i) << 32) ^ s;
1763             }
1764             if (oldSum == (oldSum = checkSum) && queues == qs)
1765                 return true;
1766         }
1767         return (mode & STOP) != 0; // recheck mode on false return
1768     }
1769 
1770     /**
1771      * Tries to decrement counts (sometimes implicitly) and possibly
1772      * arrange for a compensating worker in preparation for
1773      * blocking. May fail due to interference, in which case -1 is
1774      * returned so caller may retry. A zero return value indicates
1775      * that the caller doesn't need to re-adjust counts when later
1776      * unblocked.
1777      *
1778      * @param c incoming ctl value
1779      * @return UNCOMPENSATE: block then adjust, 0: block, -1 : retry
1780      */
1781     private int tryCompensate(long c) {
1782         Predicate<? super ForkJoinPool> sat;
1783         int md = mode, b = bounds;
1784         // counts are signed; centered at parallelism level == 0
1785         int minActive = (short)(b & SMASK),
1786             maxTotal  = b >>> SWIDTH,
1787             active    = (int)(c >> RC_SHIFT),
1788             total     = (short)(c >>> TC_SHIFT),
1789             sp        = (int)c & ~UNSIGNALLED;
1790         if ((md & SMASK) == 0)
1791             return 0;                  // cannot compensate if parallelism zero
1792         else if (total >= 0) {
1793             if (sp != 0) {                        // activate idle worker
1794                 WorkQueue[] qs; int n; WorkQueue v;
1795                 if ((qs = queues) != null && (n = qs.length) > 0 &&
1796                     (v = qs[sp & (n - 1)]) != null) {
1797                     Thread vt = v.owner;
1798                     long nc = ((long)v.stackPred & SP_MASK) | (UC_MASK & c);
1799                     if (compareAndSetCtl(c, nc)) {
1800                         v.phase = sp;
1801                         LockSupport.unpark(vt);
1802                         return UNCOMPENSATE;
1803                     }
1804                 }
1805                 return -1;                        // retry
1806             }
1807             else if (active > minActive) {        // reduce parallelism
1808                 long nc = ((RC_MASK & (c - RC_UNIT)) | (~RC_MASK & c));
1809                 return compareAndSetCtl(c, nc) ? UNCOMPENSATE : -1;
1810             }
1811         }
1812         if (total < maxTotal) {                   // expand pool
1813             long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK);
1814             return (!compareAndSetCtl(c, nc) ? -1 :
1815                     !createWorker() ? 0 : UNCOMPENSATE);
1816         }
1817         else if (!compareAndSetCtl(c, c))         // validate
1818             return -1;
1819         else if ((sat = saturate) != null && sat.test(this))
1820             return 0;
1821         else
1822             throw new RejectedExecutionException(
1823                 "Thread limit exceeded replacing blocked worker");
1824     }
1825 
1826     /**
1827      * Readjusts RC count; called from ForkJoinTask after blocking.
1828      */
1829     final void uncompensate() {
1830         getAndAddCtl(RC_UNIT);
1831     }
1832 
1833     /**
1834      * Helps if possible until the given task is done.  Scans other
1835      * queues for a task produced by one of w's stealers; returning
1836      * compensated blocking sentinel if none are found.
1837      *
1838      * @param task the task
1839      * @param w caller's WorkQueue
1840      * @param canHelp if false, compensate only
1841      * @return task status on exit, or UNCOMPENSATE for compensated blocking
1842      */
1843     final int helpJoin(ForkJoinTask<?> task, WorkQueue w, boolean canHelp) {
1844         int s = 0;
1845         if (task != null && w != null) {
1846             int wsrc = w.source, wid = w.config & SMASK, r = wid + 2;
1847             boolean scan = true;
1848             long c = 0L;                          // track ctl stability
1849             outer: for (;;) {
1850                 if ((s = task.status) < 0)
1851                     break;
1852                 else if (scan = !scan) {          // previous scan was empty
1853                     if (mode < 0)
1854                         ForkJoinTask.cancelIgnoringExceptions(task);
1855                     else if (c == (c = ctl) && (s = tryCompensate(c)) >= 0)
1856                         break;                    // block
1857                 }
1858                 else if (canHelp) {               // scan for subtasks
1859                     WorkQueue[] qs = queues;
1860                     int n = (qs == null) ? 0 : qs.length, m = n - 1;
1861                     for (int i = n; i > 0; i -= 2, r += 2) {
1862                         int j; WorkQueue q, x, y; ForkJoinTask<?>[] a;
1863                         if ((q = qs[j = r & m]) != null) {
1864                             int sq = q.source & SMASK, cap, b;
1865                             if ((a = q.array) != null && (cap = a.length) > 0) {
1866                                 int k = (cap - 1) & (b = q.base);
1867                                 int nextBase = b + 1, src = j | SRC, sx;
1868                                 ForkJoinTask<?> t = WorkQueue.getSlot(a, k);
1869                                 boolean eligible = sq == wid ||
1870                                     ((x = qs[sq & m]) != null &&   // indirect
1871                                      ((sx = (x.source & SMASK)) == wid ||
1872                                       ((y = qs[sx & m]) != null && // 2-indirect
1873                                        (y.source & SMASK) == wid)));
1874                                 if ((s = task.status) < 0)
1875                                     break outer;
1876                                 else if ((q.source & SMASK) != sq ||
1877                                          q.base != b)
1878                                     scan = true;          // inconsistent
1879                                 else if (t == null)
1880                                     scan |= (a[nextBase & (cap - 1)] != null ||
1881                                              q.top != b); // lagging
1882                                 else if (eligible) {
1883                                     if (WorkQueue.casSlotToNull(a, k, t)) {
1884                                         q.base = nextBase;
1885                                         w.source = src;
1886                                         t.doExec();
1887                                         w.source = wsrc;
1888                                     }
1889                                     scan = true;
1890                                     break;
1891                                 }
1892                             }
1893                         }
1894                     }
1895                 }
1896             }
1897         }
1898         return s;
1899     }
1900 
1901     /**
1902      * Extra helpJoin steps for CountedCompleters.  Scans for and runs
1903      * subtasks of the given root task, returning if none are found.
1904      *
1905      * @param task root of CountedCompleter computation
1906      * @param w caller's WorkQueue
1907      * @param owned true if owned by a ForkJoinWorkerThread
1908      * @return task status on exit
1909      */
1910     final int helpComplete(ForkJoinTask<?> task, WorkQueue w, boolean owned) {
1911         int s = 0;
1912         if (task != null && w != null) {
1913             int r = w.config;
1914             boolean scan = true, locals = true;
1915             long c = 0L;
1916             outer: for (;;) {
1917                 if (locals) {                     // try locals before scanning
1918                     if ((s = w.helpComplete(task, owned, 0)) < 0)
1919                         break;
1920                     locals = false;
1921                 }
1922                 else if ((s = task.status) < 0)
1923                     break;
1924                 else if (scan = !scan) {
1925                     if (c == (c = ctl))
1926                         break;
1927                 }
1928                 else {                            // scan for subtasks
1929                     WorkQueue[] qs = queues;
1930                     int n = (qs == null) ? 0 : qs.length;
1931                     for (int i = n; i > 0; --i, ++r) {
1932                         int j, cap, b; WorkQueue q; ForkJoinTask<?>[] a;
1933                         boolean eligible = false;
1934                         if ((q = qs[j = r & (n - 1)]) != null &&
1935                             (a = q.array) != null && (cap = a.length) > 0) {
1936                             int k = (cap - 1) & (b = q.base), nextBase = b + 1;
1937                             ForkJoinTask<?> t = WorkQueue.getSlot(a, k);
1938                             if (t instanceof CountedCompleter) {
1939                                 CountedCompleter<?> f = (CountedCompleter<?>)t;
1940                                 do {} while (!(eligible = (f == task)) &&
1941                                              (f = f.completer) != null);
1942                             }
1943                             if ((s = task.status) < 0)
1944                                 break outer;
1945                             else if (q.base != b)
1946                                 scan = true;       // inconsistent
1947                             else if (t == null)
1948                                 scan |= (a[nextBase & (cap - 1)] != null ||
1949                                          q.top != b);
1950                             else if (eligible) {
1951                                 if (WorkQueue.casSlotToNull(a, k, t)) {
1952                                     q.setBaseOpaque(nextBase);
1953                                     t.doExec();
1954                                     locals = true;
1955                                 }
1956                                 scan = true;
1957                                 break;
1958                             }
1959                         }
1960                     }
1961                 }
1962             }
1963         }
1964         return s;
1965     }
1966 
1967     /**
1968      * Scans for and returns a polled task, if available.  Used only
1969      * for untracked polls. Begins scan at an index (scanRover)
1970      * advanced on each call, to avoid systematic unfairness.
1971      *
1972      * @param submissionsOnly if true, only scan submission queues
1973      */
1974     private ForkJoinTask<?> pollScan(boolean submissionsOnly) {
1975         VarHandle.acquireFence();
1976         int r = scanRover += 0x61c88647; // Weyl increment; raciness OK
1977         if (submissionsOnly)             // even indices only
1978             r &= ~1;
1979         int step = (submissionsOnly) ? 2 : 1;
1980         WorkQueue[] qs; int n;
1981         while ((qs = queues) != null && (n = qs.length) > 0) {
1982             boolean scan = false;
1983             for (int i = 0; i < n; i += step) {
1984                 int j, cap, b; WorkQueue q; ForkJoinTask<?>[] a;
1985                 if ((q = qs[j = (n - 1) & (r + i)]) != null &&
1986                     (a = q.array) != null && (cap = a.length) > 0) {
1987                     int k = (cap - 1) & (b = q.base), nextBase = b + 1;
1988                     ForkJoinTask<?> t = WorkQueue.getSlot(a, k);
1989                     if (q.base != b)
1990                         scan = true;
1991                     else if (t == null)
1992                         scan |= (q.top != b || a[nextBase & (cap - 1)] != null);
1993                     else if (!WorkQueue.casSlotToNull(a, k, t))
1994                         scan = true;
1995                     else {
1996                         q.setBaseOpaque(nextBase);
1997                         return t;
1998                     }
1999                 }
2000             }
2001             if (!scan && queues == qs)
2002                 break;
2003         }
2004         return null;
2005     }
2006 
2007     /**
2008      * Runs tasks until {@code isQuiescent()}. Rather than blocking
2009      * when tasks cannot be found, rescans until all others cannot
2010      * find tasks either.
2011      *
2012      * @param nanos max wait time (Long.MAX_VALUE if effectively untimed)
2013      * @param interruptible true if return on interrupt
2014      * @return positive if quiescent, negative if interrupted, else 0
2015      */
2016     final int helpQuiescePool(WorkQueue w, long nanos, boolean interruptible) {
2017         if (w == null)
2018             return 0;
2019         long startTime = System.nanoTime(), parkTime = 0L;
2020         int prevSrc = w.source, wsrc = prevSrc, cfg = w.config, r = cfg + 1;
2021         for (boolean active = true, locals = true;;) {
2022             boolean busy = false, scan = false;
2023             if (locals) {  // run local tasks before (re)polling
2024                 locals = false;
2025                 for (ForkJoinTask<?> u; (u = w.nextLocalTask(cfg)) != null;)
2026                     u.doExec();
2027             }
2028             WorkQueue[] qs = queues;
2029             int n = (qs == null) ? 0 : qs.length;
2030             for (int i = n; i > 0; --i, ++r) {
2031                 int j, b, cap; WorkQueue q; ForkJoinTask<?>[] a;
2032                 if ((q = qs[j = (n - 1) & r]) != null && q != w &&
2033                     (a = q.array) != null && (cap = a.length) > 0) {
2034                     int k = (cap - 1) & (b = q.base);
2035                     int nextBase = b + 1, src = j | SRC;
2036                     ForkJoinTask<?> t = WorkQueue.getSlot(a, k);
2037                     if (q.base != b)
2038                         busy = scan = true;
2039                     else if (t != null) {
2040                         busy = scan = true;
2041                         if (!active) {    // increment before taking
2042                             active = true;
2043                             getAndAddCtl(RC_UNIT);
2044                         }
2045                         if (WorkQueue.casSlotToNull(a, k, t)) {
2046                             q.base = nextBase;
2047                             w.source = src;
2048                             t.doExec();
2049                             w.source = wsrc = prevSrc;
2050                             locals = true;
2051                         }
2052                         break;
2053                     }
2054                     else if (!busy) {
2055                         if (q.top != b || a[nextBase & (cap - 1)] != null)
2056                             busy = scan = true;
2057                         else if (q.source != QUIET && q.phase >= 0)
2058                             busy = true;
2059                     }
2060                 }
2061             }
2062             VarHandle.acquireFence();
2063             if (!scan && queues == qs) {
2064                 boolean interrupted;
2065                 if (!busy) {
2066                     w.source = prevSrc;
2067                     if (!active)
2068                         getAndAddCtl(RC_UNIT);
2069                     return 1;
2070                 }
2071                 if (wsrc != QUIET)
2072                     w.source = wsrc = QUIET;
2073                 if (active) {                 // decrement
2074                     active = false;
2075                     parkTime = 0L;
2076                     getAndAddCtl(RC_MASK & -RC_UNIT);
2077                 }
2078                 else if (parkTime == 0L) {
2079                     parkTime = 1L << 10; // initially about 1 usec
2080                     Thread.yield();
2081                 }
2082                 else if ((interrupted = interruptible && Thread.interrupted()) ||
2083                          System.nanoTime() - startTime > nanos) {
2084                     getAndAddCtl(RC_UNIT);
2085                     return interrupted ? -1 : 0;
2086                 }
2087                 else {
2088                     LockSupport.parkNanos(this, parkTime);
2089                     if (parkTime < nanos >>> 8 && parkTime < 1L << 20)
2090                         parkTime <<= 1;  // max sleep approx 1 sec or 1% nanos
2091                 }
2092             }
2093         }
2094     }
2095 
2096     /**
2097      * Helps quiesce from external caller until done, interrupted, or timeout
2098      *
2099      * @param nanos max wait time (Long.MAX_VALUE if effectively untimed)
2100      * @param interruptible true if return on interrupt
2101      * @return positive if quiescent, negative if interrupted, else 0
2102      */
2103     final int externalHelpQuiescePool(long nanos, boolean interruptible) {
2104         for (long startTime = System.nanoTime(), parkTime = 0L;;) {
2105             ForkJoinTask<?> t;
2106             if ((t = pollScan(false)) != null) {
2107                 t.doExec();
2108                 parkTime = 0L;
2109             }
2110             else if (canStop())
2111                 return 1;
2112             else if (parkTime == 0L) {
2113                 parkTime = 1L << 10;
2114                 Thread.yield();
2115             }
2116             else if ((System.nanoTime() - startTime) > nanos)
2117                 return 0;
2118             else if (interruptible && Thread.interrupted())
2119                 return -1;
2120             else {
2121                 LockSupport.parkNanos(this, parkTime);
2122                 if (parkTime < nanos >>> 8 && parkTime < 1L << 20)
2123                     parkTime <<= 1;
2124             }
2125         }
2126     }
2127 
2128     /**
2129      * Gets and removes a local or stolen task for the given worker.
2130      *
2131      * @return a task, if available
2132      */
2133     final ForkJoinTask<?> nextTaskFor(WorkQueue w) {
2134         ForkJoinTask<?> t;
2135         if (w == null || (t = w.nextLocalTask(w.config)) == null)
2136             t = pollScan(false);
2137         return t;
2138     }
2139 
2140     // External operations
2141 
2142     /**
2143      * Finds and locks a WorkQueue for an external submitter, or
2144      * returns null if shutdown or terminating.
2145      */
2146     final WorkQueue submissionQueue() {
2147         int r;
2148         if ((r = ThreadLocalRandom.getProbe()) == 0) {
2149             ThreadLocalRandom.localInit();           // initialize caller's probe
2150             r = ThreadLocalRandom.getProbe();
2151         }
2152         for (int id = r << 1;;) {                    // even indices only
2153             int md = mode, n, i; WorkQueue q; ReentrantLock lock;
2154             WorkQueue[] qs = queues;
2155             if ((md & SHUTDOWN) != 0 || qs == null || (n = qs.length) <= 0)
2156                 return null;
2157             else if ((q = qs[i = (n - 1) & id]) == null) {
2158                 if ((lock = registrationLock) != null) {
2159                     WorkQueue w = new WorkQueue(id | SRC);
2160                     lock.lock();                    // install under lock
2161                     if (qs[i] == null)
2162                         qs[i] = w;                  // else lost race; discard
2163                     lock.unlock();
2164                 }
2165             }
2166             else if (!q.tryLock())                  // move and restart
2167                 id = (r = ThreadLocalRandom.advanceProbe(r)) << 1;
2168             else
2169                 return q;
2170         }
2171     }
2172 
2173     /**
2174      * Adds the given task to an external submission queue, or throws
2175      * exception if shutdown or terminating.
2176      *
2177      * @param task the task. Caller must ensure non-null.
2178      */
2179     final void externalPush(ForkJoinTask<?> task) {
2180         WorkQueue q;
2181         if ((q = submissionQueue()) == null)
2182             throw new RejectedExecutionException(); // shutdown or disabled
2183         else if (q.lockedPush(task))
2184             signalWork();
2185     }
2186 
2187     /**
2188      * Pushes a possibly-external submission.
2189      */
2190     private <T> ForkJoinTask<T> externalSubmit(ForkJoinTask<T> task) {
2191         Thread t; ForkJoinWorkerThread wt; WorkQueue q;
2192         if (task == null)
2193             throw new NullPointerException();
2194         if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) &&
2195             (q = (wt = (ForkJoinWorkerThread)t).workQueue) != null &&
2196             wt.pool == this)
2197             q.push(task, this);
2198         else
2199             externalPush(task);
2200         return task;
2201     }
2202 
2203     /**
2204      * Returns common pool queue for an external thread that has
2205      * possibly ever submitted a common pool task (nonzero probe), or
2206      * null if none.
2207      */
2208     static WorkQueue commonQueue() {
2209         ForkJoinPool p; WorkQueue[] qs;
2210         int r = ThreadLocalRandom.getProbe(), n;
2211         return ((p = common) != null && (qs = p.queues) != null &&
2212                 (n = qs.length) > 0 && r != 0) ?
2213             qs[(n - 1) & (r << 1)] : null;
2214     }
2215 
2216     /**
2217      * Returns queue for an external thread, if one exists
2218      */
2219     final WorkQueue externalQueue() {
2220         WorkQueue[] qs;
2221         int r = ThreadLocalRandom.getProbe(), n;
2222         return ((qs = queues) != null && (n = qs.length) > 0 && r != 0) ?
2223             qs[(n - 1) & (r << 1)] : null;
2224     }
2225 
2226     /**
2227      * If the given executor is a ForkJoinPool, poll and execute
2228      * AsynchronousCompletionTasks from worker's queue until none are
2229      * available or blocker is released.
2230      */
2231     static void helpAsyncBlocker(Executor e, ManagedBlocker blocker) {
2232         WorkQueue w = null; Thread t; ForkJoinWorkerThread wt;
2233         if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
2234             if ((wt = (ForkJoinWorkerThread)t).pool == e)
2235                 w = wt.workQueue;
2236         }
2237         else if (e instanceof ForkJoinPool)
2238             w = ((ForkJoinPool)e).externalQueue();
2239         if (w != null)
2240             w.helpAsyncBlocker(blocker);
2241     }
2242 
2243     /**
2244      * Returns a cheap heuristic guide for task partitioning when
2245      * programmers, frameworks, tools, or languages have little or no
2246      * idea about task granularity.  In essence, by offering this
2247      * method, we ask users only about tradeoffs in overhead vs
2248      * expected throughput and its variance, rather than how finely to
2249      * partition tasks.
2250      *
2251      * In a steady state strict (tree-structured) computation, each
2252      * thread makes available for stealing enough tasks for other
2253      * threads to remain active. Inductively, if all threads play by
2254      * the same rules, each thread should make available only a
2255      * constant number of tasks.
2256      *
2257      * The minimum useful constant is just 1. But using a value of 1
2258      * would require immediate replenishment upon each steal to
2259      * maintain enough tasks, which is infeasible.  Further,
2260      * partitionings/granularities of offered tasks should minimize
2261      * steal rates, which in general means that threads nearer the top
2262      * of computation tree should generate more than those nearer the
2263      * bottom. In perfect steady state, each thread is at
2264      * approximately the same level of computation tree. However,
2265      * producing extra tasks amortizes the uncertainty of progress and
2266      * diffusion assumptions.
2267      *
2268      * So, users will want to use values larger (but not much larger)
2269      * than 1 to both smooth over transient shortages and hedge
2270      * against uneven progress; as traded off against the cost of
2271      * extra task overhead. We leave the user to pick a threshold
2272      * value to compare with the results of this call to guide
2273      * decisions, but recommend values such as 3.
2274      *
2275      * When all threads are active, it is on average OK to estimate
2276      * surplus strictly locally. In steady-state, if one thread is
2277      * maintaining say 2 surplus tasks, then so are others. So we can
2278      * just use estimated queue length.  However, this strategy alone
2279      * leads to serious mis-estimates in some non-steady-state
2280      * conditions (ramp-up, ramp-down, other stalls). We can detect
2281      * many of these by further considering the number of "idle"
2282      * threads, that are known to have zero queued tasks, so
2283      * compensate by a factor of (#idle/#active) threads.
2284      */
2285     static int getSurplusQueuedTaskCount() {
2286         Thread t; ForkJoinWorkerThread wt; ForkJoinPool pool; WorkQueue q;
2287         if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) &&
2288             (pool = (wt = (ForkJoinWorkerThread)t).pool) != null &&
2289             (q = wt.workQueue) != null) {
2290             int p = pool.mode & SMASK;
2291             int a = p + (int)(pool.ctl >> RC_SHIFT);
2292             int n = q.top - q.base;
2293             return n - (a > (p >>>= 1) ? 0 :
2294                         a > (p >>>= 1) ? 1 :
2295                         a > (p >>>= 1) ? 2 :
2296                         a > (p >>>= 1) ? 4 :
2297                         8);
2298         }
2299         return 0;
2300     }
2301 
2302     // Termination
2303 
2304     /**
2305      * Possibly initiates and/or completes termination.
2306      *
2307      * @param now if true, unconditionally terminate, else only
2308      * if no work and no active workers
2309      * @param enable if true, terminate when next possible
2310      * @return true if terminating or terminated
2311      */
2312     private boolean tryTerminate(boolean now, boolean enable) {
2313         int md; // try to set SHUTDOWN, then STOP, then help terminate
2314         if (((md = mode) & SHUTDOWN) == 0) {
2315             if (!enable)
2316                 return false;
2317             md = getAndBitwiseOrMode(SHUTDOWN);
2318         }
2319         if ((md & STOP) == 0) {
2320             if (!now && !canStop())
2321                 return false;
2322             md = getAndBitwiseOrMode(STOP);
2323         }
2324         for (boolean rescan = true;;) { // repeat until no changes
2325             boolean changed = false;
2326             for (ForkJoinTask<?> t; (t = pollScan(false)) != null; ) {
2327                 changed = true;
2328                 ForkJoinTask.cancelIgnoringExceptions(t); // help cancel
2329             }
2330             WorkQueue[] qs; int n; WorkQueue q; Thread thread;
2331             if ((qs = queues) != null && (n = qs.length) > 0) {
2332                 for (int j = 1; j < n; j += 2) { // unblock other workers
2333                     if ((q = qs[j]) != null && (thread = q.owner) != null &&
2334                         !thread.isInterrupted()) {
2335                         changed = true;
2336                         try {
2337                             thread.interrupt();
2338                         } catch (Throwable ignore) {
2339                         }
2340                     }
2341                 }
2342             }
2343             ReentrantLock lock; Condition cond; // signal when no workers
2344             if (((md = mode) & TERMINATED) == 0 &&
2345                 (md & SMASK) + (short)(ctl >>> TC_SHIFT) <= 0 &&
2346                 (getAndBitwiseOrMode(TERMINATED) & TERMINATED) == 0 &&
2347                 (lock = registrationLock) != null) {
2348                 lock.lock();
2349                 if ((cond = termination) != null)
2350                     cond.signalAll();
2351                 lock.unlock();
2352             }
2353             if (changed)
2354                 rescan = true;
2355             else if (rescan)
2356                 rescan = false;
2357             else
2358                 break;
2359         }
2360         return true;
2361     }
2362 
2363     // Exported methods
2364 
2365     // Constructors
2366 
2367     /**
2368      * Creates a {@code ForkJoinPool} with parallelism equal to {@link
2369      * java.lang.Runtime#availableProcessors}, using defaults for all
2370      * other parameters (see {@link #ForkJoinPool(int,
2371      * ForkJoinWorkerThreadFactory, UncaughtExceptionHandler, boolean,
2372      * int, int, int, Predicate, long, TimeUnit)}).
2373      *
2374      * @throws SecurityException if a security manager exists and
2375      *         the caller is not permitted to modify threads
2376      *         because it does not hold {@link
2377      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2378      */
2379     public ForkJoinPool() {
2380         this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
2381              defaultForkJoinWorkerThreadFactory, null, false,
2382              0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS);
2383     }
2384 
2385     /**
2386      * Creates a {@code ForkJoinPool} with the indicated parallelism
2387      * level, using defaults for all other parameters (see {@link
2388      * #ForkJoinPool(int, ForkJoinWorkerThreadFactory,
2389      * UncaughtExceptionHandler, boolean, int, int, int, Predicate,
2390      * long, TimeUnit)}).
2391      *
2392      * @param parallelism the parallelism level
2393      * @throws IllegalArgumentException if parallelism less than or
2394      *         equal to zero, or greater than implementation limit
2395      * @throws SecurityException if a security manager exists and
2396      *         the caller is not permitted to modify threads
2397      *         because it does not hold {@link
2398      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2399      */
2400     public ForkJoinPool(int parallelism) {
2401         this(parallelism, defaultForkJoinWorkerThreadFactory, null, false,
2402              0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS);
2403     }
2404 
2405     /**
2406      * Creates a {@code ForkJoinPool} with the given parameters (using
2407      * defaults for others -- see {@link #ForkJoinPool(int,
2408      * ForkJoinWorkerThreadFactory, UncaughtExceptionHandler, boolean,
2409      * int, int, int, Predicate, long, TimeUnit)}).
2410      *
2411      * @param parallelism the parallelism level. For default value,
2412      * use {@link java.lang.Runtime#availableProcessors}.
2413      * @param factory the factory for creating new threads. For default value,
2414      * use {@link #defaultForkJoinWorkerThreadFactory}.
2415      * @param handler the handler for internal worker threads that
2416      * terminate due to unrecoverable errors encountered while executing
2417      * tasks. For default value, use {@code null}.
2418      * @param asyncMode if true,
2419      * establishes local first-in-first-out scheduling mode for forked
2420      * tasks that are never joined. This mode may be more appropriate
2421      * than default locally stack-based mode in applications in which
2422      * worker threads only process event-style asynchronous tasks.
2423      * For default value, use {@code false}.
2424      * @throws IllegalArgumentException if parallelism less than or
2425      *         equal to zero, or greater than implementation limit
2426      * @throws NullPointerException if the factory is null
2427      * @throws SecurityException if a security manager exists and
2428      *         the caller is not permitted to modify threads
2429      *         because it does not hold {@link
2430      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2431      */
2432     public ForkJoinPool(int parallelism,
2433                         ForkJoinWorkerThreadFactory factory,
2434                         UncaughtExceptionHandler handler,
2435                         boolean asyncMode) {
2436         this(parallelism, factory, handler, asyncMode,
2437              0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS);
2438     }
2439 
2440     /**
2441      * Creates a {@code ForkJoinPool} with the given parameters.
2442      *
2443      * @param parallelism the parallelism level. For default value,
2444      * use {@link java.lang.Runtime#availableProcessors}.
2445      *
2446      * @param factory the factory for creating new threads. For
2447      * default value, use {@link #defaultForkJoinWorkerThreadFactory}.
2448      *
2449      * @param handler the handler for internal worker threads that
2450      * terminate due to unrecoverable errors encountered while
2451      * executing tasks. For default value, use {@code null}.
2452      *
2453      * @param asyncMode if true, establishes local first-in-first-out
2454      * scheduling mode for forked tasks that are never joined. This
2455      * mode may be more appropriate than default locally stack-based
2456      * mode in applications in which worker threads only process
2457      * event-style asynchronous tasks.  For default value, use {@code
2458      * false}.
2459      *
2460      * @param corePoolSize the number of threads to keep in the pool
2461      * (unless timed out after an elapsed keep-alive). Normally (and
2462      * by default) this is the same value as the parallelism level,
2463      * but may be set to a larger value to reduce dynamic overhead if
2464      * tasks regularly block. Using a smaller value (for example
2465      * {@code 0}) has the same effect as the default.
2466      *
2467      * @param maximumPoolSize the maximum number of threads allowed.
2468      * When the maximum is reached, attempts to replace blocked
2469      * threads fail.  (However, because creation and termination of
2470      * different threads may overlap, and may be managed by the given
2471      * thread factory, this value may be transiently exceeded.)  To
2472      * arrange the same value as is used by default for the common
2473      * pool, use {@code 256} plus the {@code parallelism} level. (By
2474      * default, the common pool allows a maximum of 256 spare
2475      * threads.)  Using a value (for example {@code
2476      * Integer.MAX_VALUE}) larger than the implementation's total
2477      * thread limit has the same effect as using this limit (which is
2478      * the default).
2479      *
2480      * @param minimumRunnable the minimum allowed number of core
2481      * threads not blocked by a join or {@link ManagedBlocker}.  To
2482      * ensure progress, when too few unblocked threads exist and
2483      * unexecuted tasks may exist, new threads are constructed, up to
2484      * the given maximumPoolSize.  For the default value, use {@code
2485      * 1}, that ensures liveness.  A larger value might improve
2486      * throughput in the presence of blocked activities, but might
2487      * not, due to increased overhead.  A value of zero may be
2488      * acceptable when submitted tasks cannot have dependencies
2489      * requiring additional threads.
2490      *
2491      * @param saturate if non-null, a predicate invoked upon attempts
2492      * to create more than the maximum total allowed threads.  By
2493      * default, when a thread is about to block on a join or {@link
2494      * ManagedBlocker}, but cannot be replaced because the
2495      * maximumPoolSize would be exceeded, a {@link
2496      * RejectedExecutionException} is thrown.  But if this predicate
2497      * returns {@code true}, then no exception is thrown, so the pool
2498      * continues to operate with fewer than the target number of
2499      * runnable threads, which might not ensure progress.
2500      *
2501      * @param keepAliveTime the elapsed time since last use before
2502      * a thread is terminated (and then later replaced if needed).
2503      * For the default value, use {@code 60, TimeUnit.SECONDS}.
2504      *
2505      * @param unit the time unit for the {@code keepAliveTime} argument
2506      *
2507      * @throws IllegalArgumentException if parallelism is less than or
2508      *         equal to zero, or is greater than implementation limit,
2509      *         or if maximumPoolSize is less than parallelism,
2510      *         of if the keepAliveTime is less than or equal to zero.
2511      * @throws NullPointerException if the factory is null
2512      * @throws SecurityException if a security manager exists and
2513      *         the caller is not permitted to modify threads
2514      *         because it does not hold {@link
2515      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2516      * @since 9
2517      */
2518     public ForkJoinPool(int parallelism,
2519                         ForkJoinWorkerThreadFactory factory,
2520                         UncaughtExceptionHandler handler,
2521                         boolean asyncMode,
2522                         int corePoolSize,
2523                         int maximumPoolSize,
2524                         int minimumRunnable,
2525                         Predicate<? super ForkJoinPool> saturate,
2526                         long keepAliveTime,
2527                         TimeUnit unit) {
2528         checkPermission();
2529         int p = parallelism;
2530         if (p <= 0 || p > MAX_CAP || p > maximumPoolSize || keepAliveTime <= 0L)
2531             throw new IllegalArgumentException();
2532         if (factory == null || unit == null)
2533             throw new NullPointerException();
2534         this.factory = factory;
2535         this.ueh = handler;
2536         this.saturate = saturate;
2537         this.keepAlive = Math.max(unit.toMillis(keepAliveTime), TIMEOUT_SLOP);
2538         int size = 1 << (33 - Integer.numberOfLeadingZeros(p - 1));
2539         int corep = Math.min(Math.max(corePoolSize, p), MAX_CAP);
2540         int maxSpares = Math.min(maximumPoolSize, MAX_CAP) - p;
2541         int minAvail = Math.min(Math.max(minimumRunnable, 0), MAX_CAP);
2542         this.bounds = ((minAvail - p) & SMASK) | (maxSpares << SWIDTH);
2543         this.mode = p | (asyncMode ? FIFO : 0);
2544         this.ctl = ((((long)(-corep) << TC_SHIFT) & TC_MASK) |
2545                     (((long)(-p)     << RC_SHIFT) & RC_MASK));
2546         this.registrationLock = new ReentrantLock();
2547         this.queues = new WorkQueue[size];
2548         String pid = Integer.toString(getAndAddPoolIds(1) + 1);
2549         this.workerNamePrefix = "ForkJoinPool-" + pid + "-worker-";
2550     }
2551 
2552     // helper method for commonPool constructor
2553     private static Object newInstanceFromSystemProperty(String property)
2554         throws ReflectiveOperationException {
2555         String className = System.getProperty(property);
2556         return (className == null)
2557             ? null
2558             : ClassLoader.getSystemClassLoader().loadClass(className)
2559             .getConstructor().newInstance();
2560     }
2561 
2562     /**
2563      * Constructor for common pool using parameters possibly
2564      * overridden by system properties
2565      */
2566     private ForkJoinPool(byte forCommonPoolOnly) {
2567         int parallelism = Math.max(1, Runtime.getRuntime().availableProcessors() - 1);
2568         ForkJoinWorkerThreadFactory fac = null;
2569         UncaughtExceptionHandler handler = null;
2570         try {  // ignore exceptions in accessing/parsing properties
2571             fac = (ForkJoinWorkerThreadFactory) newInstanceFromSystemProperty(
2572                 "java.util.concurrent.ForkJoinPool.common.threadFactory");
2573             handler = (UncaughtExceptionHandler) newInstanceFromSystemProperty(
2574                 "java.util.concurrent.ForkJoinPool.common.exceptionHandler");
2575             String pp = System.getProperty
2576                 ("java.util.concurrent.ForkJoinPool.common.parallelism");
2577             if (pp != null)
2578                 parallelism = Integer.parseInt(pp);
2579         } catch (Exception ignore) {
2580         }
2581         this.ueh = handler;
2582         this.keepAlive = DEFAULT_KEEPALIVE;
2583         this.saturate = null;
2584         this.workerNamePrefix = null;
2585         int p = Math.min(Math.max(parallelism, 0), MAX_CAP), size;
2586         this.mode = p;
2587         if (p > 0) {
2588             size = 1 << (33 - Integer.numberOfLeadingZeros(p - 1));
2589             this.bounds = ((1 - p) & SMASK) | (COMMON_MAX_SPARES << SWIDTH);
2590             this.ctl = ((((long)(-p) << TC_SHIFT) & TC_MASK) |
2591                         (((long)(-p) << RC_SHIFT) & RC_MASK));
2592         } else {  // zero min, max, spare counts, 1 slot
2593             size = 1;
2594             this.bounds = 0;
2595             this.ctl = 0L;
2596         }
2597         this.factory = (fac != null) ? fac :
2598             new DefaultCommonPoolForkJoinWorkerThreadFactory();
2599         this.queues = new WorkQueue[size];
2600         this.registrationLock = new ReentrantLock();
2601     }
2602 
2603     /**
2604      * Returns the common pool instance. This pool is statically
2605      * constructed; its run state is unaffected by attempts to {@link
2606      * #shutdown} or {@link #shutdownNow}. However this pool and any
2607      * ongoing processing are automatically terminated upon program
2608      * {@link System#exit}.  Any program that relies on asynchronous
2609      * task processing to complete before program termination should
2610      * invoke {@code commonPool().}{@link #awaitQuiescence awaitQuiescence},
2611      * before exit.
2612      *
2613      * @return the common pool instance
2614      * @since 1.8
2615      */
2616     public static ForkJoinPool commonPool() {
2617         // assert common != null : "static init error";
2618         return common;
2619     }
2620 
2621     // Execution methods
2622 
2623     /**
2624      * Performs the given task, returning its result upon completion.
2625      * If the computation encounters an unchecked Exception or Error,
2626      * it is rethrown as the outcome of this invocation.  Rethrown
2627      * exceptions behave in the same way as regular exceptions, but,
2628      * when possible, contain stack traces (as displayed for example
2629      * using {@code ex.printStackTrace()}) of both the current thread
2630      * as well as the thread actually encountering the exception;
2631      * minimally only the latter.
2632      *
2633      * @param task the task
2634      * @param <T> the type of the task's result
2635      * @return the task's result
2636      * @throws NullPointerException if the task is null
2637      * @throws RejectedExecutionException if the task cannot be
2638      *         scheduled for execution
2639      */
2640     public <T> T invoke(ForkJoinTask<T> task) {
2641         externalSubmit(task);
2642         return task.joinForPoolInvoke(this);
2643     }
2644 
2645     /**
2646      * Arranges for (asynchronous) execution of the given task.
2647      *
2648      * @param task the task
2649      * @throws NullPointerException if the task is null
2650      * @throws RejectedExecutionException if the task cannot be
2651      *         scheduled for execution
2652      */
2653     public void execute(ForkJoinTask<?> task) {
2654         externalSubmit(task);
2655     }
2656 
2657     // AbstractExecutorService methods
2658 
2659     /**
2660      * @throws NullPointerException if the task is null
2661      * @throws RejectedExecutionException if the task cannot be
2662      *         scheduled for execution
2663      */
2664     @Override
2665     @SuppressWarnings("unchecked")
2666     public void execute(Runnable task) {
2667         externalSubmit((task instanceof ForkJoinTask<?>)
2668                        ? (ForkJoinTask<Void>) task // avoid re-wrap
2669                        : new ForkJoinTask.RunnableExecuteAction(task));
2670     }
2671 
2672     /**
2673      * Submits a ForkJoinTask for execution.
2674      *
2675      * @param task the task to submit
2676      * @param <T> the type of the task's result
2677      * @return the task
2678      * @throws NullPointerException if the task is null
2679      * @throws RejectedExecutionException if the task cannot be
2680      *         scheduled for execution
2681      */
2682     public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
2683         return externalSubmit(task);
2684     }
2685 
2686     /**
2687      * @throws NullPointerException if the task is null
2688      * @throws RejectedExecutionException if the task cannot be
2689      *         scheduled for execution
2690      */
2691     @Override
2692     public <T> ForkJoinTask<T> submit(Callable<T> task) {
2693         return externalSubmit(new ForkJoinTask.AdaptedCallable<T>(task));
2694     }
2695 
2696     /**
2697      * @throws NullPointerException if the task is null
2698      * @throws RejectedExecutionException if the task cannot be
2699      *         scheduled for execution
2700      */
2701     @Override
2702     public <T> ForkJoinTask<T> submit(Runnable task, T result) {
2703         return externalSubmit(new ForkJoinTask.AdaptedRunnable<T>(task, result));
2704     }
2705 
2706     /**
2707      * @throws NullPointerException if the task is null
2708      * @throws RejectedExecutionException if the task cannot be
2709      *         scheduled for execution
2710      */
2711     @Override
2712     @SuppressWarnings("unchecked")
2713     public ForkJoinTask<?> submit(Runnable task) {
2714         return externalSubmit((task instanceof ForkJoinTask<?>)
2715             ? (ForkJoinTask<Void>) task // avoid re-wrap
2716             : new ForkJoinTask.AdaptedRunnableAction(task));
2717     }
2718 
2719     /**
2720      * @throws NullPointerException       {@inheritDoc}
2721      * @throws RejectedExecutionException {@inheritDoc}
2722      */
2723     @Override
2724     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
2725         ArrayList<Future<T>> futures = new ArrayList<>(tasks.size());
2726         try {
2727             for (Callable<T> t : tasks) {
2728                 ForkJoinTask<T> f =
2729                     new ForkJoinTask.AdaptedInterruptibleCallable<T>(t);
2730                 futures.add(f);
2731                 externalSubmit(f);
2732             }
2733             for (int i = futures.size() - 1; i >= 0; --i)
2734                 ((ForkJoinTask<?>)futures.get(i)).awaitPoolInvoke(this);
2735             return futures;
2736         } catch (Throwable t) {
2737             for (Future<T> e : futures)
2738                 ForkJoinTask.cancelIgnoringExceptions(e);
2739             throw t;
2740         }
2741     }
2742 
2743     @Override
2744     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
2745                                          long timeout, TimeUnit unit)
2746         throws InterruptedException {
2747         long nanos = unit.toNanos(timeout);
2748         ArrayList<Future<T>> futures = new ArrayList<>(tasks.size());
2749         try {
2750             for (Callable<T> t : tasks) {
2751                 ForkJoinTask<T> f =
2752                     new ForkJoinTask.AdaptedInterruptibleCallable<T>(t);
2753                 futures.add(f);
2754                 externalSubmit(f);
2755             }
2756             long startTime = System.nanoTime(), ns = nanos;
2757             boolean timedOut = (ns < 0L);
2758             for (int i = futures.size() - 1; i >= 0; --i) {
2759                 Future<T> f = futures.get(i);
2760                 if (!f.isDone()) {
2761                     if (timedOut)
2762                         ForkJoinTask.cancelIgnoringExceptions(f);
2763                     else {
2764                         ((ForkJoinTask<T>)f).awaitPoolInvoke(this, ns);
2765                         if ((ns = nanos - (System.nanoTime() - startTime)) < 0L)
2766                             timedOut = true;
2767                     }
2768                 }
2769             }
2770             return futures;
2771         } catch (Throwable t) {
2772             for (Future<T> e : futures)
2773                 ForkJoinTask.cancelIgnoringExceptions(e);
2774             throw t;
2775         }
2776     }
2777 
2778     // Task to hold results from InvokeAnyTasks
2779     static final class InvokeAnyRoot<E> extends ForkJoinTask<E> {
2780         private static final long serialVersionUID = 2838392045355241008L;
2781         @SuppressWarnings("serial") // Conditionally serializable
2782         volatile E result;
2783         final AtomicInteger count;  // in case all throw
2784         final ForkJoinPool pool;    // to check shutdown while collecting
2785         InvokeAnyRoot(int n, ForkJoinPool p) {
2786             pool = p;
2787             count = new AtomicInteger(n);
2788         }
2789         final void tryComplete(Callable<E> c) { // called by InvokeAnyTasks
2790             Throwable ex = null;
2791             boolean failed;
2792             if (c == null || Thread.interrupted() ||
2793                 (pool != null && pool.mode < 0))
2794                 failed = true;
2795             else if (isDone())
2796                 failed = false;
2797             else {
2798                 try {
2799                     complete(c.call());
2800                     failed = false;
2801                 } catch (Throwable tx) {
2802                     ex = tx;
2803                     failed = true;
2804                 }
2805             }
2806             if ((pool != null && pool.mode < 0) ||
2807                 (failed && count.getAndDecrement() <= 1))
2808                 trySetThrown(ex != null ? ex : new CancellationException());
2809         }
2810         public final boolean exec()         { return false; } // never forked
2811         public final E getRawResult()       { return result; }
2812         public final void setRawResult(E v) { result = v; }
2813     }
2814 
2815     // Variant of AdaptedInterruptibleCallable with results in InvokeAnyRoot
2816     static final class InvokeAnyTask<E> extends ForkJoinTask<E> {
2817         private static final long serialVersionUID = 2838392045355241008L;
2818         final InvokeAnyRoot<E> root;
2819         @SuppressWarnings("serial") // Conditionally serializable
2820         final Callable<E> callable;
2821         transient volatile Thread runner;
2822         InvokeAnyTask(InvokeAnyRoot<E> root, Callable<E> callable) {
2823             this.root = root;
2824             this.callable = callable;
2825         }
2826         public final boolean exec() {
2827             Thread.interrupted();
2828             runner = Thread.currentThread();
2829             root.tryComplete(callable);
2830             runner = null;
2831             Thread.interrupted();
2832             return true;
2833         }
2834         public final boolean cancel(boolean mayInterruptIfRunning) {
2835             Thread t;
2836             boolean stat = super.cancel(false);
2837             if (mayInterruptIfRunning && (t = runner) != null) {
2838                 try {
2839                     t.interrupt();
2840                 } catch (Throwable ignore) {
2841                 }
2842             }
2843             return stat;
2844         }
2845         public final void setRawResult(E v) {} // unused
2846         public final E getRawResult()       { return null; }
2847     }
2848 
2849     @Override
2850     public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
2851         throws InterruptedException, ExecutionException {
2852         int n = tasks.size();
2853         if (n <= 0)
2854             throw new IllegalArgumentException();
2855         InvokeAnyRoot<T> root = new InvokeAnyRoot<T>(n, this);
2856         ArrayList<InvokeAnyTask<T>> fs = new ArrayList<>(n);
2857         try {
2858             for (Callable<T> c : tasks) {
2859                 if (c == null)
2860                     throw new NullPointerException();
2861                 InvokeAnyTask<T> f = new InvokeAnyTask<T>(root, c);
2862                 fs.add(f);
2863                 externalSubmit(f);
2864                 if (root.isDone())
2865                     break;
2866             }
2867             return root.getForPoolInvoke(this);
2868         } finally {
2869             for (InvokeAnyTask<T> f : fs)
2870                 ForkJoinTask.cancelIgnoringExceptions(f);
2871         }
2872     }
2873 
2874     @Override
2875     public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
2876                            long timeout, TimeUnit unit)
2877         throws InterruptedException, ExecutionException, TimeoutException {
2878         long nanos = unit.toNanos(timeout);
2879         int n = tasks.size();
2880         if (n <= 0)
2881             throw new IllegalArgumentException();
2882         InvokeAnyRoot<T> root = new InvokeAnyRoot<T>(n, this);
2883         ArrayList<InvokeAnyTask<T>> fs = new ArrayList<>(n);
2884         try {
2885             for (Callable<T> c : tasks) {
2886                 if (c == null)
2887                     throw new NullPointerException();
2888                 InvokeAnyTask<T> f = new InvokeAnyTask<T>(root, c);
2889                 fs.add(f);
2890                 externalSubmit(f);
2891                 if (root.isDone())
2892                     break;
2893             }
2894             return root.getForPoolInvoke(this, nanos);
2895         } finally {
2896             for (InvokeAnyTask<T> f : fs)
2897                 ForkJoinTask.cancelIgnoringExceptions(f);
2898         }
2899     }
2900 
2901     /**
2902      * Returns the factory used for constructing new workers.
2903      *
2904      * @return the factory used for constructing new workers
2905      */
2906     public ForkJoinWorkerThreadFactory getFactory() {
2907         return factory;
2908     }
2909 
2910     /**
2911      * Returns the handler for internal worker threads that terminate
2912      * due to unrecoverable errors encountered while executing tasks.
2913      *
2914      * @return the handler, or {@code null} if none
2915      */
2916     public UncaughtExceptionHandler getUncaughtExceptionHandler() {
2917         return ueh;
2918     }
2919 
2920     /**
2921      * Returns the targeted parallelism level of this pool.
2922      *
2923      * @return the targeted parallelism level of this pool
2924      */
2925     public int getParallelism() {
2926         int par = mode & SMASK;
2927         return (par > 0) ? par : 1;
2928     }
2929 
2930     /**
2931      * Returns the targeted parallelism level of the common pool.
2932      *
2933      * @return the targeted parallelism level of the common pool
2934      * @since 1.8
2935      */
2936     public static int getCommonPoolParallelism() {
2937         return COMMON_PARALLELISM;
2938     }
2939 
2940     /**
2941      * Returns the number of worker threads that have started but not
2942      * yet terminated.  The result returned by this method may differ
2943      * from {@link #getParallelism} when threads are created to
2944      * maintain parallelism when others are cooperatively blocked.
2945      *
2946      * @return the number of worker threads
2947      */
2948     public int getPoolSize() {
2949         return ((mode & SMASK) + (short)(ctl >>> TC_SHIFT));
2950     }
2951 
2952     /**
2953      * Returns {@code true} if this pool uses local first-in-first-out
2954      * scheduling mode for forked tasks that are never joined.
2955      *
2956      * @return {@code true} if this pool uses async mode
2957      */
2958     public boolean getAsyncMode() {
2959         return (mode & FIFO) != 0;
2960     }
2961 
2962     /**
2963      * Returns an estimate of the number of worker threads that are
2964      * not blocked waiting to join tasks or for other managed
2965      * synchronization. This method may overestimate the
2966      * number of running threads.
2967      *
2968      * @return the number of worker threads
2969      */
2970     public int getRunningThreadCount() {
2971         VarHandle.acquireFence();
2972         WorkQueue[] qs; WorkQueue q;
2973         int rc = 0;
2974         if ((qs = queues) != null) {
2975             for (int i = 1; i < qs.length; i += 2) {
2976                 if ((q = qs[i]) != null && q.isApparentlyUnblocked())
2977                     ++rc;
2978             }
2979         }
2980         return rc;
2981     }
2982 
2983     /**
2984      * Returns an estimate of the number of threads that are currently
2985      * stealing or executing tasks. This method may overestimate the
2986      * number of active threads.
2987      *
2988      * @return the number of active threads
2989      */
2990     public int getActiveThreadCount() {
2991         int r = (mode & SMASK) + (int)(ctl >> RC_SHIFT);
2992         return (r <= 0) ? 0 : r; // suppress momentarily negative values
2993     }
2994 
2995     /**
2996      * Returns {@code true} if all worker threads are currently idle.
2997      * An idle worker is one that cannot obtain a task to execute
2998      * because none are available to steal from other threads, and
2999      * there are no pending submissions to the pool. This method is
3000      * conservative; it might not return {@code true} immediately upon
3001      * idleness of all threads, but will eventually become true if
3002      * threads remain inactive.
3003      *
3004      * @return {@code true} if all threads are currently idle
3005      */
3006     public boolean isQuiescent() {
3007         return canStop();
3008     }
3009 
3010     /**
3011      * Returns an estimate of the total number of completed tasks that
3012      * were executed by a thread other than their submitter. The
3013      * reported value underestimates the actual total number of steals
3014      * when the pool is not quiescent. This value may be useful for
3015      * monitoring and tuning fork/join programs: in general, steal
3016      * counts should be high enough to keep threads busy, but low
3017      * enough to avoid overhead and contention across threads.
3018      *
3019      * @return the number of steals
3020      */
3021     public long getStealCount() {
3022         long count = stealCount;
3023         WorkQueue[] qs; WorkQueue q;
3024         if ((qs = queues) != null) {
3025             for (int i = 1; i < qs.length; i += 2) {
3026                 if ((q = qs[i]) != null)
3027                     count += (long)q.nsteals & 0xffffffffL;
3028             }
3029         }
3030         return count;
3031     }
3032 
3033     /**
3034      * Returns an estimate of the total number of tasks currently held
3035      * in queues by worker threads (but not including tasks submitted
3036      * to the pool that have not begun executing). This value is only
3037      * an approximation, obtained by iterating across all threads in
3038      * the pool. This method may be useful for tuning task
3039      * granularities.
3040      *
3041      * @return the number of queued tasks
3042      */
3043     public long getQueuedTaskCount() {
3044         VarHandle.acquireFence();
3045         WorkQueue[] qs; WorkQueue q;
3046         int count = 0;
3047         if ((qs = queues) != null) {
3048             for (int i = 1; i < qs.length; i += 2) {
3049                 if ((q = qs[i]) != null)
3050                     count += q.queueSize();
3051             }
3052         }
3053         return count;
3054     }
3055 
3056     /**
3057      * Returns an estimate of the number of tasks submitted to this
3058      * pool that have not yet begun executing.  This method may take
3059      * time proportional to the number of submissions.
3060      *
3061      * @return the number of queued submissions
3062      */
3063     public int getQueuedSubmissionCount() {
3064         VarHandle.acquireFence();
3065         WorkQueue[] qs; WorkQueue q;
3066         int count = 0;
3067         if ((qs = queues) != null) {
3068             for (int i = 0; i < qs.length; i += 2) {
3069                 if ((q = qs[i]) != null)
3070                     count += q.queueSize();
3071             }
3072         }
3073         return count;
3074     }
3075 
3076     /**
3077      * Returns {@code true} if there are any tasks submitted to this
3078      * pool that have not yet begun executing.
3079      *
3080      * @return {@code true} if there are any queued submissions
3081      */
3082     public boolean hasQueuedSubmissions() {
3083         VarHandle.acquireFence();
3084         WorkQueue[] qs; WorkQueue q;
3085         if ((qs = queues) != null) {
3086             for (int i = 0; i < qs.length; i += 2) {
3087                 if ((q = qs[i]) != null && !q.isEmpty())
3088                     return true;
3089             }
3090         }
3091         return false;
3092     }
3093 
3094     /**
3095      * Removes and returns the next unexecuted submission if one is
3096      * available.  This method may be useful in extensions to this
3097      * class that re-assign work in systems with multiple pools.
3098      *
3099      * @return the next submission, or {@code null} if none
3100      */
3101     protected ForkJoinTask<?> pollSubmission() {
3102         return pollScan(true);
3103     }
3104 
3105     /**
3106      * Removes all available unexecuted submitted and forked tasks
3107      * from scheduling queues and adds them to the given collection,
3108      * without altering their execution status. These may include
3109      * artificially generated or wrapped tasks. This method is
3110      * designed to be invoked only when the pool is known to be
3111      * quiescent. Invocations at other times may not remove all
3112      * tasks. A failure encountered while attempting to add elements
3113      * to collection {@code c} may result in elements being in
3114      * neither, either or both collections when the associated
3115      * exception is thrown.  The behavior of this operation is
3116      * undefined if the specified collection is modified while the
3117      * operation is in progress.
3118      *
3119      * @param c the collection to transfer elements into
3120      * @return the number of elements transferred
3121      */
3122     protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
3123         int count = 0;
3124         for (ForkJoinTask<?> t; (t = pollScan(false)) != null; ) {
3125             c.add(t);
3126             ++count;
3127         }
3128         return count;
3129     }
3130 
3131     /**
3132      * Returns a string identifying this pool, as well as its state,
3133      * including indications of run state, parallelism level, and
3134      * worker and task counts.
3135      *
3136      * @return a string identifying this pool, as well as its state
3137      */
3138     public String toString() {
3139         // Use a single pass through queues to collect counts
3140         int md = mode; // read volatile fields first
3141         long c = ctl;
3142         long st = stealCount;
3143         long qt = 0L, ss = 0L; int rc = 0;
3144         WorkQueue[] qs; WorkQueue q;
3145         if ((qs = queues) != null) {
3146             for (int i = 0; i < qs.length; ++i) {
3147                 if ((q = qs[i]) != null) {
3148                     int size = q.queueSize();
3149                     if ((i & 1) == 0)
3150                         ss += size;
3151                     else {
3152                         qt += size;
3153                         st += (long)q.nsteals & 0xffffffffL;
3154                         if (q.isApparentlyUnblocked())
3155                             ++rc;
3156                     }
3157                 }
3158             }
3159         }
3160 
3161         int pc = (md & SMASK);
3162         int tc = pc + (short)(c >>> TC_SHIFT);
3163         int ac = pc + (int)(c >> RC_SHIFT);
3164         if (ac < 0) // ignore transient negative
3165             ac = 0;
3166         String level = ((md & TERMINATED) != 0 ? "Terminated" :
3167                         (md & STOP)       != 0 ? "Terminating" :
3168                         (md & SHUTDOWN)   != 0 ? "Shutting down" :
3169                         "Running");
3170         return super.toString() +
3171             "[" + level +
3172             ", parallelism = " + pc +
3173             ", size = " + tc +
3174             ", active = " + ac +
3175             ", running = " + rc +
3176             ", steals = " + st +
3177             ", tasks = " + qt +
3178             ", submissions = " + ss +
3179             "]";
3180     }
3181 
3182     /**
3183      * Possibly initiates an orderly shutdown in which previously
3184      * submitted tasks are executed, but no new tasks will be
3185      * accepted. Invocation has no effect on execution state if this
3186      * is the {@link #commonPool()}, and no additional effect if
3187      * already shut down.  Tasks that are in the process of being
3188      * submitted concurrently during the course of this method may or
3189      * may not be rejected.
3190      *
3191      * @throws SecurityException if a security manager exists and
3192      *         the caller is not permitted to modify threads
3193      *         because it does not hold {@link
3194      *         java.lang.RuntimePermission}{@code ("modifyThread")}
3195      */
3196     public void shutdown() {
3197         checkPermission();
3198         if (this != common)
3199             tryTerminate(false, true);
3200     }
3201 
3202     /**
3203      * Possibly attempts to cancel and/or stop all tasks, and reject
3204      * all subsequently submitted tasks.  Invocation has no effect on
3205      * execution state if this is the {@link #commonPool()}, and no
3206      * additional effect if already shut down. Otherwise, tasks that
3207      * are in the process of being submitted or executed concurrently
3208      * during the course of this method may or may not be
3209      * rejected. This method cancels both existing and unexecuted
3210      * tasks, in order to permit termination in the presence of task
3211      * dependencies. So the method always returns an empty list
3212      * (unlike the case for some other Executors).
3213      *
3214      * @return an empty list
3215      * @throws SecurityException if a security manager exists and
3216      *         the caller is not permitted to modify threads
3217      *         because it does not hold {@link
3218      *         java.lang.RuntimePermission}{@code ("modifyThread")}
3219      */
3220     public List<Runnable> shutdownNow() {
3221         checkPermission();
3222         if (this != common)
3223             tryTerminate(true, true);
3224         return Collections.emptyList();
3225     }
3226 
3227     /**
3228      * Returns {@code true} if all tasks have completed following shut down.
3229      *
3230      * @return {@code true} if all tasks have completed following shut down
3231      */
3232     public boolean isTerminated() {
3233         return (mode & TERMINATED) != 0;
3234     }
3235 
3236     /**
3237      * Returns {@code true} if the process of termination has
3238      * commenced but not yet completed.  This method may be useful for
3239      * debugging. A return of {@code true} reported a sufficient
3240      * period after shutdown may indicate that submitted tasks have
3241      * ignored or suppressed interruption, or are waiting for I/O,
3242      * causing this executor not to properly terminate. (See the
3243      * advisory notes for class {@link ForkJoinTask} stating that
3244      * tasks should not normally entail blocking operations.  But if
3245      * they do, they must abort them on interrupt.)
3246      *
3247      * @return {@code true} if terminating but not yet terminated
3248      */
3249     public boolean isTerminating() {
3250         return (mode & (STOP | TERMINATED)) == STOP;
3251     }
3252 
3253     /**
3254      * Returns {@code true} if this pool has been shut down.
3255      *
3256      * @return {@code true} if this pool has been shut down
3257      */
3258     public boolean isShutdown() {
3259         return (mode & SHUTDOWN) != 0;
3260     }
3261 
3262     /**
3263      * Blocks until all tasks have completed execution after a
3264      * shutdown request, or the timeout occurs, or the current thread
3265      * is interrupted, whichever happens first. Because the {@link
3266      * #commonPool()} never terminates until program shutdown, when
3267      * applied to the common pool, this method is equivalent to {@link
3268      * #awaitQuiescence(long, TimeUnit)} but always returns {@code false}.
3269      *
3270      * @param timeout the maximum time to wait
3271      * @param unit the time unit of the timeout argument
3272      * @return {@code true} if this executor terminated and
3273      *         {@code false} if the timeout elapsed before termination
3274      * @throws InterruptedException if interrupted while waiting
3275      */
3276     public boolean awaitTermination(long timeout, TimeUnit unit)
3277         throws InterruptedException {
3278         ReentrantLock lock; Condition cond;
3279         long nanos = unit.toNanos(timeout);
3280         boolean terminated = false;
3281         if (this == common) {
3282             Thread t; ForkJoinWorkerThread wt; int q;
3283             if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread &&
3284                 (wt = (ForkJoinWorkerThread)t).pool == this)
3285                 q = helpQuiescePool(wt.workQueue, nanos, true);
3286             else
3287                 q = externalHelpQuiescePool(nanos, true);
3288             if (q < 0)
3289                 throw new InterruptedException();
3290         }
3291         else if (!(terminated = ((mode & TERMINATED) != 0)) &&
3292                  (lock = registrationLock) != null) {
3293             lock.lock();
3294             try {
3295                 if ((cond = termination) == null)
3296                     termination = cond = lock.newCondition();
3297                 while (!(terminated = ((mode & TERMINATED) != 0)) && nanos > 0L)
3298                     nanos = cond.awaitNanos(nanos);
3299             } finally {
3300                 lock.unlock();
3301             }
3302         }
3303         return terminated;
3304     }
3305 
3306     /**
3307      * If called by a ForkJoinTask operating in this pool, equivalent
3308      * in effect to {@link ForkJoinTask#helpQuiesce}. Otherwise,
3309      * waits and/or attempts to assist performing tasks until this
3310      * pool {@link #isQuiescent} or the indicated timeout elapses.
3311      *
3312      * @param timeout the maximum time to wait
3313      * @param unit the time unit of the timeout argument
3314      * @return {@code true} if quiescent; {@code false} if the
3315      * timeout elapsed.
3316      */
3317     public boolean awaitQuiescence(long timeout, TimeUnit unit) {
3318         Thread t; ForkJoinWorkerThread wt; int q;
3319         long nanos = unit.toNanos(timeout);
3320         if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread &&
3321             (wt = (ForkJoinWorkerThread)t).pool == this)
3322             q = helpQuiescePool(wt.workQueue, nanos, false);
3323         else
3324             q = externalHelpQuiescePool(nanos, false);
3325         return (q > 0);
3326     }
3327 
3328     /**
3329      * Interface for extending managed parallelism for tasks running
3330      * in {@link ForkJoinPool}s.
3331      *
3332      * <p>A {@code ManagedBlocker} provides two methods.  Method
3333      * {@link #isReleasable} must return {@code true} if blocking is
3334      * not necessary. Method {@link #block} blocks the current thread
3335      * if necessary (perhaps internally invoking {@code isReleasable}
3336      * before actually blocking). These actions are performed by any
3337      * thread invoking {@link
3338      * ForkJoinPool#managedBlock(ManagedBlocker)}.  The unusual
3339      * methods in this API accommodate synchronizers that may, but
3340      * don't usually, block for long periods. Similarly, they allow
3341      * more efficient internal handling of cases in which additional
3342      * workers may be, but usually are not, needed to ensure
3343      * sufficient parallelism.  Toward this end, implementations of
3344      * method {@code isReleasable} must be amenable to repeated
3345      * invocation. Neither method is invoked after a prior invocation
3346      * of {@code isReleasable} or {@code block} returns {@code true}.
3347      *
3348      * <p>For example, here is a ManagedBlocker based on a
3349      * ReentrantLock:
3350      * <pre> {@code
3351      * class ManagedLocker implements ManagedBlocker {
3352      *   final ReentrantLock lock;
3353      *   boolean hasLock = false;
3354      *   ManagedLocker(ReentrantLock lock) { this.lock = lock; }
3355      *   public boolean block() {
3356      *     if (!hasLock)
3357      *       lock.lock();
3358      *     return true;
3359      *   }
3360      *   public boolean isReleasable() {
3361      *     return hasLock || (hasLock = lock.tryLock());
3362      *   }
3363      * }}</pre>
3364      *
3365      * <p>Here is a class that possibly blocks waiting for an
3366      * item on a given queue:
3367      * <pre> {@code
3368      * class QueueTaker<E> implements ManagedBlocker {
3369      *   final BlockingQueue<E> queue;
3370      *   volatile E item = null;
3371      *   QueueTaker(BlockingQueue<E> q) { this.queue = q; }
3372      *   public boolean block() throws InterruptedException {
3373      *     if (item == null)
3374      *       item = queue.take();
3375      *     return true;
3376      *   }
3377      *   public boolean isReleasable() {
3378      *     return item != null || (item = queue.poll()) != null;
3379      *   }
3380      *   public E getItem() { // call after pool.managedBlock completes
3381      *     return item;
3382      *   }
3383      * }}</pre>
3384      */
3385     public static interface ManagedBlocker {
3386         /**
3387          * Possibly blocks the current thread, for example waiting for
3388          * a lock or condition.
3389          *
3390          * @return {@code true} if no additional blocking is necessary
3391          * (i.e., if isReleasable would return true)
3392          * @throws InterruptedException if interrupted while waiting
3393          * (the method is not required to do so, but is allowed to)
3394          */
3395         boolean block() throws InterruptedException;
3396 
3397         /**
3398          * Returns {@code true} if blocking is unnecessary.
3399          * @return {@code true} if blocking is unnecessary
3400          */
3401         boolean isReleasable();
3402     }
3403 
3404     /**
3405      * Runs the given possibly blocking task.  When {@linkplain
3406      * ForkJoinTask#inForkJoinPool() running in a ForkJoinPool}, this
3407      * method possibly arranges for a spare thread to be activated if
3408      * necessary to ensure sufficient parallelism while the current
3409      * thread is blocked in {@link ManagedBlocker#block blocker.block()}.
3410      *
3411      * <p>This method repeatedly calls {@code blocker.isReleasable()} and
3412      * {@code blocker.block()} until either method returns {@code true}.
3413      * Every call to {@code blocker.block()} is preceded by a call to
3414      * {@code blocker.isReleasable()} that returned {@code false}.
3415      *
3416      * <p>If not running in a ForkJoinPool, this method is
3417      * behaviorally equivalent to
3418      * <pre> {@code
3419      * while (!blocker.isReleasable())
3420      *   if (blocker.block())
3421      *     break;}</pre>
3422      *
3423      * If running in a ForkJoinPool, the pool may first be expanded to
3424      * ensure sufficient parallelism available during the call to
3425      * {@code blocker.block()}.
3426      *
3427      * @param blocker the blocker task
3428      * @throws InterruptedException if {@code blocker.block()} did so
3429      */
3430     public static void managedBlock(ManagedBlocker blocker)
3431         throws InterruptedException {
3432         Thread t; ForkJoinPool p;
3433         if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread &&
3434             (p = ((ForkJoinWorkerThread)t).pool) != null)
3435             p.compensatedBlock(blocker);
3436         else
3437             unmanagedBlock(blocker);
3438     }
3439 
3440     /** ManagedBlock for ForkJoinWorkerThreads */
3441     private void compensatedBlock(ManagedBlocker blocker)
3442         throws InterruptedException {
3443         if (blocker == null) throw new NullPointerException();
3444         for (;;) {
3445             int comp; boolean done;
3446             long c = ctl;
3447             if (blocker.isReleasable())
3448                 break;
3449             if ((comp = tryCompensate(c)) >= 0) {
3450                 long post = (comp == 0) ? 0L : RC_UNIT;
3451                 try {
3452                     done = blocker.block();
3453                 } finally {
3454                     getAndAddCtl(post);
3455                 }
3456                 if (done)
3457                     break;
3458             }
3459         }
3460     }
3461 
3462     /** ManagedBlock for external threads */
3463     private static void unmanagedBlock(ManagedBlocker blocker)
3464         throws InterruptedException {
3465         if (blocker == null) throw new NullPointerException();
3466         do {} while (!blocker.isReleasable() && !blocker.block());
3467     }
3468 
3469     // AbstractExecutorService.newTaskFor overrides rely on
3470     // undocumented fact that ForkJoinTask.adapt returns ForkJoinTasks
3471     // that also implement RunnableFuture.
3472 
3473     @Override
3474     protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
3475         return new ForkJoinTask.AdaptedRunnable<T>(runnable, value);
3476     }
3477 
3478     @Override
3479     protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
3480         return new ForkJoinTask.AdaptedCallable<T>(callable);
3481     }
3482 
3483     static {
3484         try {
3485             MethodHandles.Lookup l = MethodHandles.lookup();
3486             CTL = l.findVarHandle(ForkJoinPool.class, "ctl", long.class);
3487             MODE = l.findVarHandle(ForkJoinPool.class, "mode", int.class);
3488             THREADIDS = l.findVarHandle(ForkJoinPool.class, "threadIds", int.class);
3489             POOLIDS = l.findStaticVarHandle(ForkJoinPool.class, "poolIds", int.class);
3490         } catch (ReflectiveOperationException e) {
3491             throw new ExceptionInInitializerError(e);
3492         }
3493 
3494         // Reduce the risk of rare disastrous classloading in first call to
3495         // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
3496         Class<?> ensureLoaded = LockSupport.class;
3497 
3498         int commonMaxSpares = DEFAULT_COMMON_MAX_SPARES;
3499         try {
3500             String p = System.getProperty
3501                 ("java.util.concurrent.ForkJoinPool.common.maximumSpares");
3502             if (p != null)
3503                 commonMaxSpares = Integer.parseInt(p);
3504         } catch (Exception ignore) {}
3505         COMMON_MAX_SPARES = commonMaxSpares;
3506 
3507         defaultForkJoinWorkerThreadFactory =
3508             new DefaultForkJoinWorkerThreadFactory();
3509         modifyThreadPermission = new RuntimePermission("modifyThread");
3510         @SuppressWarnings("removal")
3511         ForkJoinPool tmp = AccessController.doPrivileged(new PrivilegedAction<>() {
3512             public ForkJoinPool run() {
3513                 return new ForkJoinPool((byte)0); }});
3514         common = tmp;
3515 
3516         COMMON_PARALLELISM = Math.max(common.mode & SMASK, 1);
3517     }
3518 }
3519