View Javadoc
1   /*
2    * Copyright 2013 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License,
5    * version 2.0 (the "License"); you may not use this file except in compliance
6    * with the License. You may obtain a copy of the License at:
7    *
8    *   http://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13   * License for the specific language governing permissions and limitations
14   * under the License.
15   */
16  
17  /*
18   * Written by Doug Lea with assistance from members of JCP JSR-166
19   * Expert Group and released to the public domain, as explained at
20   * http://creativecommons.org/publicdomain/zero/1.0/
21   */
22  
23  package io.netty.util.internal.chmv8;
24  
25  import io.netty.util.internal.ThreadLocalRandom;
26  
27  import java.lang.Thread.UncaughtExceptionHandler;
28  import java.util.ArrayList;
29  import java.util.Arrays;
30  import java.util.Collection;
31  import java.util.Collections;
32  import java.util.List;
33  import java.util.concurrent.AbstractExecutorService;
34  import java.util.concurrent.Callable;
35  import java.util.concurrent.ExecutorService;
36  import java.util.concurrent.Future;
37  import java.util.concurrent.RejectedExecutionException;
38  import java.util.concurrent.RunnableFuture;
39  import java.util.concurrent.TimeUnit;
40  
41  /**
42   * An {@link ExecutorService} for running {@link ForkJoinTask}s.
43   * A {@code ForkJoinPool} provides the entry point for submissions
44   * from non-{@code ForkJoinTask} clients, as well as management and
45   * monitoring operations.
46   *
47   * <p>A {@code ForkJoinPool} differs from other kinds of {@link
48   * ExecutorService} mainly by virtue of employing
49   * <em>work-stealing</em>: all threads in the pool attempt to find and
50   * execute tasks submitted to the pool and/or created by other active
51   * tasks (eventually blocking waiting for work if none exist). This
52   * enables efficient processing when most tasks spawn other subtasks
53   * (as do most {@code ForkJoinTask}s), as well as when many small
54   * tasks are submitted to the pool from external clients.  Especially
55   * when setting <em>asyncMode</em> to true in constructors, {@code
56   * ForkJoinPool}s may also be appropriate for use with event-style
57   * tasks that are never joined.
58   *
59   * <p>A static {@link #commonPool()} is available and appropriate for
60   * most applications. The common pool is used by any ForkJoinTask that
61   * is not explicitly submitted to a specified pool. Using the common
62   * pool normally reduces resource usage (its threads are slowly
63   * reclaimed during periods of non-use, and reinstated upon subsequent
64   * use).
65   *
66   * <p>For applications that require separate or custom pools, a {@code
67   * ForkJoinPool} may be constructed with a given target parallelism
68   * level; by default, equal to the number of available processors. The
69   * pool attempts to maintain enough active (or available) threads by
70   * dynamically adding, suspending, or resuming internal worker
71   * threads, even if some tasks are stalled waiting to join others.
72   * However, no such adjustments are guaranteed in the face of blocked
73   * I/O or other unmanaged synchronization. The nested {@link
74   * ManagedBlocker} interface enables extension of the kinds of
75   * synchronization accommodated.
76   *
77   * <p>In addition to execution and lifecycle control methods, this
78   * class provides status check methods (for example
79   * {@link #getStealCount}) that are intended to aid in developing,
80   * tuning, and monitoring fork/join applications. Also, method
81   * {@link #toString} returns indications of pool state in a
82   * convenient form for informal monitoring.
83   *
84   * <p>As is the case with other ExecutorServices, there are three
85   * main task execution methods summarized in the following table.
86   * These are designed to be used primarily by clients not already
87   * engaged in fork/join computations in the current pool.  The main
88   * forms of these methods accept instances of {@code ForkJoinTask},
89   * but overloaded forms also allow mixed execution of plain {@code
90   * Runnable}- or {@code Callable}- based activities as well.  However,
91   * tasks that are already executing in a pool should normally instead
92   * use the within-computation forms listed in the table unless using
93   * async event-style tasks that are not usually joined, in which case
94   * there is little difference among choice of methods.
95   *
96   * <table BORDER CELLPADDING=3 CELLSPACING=1>
97   * <caption>Summary of task execution methods</caption>
98   *  <tr>
99   *    <td></td>
100  *    <td ALIGN=CENTER> <b>Call from non-fork/join clients</b></td>
101  *    <td ALIGN=CENTER> <b>Call from within fork/join computations</b></td>
102  *  </tr>
103  *  <tr>
104  *    <td> <b>Arrange async execution</b></td>
105  *    <td> {@link #execute(ForkJoinTask)}</td>
106  *    <td> {@link ForkJoinTask#fork}</td>
107  *  </tr>
108  *  <tr>
109  *    <td> <b>Await and obtain result</b></td>
110  *    <td> {@link #invoke(ForkJoinTask)}</td>
111  *    <td> {@link ForkJoinTask#invoke}</td>
112  *  </tr>
113  *  <tr>
114  *    <td> <b>Arrange exec and obtain Future</b></td>
115  *    <td> {@link #submit(ForkJoinTask)}</td>
116  *    <td> {@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td>
117  *  </tr>
118  * </table>
119  *
120  * <p>The common pool is by default constructed with default
121  * parameters, but these may be controlled by setting three
122  * {@linkplain System#getProperty system properties}:
123  * <ul>
124  * <li>{@code java.util.concurrent.ForkJoinPool.common.parallelism}
125  * - the parallelism level, a non-negative integer
126  * <li>{@code java.util.concurrent.ForkJoinPool.common.threadFactory}
127  * - the class name of a {@link ForkJoinWorkerThreadFactory}
128  * <li>{@code java.util.concurrent.ForkJoinPool.common.exceptionHandler}
129  * - the class name of a {@link UncaughtExceptionHandler}
130  * </ul>
131  * The system class loader is used to load these classes.
132  * Upon any error in establishing these settings, default parameters
133  * are used. It is possible to disable or limit the use of threads in
134  * the common pool by setting the parallelism property to zero, and/or
135  * using a factory that may return {@code null}.
136  *
137  * <p><b>Implementation notes</b>: This implementation restricts the
138  * maximum number of running threads to 32767. Attempts to create
139  * pools with greater than the maximum number result in
140  * {@code IllegalArgumentException}.
141  *
142  * <p>This implementation rejects submitted tasks (that is, by throwing
143  * {@link RejectedExecutionException}) only when the pool is shut down
144  * or internal resources have been exhausted.
145  *
146  * @since 1.7
147  * @author Doug Lea
148  */
149 @SuppressWarnings("all")
150 public class ForkJoinPool extends AbstractExecutorService {
151 
152     /*
153      * Implementation Overview
154      *
155      * This class and its nested classes provide the main
156      * functionality and control for a set of worker threads:
157      * Submissions from non-FJ threads enter into submission queues.
158      * Workers take these tasks and typically split them into subtasks
159      * that may be stolen by other workers.  Preference rules give
160      * first priority to processing tasks from their own queues (LIFO
161      * or FIFO, depending on mode), then to randomized FIFO steals of
162      * tasks in other queues.
163      *
164      * WorkQueues
165      * ==========
166      *
167      * Most operations occur within work-stealing queues (in nested
168      * class WorkQueue).  These are special forms of Deques that
169      * support only three of the four possible end-operations -- push,
170      * pop, and poll (aka steal), under the further constraints that
171      * push and pop are called only from the owning thread (or, as
172      * extended here, under a lock), while poll may be called from
173      * other threads.  (If you are unfamiliar with them, you probably
174      * want to read Herlihy and Shavit's book "The Art of
175      * Multiprocessor programming", chapter 16 describing these in
176      * more detail before proceeding.)  The main work-stealing queue
177      * design is roughly similar to those in the papers "Dynamic
178      * Circular Work-Stealing Deque" by Chase and Lev, SPAA 2005
179      * (http://research.sun.com/scalable/pubs/index.html) and
180      * "Idempotent work stealing" by Michael, Saraswat, and Vechev,
181      * PPoPP 2009 (http://portal.acm.org/citation.cfm?id=1504186).
182      * See also "Correct and Efficient Work-Stealing for Weak Memory
183      * Models" by Le, Pop, Cohen, and Nardelli, PPoPP 2013
184      * (http://www.di.ens.fr/~zappa/readings/ppopp13.pdf) for an
185      * analysis of memory ordering (atomic, volatile etc) issues.  The
186      * main differences ultimately stem from GC requirements that we
187      * null out taken slots as soon as we can, to maintain as small a
188      * footprint as possible even in programs generating huge numbers
189      * of tasks. To accomplish this, we shift the CAS arbitrating pop
190      * vs poll (steal) from being on the indices ("base" and "top") to
191      * the slots themselves.  So, both a successful pop and poll
192      * mainly entail a CAS of a slot from non-null to null.  Because
193      * we rely on CASes of references, we do not need tag bits on base
194      * or top.  They are simple ints as used in any circular
195      * array-based queue (see for example ArrayDeque).  Updates to the
196      * indices must still be ordered in a way that guarantees that top
197      * == base means the queue is empty, but otherwise may err on the
198      * side of possibly making the queue appear nonempty when a push,
199      * pop, or poll have not fully committed. Note that this means
200      * that the poll operation, considered individually, is not
201      * wait-free. One thief cannot successfully continue until another
202      * in-progress one (or, if previously empty, a push) completes.
203      * However, in the aggregate, we ensure at least probabilistic
204      * non-blockingness.  If an attempted steal fails, a thief always
205      * chooses a different random victim target to try next. So, in
206      * order for one thief to progress, it suffices for any
207      * in-progress poll or new push on any empty queue to
208      * complete. (This is why we normally use method pollAt and its
209      * variants that try once at the apparent base index, else
210      * consider alternative actions, rather than method poll.)
211      *
212      * This approach also enables support of a user mode in which local
213      * task processing is in FIFO, not LIFO order, simply by using
214      * poll rather than pop.  This can be useful in message-passing
215      * frameworks in which tasks are never joined.  However neither
216      * mode considers affinities, loads, cache localities, etc, so
217      * rarely provide the best possible performance on a given
218      * machine, but portably provide good throughput by averaging over
219      * these factors.  (Further, even if we did try to use such
220      * information, we do not usually have a basis for exploiting it.
221      * For example, some sets of tasks profit from cache affinities,
222      * but others are harmed by cache pollution effects.)
223      *
224      * WorkQueues are also used in a similar way for tasks submitted
225      * to the pool. We cannot mix these tasks in the same queues used
226      * for work-stealing (this would contaminate lifo/fifo
227      * processing). Instead, we randomly associate submission queues
228      * with submitting threads, using a form of hashing.  The
229      * Submitter probe value serves as a hash code for
230      * choosing existing queues, and may be randomly repositioned upon
231      * contention with other submitters.  In essence, submitters act
232      * like workers except that they are restricted to executing local
233      * tasks that they submitted (or in the case of CountedCompleters,
234      * others with the same root task).  However, because most
235      * shared/external queue operations are more expensive than
236      * internal, and because, at steady state, external submitters
237      * will compete for CPU with workers, ForkJoinTask.join and
238      * related methods disable them from repeatedly helping to process
239      * tasks if all workers are active.  Insertion of tasks in shared
240      * mode requires a lock (mainly to protect in the case of
241      * resizing) but we use only a simple spinlock (using bits in
242      * field qlock), because submitters encountering a busy queue move
243      * on to try or create other queues -- they block only when
244      * creating and registering new queues.
245      *
246      * Management
247      * ==========
248      *
249      * The main throughput advantages of work-stealing stem from
250      * decentralized control -- workers mostly take tasks from
251      * themselves or each other. We cannot negate this in the
252      * implementation of other management responsibilities. The main
253      * tactic for avoiding bottlenecks is packing nearly all
254      * essentially atomic control state into two volatile variables
255      * that are by far most often read (not written) as status and
256      * consistency checks.
257      *
258      * Field "ctl" contains 64 bits holding all the information needed
259      * to atomically decide to add, inactivate, enqueue (on an event
260      * queue), dequeue, and/or re-activate workers.  To enable this
261      * packing, we restrict maximum parallelism to (1<<15)-1 (which is
262      * far in excess of normal operating range) to allow ids, counts,
263      * and their negations (used for thresholding) to fit into 16bit
264      * fields.
265      *
266      * Field "plock" is a form of sequence lock with a saturating
267      * shutdown bit (similarly for per-queue "qlocks"), mainly
268      * protecting updates to the workQueues array, as well as to
269      * enable shutdown.  When used as a lock, it is normally only very
270      * briefly held, so is nearly always available after at most a
271      * brief spin, but we use a monitor-based backup strategy to
272      * block when needed.
273      *
274      * Recording WorkQueues.  WorkQueues are recorded in the
275      * "workQueues" array that is created upon first use and expanded
276      * if necessary.  Updates to the array while recording new workers
277      * and unrecording terminated ones are protected from each other
278      * by a lock but the array is otherwise concurrently readable, and
279      * accessed directly.  To simplify index-based operations, the
280      * array size is always a power of two, and all readers must
281      * tolerate null slots. Worker queues are at odd indices. Shared
282      * (submission) queues are at even indices, up to a maximum of 64
283      * slots, to limit growth even if array needs to expand to add
284      * more workers. Grouping them together in this way simplifies and
285      * speeds up task scanning.
286      *
287      * All worker thread creation is on-demand, triggered by task
288      * submissions, replacement of terminated workers, and/or
289      * compensation for blocked workers. However, all other support
290      * code is set up to work with other policies.  To ensure that we
291      * do not hold on to worker references that would prevent GC, ALL
292      * accesses to workQueues are via indices into the workQueues
293      * array (which is one source of some of the messy code
294      * constructions here). In essence, the workQueues array serves as
295      * a weak reference mechanism. Thus for example the wait queue
296      * field of ctl stores indices, not references.  Access to the
297      * workQueues in associated methods (for example signalWork) must
298      * both index-check and null-check the IDs. All such accesses
299      * ignore bad IDs by returning out early from what they are doing,
300      * since this can only be associated with termination, in which
301      * case it is OK to give up.  All uses of the workQueues array
302      * also check that it is non-null (even if previously
303      * non-null). This allows nulling during termination, which is
304      * currently not necessary, but remains an option for
305      * resource-revocation-based shutdown schemes. It also helps
306      * reduce JIT issuance of uncommon-trap code, which tends to
307      * unnecessarily complicate control flow in some methods.
308      *
309      * Event Queuing. Unlike HPC work-stealing frameworks, we cannot
310      * let workers spin indefinitely scanning for tasks when none can
311      * be found immediately, and we cannot start/resume workers unless
312      * there appear to be tasks available.  On the other hand, we must
313      * quickly prod them into action when new tasks are submitted or
314      * generated. In many usages, ramp-up time to activate workers is
315      * the main limiting factor in overall performance (this is
316      * compounded at program start-up by JIT compilation and
317      * allocation). So we try to streamline this as much as possible.
318      * We park/unpark workers after placing in an event wait queue
319      * when they cannot find work. This "queue" is actually a simple
320      * Treiber stack, headed by the "id" field of ctl, plus a 15bit
321      * counter value (that reflects the number of times a worker has
322      * been inactivated) to avoid ABA effects (we need only as many
323      * version numbers as worker threads). Successors are held in
324      * field WorkQueue.nextWait.  Queuing deals with several intrinsic
325      * races, mainly that a task-producing thread can miss seeing (and
326      * signalling) another thread that gave up looking for work but
327      * has not yet entered the wait queue. We solve this by requiring
328      * a full sweep of all workers (via repeated calls to method
329      * scan()) both before and after a newly waiting worker is added
330      * to the wait queue.  Because enqueued workers may actually be
331      * rescanning rather than waiting, we set and clear the "parker"
332      * field of WorkQueues to reduce unnecessary calls to unpark.
333      * (This requires a secondary recheck to avoid missed signals.)
334      * Note the unusual conventions about Thread.interrupts
335      * surrounding parking and other blocking: Because interrupts are
336      * used solely to alert threads to check termination, which is
337      * checked anyway upon blocking, we clear status (using
338      * Thread.interrupted) before any call to park, so that park does
339      * not immediately return due to status being set via some other
340      * unrelated call to interrupt in user code.
341      *
342      * Signalling.  We create or wake up workers only when there
343      * appears to be at least one task they might be able to find and
344      * execute.  When a submission is added or another worker adds a
345      * task to a queue that has fewer than two tasks, they signal
346      * waiting workers (or trigger creation of new ones if fewer than
347      * the given parallelism level -- signalWork).  These primary
348      * signals are buttressed by others whenever other threads remove
349      * a task from a queue and notice that there are other tasks there
350      * as well.  So in general, pools will be over-signalled. On most
351      * platforms, signalling (unpark) overhead time is noticeably
352      * long, and the time between signalling a thread and it actually
353      * making progress can be very noticeably long, so it is worth
354      * offloading these delays from critical paths as much as
355      * possible. Additionally, workers spin-down gradually, by staying
356      * alive so long as they see the ctl state changing.  Similar
357      * stability-sensing techniques are also used before blocking in
358      * awaitJoin and helpComplete.
359      *
360      * Trimming workers. To release resources after periods of lack of
361      * use, a worker starting to wait when the pool is quiescent will
362      * time out and terminate if the pool has remained quiescent for a
363      * given period -- a short period if there are more threads than
364      * parallelism, longer as the number of threads decreases. This
365      * will slowly propagate, eventually terminating all workers after
366      * periods of non-use.
367      *
368      * Shutdown and Termination. A call to shutdownNow atomically sets
369      * a plock bit and then (non-atomically) sets each worker's
370      * qlock status, cancels all unprocessed tasks, and wakes up
371      * all waiting workers.  Detecting whether termination should
372      * commence after a non-abrupt shutdown() call requires more work
373      * and bookkeeping. We need consensus about quiescence (i.e., that
374      * there is no more work). The active count provides a primary
375      * indication but non-abrupt shutdown still requires a rechecking
376      * scan for any workers that are inactive but not queued.
377      *
378      * Joining Tasks
379      * =============
380      *
381      * Any of several actions may be taken when one worker is waiting
382      * to join a task stolen (or always held) by another.  Because we
383      * are multiplexing many tasks on to a pool of workers, we can't
384      * just let them block (as in Thread.join).  We also cannot just
385      * reassign the joiner's run-time stack with another and replace
386      * it later, which would be a form of "continuation", that even if
387      * possible is not necessarily a good idea since we sometimes need
388      * both an unblocked task and its continuation to progress.
389      * Instead we combine two tactics:
390      *
391      *   Helping: Arranging for the joiner to execute some task that it
392      *      would be running if the steal had not occurred.
393      *
394      *   Compensating: Unless there are already enough live threads,
395      *      method tryCompensate() may create or re-activate a spare
396      *      thread to compensate for blocked joiners until they unblock.
397      *
398      * A third form (implemented in tryRemoveAndExec) amounts to
399      * helping a hypothetical compensator: If we can readily tell that
400      * a possible action of a compensator is to steal and execute the
401      * task being joined, the joining thread can do so directly,
402      * without the need for a compensation thread (although at the
403      * expense of larger run-time stacks, but the tradeoff is
404      * typically worthwhile).
405      *
406      * The ManagedBlocker extension API can't use helping so relies
407      * only on compensation in method awaitBlocker.
408      *
409      * The algorithm in tryHelpStealer entails a form of "linear"
410      * helping: Each worker records (in field currentSteal) the most
411      * recent task it stole from some other worker. Plus, it records
412      * (in field currentJoin) the task it is currently actively
413      * joining. Method tryHelpStealer uses these markers to try to
414      * find a worker to help (i.e., steal back a task from and execute
415      * it) that could hasten completion of the actively joined task.
416      * In essence, the joiner executes a task that would be on its own
417      * local deque had the to-be-joined task not been stolen. This may
418      * be seen as a conservative variant of the approach in Wagner &
419      * Calder "Leapfrogging: a portable technique for implementing
420      * efficient futures" SIGPLAN Notices, 1993
421      * (http://portal.acm.org/citation.cfm?id=155354). It differs in
422      * that: (1) We only maintain dependency links across workers upon
423      * steals, rather than use per-task bookkeeping.  This sometimes
424      * requires a linear scan of workQueues array to locate stealers,
425      * but often doesn't because stealers leave hints (that may become
426      * stale/wrong) of where to locate them.  It is only a hint
427      * because a worker might have had multiple steals and the hint
428      * records only one of them (usually the most current).  Hinting
429      * isolates cost to when it is needed, rather than adding to
430      * per-task overhead.  (2) It is "shallow", ignoring nesting and
431      * potentially cyclic mutual steals.  (3) It is intentionally
432      * racy: field currentJoin is updated only while actively joining,
433      * which means that we miss links in the chain during long-lived
434      * tasks, GC stalls etc (which is OK since blocking in such cases
435      * is usually a good idea).  (4) We bound the number of attempts
436      * to find work (see MAX_HELP) and fall back to suspending the
437      * worker and if necessary replacing it with another.
438      *
439      * Helping actions for CountedCompleters are much simpler: Method
440      * helpComplete can take and execute any task with the same root
441      * as the task being waited on. However, this still entails some
442      * traversal of completer chains, so is less efficient than using
443      * CountedCompleters without explicit joins.
444      *
445      * It is impossible to keep exactly the target parallelism number
446      * of threads running at any given time.  Determining the
447      * existence of conservatively safe helping targets, the
448      * availability of already-created spares, and the apparent need
449      * to create new spares are all racy, so we rely on multiple
450      * retries of each.  Compensation in the apparent absence of
451      * helping opportunities is challenging to control on JVMs, where
452      * GC and other activities can stall progress of tasks that in
453      * turn stall out many other dependent tasks, without us being
454      * able to determine whether they will ever require compensation.
455      * Even though work-stealing otherwise encounters little
456      * degradation in the presence of more threads than cores,
457      * aggressively adding new threads in such cases entails risk of
458      * unwanted positive feedback control loops in which more threads
459      * cause more dependent stalls (as well as delayed progress of
460      * unblocked threads to the point that we know they are available)
461      * leading to more situations requiring more threads, and so
462      * on. This aspect of control can be seen as an (analytically
463      * intractable) game with an opponent that may choose the worst
464      * (for us) active thread to stall at any time.  We take several
465      * precautions to bound losses (and thus bound gains), mainly in
466      * methods tryCompensate and awaitJoin.
467      *
468      * Common Pool
469      * ===========
470      *
471      * The static common pool always exists after static
472      * initialization.  Since it (or any other created pool) need
473      * never be used, we minimize initial construction overhead and
474      * footprint to the setup of about a dozen fields, with no nested
475      * allocation. Most bootstrapping occurs within method
476      * fullExternalPush during the first submission to the pool.
477      *
478      * When external threads submit to the common pool, they can
479      * perform subtask processing (see externalHelpJoin and related
480      * methods).  This caller-helps policy makes it sensible to set
481      * common pool parallelism level to one (or more) less than the
482      * total number of available cores, or even zero for pure
483      * caller-runs.  We do not need to record whether external
484      * submissions are to the common pool -- if not, externalHelpJoin
485      * returns quickly (at the most helping to signal some common pool
486      * workers). These submitters would otherwise be blocked waiting
487      * for completion, so the extra effort (with liberally sprinkled
488      * task status checks) in inapplicable cases amounts to an odd
489      * form of limited spin-wait before blocking in ForkJoinTask.join.
490      *
491      * Style notes
492      * ===========
493      *
494      * There is a lot of representation-level coupling among classes
495      * ForkJoinPool, ForkJoinWorkerThread, and ForkJoinTask.  The
496      * fields of WorkQueue maintain data structures managed by
497      * ForkJoinPool, so are directly accessed.  There is little point
498      * trying to reduce this, since any associated future changes in
499      * representations will need to be accompanied by algorithmic
500      * changes anyway. Several methods intrinsically sprawl because
501      * they must accumulate sets of consistent reads of volatiles held
502      * in local variables.  Methods signalWork() and scan() are the
503      * main bottlenecks, so are especially heavily
504      * micro-optimized/mangled.  There are lots of inline assignments
505      * (of form "while ((local = field) != 0)") which are usually the
506      * simplest way to ensure the required read orderings (which are
507      * sometimes critical). This leads to a "C"-like style of listing
508      * declarations of these locals at the heads of methods or blocks.
509      * There are several occurrences of the unusual "do {} while
510      * (!cas...)"  which is the simplest way to force an update of a
511      * CAS'ed variable. There are also other coding oddities (including
512      * several unnecessary-looking hoisted null checks) that help
513      * some methods perform reasonably even when interpreted (not
514      * compiled).
515      *
516      * The order of declarations in this file is:
517      * (1) Static utility functions
518      * (2) Nested (static) classes
519      * (3) Static fields
520      * (4) Fields, along with constants used when unpacking some of them
521      * (5) Internal control methods
522      * (6) Callbacks and other support for ForkJoinTask methods
523      * (7) Exported methods
524      * (8) Static block initializing statics in minimally dependent order
525      */
526 
527     // Static utilities
528 
529     /**
530      * If there is a security manager, makes sure caller has
531      * permission to modify threads.
532      */
533     private static void checkPermission() {
534         SecurityManager security = System.getSecurityManager();
535         if (security != null)
536             security.checkPermission(modifyThreadPermission);
537     }
538 
539     // Nested classes
540 
541     /**
542      * Factory for creating new {@link ForkJoinWorkerThread}s.
543      * A {@code ForkJoinWorkerThreadFactory} must be defined and used
544      * for {@code ForkJoinWorkerThread} subclasses that extend base
545      * functionality or initialize threads with different contexts.
546      */
547     public static interface ForkJoinWorkerThreadFactory {
548         /**
549          * Returns a new worker thread operating in the given pool.
550          *
551          * @param pool the pool this thread works in
552          * @throws NullPointerException if the pool is null
553          * @return the new worker thread
554          */
555         public ForkJoinWorkerThread newThread(ForkJoinPool pool);
556     }
557 
558     /**
559      * Default ForkJoinWorkerThreadFactory implementation; creates a
560      * new ForkJoinWorkerThread.
561      */
562     static final class DefaultForkJoinWorkerThreadFactory
563             implements ForkJoinWorkerThreadFactory {
564         public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
565             return new ForkJoinWorkerThread(pool);
566         }
567     }
568 
569     /**
570      * Class for artificial tasks that are used to replace the target
571      * of local joins if they are removed from an interior queue slot
572      * in WorkQueue.tryRemoveAndExec. We don't need the proxy to
573      * actually do anything beyond having a unique identity.
574      */
575     static final class EmptyTask extends ForkJoinTask<Void> {
576         private static final long serialVersionUID = -7721805057305804111L;
577         EmptyTask() { status = ForkJoinTask.NORMAL; } // force done
578         public final Void getRawResult() { return null; }
579         public final void setRawResult(Void x) {}
580         public final boolean exec() { return true; }
581     }
582 
583     /**
584      * Queues supporting work-stealing as well as external task
585      * submission. See above for main rationale and algorithms.
586      * Implementation relies heavily on "Unsafe" intrinsics
587      * and selective use of "volatile":
588      *
589      * Field "base" is the index (mod array.length) of the least valid
590      * queue slot, which is always the next position to steal (poll)
591      * from if nonempty. Reads and writes require volatile orderings
592      * but not CAS, because updates are only performed after slot
593      * CASes.
594      *
595      * Field "top" is the index (mod array.length) of the next queue
596      * slot to push to or pop from. It is written only by owner thread
597      * for push, or under lock for external/shared push, and accessed
598      * by other threads only after reading (volatile) base.  Both top
599      * and base are allowed to wrap around on overflow, but (top -
600      * base) (or more commonly -(base - top) to force volatile read of
601      * base before top) still estimates size. The lock ("qlock") is
602      * forced to -1 on termination, causing all further lock attempts
603      * to fail. (Note: we don't need CAS for termination state because
604      * upon pool shutdown, all shared-queues will stop being used
605      * anyway.)  Nearly all lock bodies are set up so that exceptions
606      * within lock bodies are "impossible" (modulo JVM errors that
607      * would cause failure anyway.)
608      *
609      * The array slots are read and written using the emulation of
610      * volatiles/atomics provided by Unsafe. Insertions must in
611      * general use putOrderedObject as a form of releasing store to
612      * ensure that all writes to the task object are ordered before
613      * its publication in the queue.  All removals entail a CAS to
614      * null.  The array is always a power of two. To ensure safety of
615      * Unsafe array operations, all accesses perform explicit null
616      * checks and implicit bounds checks via power-of-two masking.
617      *
618      * In addition to basic queuing support, this class contains
619      * fields described elsewhere to control execution. It turns out
620      * to work better memory-layout-wise to include them in this class
621      * rather than a separate class.
622      *
623      * Performance on most platforms is very sensitive to placement of
624      * instances of both WorkQueues and their arrays -- we absolutely
625      * do not want multiple WorkQueue instances or multiple queue
626      * arrays sharing cache lines. (It would be best for queue objects
627      * and their arrays to share, but there is nothing available to
628      * help arrange that). The @Contended annotation alerts JVMs to
629      * try to keep instances apart.
630      */
631     static final class WorkQueue {
632         /**
633          * Capacity of work-stealing queue array upon initialization.
634          * Must be a power of two; at least 4, but should be larger to
635          * reduce or eliminate cacheline sharing among queues.
636          * Currently, it is much larger, as a partial workaround for
637          * the fact that JVMs often place arrays in locations that
638          * share GC bookkeeping (especially cardmarks) such that
639          * per-write accesses encounter serious memory contention.
640          */
641         static final int INITIAL_QUEUE_CAPACITY = 1 << 13;
642 
643         /**
644          * Maximum size for queue arrays. Must be a power of two less
645          * than or equal to 1 << (31 - width of array entry) to ensure
646          * lack of wraparound of index calculations, but defined to a
647          * value a bit less than this to help users trap runaway
648          * programs before saturating systems.
649          */
650         static final int MAXIMUM_QUEUE_CAPACITY = 1 << 26; // 64M
651 
652         // Heuristic padding to ameliorate unfortunate memory placements
653         volatile long pad00, pad01, pad02, pad03, pad04, pad05, pad06;
654 
655         volatile int eventCount;   // encoded inactivation count; < 0 if inactive
656         int nextWait;              // encoded record of next event waiter
657         int nsteals;               // number of steals
658         int hint;                  // steal index hint
659         short poolIndex;           // index of this queue in pool
660         final short mode;          // 0: lifo, > 0: fifo, < 0: shared
661         volatile int qlock;        // 1: locked, -1: terminate; else 0
662         volatile int base;         // index of next slot for poll
663         int top;                   // index of next slot for push
664         ForkJoinTask<?>[] array;   // the elements (initially unallocated)
665         final ForkJoinPool pool;   // the containing pool (may be null)
666         final ForkJoinWorkerThread owner; // owning thread or null if shared
667         volatile Thread parker;    // == owner during call to park; else null
668         volatile ForkJoinTask<?> currentJoin;  // task being joined in awaitJoin
669         ForkJoinTask<?> currentSteal; // current non-local task being executed
670 
671         volatile Object pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17;
672         volatile Object pad18, pad19, pad1a, pad1b, pad1c, pad1d;
673 
674         WorkQueue(ForkJoinPool pool, ForkJoinWorkerThread owner, int mode,
675                   int seed) {
676             this.pool = pool;
677             this.owner = owner;
678             this.mode = (short)mode;
679             this.hint = seed; // store initial seed for runWorker
680             // Place indices in the center of array (that is not yet allocated)
681             base = top = INITIAL_QUEUE_CAPACITY >>> 1;
682         }
683 
684         /**
685          * Returns the approximate number of tasks in the queue.
686          */
687         final int queueSize() {
688             int n = base - top;       // non-owner callers must read base first
689             return (n >= 0) ? 0 : -n; // ignore transient negative
690         }
691 
692         /**
693          * Provides a more accurate estimate of whether this queue has
694          * any tasks than does queueSize, by checking whether a
695          * near-empty queue has at least one unclaimed task.
696          */
697         final boolean isEmpty() {
698             ForkJoinTask<?>[] a; int m, s;
699             int n = base - (s = top);
700             return (n >= 0 ||
701                     (n == -1 &&
702                             ((a = array) == null ||
703                                     (m = a.length - 1) < 0 ||
704                                     U.getObject
705                                             (a, (long)((m & (s - 1)) << ASHIFT) + ABASE) == null)));
706         }
707 
708         /**
709          * Pushes a task. Call only by owner in unshared queues.  (The
710          * shared-queue version is embedded in method externalPush.)
711          *
712          * @param task the task. Caller must ensure non-null.
713          * @throws RejectedExecutionException if array cannot be resized
714          */
715         final void push(ForkJoinTask<?> task) {
716             ForkJoinTask<?>[] a; ForkJoinPool p;
717             int s = top, n;
718             if ((a = array) != null) {    // ignore if queue removed
719                 int m = a.length - 1;
720                 U.putOrderedObject(a, ((m & s) << ASHIFT) + ABASE, task);
721                 if ((n = (top = s + 1) - base) <= 2)
722                     (p = pool).signalWork(p.workQueues, this);
723                 else if (n >= m)
724                     growArray();
725             }
726         }
727 
728         /**
729          * Initializes or doubles the capacity of array. Call either
730          * by owner or with lock held -- it is OK for base, but not
731          * top, to move while resizings are in progress.
732          */
733         final ForkJoinTask<?>[] growArray() {
734             ForkJoinTask<?>[] oldA = array;
735             int size = oldA != null ? oldA.length << 1 : INITIAL_QUEUE_CAPACITY;
736             if (size > MAXIMUM_QUEUE_CAPACITY)
737                 throw new RejectedExecutionException("Queue capacity exceeded");
738             int oldMask, t, b;
739             ForkJoinTask<?>[] a = array = new ForkJoinTask<?>[size];
740             if (oldA != null && (oldMask = oldA.length - 1) >= 0 &&
741                     (t = top) - (b = base) > 0) {
742                 int mask = size - 1;
743                 do {
744                     ForkJoinTask<?> x;
745                     int oldj = ((b & oldMask) << ASHIFT) + ABASE;
746                     int j    = ((b &    mask) << ASHIFT) + ABASE;
747                     x = (ForkJoinTask<?>)U.getObjectVolatile(oldA, oldj);
748                     if (x != null &&
749                             U.compareAndSwapObject(oldA, oldj, x, null))
750                         U.putObjectVolatile(a, j, x);
751                 } while (++b != t);
752             }
753             return a;
754         }
755 
756         /**
757          * Takes next task, if one exists, in LIFO order.  Call only
758          * by owner in unshared queues.
759          */
760         final ForkJoinTask<?> pop() {
761             ForkJoinTask<?>[] a; ForkJoinTask<?> t; int m;
762             if ((a = array) != null && (m = a.length - 1) >= 0) {
763                 for (int s; (s = top - 1) - base >= 0;) {
764                     long j = ((m & s) << ASHIFT) + ABASE;
765                     if ((t = (ForkJoinTask<?>)U.getObject(a, j)) == null)
766                         break;
767                     if (U.compareAndSwapObject(a, j, t, null)) {
768                         top = s;
769                         return t;
770                     }
771                 }
772             }
773             return null;
774         }
775 
776         /**
777          * Takes a task in FIFO order if b is base of queue and a task
778          * can be claimed without contention. Specialized versions
779          * appear in ForkJoinPool methods scan and tryHelpStealer.
780          */
781         final ForkJoinTask<?> pollAt(int b) {
782             ForkJoinTask<?> t; ForkJoinTask<?>[] a;
783             if ((a = array) != null) {
784                 int j = (((a.length - 1) & b) << ASHIFT) + ABASE;
785                 if ((t = (ForkJoinTask<?>)U.getObjectVolatile(a, j)) != null &&
786                         base == b && U.compareAndSwapObject(a, j, t, null)) {
787                     U.putOrderedInt(this, QBASE, b + 1);
788                     return t;
789                 }
790             }
791             return null;
792         }
793 
794         /**
795          * Takes next task, if one exists, in FIFO order.
796          */
797         final ForkJoinTask<?> poll() {
798             ForkJoinTask<?>[] a; int b; ForkJoinTask<?> t;
799             while ((b = base) - top < 0 && (a = array) != null) {
800                 int j = (((a.length - 1) & b) << ASHIFT) + ABASE;
801                 t = (ForkJoinTask<?>)U.getObjectVolatile(a, j);
802                 if (t != null) {
803                     if (U.compareAndSwapObject(a, j, t, null)) {
804                         U.putOrderedInt(this, QBASE, b + 1);
805                         return t;
806                     }
807                 }
808                 else if (base == b) {
809                     if (b + 1 == top)
810                         break;
811                     Thread.yield(); // wait for lagging update (very rare)
812                 }
813             }
814             return null;
815         }
816 
817         /**
818          * Takes next task, if one exists, in order specified by mode.
819          */
820         final ForkJoinTask<?> nextLocalTask() {
821             return mode == 0 ? pop() : poll();
822         }
823 
824         /**
825          * Returns next task, if one exists, in order specified by mode.
826          */
827         final ForkJoinTask<?> peek() {
828             ForkJoinTask<?>[] a = array; int m;
829             if (a == null || (m = a.length - 1) < 0)
830                 return null;
831             int i = mode == 0 ? top - 1 : base;
832             int j = ((i & m) << ASHIFT) + ABASE;
833             return (ForkJoinTask<?>)U.getObjectVolatile(a, j);
834         }
835 
836         /**
837          * Pops the given task only if it is at the current top.
838          * (A shared version is available only via FJP.tryExternalUnpush)
839          */
840         final boolean tryUnpush(ForkJoinTask<?> t) {
841             ForkJoinTask<?>[] a; int s;
842             if ((a = array) != null && (s = top) != base &&
843                     U.compareAndSwapObject
844                             (a, (((a.length - 1) & --s) << ASHIFT) + ABASE, t, null)) {
845                 top = s;
846                 return true;
847             }
848             return false;
849         }
850 
851         /**
852          * Removes and cancels all known tasks, ignoring any exceptions.
853          */
854         final void cancelAll() {
855             ForkJoinTask.cancelIgnoringExceptions(currentJoin);
856             ForkJoinTask.cancelIgnoringExceptions(currentSteal);
857             for (ForkJoinTask<?> t; (t = poll()) != null; )
858                 ForkJoinTask.cancelIgnoringExceptions(t);
859         }
860 
861         // Specialized execution methods
862 
863         /**
864          * Polls and runs tasks until empty.
865          */
866         final void pollAndExecAll() {
867             for (ForkJoinTask<?> t; (t = poll()) != null;)
868                 t.doExec();
869         }
870 
871         /**
872          * Executes a top-level task and any local tasks remaining
873          * after execution.
874          */
875         final void runTask(ForkJoinTask<?> task) {
876             if ((currentSteal = task) != null) {
877                 task.doExec();
878                 ForkJoinTask<?>[] a = array;
879                 int md = mode;
880                 ++nsteals;
881                 currentSteal = null;
882                 if (md != 0)
883                     pollAndExecAll();
884                 else if (a != null) {
885                     int s, m = a.length - 1;
886                     while ((s = top - 1) - base >= 0) {
887                         long i = ((m & s) << ASHIFT) + ABASE;
888                         ForkJoinTask<?> t = (ForkJoinTask<?>)U.getObject(a, i);
889                         if (t == null)
890                             break;
891                         if (U.compareAndSwapObject(a, i, t, null)) {
892                             top = s;
893                             t.doExec();
894                         }
895                     }
896                 }
897             }
898         }
899 
900         /**
901          * If present, removes from queue and executes the given task,
902          * or any other cancelled task. Returns (true) on any CAS
903          * or consistency check failure so caller can retry.
904          *
905          * @return false if no progress can be made, else true
906          */
907         final boolean tryRemoveAndExec(ForkJoinTask<?> task) {
908             boolean stat;
909             ForkJoinTask<?>[] a; int m, s, b, n;
910             if (task != null && (a = array) != null && (m = a.length - 1) >= 0 &&
911                     (n = (s = top) - (b = base)) > 0) {
912                 boolean removed = false, empty = true;
913                 stat = true;
914                 for (ForkJoinTask<?> t;;) {           // traverse from s to b
915                     long j = ((--s & m) << ASHIFT) + ABASE;
916                     t = (ForkJoinTask<?>)U.getObject(a, j);
917                     if (t == null)                    // inconsistent length
918                         break;
919                     else if (t == task) {
920                         if (s + 1 == top) {           // pop
921                             if (!U.compareAndSwapObject(a, j, task, null))
922                                 break;
923                             top = s;
924                             removed = true;
925                         }
926                         else if (base == b)           // replace with proxy
927                             removed = U.compareAndSwapObject(a, j, task,
928                                     new EmptyTask());
929                         break;
930                     }
931                     else if (t.status >= 0)
932                         empty = false;
933                     else if (s + 1 == top) {          // pop and throw away
934                         if (U.compareAndSwapObject(a, j, t, null))
935                             top = s;
936                         break;
937                     }
938                     if (--n == 0) {
939                         if (!empty && base == b)
940                             stat = false;
941                         break;
942                     }
943                 }
944                 if (removed)
945                     task.doExec();
946             }
947             else
948                 stat = false;
949             return stat;
950         }
951 
952         /**
953          * Tries to poll for and execute the given task or any other
954          * task in its CountedCompleter computation.
955          */
956         final boolean pollAndExecCC(CountedCompleter<?> root) {
957             ForkJoinTask<?>[] a; int b; Object o; CountedCompleter<?> t, r;
958             if ((b = base) - top < 0 && (a = array) != null) {
959                 long j = (((a.length - 1) & b) << ASHIFT) + ABASE;
960                 if ((o = U.getObjectVolatile(a, j)) == null)
961                     return true; // retry
962                 if (o instanceof CountedCompleter) {
963                     for (t = (CountedCompleter<?>)o, r = t;;) {
964                         if (r == root) {
965                             if (base == b &&
966                                     U.compareAndSwapObject(a, j, t, null)) {
967                                 U.putOrderedInt(this, QBASE, b + 1);
968                                 t.doExec();
969                             }
970                             return true;
971                         }
972                         else if ((r = r.completer) == null)
973                             break; // not part of root computation
974                     }
975                 }
976             }
977             return false;
978         }
979 
980         /**
981          * Tries to pop and execute the given task or any other task
982          * in its CountedCompleter computation.
983          */
984         final boolean externalPopAndExecCC(CountedCompleter<?> root) {
985             ForkJoinTask<?>[] a; int s; Object o; CountedCompleter<?> t, r;
986             if (base - (s = top) < 0 && (a = array) != null) {
987                 long j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE;
988                 if ((o = U.getObject(a, j)) instanceof CountedCompleter) {
989                     for (t = (CountedCompleter<?>)o, r = t;;) {
990                         if (r == root) {
991                             if (U.compareAndSwapInt(this, QLOCK, 0, 1)) {
992                                 if (top == s && array == a &&
993                                         U.compareAndSwapObject(a, j, t, null)) {
994                                     top = s - 1;
995                                     qlock = 0;
996                                     t.doExec();
997                                 }
998                                 else
999                                     qlock = 0;
1000                             }
1001                             return true;
1002                         }
1003                         else if ((r = r.completer) == null)
1004                             break;
1005                     }
1006                 }
1007             }
1008             return false;
1009         }
1010 
1011         /**
1012          * Internal version
1013          */
1014         final boolean internalPopAndExecCC(CountedCompleter<?> root) {
1015             ForkJoinTask<?>[] a; int s; Object o; CountedCompleter<?> t, r;
1016             if (base - (s = top) < 0 && (a = array) != null) {
1017                 long j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE;
1018                 if ((o = U.getObject(a, j)) instanceof CountedCompleter) {
1019                     for (t = (CountedCompleter<?>)o, r = t;;) {
1020                         if (r == root) {
1021                             if (U.compareAndSwapObject(a, j, t, null)) {
1022                                 top = s - 1;
1023                                 t.doExec();
1024                             }
1025                             return true;
1026                         }
1027                         else if ((r = r.completer) == null)
1028                             break;
1029                     }
1030                 }
1031             }
1032             return false;
1033         }
1034 
1035         /**
1036          * Returns true if owned and not known to be blocked.
1037          */
1038         final boolean isApparentlyUnblocked() {
1039             Thread wt; Thread.State s;
1040             return (eventCount >= 0 &&
1041                     (wt = owner) != null &&
1042                     (s = wt.getState()) != Thread.State.BLOCKED &&
1043                     s != Thread.State.WAITING &&
1044                     s != Thread.State.TIMED_WAITING);
1045         }
1046 
1047         // Unsafe mechanics
1048         private static final sun.misc.Unsafe U;
1049         private static final long QBASE;
1050         private static final long QLOCK;
1051         private static final int ABASE;
1052         private static final int ASHIFT;
1053         static {
1054             try {
1055                 U = getUnsafe();
1056                 Class<?> k = WorkQueue.class;
1057                 Class<?> ak = ForkJoinTask[].class;
1058                 QBASE = U.objectFieldOffset
1059                         (k.getDeclaredField("base"));
1060                 QLOCK = U.objectFieldOffset
1061                         (k.getDeclaredField("qlock"));
1062                 ABASE = U.arrayBaseOffset(ak);
1063                 int scale = U.arrayIndexScale(ak);
1064                 if ((scale & (scale - 1)) != 0)
1065                     throw new Error("data type scale not a power of two");
1066                 ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
1067             } catch (Exception e) {
1068                 throw new Error(e);
1069             }
1070         }
1071     }
1072 
1073     // static fields (initialized in static initializer below)
1074 
1075     /**
1076      * Per-thread submission bookkeeping. Shared across all pools
1077      * to reduce ThreadLocal pollution and because random motion
1078      * to avoid contention in one pool is likely to hold for others.
1079      * Lazily initialized on first submission (but null-checked
1080      * in other contexts to avoid unnecessary initialization).
1081      */
1082     static final ThreadLocal<Submitter> submitters;
1083 
1084     /**
1085      * Creates a new ForkJoinWorkerThread. This factory is used unless
1086      * overridden in ForkJoinPool constructors.
1087      */
1088     public static final ForkJoinWorkerThreadFactory
1089             defaultForkJoinWorkerThreadFactory;
1090 
1091     /**
1092      * Permission required for callers of methods that may start or
1093      * kill threads.
1094      */
1095     private static final RuntimePermission modifyThreadPermission;
1096 
1097     /**
1098      * Common (static) pool. Non-null for public use unless a static
1099      * construction exception, but internal usages null-check on use
1100      * to paranoically avoid potential initialization circularities
1101      * as well as to simplify generated code.
1102      */
1103     static final ForkJoinPool common;
1104 
1105     /**
1106      * Common pool parallelism. To allow simpler use and management
1107      * when common pool threads are disabled, we allow the underlying
1108      * common.parallelism field to be zero, but in that case still report
1109      * parallelism as 1 to reflect resulting caller-runs mechanics.
1110      */
1111     static final int commonParallelism;
1112 
1113     /**
1114      * Sequence number for creating workerNamePrefix.
1115      */
1116     private static int poolNumberSequence;
1117 
1118     /**
1119      * Returns the next sequence number. We don't expect this to
1120      * ever contend, so use simple builtin sync.
1121      */
1122     private static final synchronized int nextPoolId() {
1123         return ++poolNumberSequence;
1124     }
1125 
1126     // static constants
1127 
1128     /**
1129      * Initial timeout value (in nanoseconds) for the thread
1130      * triggering quiescence to park waiting for new work. On timeout,
1131      * the thread will instead try to shrink the number of
1132      * workers. The value should be large enough to avoid overly
1133      * aggressive shrinkage during most transient stalls (long GCs
1134      * etc).
1135      */
1136     private static final long IDLE_TIMEOUT      = 2000L * 1000L * 1000L; // 2sec
1137 
1138     /**
1139      * Timeout value when there are more threads than parallelism level
1140      */
1141     private static final long FAST_IDLE_TIMEOUT =  200L * 1000L * 1000L;
1142 
1143     /**
1144      * Tolerance for idle timeouts, to cope with timer undershoots
1145      */
1146     private static final long TIMEOUT_SLOP = 2000000L;
1147 
1148     /**
1149      * The maximum stolen->joining link depth allowed in method
1150      * tryHelpStealer.  Must be a power of two.  Depths for legitimate
1151      * chains are unbounded, but we use a fixed constant to avoid
1152      * (otherwise unchecked) cycles and to bound staleness of
1153      * traversal parameters at the expense of sometimes blocking when
1154      * we could be helping.
1155      */
1156     private static final int MAX_HELP = 64;
1157 
1158     /**
1159      * Increment for seed generators. See class ThreadLocal for
1160      * explanation.
1161      */
1162     private static final int SEED_INCREMENT = 0x61c88647;
1163 
1164     /*
1165      * Bits and masks for control variables
1166      *
1167      * Field ctl is a long packed with:
1168      * AC: Number of active running workers minus target parallelism (16 bits)
1169      * TC: Number of total workers minus target parallelism (16 bits)
1170      * ST: true if pool is terminating (1 bit)
1171      * EC: the wait count of top waiting thread (15 bits)
1172      * ID: poolIndex of top of Treiber stack of waiters (16 bits)
1173      *
1174      * When convenient, we can extract the upper 32 bits of counts and
1175      * the lower 32 bits of queue state, u = (int)(ctl >>> 32) and e =
1176      * (int)ctl.  The ec field is never accessed alone, but always
1177      * together with id and st. The offsets of counts by the target
1178      * parallelism and the positionings of fields makes it possible to
1179      * perform the most common checks via sign tests of fields: When
1180      * ac is negative, there are not enough active workers, when tc is
1181      * negative, there are not enough total workers, and when e is
1182      * negative, the pool is terminating.  To deal with these possibly
1183      * negative fields, we use casts in and out of "short" and/or
1184      * signed shifts to maintain signedness.
1185      *
1186      * When a thread is queued (inactivated), its eventCount field is
1187      * set negative, which is the only way to tell if a worker is
1188      * prevented from executing tasks, even though it must continue to
1189      * scan for them to avoid queuing races. Note however that
1190      * eventCount updates lag releases so usage requires care.
1191      *
1192      * Field plock is an int packed with:
1193      * SHUTDOWN: true if shutdown is enabled (1 bit)
1194      * SEQ:  a sequence lock, with PL_LOCK bit set if locked (30 bits)
1195      * SIGNAL: set when threads may be waiting on the lock (1 bit)
1196      *
1197      * The sequence number enables simple consistency checks:
1198      * Staleness of read-only operations on the workQueues array can
1199      * be checked by comparing plock before vs after the reads.
1200      */
1201 
1202     // bit positions/shifts for fields
1203     private static final int  AC_SHIFT   = 48;
1204     private static final int  TC_SHIFT   = 32;
1205     private static final int  ST_SHIFT   = 31;
1206     private static final int  EC_SHIFT   = 16;
1207 
1208     // bounds
1209     private static final int  SMASK      = 0xffff;  // short bits
1210     private static final int  MAX_CAP    = 0x7fff;  // max #workers - 1
1211     private static final int  EVENMASK   = 0xfffe;  // even short bits
1212     private static final int  SQMASK     = 0x007e;  // max 64 (even) slots
1213     private static final int  SHORT_SIGN = 1 << 15;
1214     private static final int  INT_SIGN   = 1 << 31;
1215 
1216     // masks
1217     private static final long STOP_BIT   = 0x0001L << ST_SHIFT;
1218     private static final long AC_MASK    = ((long)SMASK) << AC_SHIFT;
1219     private static final long TC_MASK    = ((long)SMASK) << TC_SHIFT;
1220 
1221     // units for incrementing and decrementing
1222     private static final long TC_UNIT    = 1L << TC_SHIFT;
1223     private static final long AC_UNIT    = 1L << AC_SHIFT;
1224 
1225     // masks and units for dealing with u = (int)(ctl >>> 32)
1226     private static final int  UAC_SHIFT  = AC_SHIFT - 32;
1227     private static final int  UTC_SHIFT  = TC_SHIFT - 32;
1228     private static final int  UAC_MASK   = SMASK << UAC_SHIFT;
1229     private static final int  UTC_MASK   = SMASK << UTC_SHIFT;
1230     private static final int  UAC_UNIT   = 1 << UAC_SHIFT;
1231     private static final int  UTC_UNIT   = 1 << UTC_SHIFT;
1232 
1233     // masks and units for dealing with e = (int)ctl
1234     private static final int E_MASK      = 0x7fffffff; // no STOP_BIT
1235     private static final int E_SEQ       = 1 << EC_SHIFT;
1236 
1237     // plock bits
1238     private static final int SHUTDOWN    = 1 << 31;
1239     private static final int PL_LOCK     = 2;
1240     private static final int PL_SIGNAL   = 1;
1241     private static final int PL_SPINS    = 1 << 8;
1242 
1243     // access mode for WorkQueue
1244     static final int LIFO_QUEUE          =  0;
1245     static final int FIFO_QUEUE          =  1;
1246     static final int SHARED_QUEUE        = -1;
1247 
1248     // Heuristic padding to ameliorate unfortunate memory placements
1249     volatile long pad00, pad01, pad02, pad03, pad04, pad05, pad06;
1250 
1251     // Instance fields
1252     volatile long stealCount;                  // collects worker counts
1253     volatile long ctl;                         // main pool control
1254     volatile int plock;                        // shutdown status and seqLock
1255     volatile int indexSeed;                    // worker/submitter index seed
1256     final short parallelism;                   // parallelism level
1257     final short mode;                          // LIFO/FIFO
1258     WorkQueue[] workQueues;                    // main registry
1259     final ForkJoinWorkerThreadFactory factory;
1260     final UncaughtExceptionHandler ueh;        // per-worker UEH
1261     final String workerNamePrefix;             // to create worker name string
1262 
1263     volatile Object pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17;
1264     volatile Object pad18, pad19, pad1a, pad1b;
1265 
1266     /**
1267      * Acquires the plock lock to protect worker array and related
1268      * updates. This method is called only if an initial CAS on plock
1269      * fails. This acts as a spinlock for normal cases, but falls back
1270      * to builtin monitor to block when (rarely) needed. This would be
1271      * a terrible idea for a highly contended lock, but works fine as
1272      * a more conservative alternative to a pure spinlock.
1273      */
1274     private int acquirePlock() {
1275         int spins = PL_SPINS, ps, nps;
1276         for (;;) {
1277             if (((ps = plock) & PL_LOCK) == 0 &&
1278                     U.compareAndSwapInt(this, PLOCK, ps, nps = ps + PL_LOCK))
1279                 return nps;
1280             else if (spins >= 0) {
1281                 if (ThreadLocalRandom.current().nextInt() >= 0)
1282                     --spins;
1283             }
1284             else if (U.compareAndSwapInt(this, PLOCK, ps, ps | PL_SIGNAL)) {
1285                 synchronized (this) {
1286                     if ((plock & PL_SIGNAL) != 0) {
1287                         try {
1288                             wait();
1289                         } catch (InterruptedException ie) {
1290                             try {
1291                                 Thread.currentThread().interrupt();
1292                             } catch (SecurityException ignore) {
1293                             }
1294                         }
1295                     }
1296                     else
1297                         notifyAll();
1298                 }
1299             }
1300         }
1301     }
1302 
1303     /**
1304      * Unlocks and signals any thread waiting for plock. Called only
1305      * when CAS of seq value for unlock fails.
1306      */
1307     private void releasePlock(int ps) {
1308         plock = ps;
1309         synchronized (this) { notifyAll(); }
1310     }
1311 
1312     /**
1313      * Tries to create and start one worker if fewer than target
1314      * parallelism level exist. Adjusts counts etc on failure.
1315      */
1316     private void tryAddWorker() {
1317         long c; int u, e;
1318         while ((u = (int)((c = ctl) >>> 32)) < 0 &&
1319                 (u & SHORT_SIGN) != 0 && (e = (int)c) >= 0) {
1320             long nc = ((long)(((u + UTC_UNIT) & UTC_MASK) |
1321                     ((u + UAC_UNIT) & UAC_MASK)) << 32) | (long)e;
1322             if (U.compareAndSwapLong(this, CTL, c, nc)) {
1323                 ForkJoinWorkerThreadFactory fac;
1324                 Throwable ex = null;
1325                 ForkJoinWorkerThread wt = null;
1326                 try {
1327                     if ((fac = factory) != null &&
1328                             (wt = fac.newThread(this)) != null) {
1329                         wt.start();
1330                         break;
1331                     }
1332                 } catch (Throwable rex) {
1333                     ex = rex;
1334                 }
1335                 deregisterWorker(wt, ex);
1336                 break;
1337             }
1338         }
1339     }
1340 
1341     //  Registering and deregistering workers
1342 
1343     /**
1344      * Callback from ForkJoinWorkerThread to establish and record its
1345      * WorkQueue. To avoid scanning bias due to packing entries in
1346      * front of the workQueues array, we treat the array as a simple
1347      * power-of-two hash table using per-thread seed as hash,
1348      * expanding as needed.
1349      *
1350      * @param wt the worker thread
1351      * @return the worker's queue
1352      */
1353     final WorkQueue registerWorker(ForkJoinWorkerThread wt) {
1354         UncaughtExceptionHandler handler; WorkQueue[] ws; int s, ps;
1355         wt.setDaemon(true);
1356         if ((handler = ueh) != null)
1357             wt.setUncaughtExceptionHandler(handler);
1358         do {} while (!U.compareAndSwapInt(this, INDEXSEED, s = indexSeed,
1359                 s += SEED_INCREMENT) ||
1360                 s == 0); // skip 0
1361         WorkQueue w = new WorkQueue(this, wt, mode, s);
1362         if (((ps = plock) & PL_LOCK) != 0 ||
1363                 !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
1364             ps = acquirePlock();
1365         int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
1366         try {
1367             if ((ws = workQueues) != null) {    // skip if shutting down
1368                 int n = ws.length, m = n - 1;
1369                 int r = (s << 1) | 1;           // use odd-numbered indices
1370                 if (ws[r &= m] != null) {       // collision
1371                     int probes = 0;             // step by approx half size
1372                     int step = (n <= 4) ? 2 : ((n >>> 1) & EVENMASK) + 2;
1373                     while (ws[r = (r + step) & m] != null) {
1374                         if (++probes >= n) {
1375                             workQueues = ws = Arrays.copyOf(ws, n <<= 1);
1376                             m = n - 1;
1377                             probes = 0;
1378                         }
1379                     }
1380                 }
1381                 w.poolIndex = (short)r;
1382                 w.eventCount = r; // volatile write orders
1383                 ws[r] = w;
1384             }
1385         } finally {
1386             if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
1387                 releasePlock(nps);
1388         }
1389         wt.setName(workerNamePrefix.concat(Integer.toString(w.poolIndex >>> 1)));
1390         return w;
1391     }
1392 
1393     /**
1394      * Final callback from terminating worker, as well as upon failure
1395      * to construct or start a worker.  Removes record of worker from
1396      * array, and adjusts counts. If pool is shutting down, tries to
1397      * complete termination.
1398      *
1399      * @param wt the worker thread, or null if construction failed
1400      * @param ex the exception causing failure, or null if none
1401      */
1402     final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) {
1403         WorkQueue w = null;
1404         if (wt != null && (w = wt.workQueue) != null) {
1405             int ps; long sc;
1406             w.qlock = -1;                // ensure set
1407             do {} while (!U.compareAndSwapLong(this, STEALCOUNT,
1408                     sc = stealCount,
1409                     sc + w.nsteals));
1410             if (((ps = plock) & PL_LOCK) != 0 ||
1411                     !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
1412                 ps = acquirePlock();
1413             int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
1414             try {
1415                 int idx = w.poolIndex;
1416                 WorkQueue[] ws = workQueues;
1417                 if (ws != null && idx >= 0 && idx < ws.length && ws[idx] == w)
1418                     ws[idx] = null;
1419             } finally {
1420                 if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
1421                     releasePlock(nps);
1422             }
1423         }
1424 
1425         long c;                          // adjust ctl counts
1426         do {} while (!U.compareAndSwapLong
1427                 (this, CTL, c = ctl, (((c - AC_UNIT) & AC_MASK) |
1428                         ((c - TC_UNIT) & TC_MASK) |
1429                         (c & ~(AC_MASK|TC_MASK)))));
1430 
1431         if (!tryTerminate(false, false) && w != null && w.array != null) {
1432             w.cancelAll();               // cancel remaining tasks
1433             WorkQueue[] ws; WorkQueue v; Thread p; int u, i, e;
1434             while ((u = (int)((c = ctl) >>> 32)) < 0 && (e = (int)c) >= 0) {
1435                 if (e > 0) {             // activate or create replacement
1436                     if ((ws = workQueues) == null ||
1437                             (i = e & SMASK) >= ws.length ||
1438                             (v = ws[i]) == null)
1439                         break;
1440                     long nc = (((long)(v.nextWait & E_MASK)) |
1441                             ((long)(u + UAC_UNIT) << 32));
1442                     if (v.eventCount != (e | INT_SIGN))
1443                         break;
1444                     if (U.compareAndSwapLong(this, CTL, c, nc)) {
1445                         v.eventCount = (e + E_SEQ) & E_MASK;
1446                         if ((p = v.parker) != null)
1447                             U.unpark(p);
1448                         break;
1449                     }
1450                 }
1451                 else {
1452                     if ((short)u < 0)
1453                         tryAddWorker();
1454                     break;
1455                 }
1456             }
1457         }
1458         if (ex == null)                     // help clean refs on way out
1459             ForkJoinTask.helpExpungeStaleExceptions();
1460         else                                // rethrow
1461             ForkJoinTask.rethrow(ex);
1462     }
1463 
1464     // Submissions
1465 
1466     /**
1467      * Per-thread records for threads that submit to pools. Currently
1468      * holds only pseudo-random seed / index that is used to choose
1469      * submission queues in method externalPush. In the future, this may
1470      * also incorporate a means to implement different task rejection
1471      * and resubmission policies.
1472      *
1473      * Seeds for submitters and workers/workQueues work in basically
1474      * the same way but are initialized and updated using slightly
1475      * different mechanics. Both are initialized using the same
1476      * approach as in class ThreadLocal, where successive values are
1477      * unlikely to collide with previous values. Seeds are then
1478      * randomly modified upon collisions using xorshifts, which
1479      * requires a non-zero seed.
1480      */
1481     static final class Submitter {
1482         int seed;
1483         Submitter(int s) { seed = s; }
1484     }
1485 
1486     /**
1487      * Unless shutting down, adds the given task to a submission queue
1488      * at submitter's current queue index (modulo submission
1489      * range). Only the most common path is directly handled in this
1490      * method. All others are relayed to fullExternalPush.
1491      *
1492      * @param task the task. Caller must ensure non-null.
1493      */
1494     final void externalPush(ForkJoinTask<?> task) {
1495         Submitter z = submitters.get();
1496         WorkQueue q; int r, m, s, n, am; ForkJoinTask<?>[] a;
1497         int ps = plock;
1498         WorkQueue[] ws = workQueues;
1499         if (z != null && ps > 0 && ws != null && (m = (ws.length - 1)) >= 0 &&
1500                 (q = ws[m & (r = z.seed) & SQMASK]) != null && r != 0 &&
1501                 U.compareAndSwapInt(q, QLOCK, 0, 1)) { // lock
1502             if ((a = q.array) != null &&
1503                     (am = a.length - 1) > (n = (s = q.top) - q.base)) {
1504                 int j = ((am & s) << ASHIFT) + ABASE;
1505                 U.putOrderedObject(a, j, task);
1506                 q.top = s + 1;                     // push on to deque
1507                 q.qlock = 0;
1508                 if (n <= 1)
1509                     signalWork(ws, q);
1510                 return;
1511             }
1512             q.qlock = 0;
1513         }
1514         fullExternalPush(task);
1515     }
1516 
1517     /**
1518      * Full version of externalPush. This method is called, among
1519      * other times, upon the first submission of the first task to the
1520      * pool, so must perform secondary initialization.  It also
1521      * detects first submission by an external thread by looking up
1522      * its ThreadLocal, and creates a new shared queue if the one at
1523      * index if empty or contended. The plock lock body must be
1524      * exception-free (so no try/finally) so we optimistically
1525      * allocate new queues outside the lock and throw them away if
1526      * (very rarely) not needed.
1527      *
1528      * Secondary initialization occurs when plock is zero, to create
1529      * workQueue array and set plock to a valid value.  This lock body
1530      * must also be exception-free. Because the plock seq value can
1531      * eventually wrap around zero, this method harmlessly fails to
1532      * reinitialize if workQueues exists, while still advancing plock.
1533      */
1534     private void fullExternalPush(ForkJoinTask<?> task) {
1535         int r = 0; // random index seed
1536         for (Submitter z = submitters.get();;) {
1537             WorkQueue[] ws; WorkQueue q; int ps, m, k;
1538             if (z == null) {
1539                 if (U.compareAndSwapInt(this, INDEXSEED, r = indexSeed,
1540                         r += SEED_INCREMENT) && r != 0)
1541                     submitters.set(z = new Submitter(r));
1542             }
1543             else if (r == 0) {                  // move to a different index
1544                 r = z.seed;
1545                 r ^= r << 13;                   // same xorshift as WorkQueues
1546                 r ^= r >>> 17;
1547                 z.seed = r ^= (r << 5);
1548             }
1549             if ((ps = plock) < 0)
1550                 throw new RejectedExecutionException();
1551             else if (ps == 0 || (ws = workQueues) == null ||
1552                     (m = ws.length - 1) < 0) { // initialize workQueues
1553                 int p = parallelism;            // find power of two table size
1554                 int n = (p > 1) ? p - 1 : 1;    // ensure at least 2 slots
1555                 n |= n >>> 1; n |= n >>> 2;  n |= n >>> 4;
1556                 n |= n >>> 8; n |= n >>> 16; n = (n + 1) << 1;
1557                 WorkQueue[] nws = ((ws = workQueues) == null || ws.length == 0 ?
1558                         new WorkQueue[n] : null);
1559                 if (((ps = plock) & PL_LOCK) != 0 ||
1560                         !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
1561                     ps = acquirePlock();
1562                 if (((ws = workQueues) == null || ws.length == 0) && nws != null)
1563                     workQueues = nws;
1564                 int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
1565                 if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
1566                     releasePlock(nps);
1567             }
1568             else if ((q = ws[k = r & m & SQMASK]) != null) {
1569                 if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) {
1570                     ForkJoinTask<?>[] a = q.array;
1571                     int s = q.top;
1572                     boolean submitted = false;
1573                     try {                      // locked version of push
1574                         if ((a != null && a.length > s + 1 - q.base) ||
1575                                 (a = q.growArray()) != null) {   // must presize
1576                             int j = (((a.length - 1) & s) << ASHIFT) + ABASE;
1577                             U.putOrderedObject(a, j, task);
1578                             q.top = s + 1;
1579                             submitted = true;
1580                         }
1581                     } finally {
1582                         q.qlock = 0;  // unlock
1583                     }
1584                     if (submitted) {
1585                         signalWork(ws, q);
1586                         return;
1587                     }
1588                 }
1589                 r = 0; // move on failure
1590             }
1591             else if (((ps = plock) & PL_LOCK) == 0) { // create new queue
1592                 q = new WorkQueue(this, null, SHARED_QUEUE, r);
1593                 q.poolIndex = (short)k;
1594                 if (((ps = plock) & PL_LOCK) != 0 ||
1595                         !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
1596                     ps = acquirePlock();
1597                 if ((ws = workQueues) != null && k < ws.length && ws[k] == null)
1598                     ws[k] = q;
1599                 int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
1600                 if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
1601                     releasePlock(nps);
1602             }
1603             else
1604                 r = 0;
1605         }
1606     }
1607 
1608     // Maintaining ctl counts
1609 
1610     /**
1611      * Increments active count; mainly called upon return from blocking.
1612      */
1613     final void incrementActiveCount() {
1614         long c;
1615         do {} while (!U.compareAndSwapLong
1616                 (this, CTL, c = ctl, ((c & ~AC_MASK) |
1617                         ((c & AC_MASK) + AC_UNIT))));
1618     }
1619 
1620     /**
1621      * Tries to create or activate a worker if too few are active.
1622      *
1623      * @param ws the worker array to use to find signallees
1624      * @param q if non-null, the queue holding tasks to be processed
1625      */
1626     final void signalWork(WorkQueue[] ws, WorkQueue q) {
1627         for (;;) {
1628             long c; int e, u, i; WorkQueue w; Thread p;
1629             if ((u = (int)((c = ctl) >>> 32)) >= 0)
1630                 break;
1631             if ((e = (int)c) <= 0) {
1632                 if ((short)u < 0)
1633                     tryAddWorker();
1634                 break;
1635             }
1636             if (ws == null || ws.length <= (i = e & SMASK) ||
1637                     (w = ws[i]) == null)
1638                 break;
1639             long nc = (((long)(w.nextWait & E_MASK)) |
1640                     ((long)(u + UAC_UNIT)) << 32);
1641             int ne = (e + E_SEQ) & E_MASK;
1642             if (w.eventCount == (e | INT_SIGN) &&
1643                     U.compareAndSwapLong(this, CTL, c, nc)) {
1644                 w.eventCount = ne;
1645                 if ((p = w.parker) != null)
1646                     U.unpark(p);
1647                 break;
1648             }
1649             if (q != null && q.base >= q.top)
1650                 break;
1651         }
1652     }
1653 
1654     // Scanning for tasks
1655 
1656     /**
1657      * Top-level runloop for workers, called by ForkJoinWorkerThread.run.
1658      */
1659     final void runWorker(WorkQueue w) {
1660         w.growArray(); // allocate queue
1661         for (int r = w.hint; scan(w, r) == 0; ) {
1662             r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift
1663         }
1664     }
1665 
1666     /**
1667      * Scans for and, if found, runs one task, else possibly
1668      * inactivates the worker. This method operates on single reads of
1669      * volatile state and is designed to be re-invoked continuously,
1670      * in part because it returns upon detecting inconsistencies,
1671      * contention, or state changes that indicate possible success on
1672      * re-invocation.
1673      *
1674      * The scan searches for tasks across queues starting at a random
1675      * index, checking each at least twice.  The scan terminates upon
1676      * either finding a non-empty queue, or completing the sweep. If
1677      * the worker is not inactivated, it takes and runs a task from
1678      * this queue. Otherwise, if not activated, it tries to activate
1679      * itself or some other worker by signalling. On failure to find a
1680      * task, returns (for retry) if pool state may have changed during
1681      * an empty scan, or tries to inactivate if active, else possibly
1682      * blocks or terminates via method awaitWork.
1683      *
1684      * @param w the worker (via its WorkQueue)
1685      * @param r a random seed
1686      * @return worker qlock status if would have waited, else 0
1687      */
1688     private final int scan(WorkQueue w, int r) {
1689         WorkQueue[] ws; int m;
1690         long c = ctl;                            // for consistency check
1691         if ((ws = workQueues) != null && (m = ws.length - 1) >= 0 && w != null) {
1692             for (int j = m + m + 1, ec = w.eventCount;;) {
1693                 WorkQueue q; int b, e; ForkJoinTask<?>[] a; ForkJoinTask<?> t;
1694                 if ((q = ws[(r - j) & m]) != null &&
1695                         (b = q.base) - q.top < 0 && (a = q.array) != null) {
1696                     long i = (((a.length - 1) & b) << ASHIFT) + ABASE;
1697                     if ((t = ((ForkJoinTask<?>)
1698                             U.getObjectVolatile(a, i))) != null) {
1699                         if (ec < 0)
1700                             helpRelease(c, ws, w, q, b);
1701                         else if (q.base == b &&
1702                                 U.compareAndSwapObject(a, i, t, null)) {
1703                             U.putOrderedInt(q, QBASE, b + 1);
1704                             if ((b + 1) - q.top < 0)
1705                                 signalWork(ws, q);
1706                             w.runTask(t);
1707                         }
1708                     }
1709                     break;
1710                 }
1711                 else if (--j < 0) {
1712                     if ((ec | (e = (int)c)) < 0) // inactive or terminating
1713                         return awaitWork(w, c, ec);
1714                     else if (ctl == c) {         // try to inactivate and enqueue
1715                         long nc = (long)ec | ((c - AC_UNIT) & (AC_MASK|TC_MASK));
1716                         w.nextWait = e;
1717                         w.eventCount = ec | INT_SIGN;
1718                         if (!U.compareAndSwapLong(this, CTL, c, nc))
1719                             w.eventCount = ec;   // back out
1720                     }
1721                     break;
1722                 }
1723             }
1724         }
1725         return 0;
1726     }
1727 
1728     /**
1729      * A continuation of scan(), possibly blocking or terminating
1730      * worker w. Returns without blocking if pool state has apparently
1731      * changed since last invocation.  Also, if inactivating w has
1732      * caused the pool to become quiescent, checks for pool
1733      * termination, and, so long as this is not the only worker, waits
1734      * for event for up to a given duration.  On timeout, if ctl has
1735      * not changed, terminates the worker, which will in turn wake up
1736      * another worker to possibly repeat this process.
1737      *
1738      * @param w the calling worker
1739      * @param c the ctl value on entry to scan
1740      * @param ec the worker's eventCount on entry to scan
1741      */
1742     private final int awaitWork(WorkQueue w, long c, int ec) {
1743         int stat, ns; long parkTime, deadline;
1744         if ((stat = w.qlock) >= 0 && w.eventCount == ec && ctl == c &&
1745                 !Thread.interrupted()) {
1746             int e = (int)c;
1747             int u = (int)(c >>> 32);
1748             int d = (u >> UAC_SHIFT) + parallelism; // active count
1749 
1750             if (e < 0 || (d <= 0 && tryTerminate(false, false)))
1751                 stat = w.qlock = -1;          // pool is terminating
1752             else if ((ns = w.nsteals) != 0) { // collect steals and retry
1753                 long sc;
1754                 w.nsteals = 0;
1755                 do {} while (!U.compareAndSwapLong(this, STEALCOUNT,
1756                         sc = stealCount, sc + ns));
1757             }
1758             else {
1759                 long pc = ((d > 0 || ec != (e | INT_SIGN)) ? 0L :
1760                         ((long)(w.nextWait & E_MASK)) | // ctl to restore
1761                                 ((long)(u + UAC_UNIT)) << 32);
1762                 if (pc != 0L) {               // timed wait if last waiter
1763                     int dc = -(short)(c >>> TC_SHIFT);
1764                     parkTime = (dc < 0 ? FAST_IDLE_TIMEOUT:
1765                             (dc + 1) * IDLE_TIMEOUT);
1766                     deadline = System.nanoTime() + parkTime - TIMEOUT_SLOP;
1767                 }
1768                 else
1769                     parkTime = deadline = 0L;
1770                 if (w.eventCount == ec && ctl == c) {
1771                     Thread wt = Thread.currentThread();
1772                     U.putObject(wt, PARKBLOCKER, this);
1773                     w.parker = wt;            // emulate LockSupport.park
1774                     if (w.eventCount == ec && ctl == c)
1775                         U.park(false, parkTime);  // must recheck before park
1776                     w.parker = null;
1777                     U.putObject(wt, PARKBLOCKER, null);
1778                     if (parkTime != 0L && ctl == c &&
1779                             deadline - System.nanoTime() <= 0L &&
1780                             U.compareAndSwapLong(this, CTL, c, pc))
1781                         stat = w.qlock = -1;  // shrink pool
1782                 }
1783             }
1784         }
1785         return stat;
1786     }
1787 
1788     /**
1789      * Possibly releases (signals) a worker. Called only from scan()
1790      * when a worker with apparently inactive status finds a non-empty
1791      * queue. This requires revalidating all of the associated state
1792      * from caller.
1793      */
1794     private final void helpRelease(long c, WorkQueue[] ws, WorkQueue w,
1795                                    WorkQueue q, int b) {
1796         WorkQueue v; int e, i; Thread p;
1797         if (w != null && w.eventCount < 0 && (e = (int)c) > 0 &&
1798                 ws != null && ws.length > (i = e & SMASK) &&
1799                 (v = ws[i]) != null && ctl == c) {
1800             long nc = (((long)(v.nextWait & E_MASK)) |
1801                     ((long)((int)(c >>> 32) + UAC_UNIT)) << 32);
1802             int ne = (e + E_SEQ) & E_MASK;
1803             if (q != null && q.base == b && w.eventCount < 0 &&
1804                     v.eventCount == (e | INT_SIGN) &&
1805                     U.compareAndSwapLong(this, CTL, c, nc)) {
1806                 v.eventCount = ne;
1807                 if ((p = v.parker) != null)
1808                     U.unpark(p);
1809             }
1810         }
1811     }
1812 
1813     /**
1814      * Tries to locate and execute tasks for a stealer of the given
1815      * task, or in turn one of its stealers, Traces currentSteal ->
1816      * currentJoin links looking for a thread working on a descendant
1817      * of the given task and with a non-empty queue to steal back and
1818      * execute tasks from. The first call to this method upon a
1819      * waiting join will often entail scanning/search, (which is OK
1820      * because the joiner has nothing better to do), but this method
1821      * leaves hints in workers to speed up subsequent calls. The
1822      * implementation is very branchy to cope with potential
1823      * inconsistencies or loops encountering chains that are stale,
1824      * unknown, or so long that they are likely cyclic.
1825      *
1826      * @param joiner the joining worker
1827      * @param task the task to join
1828      * @return 0 if no progress can be made, negative if task
1829      * known complete, else positive
1830      */
1831     private int tryHelpStealer(WorkQueue joiner, ForkJoinTask<?> task) {
1832         int stat = 0, steps = 0;                    // bound to avoid cycles
1833         if (task != null && joiner != null &&
1834                 joiner.base - joiner.top >= 0) {        // hoist checks
1835             restart: for (;;) {
1836                 ForkJoinTask<?> subtask = task;     // current target
1837                 for (WorkQueue j = joiner, v;;) {   // v is stealer of subtask
1838                     WorkQueue[] ws; int m, s, h;
1839                     if ((s = task.status) < 0) {
1840                         stat = s;
1841                         break restart;
1842                     }
1843                     if ((ws = workQueues) == null || (m = ws.length - 1) <= 0)
1844                         break restart;              // shutting down
1845                     if ((v = ws[h = (j.hint | 1) & m]) == null ||
1846                             v.currentSteal != subtask) {
1847                         for (int origin = h;;) {    // find stealer
1848                             if (((h = (h + 2) & m) & 15) == 1 &&
1849                                     (subtask.status < 0 || j.currentJoin != subtask))
1850                                 continue restart;   // occasional staleness check
1851                             if ((v = ws[h]) != null &&
1852                                     v.currentSteal == subtask) {
1853                                 j.hint = h;        // save hint
1854                                 break;
1855                             }
1856                             if (h == origin)
1857                                 break restart;      // cannot find stealer
1858                         }
1859                     }
1860                     for (;;) { // help stealer or descend to its stealer
1861                         ForkJoinTask[] a; int b;
1862                         if (subtask.status < 0)     // surround probes with
1863                             continue restart;       //   consistency checks
1864                         if ((b = v.base) - v.top < 0 && (a = v.array) != null) {
1865                             int i = (((a.length - 1) & b) << ASHIFT) + ABASE;
1866                             ForkJoinTask<?> t =
1867                                     (ForkJoinTask<?>)U.getObjectVolatile(a, i);
1868                             if (subtask.status < 0 || j.currentJoin != subtask ||
1869                                     v.currentSteal != subtask)
1870                                 continue restart;   // stale
1871                             stat = 1;               // apparent progress
1872                             if (v.base == b) {
1873                                 if (t == null)
1874                                     break restart;
1875                                 if (U.compareAndSwapObject(a, i, t, null)) {
1876                                     U.putOrderedInt(v, QBASE, b + 1);
1877                                     ForkJoinTask<?> ps = joiner.currentSteal;
1878                                     int jt = joiner.top;
1879                                     do {
1880                                         joiner.currentSteal = t;
1881                                         t.doExec(); // clear local tasks too
1882                                     } while (task.status >= 0 &&
1883                                             joiner.top != jt &&
1884                                             (t = joiner.pop()) != null);
1885                                     joiner.currentSteal = ps;
1886                                     break restart;
1887                                 }
1888                             }
1889                         }
1890                         else {                      // empty -- try to descend
1891                             ForkJoinTask<?> next = v.currentJoin;
1892                             if (subtask.status < 0 || j.currentJoin != subtask ||
1893                                     v.currentSteal != subtask)
1894                                 continue restart;   // stale
1895                             else if (next == null || ++steps == MAX_HELP)
1896                                 break restart;      // dead-end or maybe cyclic
1897                             else {
1898                                 subtask = next;
1899                                 j = v;
1900                                 break;
1901                             }
1902                         }
1903                     }
1904                 }
1905             }
1906         }
1907         return stat;
1908     }
1909 
1910     /**
1911      * Analog of tryHelpStealer for CountedCompleters. Tries to steal
1912      * and run tasks within the target's computation.
1913      *
1914      * @param task the task to join
1915      */
1916     private int helpComplete(WorkQueue joiner, CountedCompleter<?> task) {
1917         WorkQueue[] ws; int m;
1918         int s = 0;
1919         if ((ws = workQueues) != null && (m = ws.length - 1) >= 0 &&
1920                 joiner != null && task != null) {
1921             int j = joiner.poolIndex;
1922             int scans = m + m + 1;
1923             long c = 0L;              // for stability check
1924             for (int k = scans; ; j += 2) {
1925                 WorkQueue q;
1926                 if ((s = task.status) < 0)
1927                     break;
1928                 else if (joiner.internalPopAndExecCC(task))
1929                     k = scans;
1930                 else if ((s = task.status) < 0)
1931                     break;
1932                 else if ((q = ws[j & m]) != null && q.pollAndExecCC(task))
1933                     k = scans;
1934                 else if (--k < 0) {
1935                     if (c == (c = ctl))
1936                         break;
1937                     k = scans;
1938                 }
1939             }
1940         }
1941         return s;
1942     }
1943 
1944     /**
1945      * Tries to decrement active count (sometimes implicitly) and
1946      * possibly release or create a compensating worker in preparation
1947      * for blocking. Fails on contention or termination. Otherwise,
1948      * adds a new thread if no idle workers are available and pool
1949      * may become starved.
1950      *
1951      * @param c the assumed ctl value
1952      */
1953     final boolean tryCompensate(long c) {
1954         WorkQueue[] ws = workQueues;
1955         int pc = parallelism, e = (int)c, m, tc;
1956         if (ws != null && (m = ws.length - 1) >= 0 && e >= 0 && ctl == c) {
1957             WorkQueue w = ws[e & m];
1958             if (e != 0 && w != null) {
1959                 Thread p;
1960                 long nc = ((long)(w.nextWait & E_MASK) |
1961                         (c & (AC_MASK|TC_MASK)));
1962                 int ne = (e + E_SEQ) & E_MASK;
1963                 if (w.eventCount == (e | INT_SIGN) &&
1964                         U.compareAndSwapLong(this, CTL, c, nc)) {
1965                     w.eventCount = ne;
1966                     if ((p = w.parker) != null)
1967                         U.unpark(p);
1968                     return true;   // replace with idle worker
1969                 }
1970             }
1971             else if ((tc = (short)(c >>> TC_SHIFT)) >= 0 &&
1972                     (int)(c >> AC_SHIFT) + pc > 1) {
1973                 long nc = ((c - AC_UNIT) & AC_MASK) | (c & ~AC_MASK);
1974                 if (U.compareAndSwapLong(this, CTL, c, nc))
1975                     return true;   // no compensation
1976             }
1977             else if (tc + pc < MAX_CAP) {
1978                 long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK);
1979                 if (U.compareAndSwapLong(this, CTL, c, nc)) {
1980                     ForkJoinWorkerThreadFactory fac;
1981                     Throwable ex = null;
1982                     ForkJoinWorkerThread wt = null;
1983                     try {
1984                         if ((fac = factory) != null &&
1985                                 (wt = fac.newThread(this)) != null) {
1986                             wt.start();
1987                             return true;
1988                         }
1989                     } catch (Throwable rex) {
1990                         ex = rex;
1991                     }
1992                     deregisterWorker(wt, ex); // clean up and return false
1993                 }
1994             }
1995         }
1996         return false;
1997     }
1998 
1999     /**
2000      * Helps and/or blocks until the given task is done.
2001      *
2002      * @param joiner the joining worker
2003      * @param task the task
2004      * @return task status on exit
2005      */
2006     final int awaitJoin(WorkQueue joiner, ForkJoinTask<?> task) {
2007         int s = 0;
2008         if (task != null && (s = task.status) >= 0 && joiner != null) {
2009             ForkJoinTask<?> prevJoin = joiner.currentJoin;
2010             joiner.currentJoin = task;
2011             do {} while (joiner.tryRemoveAndExec(task) && // process local tasks
2012                     (s = task.status) >= 0);
2013             if (s >= 0 && (task instanceof CountedCompleter))
2014                 s = helpComplete(joiner, (CountedCompleter<?>)task);
2015             long cc = 0;        // for stability checks
2016             while (s >= 0 && (s = task.status) >= 0) {
2017                 if ((s = tryHelpStealer(joiner, task)) == 0 &&
2018                         (s = task.status) >= 0) {
2019                     if (!tryCompensate(cc))
2020                         cc = ctl;
2021                     else {
2022                         if (task.trySetSignal() && (s = task.status) >= 0) {
2023                             synchronized (task) {
2024                                 if (task.status >= 0) {
2025                                     try {                // see ForkJoinTask
2026                                         task.wait();     //  for explanation
2027                                     } catch (InterruptedException ie) {
2028                                     }
2029                                 }
2030                                 else
2031                                     task.notifyAll();
2032                             }
2033                         }
2034                         long c; // reactivate
2035                         do {} while (!U.compareAndSwapLong
2036                                 (this, CTL, c = ctl,
2037                                         ((c & ~AC_MASK) |
2038                                                 ((c & AC_MASK) + AC_UNIT))));
2039                     }
2040                 }
2041             }
2042             joiner.currentJoin = prevJoin;
2043         }
2044         return s;
2045     }
2046 
2047     /**
2048      * Stripped-down variant of awaitJoin used by timed joins. Tries
2049      * to help join only while there is continuous progress. (Caller
2050      * will then enter a timed wait.)
2051      *
2052      * @param joiner the joining worker
2053      * @param task the task
2054      */
2055     final void helpJoinOnce(WorkQueue joiner, ForkJoinTask<?> task) {
2056         int s;
2057         if (joiner != null && task != null && (s = task.status) >= 0) {
2058             ForkJoinTask<?> prevJoin = joiner.currentJoin;
2059             joiner.currentJoin = task;
2060             do {} while (joiner.tryRemoveAndExec(task) && // process local tasks
2061                     (s = task.status) >= 0);
2062             if (s >= 0) {
2063                 if (task instanceof CountedCompleter)
2064                     helpComplete(joiner, (CountedCompleter<?>)task);
2065                 do {} while (task.status >= 0 &&
2066                         tryHelpStealer(joiner, task) > 0);
2067             }
2068             joiner.currentJoin = prevJoin;
2069         }
2070     }
2071 
2072     /**
2073      * Returns a (probably) non-empty steal queue, if one is found
2074      * during a scan, else null.  This method must be retried by
2075      * caller if, by the time it tries to use the queue, it is empty.
2076      */
2077     private WorkQueue findNonEmptyStealQueue() {
2078         int r = ThreadLocalRandom.current().nextInt();
2079         for (;;) {
2080             int ps = plock, m; WorkQueue[] ws; WorkQueue q;
2081             if ((ws = workQueues) != null && (m = ws.length - 1) >= 0) {
2082                 for (int j = (m + 1) << 2; j >= 0; --j) {
2083                     if ((q = ws[(((r - j) << 1) | 1) & m]) != null &&
2084                             q.base - q.top < 0)
2085                         return q;
2086                 }
2087             }
2088             if (plock == ps)
2089                 return null;
2090         }
2091     }
2092 
2093     /**
2094      * Runs tasks until {@code isQuiescent()}. We piggyback on
2095      * active count ctl maintenance, but rather than blocking
2096      * when tasks cannot be found, we rescan until all others cannot
2097      * find tasks either.
2098      */
2099     final void helpQuiescePool(WorkQueue w) {
2100         ForkJoinTask<?> ps = w.currentSteal;
2101         for (boolean active = true;;) {
2102             long c; WorkQueue q; ForkJoinTask<?> t; int b;
2103             while ((t = w.nextLocalTask()) != null)
2104                 t.doExec();
2105             if ((q = findNonEmptyStealQueue()) != null) {
2106                 if (!active) {      // re-establish active count
2107                     active = true;
2108                     do {} while (!U.compareAndSwapLong
2109                             (this, CTL, c = ctl,
2110                                     ((c & ~AC_MASK) |
2111                                             ((c & AC_MASK) + AC_UNIT))));
2112                 }
2113                 if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null) {
2114                     (w.currentSteal = t).doExec();
2115                     w.currentSteal = ps;
2116                 }
2117             }
2118             else if (active) {       // decrement active count without queuing
2119                 long nc = ((c = ctl) & ~AC_MASK) | ((c & AC_MASK) - AC_UNIT);
2120                 if ((int)(nc >> AC_SHIFT) + parallelism == 0)
2121                     break;          // bypass decrement-then-increment
2122                 if (U.compareAndSwapLong(this, CTL, c, nc))
2123                     active = false;
2124             }
2125             else if ((int)((c = ctl) >> AC_SHIFT) + parallelism <= 0 &&
2126                     U.compareAndSwapLong
2127                             (this, CTL, c, ((c & ~AC_MASK) |
2128                                     ((c & AC_MASK) + AC_UNIT))))
2129                 break;
2130         }
2131     }
2132 
2133     /**
2134      * Gets and removes a local or stolen task for the given worker.
2135      *
2136      * @return a task, if available
2137      */
2138     final ForkJoinTask<?> nextTaskFor(WorkQueue w) {
2139         for (ForkJoinTask<?> t;;) {
2140             WorkQueue q; int b;
2141             if ((t = w.nextLocalTask()) != null)
2142                 return t;
2143             if ((q = findNonEmptyStealQueue()) == null)
2144                 return null;
2145             if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null)
2146                 return t;
2147         }
2148     }
2149 
2150     /**
2151      * Returns a cheap heuristic guide for task partitioning when
2152      * programmers, frameworks, tools, or languages have little or no
2153      * idea about task granularity.  In essence by offering this
2154      * method, we ask users only about tradeoffs in overhead vs
2155      * expected throughput and its variance, rather than how finely to
2156      * partition tasks.
2157      *
2158      * In a steady state strict (tree-structured) computation, each
2159      * thread makes available for stealing enough tasks for other
2160      * threads to remain active. Inductively, if all threads play by
2161      * the same rules, each thread should make available only a
2162      * constant number of tasks.
2163      *
2164      * The minimum useful constant is just 1. But using a value of 1
2165      * would require immediate replenishment upon each steal to
2166      * maintain enough tasks, which is infeasible.  Further,
2167      * partitionings/granularities of offered tasks should minimize
2168      * steal rates, which in general means that threads nearer the top
2169      * of computation tree should generate more than those nearer the
2170      * bottom. In perfect steady state, each thread is at
2171      * approximately the same level of computation tree. However,
2172      * producing extra tasks amortizes the uncertainty of progress and
2173      * diffusion assumptions.
2174      *
2175      * So, users will want to use values larger (but not much larger)
2176      * than 1 to both smooth over transient shortages and hedge
2177      * against uneven progress; as traded off against the cost of
2178      * extra task overhead. We leave the user to pick a threshold
2179      * value to compare with the results of this call to guide
2180      * decisions, but recommend values such as 3.
2181      *
2182      * When all threads are active, it is on average OK to estimate
2183      * surplus strictly locally. In steady-state, if one thread is
2184      * maintaining say 2 surplus tasks, then so are others. So we can
2185      * just use estimated queue length.  However, this strategy alone
2186      * leads to serious mis-estimates in some non-steady-state
2187      * conditions (ramp-up, ramp-down, other stalls). We can detect
2188      * many of these by further considering the number of "idle"
2189      * threads, that are known to have zero queued tasks, so
2190      * compensate by a factor of (#idle/#active) threads.
2191      *
2192      * Note: The approximation of #busy workers as #active workers is
2193      * not very good under current signalling scheme, and should be
2194      * improved.
2195      */
2196     static int getSurplusQueuedTaskCount() {
2197         Thread t; ForkJoinWorkerThread wt; ForkJoinPool pool; WorkQueue q;
2198         if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) {
2199             int p = (pool = (wt = (ForkJoinWorkerThread)t).pool).parallelism;
2200             int n = (q = wt.workQueue).top - q.base;
2201             int a = (int)(pool.ctl >> AC_SHIFT) + p;
2202             return n - (a > (p >>>= 1) ? 0 :
2203                     a > (p >>>= 1) ? 1 :
2204                             a > (p >>>= 1) ? 2 :
2205                                     a > (p >>>= 1) ? 4 :
2206                                             8);
2207         }
2208         return 0;
2209     }
2210 
2211     //  Termination
2212 
2213     /**
2214      * Possibly initiates and/or completes termination.  The caller
2215      * triggering termination runs three passes through workQueues:
2216      * (0) Setting termination status, followed by wakeups of queued
2217      * workers; (1) cancelling all tasks; (2) interrupting lagging
2218      * threads (likely in external tasks, but possibly also blocked in
2219      * joins).  Each pass repeats previous steps because of potential
2220      * lagging thread creation.
2221      *
2222      * @param now if true, unconditionally terminate, else only
2223      * if no work and no active workers
2224      * @param enable if true, enable shutdown when next possible
2225      * @return true if now terminating or terminated
2226      */
2227     private boolean tryTerminate(boolean now, boolean enable) {
2228         int ps;
2229         if (this == common)                        // cannot shut down
2230             return false;
2231         if ((ps = plock) >= 0) {                   // enable by setting plock
2232             if (!enable)
2233                 return false;
2234             if ((ps & PL_LOCK) != 0 ||
2235                     !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
2236                 ps = acquirePlock();
2237             int nps = ((ps + PL_LOCK) & ~SHUTDOWN) | SHUTDOWN;
2238             if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
2239                 releasePlock(nps);
2240         }
2241         for (long c;;) {
2242             if (((c = ctl) & STOP_BIT) != 0) {     // already terminating
2243                 if ((short)(c >>> TC_SHIFT) + parallelism <= 0) {
2244                     synchronized (this) {
2245                         notifyAll();               // signal when 0 workers
2246                     }
2247                 }
2248                 return true;
2249             }
2250             if (!now) {                            // check if idle & no tasks
2251                 WorkQueue[] ws; WorkQueue w;
2252                 if ((int)(c >> AC_SHIFT) + parallelism > 0)
2253                     return false;
2254                 if ((ws = workQueues) != null) {
2255                     for (int i = 0; i < ws.length; ++i) {
2256                         if ((w = ws[i]) != null &&
2257                                 (!w.isEmpty() ||
2258                                         ((i & 1) != 0 && w.eventCount >= 0))) {
2259                             signalWork(ws, w);
2260                             return false;
2261                         }
2262                     }
2263                 }
2264             }
2265             if (U.compareAndSwapLong(this, CTL, c, c | STOP_BIT)) {
2266                 for (int pass = 0; pass < 3; ++pass) {
2267                     WorkQueue[] ws; WorkQueue w; Thread wt;
2268                     if ((ws = workQueues) != null) {
2269                         int n = ws.length;
2270                         for (int i = 0; i < n; ++i) {
2271                             if ((w = ws[i]) != null) {
2272                                 w.qlock = -1;
2273                                 if (pass > 0) {
2274                                     w.cancelAll();
2275                                     if (pass > 1 && (wt = w.owner) != null) {
2276                                         if (!wt.isInterrupted()) {
2277                                             try {
2278                                                 wt.interrupt();
2279                                             } catch (Throwable ignore) {
2280                                             }
2281                                         }
2282                                         U.unpark(wt);
2283                                     }
2284                                 }
2285                             }
2286                         }
2287                         // Wake up workers parked on event queue
2288                         int i, e; long cc; Thread p;
2289                         while ((e = (int)(cc = ctl) & E_MASK) != 0 &&
2290                                 (i = e & SMASK) < n && i >= 0 &&
2291                                 (w = ws[i]) != null) {
2292                             long nc = ((long)(w.nextWait & E_MASK) |
2293                                     ((cc + AC_UNIT) & AC_MASK) |
2294                                     (cc & (TC_MASK|STOP_BIT)));
2295                             if (w.eventCount == (e | INT_SIGN) &&
2296                                     U.compareAndSwapLong(this, CTL, cc, nc)) {
2297                                 w.eventCount = (e + E_SEQ) & E_MASK;
2298                                 w.qlock = -1;
2299                                 if ((p = w.parker) != null)
2300                                     U.unpark(p);
2301                             }
2302                         }
2303                     }
2304                 }
2305             }
2306         }
2307     }
2308 
2309     // external operations on common pool
2310 
2311     /**
2312      * Returns common pool queue for a thread that has submitted at
2313      * least one task.
2314      */
2315     static WorkQueue commonSubmitterQueue() {
2316         Submitter z; ForkJoinPool p; WorkQueue[] ws; int m, r;
2317         return ((z = submitters.get()) != null &&
2318                 (p = common) != null &&
2319                 (ws = p.workQueues) != null &&
2320                 (m = ws.length - 1) >= 0) ?
2321                 ws[m & z.seed & SQMASK] : null;
2322     }
2323 
2324     /**
2325      * Tries to pop the given task from submitter's queue in common pool.
2326      */
2327     final boolean tryExternalUnpush(ForkJoinTask<?> task) {
2328         WorkQueue joiner; ForkJoinTask<?>[] a; int m, s;
2329         Submitter z = submitters.get();
2330         WorkQueue[] ws = workQueues;
2331         boolean popped = false;
2332         if (z != null && ws != null && (m = ws.length - 1) >= 0 &&
2333                 (joiner = ws[z.seed & m & SQMASK]) != null &&
2334                 joiner.base != (s = joiner.top) &&
2335                 (a = joiner.array) != null) {
2336             long j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE;
2337             if (U.getObject(a, j) == task &&
2338                     U.compareAndSwapInt(joiner, QLOCK, 0, 1)) {
2339                 if (joiner.top == s && joiner.array == a &&
2340                         U.compareAndSwapObject(a, j, task, null)) {
2341                     joiner.top = s - 1;
2342                     popped = true;
2343                 }
2344                 joiner.qlock = 0;
2345             }
2346         }
2347         return popped;
2348     }
2349 
2350     final int externalHelpComplete(CountedCompleter<?> task) {
2351         WorkQueue joiner; int m, j;
2352         Submitter z = submitters.get();
2353         WorkQueue[] ws = workQueues;
2354         int s = 0;
2355         if (z != null && ws != null && (m = ws.length - 1) >= 0 &&
2356                 (joiner = ws[(j = z.seed) & m & SQMASK]) != null && task != null) {
2357             int scans = m + m + 1;
2358             long c = 0L;             // for stability check
2359             j |= 1;                  // poll odd queues
2360             for (int k = scans; ; j += 2) {
2361                 WorkQueue q;
2362                 if ((s = task.status) < 0)
2363                     break;
2364                 else if (joiner.externalPopAndExecCC(task))
2365                     k = scans;
2366                 else if ((s = task.status) < 0)
2367                     break;
2368                 else if ((q = ws[j & m]) != null && q.pollAndExecCC(task))
2369                     k = scans;
2370                 else if (--k < 0) {
2371                     if (c == (c = ctl))
2372                         break;
2373                     k = scans;
2374                 }
2375             }
2376         }
2377         return s;
2378     }
2379 
2380     // Exported methods
2381 
2382     // Constructors
2383 
2384     /**
2385      * Creates a {@code ForkJoinPool} with parallelism equal to {@link
2386      * java.lang.Runtime#availableProcessors}, using the {@linkplain
2387      * #defaultForkJoinWorkerThreadFactory default thread factory},
2388      * no UncaughtExceptionHandler, and non-async LIFO processing mode.
2389      *
2390      * @throws SecurityException if a security manager exists and
2391      *         the caller is not permitted to modify threads
2392      *         because it does not hold {@link
2393      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2394      */
2395     public ForkJoinPool() {
2396         this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
2397                 defaultForkJoinWorkerThreadFactory, null, false);
2398     }
2399 
2400     /**
2401      * Creates a {@code ForkJoinPool} with the indicated parallelism
2402      * level, the {@linkplain
2403      * #defaultForkJoinWorkerThreadFactory default thread factory},
2404      * no UncaughtExceptionHandler, and non-async LIFO processing mode.
2405      *
2406      * @param parallelism the parallelism level
2407      * @throws IllegalArgumentException if parallelism less than or
2408      *         equal to zero, or greater than implementation limit
2409      * @throws SecurityException if a security manager exists and
2410      *         the caller is not permitted to modify threads
2411      *         because it does not hold {@link
2412      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2413      */
2414     public ForkJoinPool(int parallelism) {
2415         this(parallelism, defaultForkJoinWorkerThreadFactory, null, false);
2416     }
2417 
2418     /**
2419      * Creates a {@code ForkJoinPool} with the given parameters.
2420      *
2421      * @param parallelism the parallelism level. For default value,
2422      * use {@link java.lang.Runtime#availableProcessors}.
2423      * @param factory the factory for creating new threads. For default value,
2424      * use {@link #defaultForkJoinWorkerThreadFactory}.
2425      * @param handler the handler for internal worker threads that
2426      * terminate due to unrecoverable errors encountered while executing
2427      * tasks. For default value, use {@code null}.
2428      * @param asyncMode if true,
2429      * establishes local first-in-first-out scheduling mode for forked
2430      * tasks that are never joined. This mode may be more appropriate
2431      * than default locally stack-based mode in applications in which
2432      * worker threads only process event-style asynchronous tasks.
2433      * For default value, use {@code false}.
2434      * @throws IllegalArgumentException if parallelism less than or
2435      *         equal to zero, or greater than implementation limit
2436      * @throws NullPointerException if the factory is null
2437      * @throws SecurityException if a security manager exists and
2438      *         the caller is not permitted to modify threads
2439      *         because it does not hold {@link
2440      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2441      */
2442     public ForkJoinPool(int parallelism,
2443                         ForkJoinWorkerThreadFactory factory,
2444                         UncaughtExceptionHandler handler,
2445                         boolean asyncMode) {
2446         this(checkParallelism(parallelism),
2447                 checkFactory(factory),
2448                 handler,
2449                 (asyncMode ? FIFO_QUEUE : LIFO_QUEUE),
2450                 "ForkJoinPool-" + nextPoolId() + "-worker-");
2451         checkPermission();
2452     }
2453 
2454     private static int checkParallelism(int parallelism) {
2455         if (parallelism <= 0 || parallelism > MAX_CAP)
2456             throw new IllegalArgumentException();
2457         return parallelism;
2458     }
2459 
2460     private static ForkJoinWorkerThreadFactory checkFactory
2461             (ForkJoinWorkerThreadFactory factory) {
2462         if (factory == null)
2463             throw new NullPointerException();
2464         return factory;
2465     }
2466 
2467     /**
2468      * Creates a {@code ForkJoinPool} with the given parameters, without
2469      * any security checks or parameter validation.  Invoked directly by
2470      * makeCommonPool.
2471      */
2472     private ForkJoinPool(int parallelism,
2473                          ForkJoinWorkerThreadFactory factory,
2474                          UncaughtExceptionHandler handler,
2475                          int mode,
2476                          String workerNamePrefix) {
2477         this.workerNamePrefix = workerNamePrefix;
2478         this.factory = factory;
2479         this.ueh = handler;
2480         this.mode = (short)mode;
2481         this.parallelism = (short)parallelism;
2482         long np = (long)(-parallelism); // offset ctl counts
2483         this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
2484     }
2485 
2486     /**
2487      * Returns the common pool instance. This pool is statically
2488      * constructed; its run state is unaffected by attempts to {@link
2489      * #shutdown} or {@link #shutdownNow}. However this pool and any
2490      * ongoing processing are automatically terminated upon program
2491      * {@link System#exit}.  Any program that relies on asynchronous
2492      * task processing to complete before program termination should
2493      * invoke {@code commonPool().}{@link #awaitQuiescence awaitQuiescence},
2494      * before exit.
2495      *
2496      * @return the common pool instance
2497      * @since 1.8
2498      */
2499     public static ForkJoinPool commonPool() {
2500         // assert common != null : "static init error";
2501         return common;
2502     }
2503 
2504     // Execution methods
2505 
2506     /**
2507      * Performs the given task, returning its result upon completion.
2508      * If the computation encounters an unchecked Exception or Error,
2509      * it is rethrown as the outcome of this invocation.  Rethrown
2510      * exceptions behave in the same way as regular exceptions, but,
2511      * when possible, contain stack traces (as displayed for example
2512      * using {@code ex.printStackTrace()}) of both the current thread
2513      * as well as the thread actually encountering the exception;
2514      * minimally only the latter.
2515      *
2516      * @param task the task
2517      * @return the task's result
2518      * @throws NullPointerException if the task is null
2519      * @throws RejectedExecutionException if the task cannot be
2520      *         scheduled for execution
2521      */
2522     public <T> T invoke(ForkJoinTask<T> task) {
2523         if (task == null)
2524             throw new NullPointerException();
2525         externalPush(task);
2526         return task.join();
2527     }
2528 
2529     /**
2530      * Arranges for (asynchronous) execution of the given task.
2531      *
2532      * @param task the task
2533      * @throws NullPointerException if the task is null
2534      * @throws RejectedExecutionException if the task cannot be
2535      *         scheduled for execution
2536      */
2537     public void execute(ForkJoinTask<?> task) {
2538         if (task == null)
2539             throw new NullPointerException();
2540         externalPush(task);
2541     }
2542 
2543     // AbstractExecutorService methods
2544 
2545     /**
2546      * @throws NullPointerException if the task is null
2547      * @throws RejectedExecutionException if the task cannot be
2548      *         scheduled for execution
2549      */
2550     public void execute(Runnable task) {
2551         if (task == null)
2552             throw new NullPointerException();
2553         ForkJoinTask<?> job;
2554         if (task instanceof ForkJoinTask<?>) // avoid re-wrap
2555             job = (ForkJoinTask<?>) task;
2556         else
2557             job = new ForkJoinTask.RunnableExecuteAction(task);
2558         externalPush(job);
2559     }
2560 
2561     /**
2562      * Submits a ForkJoinTask for execution.
2563      *
2564      * @param task the task to submit
2565      * @return the task
2566      * @throws NullPointerException if the task is null
2567      * @throws RejectedExecutionException if the task cannot be
2568      *         scheduled for execution
2569      */
2570     public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
2571         if (task == null)
2572             throw new NullPointerException();
2573         externalPush(task);
2574         return task;
2575     }
2576 
2577     /**
2578      * @throws NullPointerException if the task is null
2579      * @throws RejectedExecutionException if the task cannot be
2580      *         scheduled for execution
2581      */
2582     public <T> ForkJoinTask<T> submit(Callable<T> task) {
2583         ForkJoinTask<T> job = new ForkJoinTask.AdaptedCallable<T>(task);
2584         externalPush(job);
2585         return job;
2586     }
2587 
2588     /**
2589      * @throws NullPointerException if the task is null
2590      * @throws RejectedExecutionException if the task cannot be
2591      *         scheduled for execution
2592      */
2593     public <T> ForkJoinTask<T> submit(Runnable task, T result) {
2594         ForkJoinTask<T> job = new ForkJoinTask.AdaptedRunnable<T>(task, result);
2595         externalPush(job);
2596         return job;
2597     }
2598 
2599     /**
2600      * @throws NullPointerException if the task is null
2601      * @throws RejectedExecutionException if the task cannot be
2602      *         scheduled for execution
2603      */
2604     public ForkJoinTask<?> submit(Runnable task) {
2605         if (task == null)
2606             throw new NullPointerException();
2607         ForkJoinTask<?> job;
2608         if (task instanceof ForkJoinTask<?>) // avoid re-wrap
2609             job = (ForkJoinTask<?>) task;
2610         else
2611             job = new ForkJoinTask.AdaptedRunnableAction(task);
2612         externalPush(job);
2613         return job;
2614     }
2615 
2616     /**
2617      * @throws NullPointerException       {@inheritDoc}
2618      * @throws RejectedExecutionException {@inheritDoc}
2619      */
2620     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
2621         // In previous versions of this class, this method constructed
2622         // a task to run ForkJoinTask.invokeAll, but now external
2623         // invocation of multiple tasks is at least as efficient.
2624         ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
2625 
2626         boolean done = false;
2627         try {
2628             for (Callable<T> t : tasks) {
2629                 ForkJoinTask<T> f = new ForkJoinTask.AdaptedCallable<T>(t);
2630                 futures.add(f);
2631                 externalPush(f);
2632             }
2633             for (int i = 0, size = futures.size(); i < size; i++)
2634                 ((ForkJoinTask<?>)futures.get(i)).quietlyJoin();
2635             done = true;
2636             return futures;
2637         } finally {
2638             if (!done)
2639                 for (int i = 0, size = futures.size(); i < size; i++)
2640                     futures.get(i).cancel(false);
2641         }
2642     }
2643 
2644     /**
2645      * Returns the factory used for constructing new workers.
2646      *
2647      * @return the factory used for constructing new workers
2648      */
2649     public ForkJoinWorkerThreadFactory getFactory() {
2650         return factory;
2651     }
2652 
2653     /**
2654      * Returns the handler for internal worker threads that terminate
2655      * due to unrecoverable errors encountered while executing tasks.
2656      *
2657      * @return the handler, or {@code null} if none
2658      */
2659     public UncaughtExceptionHandler getUncaughtExceptionHandler() {
2660         return ueh;
2661     }
2662 
2663     /**
2664      * Returns the targeted parallelism level of this pool.
2665      *
2666      * @return the targeted parallelism level of this pool
2667      */
2668     public int getParallelism() {
2669         int par;
2670         return ((par = parallelism) > 0) ? par : 1;
2671     }
2672 
2673     /**
2674      * Returns the targeted parallelism level of the common pool.
2675      *
2676      * @return the targeted parallelism level of the common pool
2677      * @since 1.8
2678      */
2679     public static int getCommonPoolParallelism() {
2680         return commonParallelism;
2681     }
2682 
2683     /**
2684      * Returns the number of worker threads that have started but not
2685      * yet terminated.  The result returned by this method may differ
2686      * from {@link #getParallelism} when threads are created to
2687      * maintain parallelism when others are cooperatively blocked.
2688      *
2689      * @return the number of worker threads
2690      */
2691     public int getPoolSize() {
2692         return parallelism + (short)(ctl >>> TC_SHIFT);
2693     }
2694 
2695     /**
2696      * Returns {@code true} if this pool uses local first-in-first-out
2697      * scheduling mode for forked tasks that are never joined.
2698      *
2699      * @return {@code true} if this pool uses async mode
2700      */
2701     public boolean getAsyncMode() {
2702         return mode == FIFO_QUEUE;
2703     }
2704 
2705     /**
2706      * Returns an estimate of the number of worker threads that are
2707      * not blocked waiting to join tasks or for other managed
2708      * synchronization. This method may overestimate the
2709      * number of running threads.
2710      *
2711      * @return the number of worker threads
2712      */
2713     public int getRunningThreadCount() {
2714         int rc = 0;
2715         WorkQueue[] ws; WorkQueue w;
2716         if ((ws = workQueues) != null) {
2717             for (int i = 1; i < ws.length; i += 2) {
2718                 if ((w = ws[i]) != null && w.isApparentlyUnblocked())
2719                     ++rc;
2720             }
2721         }
2722         return rc;
2723     }
2724 
2725     /**
2726      * Returns an estimate of the number of threads that are currently
2727      * stealing or executing tasks. This method may overestimate the
2728      * number of active threads.
2729      *
2730      * @return the number of active threads
2731      */
2732     public int getActiveThreadCount() {
2733         int r = parallelism + (int)(ctl >> AC_SHIFT);
2734         return (r <= 0) ? 0 : r; // suppress momentarily negative values
2735     }
2736 
2737     /**
2738      * Returns {@code true} if all worker threads are currently idle.
2739      * An idle worker is one that cannot obtain a task to execute
2740      * because none are available to steal from other threads, and
2741      * there are no pending submissions to the pool. This method is
2742      * conservative; it might not return {@code true} immediately upon
2743      * idleness of all threads, but will eventually become true if
2744      * threads remain inactive.
2745      *
2746      * @return {@code true} if all threads are currently idle
2747      */
2748     public boolean isQuiescent() {
2749         return parallelism + (int)(ctl >> AC_SHIFT) <= 0;
2750     }
2751 
2752     /**
2753      * Returns an estimate of the total number of tasks stolen from
2754      * one thread's work queue by another. The reported value
2755      * underestimates the actual total number of steals when the pool
2756      * is not quiescent. This value may be useful for monitoring and
2757      * tuning fork/join programs: in general, steal counts should be
2758      * high enough to keep threads busy, but low enough to avoid
2759      * overhead and contention across threads.
2760      *
2761      * @return the number of steals
2762      */
2763     public long getStealCount() {
2764         long count = stealCount;
2765         WorkQueue[] ws; WorkQueue w;
2766         if ((ws = workQueues) != null) {
2767             for (int i = 1; i < ws.length; i += 2) {
2768                 if ((w = ws[i]) != null)
2769                     count += w.nsteals;
2770             }
2771         }
2772         return count;
2773     }
2774 
2775     /**
2776      * Returns an estimate of the total number of tasks currently held
2777      * in queues by worker threads (but not including tasks submitted
2778      * to the pool that have not begun executing). This value is only
2779      * an approximation, obtained by iterating across all threads in
2780      * the pool. This method may be useful for tuning task
2781      * granularities.
2782      *
2783      * @return the number of queued tasks
2784      */
2785     public long getQueuedTaskCount() {
2786         long count = 0;
2787         WorkQueue[] ws; WorkQueue w;
2788         if ((ws = workQueues) != null) {
2789             for (int i = 1; i < ws.length; i += 2) {
2790                 if ((w = ws[i]) != null)
2791                     count += w.queueSize();
2792             }
2793         }
2794         return count;
2795     }
2796 
2797     /**
2798      * Returns an estimate of the number of tasks submitted to this
2799      * pool that have not yet begun executing.  This method may take
2800      * time proportional to the number of submissions.
2801      *
2802      * @return the number of queued submissions
2803      */
2804     public int getQueuedSubmissionCount() {
2805         int count = 0;
2806         WorkQueue[] ws; WorkQueue w;
2807         if ((ws = workQueues) != null) {
2808             for (int i = 0; i < ws.length; i += 2) {
2809                 if ((w = ws[i]) != null)
2810                     count += w.queueSize();
2811             }
2812         }
2813         return count;
2814     }
2815 
2816     /**
2817      * Returns {@code true} if there are any tasks submitted to this
2818      * pool that have not yet begun executing.
2819      *
2820      * @return {@code true} if there are any queued submissions
2821      */
2822     public boolean hasQueuedSubmissions() {
2823         WorkQueue[] ws; WorkQueue w;
2824         if ((ws = workQueues) != null) {
2825             for (int i = 0; i < ws.length; i += 2) {
2826                 if ((w = ws[i]) != null && !w.isEmpty())
2827                     return true;
2828             }
2829         }
2830         return false;
2831     }
2832 
2833     /**
2834      * Removes and returns the next unexecuted submission if one is
2835      * available.  This method may be useful in extensions to this
2836      * class that re-assign work in systems with multiple pools.
2837      *
2838      * @return the next submission, or {@code null} if none
2839      */
2840     protected ForkJoinTask<?> pollSubmission() {
2841         WorkQueue[] ws; WorkQueue w; ForkJoinTask<?> t;
2842         if ((ws = workQueues) != null) {
2843             for (int i = 0; i < ws.length; i += 2) {
2844                 if ((w = ws[i]) != null && (t = w.poll()) != null)
2845                     return t;
2846             }
2847         }
2848         return null;
2849     }
2850 
2851     /**
2852      * Removes all available unexecuted submitted and forked tasks
2853      * from scheduling queues and adds them to the given collection,
2854      * without altering their execution status. These may include
2855      * artificially generated or wrapped tasks. This method is
2856      * designed to be invoked only when the pool is known to be
2857      * quiescent. Invocations at other times may not remove all
2858      * tasks. A failure encountered while attempting to add elements
2859      * to collection {@code c} may result in elements being in
2860      * neither, either or both collections when the associated
2861      * exception is thrown.  The behavior of this operation is
2862      * undefined if the specified collection is modified while the
2863      * operation is in progress.
2864      *
2865      * @param c the collection to transfer elements into
2866      * @return the number of elements transferred
2867      */
2868     protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
2869         int count = 0;
2870         WorkQueue[] ws; WorkQueue w; ForkJoinTask<?> t;
2871         if ((ws = workQueues) != null) {
2872             for (int i = 0; i < ws.length; ++i) {
2873                 if ((w = ws[i]) != null) {
2874                     while ((t = w.poll()) != null) {
2875                         c.add(t);
2876                         ++count;
2877                     }
2878                 }
2879             }
2880         }
2881         return count;
2882     }
2883 
2884     /**
2885      * Returns a string identifying this pool, as well as its state,
2886      * including indications of run state, parallelism level, and
2887      * worker and task counts.
2888      *
2889      * @return a string identifying this pool, as well as its state
2890      */
2891     public String toString() {
2892         // Use a single pass through workQueues to collect counts
2893         long qt = 0L, qs = 0L; int rc = 0;
2894         long st = stealCount;
2895         long c = ctl;
2896         WorkQueue[] ws; WorkQueue w;
2897         if ((ws = workQueues) != null) {
2898             for (int i = 0; i < ws.length; ++i) {
2899                 if ((w = ws[i]) != null) {
2900                     int size = w.queueSize();
2901                     if ((i & 1) == 0)
2902                         qs += size;
2903                     else {
2904                         qt += size;
2905                         st += w.nsteals;
2906                         if (w.isApparentlyUnblocked())
2907                             ++rc;
2908                     }
2909                 }
2910             }
2911         }
2912         int pc = parallelism;
2913         int tc = pc + (short)(c >>> TC_SHIFT);
2914         int ac = pc + (int)(c >> AC_SHIFT);
2915         if (ac < 0) // ignore transient negative
2916             ac = 0;
2917         String level;
2918         if ((c & STOP_BIT) != 0)
2919             level = (tc == 0) ? "Terminated" : "Terminating";
2920         else
2921             level = plock < 0 ? "Shutting down" : "Running";
2922         return super.toString() +
2923                 "[" + level +
2924                 ", parallelism = " + pc +
2925                 ", size = " + tc +
2926                 ", active = " + ac +
2927                 ", running = " + rc +
2928                 ", steals = " + st +
2929                 ", tasks = " + qt +
2930                 ", submissions = " + qs +
2931                 "]";
2932     }
2933 
2934     /**
2935      * Possibly initiates an orderly shutdown in which previously
2936      * submitted tasks are executed, but no new tasks will be
2937      * accepted. Invocation has no effect on execution state if this
2938      * is the {@link #commonPool()}, and no additional effect if
2939      * already shut down.  Tasks that are in the process of being
2940      * submitted concurrently during the course of this method may or
2941      * may not be rejected.
2942      *
2943      * @throws SecurityException if a security manager exists and
2944      *         the caller is not permitted to modify threads
2945      *         because it does not hold {@link
2946      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2947      */
2948     public void shutdown() {
2949         checkPermission();
2950         tryTerminate(false, true);
2951     }
2952 
2953     /**
2954      * Possibly attempts to cancel and/or stop all tasks, and reject
2955      * all subsequently submitted tasks.  Invocation has no effect on
2956      * execution state if this is the {@link #commonPool()}, and no
2957      * additional effect if already shut down. Otherwise, tasks that
2958      * are in the process of being submitted or executed concurrently
2959      * during the course of this method may or may not be
2960      * rejected. This method cancels both existing and unexecuted
2961      * tasks, in order to permit termination in the presence of task
2962      * dependencies. So the method always returns an empty list
2963      * (unlike the case for some other Executors).
2964      *
2965      * @return an empty list
2966      * @throws SecurityException if a security manager exists and
2967      *         the caller is not permitted to modify threads
2968      *         because it does not hold {@link
2969      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2970      */
2971     public List<Runnable> shutdownNow() {
2972         checkPermission();
2973         tryTerminate(true, true);
2974         return Collections.emptyList();
2975     }
2976 
2977     /**
2978      * Returns {@code true} if all tasks have completed following shut down.
2979      *
2980      * @return {@code true} if all tasks have completed following shut down
2981      */
2982     public boolean isTerminated() {
2983         long c = ctl;
2984         return ((c & STOP_BIT) != 0L &&
2985                 (short)(c >>> TC_SHIFT) + parallelism <= 0);
2986     }
2987 
2988     /**
2989      * Returns {@code true} if the process of termination has
2990      * commenced but not yet completed.  This method may be useful for
2991      * debugging. A return of {@code true} reported a sufficient
2992      * period after shutdown may indicate that submitted tasks have
2993      * ignored or suppressed interruption, or are waiting for I/O,
2994      * causing this executor not to properly terminate. (See the
2995      * advisory notes for class {@link ForkJoinTask} stating that
2996      * tasks should not normally entail blocking operations.  But if
2997      * they do, they must abort them on interrupt.)
2998      *
2999      * @return {@code true} if terminating but not yet terminated
3000      */
3001     public boolean isTerminating() {
3002         long c = ctl;
3003         return ((c & STOP_BIT) != 0L &&
3004                 (short)(c >>> TC_SHIFT) + parallelism > 0);
3005     }
3006 
3007     /**
3008      * Returns {@code true} if this pool has been shut down.
3009      *
3010      * @return {@code true} if this pool has been shut down
3011      */
3012     public boolean isShutdown() {
3013         return plock < 0;
3014     }
3015 
3016     /**
3017      * Blocks until all tasks have completed execution after a
3018      * shutdown request, or the timeout occurs, or the current thread
3019      * is interrupted, whichever happens first. Because the {@link
3020      * #commonPool()} never terminates until program shutdown, when
3021      * applied to the common pool, this method is equivalent to {@link
3022      * #awaitQuiescence(long, TimeUnit)} but always returns {@code false}.
3023      *
3024      * @param timeout the maximum time to wait
3025      * @param unit the time unit of the timeout argument
3026      * @return {@code true} if this executor terminated and
3027      *         {@code false} if the timeout elapsed before termination
3028      * @throws InterruptedException if interrupted while waiting
3029      */
3030     public boolean awaitTermination(long timeout, TimeUnit unit)
3031             throws InterruptedException {
3032         if (Thread.interrupted())
3033             throw new InterruptedException();
3034         if (this == common) {
3035             awaitQuiescence(timeout, unit);
3036             return false;
3037         }
3038         long nanos = unit.toNanos(timeout);
3039         if (isTerminated())
3040             return true;
3041         if (nanos <= 0L)
3042             return false;
3043         long deadline = System.nanoTime() + nanos;
3044         synchronized (this) {
3045             for (;;) {
3046                 if (isTerminated())
3047                     return true;
3048                 if (nanos <= 0L)
3049                     return false;
3050                 long millis = TimeUnit.NANOSECONDS.toMillis(nanos);
3051                 wait(millis > 0L ? millis : 1L);
3052                 nanos = deadline - System.nanoTime();
3053             }
3054         }
3055     }
3056 
3057     /**
3058      * If called by a ForkJoinTask operating in this pool, equivalent
3059      * in effect to {@link ForkJoinTask#helpQuiesce}. Otherwise,
3060      * waits and/or attempts to assist performing tasks until this
3061      * pool {@link #isQuiescent} or the indicated timeout elapses.
3062      *
3063      * @param timeout the maximum time to wait
3064      * @param unit the time unit of the timeout argument
3065      * @return {@code true} if quiescent; {@code false} if the
3066      * timeout elapsed.
3067      */
3068     public boolean awaitQuiescence(long timeout, TimeUnit unit) {
3069         long nanos = unit.toNanos(timeout);
3070         ForkJoinWorkerThread wt;
3071         Thread thread = Thread.currentThread();
3072         if ((thread instanceof ForkJoinWorkerThread) &&
3073                 (wt = (ForkJoinWorkerThread)thread).pool == this) {
3074             helpQuiescePool(wt.workQueue);
3075             return true;
3076         }
3077         long startTime = System.nanoTime();
3078         WorkQueue[] ws;
3079         int r = 0, m;
3080         boolean found = true;
3081         while (!isQuiescent() && (ws = workQueues) != null &&
3082                 (m = ws.length - 1) >= 0) {
3083             if (!found) {
3084                 if ((System.nanoTime() - startTime) > nanos)
3085                     return false;
3086                 Thread.yield(); // cannot block
3087             }
3088             found = false;
3089             for (int j = (m + 1) << 2; j >= 0; --j) {
3090                 ForkJoinTask<?> t; WorkQueue q; int b;
3091                 if ((q = ws[r++ & m]) != null && (b = q.base) - q.top < 0) {
3092                     found = true;
3093                     if ((t = q.pollAt(b)) != null)
3094                         t.doExec();
3095                     break;
3096                 }
3097             }
3098         }
3099         return true;
3100     }
3101 
3102     /**
3103      * Waits and/or attempts to assist performing tasks indefinitely
3104      * until the {@link #commonPool()} {@link #isQuiescent}.
3105      */
3106     static void quiesceCommonPool() {
3107         common.awaitQuiescence(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
3108     }
3109 
3110     /**
3111      * Interface for extending managed parallelism for tasks running
3112      * in {@link ForkJoinPool}s.
3113      *
3114      * <p>A {@code ManagedBlocker} provides two methods.  Method
3115      * {@code isReleasable} must return {@code true} if blocking is
3116      * not necessary. Method {@code block} blocks the current thread
3117      * if necessary (perhaps internally invoking {@code isReleasable}
3118      * before actually blocking). These actions are performed by any
3119      * thread invoking {@link ForkJoinPool#managedBlock(ManagedBlocker)}.
3120      * The unusual methods in this API accommodate synchronizers that
3121      * may, but don't usually, block for long periods. Similarly, they
3122      * allow more efficient internal handling of cases in which
3123      * additional workers may be, but usually are not, needed to
3124      * ensure sufficient parallelism.  Toward this end,
3125      * implementations of method {@code isReleasable} must be amenable
3126      * to repeated invocation.
3127      *
3128      * <p>For example, here is a ManagedBlocker based on a
3129      * ReentrantLock:
3130      *  <pre> {@code
3131      * class ManagedLocker implements ManagedBlocker {
3132      *   final ReentrantLock lock;
3133      *   boolean hasLock = false;
3134      *   ManagedLocker(ReentrantLock lock) { this.lock = lock; }
3135      *   public boolean block() {
3136      *     if (!hasLock)
3137      *       lock.lock();
3138      *     return true;
3139      *   }
3140      *   public boolean isReleasable() {
3141      *     return hasLock || (hasLock = lock.tryLock());
3142      *   }
3143      * }}</pre>
3144      *
3145      * <p>Here is a class that possibly blocks waiting for an
3146      * item on a given queue:
3147      *  <pre> {@code
3148      * class QueueTaker<E> implements ManagedBlocker {
3149      *   final BlockingQueue<E> queue;
3150      *   volatile E item = null;
3151      *   QueueTaker(BlockingQueue<E> q) { this.queue = q; }
3152      *   public boolean block() throws InterruptedException {
3153      *     if (item == null)
3154      *       item = queue.take();
3155      *     return true;
3156      *   }
3157      *   public boolean isReleasable() {
3158      *     return item != null || (item = queue.poll()) != null;
3159      *   }
3160      *   public E getItem() { // call after pool.managedBlock completes
3161      *     return item;
3162      *   }
3163      * }}</pre>
3164      */
3165     public static interface ManagedBlocker {
3166         /**
3167          * Possibly blocks the current thread, for example waiting for
3168          * a lock or condition.
3169          *
3170          * @return {@code true} if no additional blocking is necessary
3171          * (i.e., if isReleasable would return true)
3172          * @throws InterruptedException if interrupted while waiting
3173          * (the method is not required to do so, but is allowed to)
3174          */
3175         boolean block() throws InterruptedException;
3176 
3177         /**
3178          * Returns {@code true} if blocking is unnecessary.
3179          * @return {@code true} if blocking is unnecessary
3180          */
3181         boolean isReleasable();
3182     }
3183 
3184     /**
3185      * Blocks in accord with the given blocker.  If the current thread
3186      * is a {@link ForkJoinWorkerThread}, this method possibly
3187      * arranges for a spare thread to be activated if necessary to
3188      * ensure sufficient parallelism while the current thread is blocked.
3189      *
3190      * <p>If the caller is not a {@link ForkJoinTask}, this method is
3191      * behaviorally equivalent to
3192      *  <pre> {@code
3193      * while (!blocker.isReleasable())
3194      *   if (blocker.block())
3195      *     return;
3196      * }</pre>
3197      *
3198      * If the caller is a {@code ForkJoinTask}, then the pool may
3199      * first be expanded to ensure parallelism, and later adjusted.
3200      *
3201      * @param blocker the blocker
3202      * @throws InterruptedException if blocker.block did so
3203      */
3204     public static void managedBlock(ManagedBlocker blocker)
3205             throws InterruptedException {
3206         Thread t = Thread.currentThread();
3207         if (t instanceof ForkJoinWorkerThread) {
3208             ForkJoinPool p = ((ForkJoinWorkerThread)t).pool;
3209             while (!blocker.isReleasable()) {
3210                 if (p.tryCompensate(p.ctl)) {
3211                     try {
3212                         do {} while (!blocker.isReleasable() &&
3213                                 !blocker.block());
3214                     } finally {
3215                         p.incrementActiveCount();
3216                     }
3217                     break;
3218                 }
3219             }
3220         }
3221         else {
3222             do {} while (!blocker.isReleasable() &&
3223                     !blocker.block());
3224         }
3225     }
3226 
3227     // AbstractExecutorService overrides.  These rely on undocumented
3228     // fact that ForkJoinTask.adapt returns ForkJoinTasks that also
3229     // implement RunnableFuture.
3230 
3231     protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
3232         return new ForkJoinTask.AdaptedRunnable<T>(runnable, value);
3233     }
3234 
3235     protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
3236         return new ForkJoinTask.AdaptedCallable<T>(callable);
3237     }
3238 
3239     // Unsafe mechanics
3240     private static final sun.misc.Unsafe U;
3241     private static final long CTL;
3242     private static final long PARKBLOCKER;
3243     private static final int ABASE;
3244     private static final int ASHIFT;
3245     private static final long STEALCOUNT;
3246     private static final long PLOCK;
3247     private static final long INDEXSEED;
3248     private static final long QBASE;
3249     private static final long QLOCK;
3250 
3251     static {
3252         // initialize field offsets for CAS etc
3253         try {
3254             U = getUnsafe();
3255             Class<?> k = ForkJoinPool.class;
3256             CTL = U.objectFieldOffset
3257                     (k.getDeclaredField("ctl"));
3258             STEALCOUNT = U.objectFieldOffset
3259                     (k.getDeclaredField("stealCount"));
3260             PLOCK = U.objectFieldOffset
3261                     (k.getDeclaredField("plock"));
3262             INDEXSEED = U.objectFieldOffset
3263                     (k.getDeclaredField("indexSeed"));
3264             Class<?> tk = Thread.class;
3265             PARKBLOCKER = U.objectFieldOffset
3266                     (tk.getDeclaredField("parkBlocker"));
3267             Class<?> wk = WorkQueue.class;
3268             QBASE = U.objectFieldOffset
3269                     (wk.getDeclaredField("base"));
3270             QLOCK = U.objectFieldOffset
3271                     (wk.getDeclaredField("qlock"));
3272             Class<?> ak = ForkJoinTask[].class;
3273             ABASE = U.arrayBaseOffset(ak);
3274             int scale = U.arrayIndexScale(ak);
3275             if ((scale & (scale - 1)) != 0)
3276                 throw new Error("data type scale not a power of two");
3277             ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
3278         } catch (Exception e) {
3279             throw new Error(e);
3280         }
3281 
3282         submitters = new ThreadLocal<Submitter>();
3283         defaultForkJoinWorkerThreadFactory =
3284                 new DefaultForkJoinWorkerThreadFactory();
3285         modifyThreadPermission = new RuntimePermission("modifyThread");
3286 
3287         common = java.security.AccessController.doPrivileged
3288                 (new java.security.PrivilegedAction<ForkJoinPool>() {
3289                     public ForkJoinPool run() { return makeCommonPool(); }});
3290         int par = common.parallelism; // report 1 even if threads disabled
3291         commonParallelism = par > 0 ? par : 1;
3292     }
3293 
3294     /**
3295      * Creates and returns the common pool, respecting user settings
3296      * specified via system properties.
3297      */
3298     private static ForkJoinPool makeCommonPool() {
3299         int parallelism = -1;
3300         ForkJoinWorkerThreadFactory factory
3301                 = defaultForkJoinWorkerThreadFactory;
3302         UncaughtExceptionHandler handler = null;
3303         try {  // ignore exceptions in accessing/parsing properties
3304             String pp = System.getProperty
3305                     ("java.util.concurrent.ForkJoinPool.common.parallelism");
3306             String fp = System.getProperty
3307                     ("java.util.concurrent.ForkJoinPool.common.threadFactory");
3308             String hp = System.getProperty
3309                     ("java.util.concurrent.ForkJoinPool.common.exceptionHandler");
3310             if (pp != null)
3311                 parallelism = Integer.parseInt(pp);
3312             if (fp != null)
3313                 factory = ((ForkJoinWorkerThreadFactory)ClassLoader.
3314                         getSystemClassLoader().loadClass(fp).newInstance());
3315             if (hp != null)
3316                 handler = ((UncaughtExceptionHandler)ClassLoader.
3317                         getSystemClassLoader().loadClass(hp).newInstance());
3318         } catch (Exception ignore) {
3319         }
3320 
3321         if (parallelism < 0 && // default 1 less than #cores
3322                 (parallelism = Runtime.getRuntime().availableProcessors() - 1) < 0)
3323             parallelism = 0;
3324         if (parallelism > MAX_CAP)
3325             parallelism = MAX_CAP;
3326         return new ForkJoinPool(parallelism, factory, handler, LIFO_QUEUE,
3327                 "ForkJoinPool.commonPool-worker-");
3328     }
3329 
3330     /**
3331      * Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
3332      * Replace with a simple call to Unsafe.getUnsafe when integrating
3333      * into a jdk.
3334      *
3335      * @return a sun.misc.Unsafe
3336      */
3337     private static sun.misc.Unsafe getUnsafe() {
3338         try {
3339             return sun.misc.Unsafe.getUnsafe();
3340         } catch (SecurityException tryReflectionInstead) {}
3341         try {
3342             return java.security.AccessController.doPrivileged
3343                     (new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
3344                         public sun.misc.Unsafe run() throws Exception {
3345                             Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class;
3346                             for (java.lang.reflect.Field f : k.getDeclaredFields()) {
3347                                 f.setAccessible(true);
3348                                 Object x = f.get(null);
3349                                 if (k.isInstance(x))
3350                                     return k.cast(x);
3351                             }
3352                             throw new NoSuchFieldError("the Unsafe");
3353                         }});
3354         } catch (java.security.PrivilegedActionException e) {
3355             throw new RuntimeException("Could not initialize intrinsics",
3356                     e.getCause());
3357         }
3358     }
3359 }