8246585: ForkJoin updates
Reviewed-by: martin

   1 /*
   2  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   3  *
   4  * This code is free software; you can redistribute it and/or modify it
   5  * under the terms of the GNU General Public License version 2 only, as
   6  * published by the Free Software Foundation.  Oracle designates this
   7  * particular file as subject to the "Classpath" exception as provided
   8  * by Oracle in the LICENSE file that accompanied this code.
   9  *
  10  * This code is distributed in the hope that it will be useful, but WITHOUT
  11  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  13  * version 2 for more details (a copy is included in the LICENSE file that
  14  * accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License version
  17  * 2 along with this work; if not, write to the Free Software Foundation,
  18  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  19  *
  20  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  21  * or visit www.oracle.com if you need additional information or have any
  22  * questions.
  23  */
  24 
  25 /*
  26  * This file is available under and governed by the GNU General Public
  27  * License version 2 only, as published by the Free Software Foundation.
  28  * However, the following notice accompanied the original version of this
  29  * file:
  30  *
  31  * Written by Doug Lea with assistance from members of JCP JSR-166
  32  * Expert Group and released to the public domain, as explained at
  33  * http://creativecommons.org/publicdomain/zero/1.0/
  34  */
  35 
  36 package java.util.concurrent;
  37 
  38 import java.lang.Thread.UncaughtExceptionHandler;
  39 import java.lang.invoke.MethodHandles;
  40 import java.lang.invoke.VarHandle;
  41 import java.security.AccessController;
  42 import java.security.AccessControlContext;
  43 import java.security.Permission;
  44 import java.security.Permissions;
  45 import java.security.PrivilegedAction;
  46 import java.security.ProtectionDomain;
  47 import java.util.ArrayList;
  48 import java.util.Collection;
  49 import java.util.Collections;
  50 import java.util.List;
  51 import java.util.function.Predicate;
  52 import java.util.concurrent.atomic.AtomicInteger;
  53 import java.util.concurrent.locks.LockSupport;
  54 import java.util.concurrent.locks.ReentrantLock;
  55 import java.util.concurrent.locks.Condition;
  56 
  57 /**
  58  * An {@link ExecutorService} for running {@link ForkJoinTask}s.
  59  * A {@code ForkJoinPool} provides the entry point for submissions
  60  * from non-{@code ForkJoinTask} clients, as well as management and
  61  * monitoring operations.
  62  *
  63  * <p>A {@code ForkJoinPool} differs from other kinds of {@link
  64  * ExecutorService} mainly by virtue of employing
  65  * <em>work-stealing</em>: all threads in the pool attempt to find and
  66  * execute tasks submitted to the pool and/or created by other active
  67  * tasks (eventually blocking waiting for work if none exist). This
  68  * enables efficient processing when most tasks spawn other subtasks
  69  * (as do most {@code ForkJoinTask}s), as well as when many small
  70  * tasks are submitted to the pool from external clients.  Especially
  71  * when setting <em>asyncMode</em> to true in constructors, {@code
  72  * ForkJoinPool}s may also be appropriate for use with event-style
  73  * tasks that are never joined. All worker threads are initialized
  74  * with {@link Thread#isDaemon} set {@code true}.
  75  *
  76  * <p>A static {@link #commonPool()} is available and appropriate for
  77  * most applications. The common pool is used by any ForkJoinTask that
  78  * is not explicitly submitted to a specified pool. Using the common
  79  * pool normally reduces resource usage (its threads are slowly
  80  * reclaimed during periods of non-use, and reinstated upon subsequent
  81  * use).
  82  *
  83  * <p>For applications that require separate or custom pools, a {@code
  84  * ForkJoinPool} may be constructed with a given target parallelism
  85  * level; by default, equal to the number of available processors.
  86  * The pool attempts to maintain enough active (or available) threads
  87  * by dynamically adding, suspending, or resuming internal worker
  88  * threads, even if some tasks are stalled waiting to join others.
  89  * However, no such adjustments are guaranteed in the face of blocked
  90  * I/O or other unmanaged synchronization. The nested {@link
  91  * ManagedBlocker} interface enables extension of the kinds of
  92  * synchronization accommodated. The default policies may be
  93  * overridden using a constructor with parameters corresponding to
  94  * those documented in class {@link ThreadPoolExecutor}.
  95  *
  96  * <p>In addition to execution and lifecycle control methods, this
  97  * class provides status check methods (for example
  98  * {@link #getStealCount}) that are intended to aid in developing,
  99  * tuning, and monitoring fork/join applications. Also, method
 100  * {@link #toString} returns indications of pool state in a
 101  * convenient form for informal monitoring.
 102  *
 103  * <p>As is the case with other ExecutorServices, there are three
 104  * main task execution methods summarized in the following table.
 105  * These are designed to be used primarily by clients not already
 106  * engaged in fork/join computations in the current pool.  The main
 107  * forms of these methods accept instances of {@code ForkJoinTask},
 108  * but overloaded forms also allow mixed execution of plain {@code
 109  * Runnable}- or {@code Callable}- based activities as well.  However,
 110  * tasks that are already executing in a pool should normally instead
 111  * use the within-computation forms listed in the table unless using
 112  * async event-style tasks that are not usually joined, in which case
 113  * there is little difference among choice of methods.
 114  *
 115  * <table class="plain">
 116  * <caption>Summary of task execution methods</caption>
 117  *  <tr>
 118  *    <td></td>
 119  *    <th scope="col"> Call from non-fork/join clients</th>
 120  *    <th scope="col"> Call from within fork/join computations</th>
 121  *  </tr>
 122  *  <tr>
 123  *    <th scope="row" style="text-align:left"> Arrange async execution</th>
 124  *    <td> {@link #execute(ForkJoinTask)}</td>
 125  *    <td> {@link ForkJoinTask#fork}</td>
 126  *  </tr>
 127  *  <tr>
 128  *    <th scope="row" style="text-align:left"> Await and obtain result</th>
 129  *    <td> {@link #invoke(ForkJoinTask)}</td>
 130  *    <td> {@link ForkJoinTask#invoke}</td>
 131  *  </tr>
 132  *  <tr>
 133  *    <th scope="row" style="text-align:left"> Arrange exec and obtain Future</th>
 134  *    <td> {@link #submit(ForkJoinTask)}</td>
 135  *    <td> {@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td>
 136  *  </tr>
 137  * </table>
 138  *
 139  * <p>The parameters used to construct the common pool may be controlled by
 140  * setting the following {@linkplain System#getProperty system properties}:
 141  * <ul>
 142  * <li>{@systemProperty java.util.concurrent.ForkJoinPool.common.parallelism}
 143  * - the parallelism level, a non-negative integer
 144  * <li>{@systemProperty java.util.concurrent.ForkJoinPool.common.threadFactory}
 145  * - the class name of a {@link ForkJoinWorkerThreadFactory}.
 146  * The {@linkplain ClassLoader#getSystemClassLoader() system class loader}
 147  * is used to load this class.
 148  * <li>{@systemProperty java.util.concurrent.ForkJoinPool.common.exceptionHandler}
 149  * - the class name of a {@link UncaughtExceptionHandler}.
 150  * The {@linkplain ClassLoader#getSystemClassLoader() system class loader}
 151  * is used to load this class.
 152  * <li>{@systemProperty java.util.concurrent.ForkJoinPool.common.maximumSpares}
 153  * - the maximum number of allowed extra threads to maintain target
 154  * parallelism (default 256).
 155  * </ul>
 156  * If no thread factory is supplied via a system property, then the
 157  * common pool uses a factory that uses the system class loader as the
 158  * {@linkplain Thread#getContextClassLoader() thread context class loader}.
 159  * In addition, if a {@link SecurityManager} is present, then
 160  * the common pool uses a factory supplying threads that have no
 161  * {@link Permissions} enabled.
 162  *
 163  * Upon any error in establishing these settings, default parameters
 164  * are used. It is possible to disable or limit the use of threads in
 165  * the common pool by setting the parallelism property to zero, and/or
 166  * using a factory that may return {@code null}. However doing so may
 167  * cause unjoined tasks to never be executed.
 168  *
 169  * <p><b>Implementation notes:</b> This implementation restricts the
 170  * maximum number of running threads to 32767. Attempts to create
 171  * pools with greater than the maximum number result in
 172  * {@code IllegalArgumentException}.
 173  *
 174  * <p>This implementation rejects submitted tasks (that is, by throwing
 175  * {@link RejectedExecutionException}) only when the pool is shut down
 176  * or internal resources have been exhausted.
 177  *
 178  * @since 1.7
 179  * @author Doug Lea
 180  */
 181 public class ForkJoinPool extends AbstractExecutorService {
 182 
 183     /*
 184      * Implementation Overview
 185      *
 186      * This class and its nested classes provide the main
 187      * functionality and control for a set of worker threads:
 188      * Submissions from non-FJ threads enter into submission queues.
 189      * Workers take these tasks and typically split them into subtasks
 190      * that may be stolen by other workers. Work-stealing based on
 191      * randomized scans generally leads to better throughput than
 192      * "work dealing" in which producers assign tasks to idle threads,
 193      * in part because threads that have finished other tasks before
 194      * the signalled thread wakes up (which can be a long time) can
 195      * take the task instead.  Preference rules give first priority to
 196      * processing tasks from their own queues (LIFO or FIFO, depending
 197      * on mode), then to randomized FIFO steals of tasks in other
 198      * queues.  This framework began as vehicle for supporting
 199      * tree-structured parallelism using work-stealing.  Over time,
 200      * its scalability advantages led to extensions and changes to
 201      * better support more diverse usage contexts.  Because most
 202      * internal methods and nested classes are interrelated, their
 203      * main rationale and descriptions are presented here; individual
 204      * methods and nested classes contain only brief comments about
 205      * details.
 206      *
 207      * WorkQueues
 208      * ==========
 209      *
 210      * Most operations occur within work-stealing queues (in nested
 211      * class WorkQueue).  These are special forms of Deques that
 212      * support only three of the four possible end-operations -- push,
 213      * pop, and poll (aka steal), under the further constraints that
 214      * push and pop are called only from the owning thread (or, as
 215      * extended here, under a lock), while poll may be called from
 216      * other threads.  (If you are unfamiliar with them, you probably
 217      * want to read Herlihy and Shavit's book "The Art of
 218      * Multiprocessor programming", chapter 16 describing these in
 219      * more detail before proceeding.)  The main work-stealing queue
 220      * design is roughly similar to those in the papers "Dynamic
 221      * Circular Work-Stealing Deque" by Chase and Lev, SPAA 2005
 222      * (http://research.sun.com/scalable/pubs/index.html) and
 223      * "Idempotent work stealing" by Michael, Saraswat, and Vechev,
 224      * PPoPP 2009 (http://portal.acm.org/citation.cfm?id=1504186).
 225      * The main differences ultimately stem from GC requirements that
 226      * we null out taken slots as soon as we can, to maintain as small
 227      * a footprint as possible even in programs generating huge
 228      * numbers of tasks. To accomplish this, we shift the CAS
 229      * arbitrating pop vs poll (steal) from being on the indices
 230      * ("base" and "top") to the slots themselves.
 231      *
 232      * Adding tasks then takes the form of a classic array push(task)
 233      * in a circular buffer:
 234      *    q.array[q.top++ % length] = task;
 235      *
 236      * The actual code needs to null-check and size-check the array,
 237      * uses masking, not mod, for indexing a power-of-two-sized array,
 238      * enforces memory ordering, supports resizing, and possibly
 239      * signals waiting workers to start scanning -- see below.
 240      *
 241      * The pop operation (always performed by owner) is of the form:
 242      *   if ((task = getAndSet(q.array, (q.top-1) % length, null)) != null)



 243      *        decrement top and return task;
 244      * If this fails, the queue is empty.
 245      *
 246      * The poll operation by another stealer thread is, basically:
 247      *   if (CAS nonnull task at q.array[q.base % length] to null)

 248      *       increment base and return task;
 249      *
 250      * This may fail due to contention, and may be retried.
 251      * Implementations must ensure a consistent snapshot of the base
 252      * index and the task (by looping or trying elsewhere) before
 253      * trying CAS.  There isn't actually a method of this form,
 254      * because failure due to inconsistency or contention is handled
 255      * in different ways in different contexts, normally by first
 256      * trying other queues. (For the most straightforward example, see
 257      * method pollScan.) There are further variants for cases
 258      * requiring inspection of elements before extracting them, so
 259      * must interleave these with variants of this code.  Also, a more
 260      * efficient version (nextLocalTask) is used for polls by owners.
 261      * It avoids some overhead because the queue cannot be growing
 262      * during call.
 263      *
 264      * Memory ordering.  See "Correct and Efficient Work-Stealing for
 265      * Weak Memory Models" by Le, Pop, Cohen, and Nardelli, PPoPP 2013
 266      * (http://www.di.ens.fr/~zappa/readings/ppopp13.pdf) for an
 267      * analysis of memory ordering requirements in work-stealing
 268      * algorithms similar to the one used here.  Inserting and
 269      * extracting tasks in array slots via volatile or atomic accesses
 270      * or explicit fences provides primary synchronization.
 271      *
 272      * Operations on deque elements require reads and writes of both
 273      * indices and slots. When possible, we allow these to occur in
 274      * any order.  Because the base and top indices (along with other
 275      * pool or array fields accessed in many methods) only imprecisely
 276      * guide where to extract from, we let accesses other than the
 277      * element getAndSet/CAS/setVolatile appear in any order, using
 278      * plain mode. But we must still preface some methods (mainly
 279      * those that may be accessed externally) with an acquireFence to
 280      * avoid unbounded staleness. This is equivalent to acting as if
 281      * callers use an acquiring read of the reference to the pool or
 282      * queue when invoking the method, even when they do not. We use
 283      * explicit acquiring reads (getSlot) rather than plain array
 284      * access when acquire mode is required but not otherwise ensured
 285      * by context. To reduce stalls by other stealers, we encourage
 286      * timely writes to the base index by immediately following
 287      * updates with a write of a volatile field that must be updated
 288      * anyway, or an Opaque-mode write if there is no such
 289      * opportunity.
 290      *
 291      * Because indices and slot contents cannot always be consistent,
 292      * the emptiness check base == top is only quiescently accurate
 293      * (and so used where this suffices). Otherwise, it may err on the
 294      * side of possibly making the queue appear nonempty when a push,
 295      * pop, or poll have not fully committed, or making it appear
 296      * empty when an update of top or base has not yet been seen.
 297      * Similarly, the check in push for the queue array being full may
 298      * trigger when not completely full, causing a resize earlier than
 299      * required.
 300      *
 301      * Mainly because of these potential inconsistencies among slots
 302      * vs indices, the poll operation, considered individually, is not
 303      * wait-free. One thief cannot successfully continue until another
 304      * in-progress one (or, if previously empty, a push) visibly
 305      * completes.  This can stall threads when required to consume
 306      * from a given queue (which may spin).  However, in the
 307      * aggregate, we ensure probabilistic non-blockingness at least
 308      * until checking quiescence (which is intrinsically blocking):
 309      * If an attempted steal fails, a scanning thief chooses a
 310      * different victim target to try next. So, in order for one thief
 311      * to progress, it suffices for any in-progress poll or new push
 312      * on any empty queue to complete. The worst cases occur when many
 313      * threads are looking for tasks being produced by a stalled
 314      * producer.
 315      *
 316      * This approach also enables support of a user mode in which
 317      * local task processing is in FIFO, not LIFO order, simply by
 318      * using poll rather than pop.  This can be useful in
 319      * message-passing frameworks in which tasks are never joined,
 320      * although with increased contention among task producers and
 321      * consumers.
 322      *
 323      * WorkQueues are also used in a similar way for tasks submitted
 324      * to the pool. We cannot mix these tasks in the same queues used
 325      * by workers. Instead, we randomly associate submission queues
 326      * with submitting threads, using a form of hashing.  The
 327      * ThreadLocalRandom probe value serves as a hash code for
 328      * choosing existing queues, and may be randomly repositioned upon
 329      * contention with other submitters.  In essence, submitters act
 330      * like workers except that they are restricted to executing local
 331      * tasks that they submitted (or when known, subtasks thereof).
 332      * Insertion of tasks in shared mode requires a lock. We use only
 333      * a simple spinlock (using field "source"), because submitters
 334      * encountering a busy queue move to a different position to use
 335      * or create other queues. They block only when registering new
 336      * queues.

 337      *
 338      * Management
 339      * ==========
 340      *
 341      * The main throughput advantages of work-stealing stem from
 342      * decentralized control -- workers mostly take tasks from
 343      * themselves or each other, at rates that can exceed a billion
 344      * per second.  Most non-atomic control is performed by some form
 345      * of scanning across or within queues.  The pool itself creates,
 346      * activates (enables scanning for and running tasks),
 347      * deactivates, blocks, and terminates threads, all with minimal
 348      * central information.  There are only a few properties that we
 349      * can globally track or maintain, so we pack them into a small
 350      * number of variables, often maintaining atomicity without
 351      * blocking or locking.  Nearly all essentially atomic control
 352      * state is held in a few volatile variables that are by far most
 353      * often read (not written) as status and consistency checks. We
 354      * pack as much information into them as we can.
 355      *
 356      * Field "ctl" contains 64 bits holding information needed to
 357      * atomically decide to add, enqueue (on an event queue), and
 358      * dequeue and release workers.  To enable this packing, we
 359      * restrict maximum parallelism to (1<<15)-1 (which is far in
 360      * excess of normal operating range) to allow ids, counts, and
 361      * their negations (used for thresholding) to fit into 16bit
 362      * subfields.
 363      *
 364      * Field "mode" holds configuration parameters as well as lifetime
 365      * status, atomically and monotonically setting SHUTDOWN, STOP,
 366      * and finally TERMINATED bits. It is updated only via bitwise
 367      * atomics (getAndBitwiseOr).
 368      *
 369      * Array "queues" holds references to WorkQueues.  It is updated
 370      * (only during worker creation and termination) under the
 371      * registrationLock, but is otherwise concurrently readable, and
 372      * accessed directly (although always prefaced by acquireFences or
 373      * other acquiring reads). To simplify index-based operations, the
 374      * array size is always a power of two, and all readers must
 375      * tolerate null slots.  Worker queues are at odd indices. Worker
 376      * ids masked with SMASK match their index. Shared (submission)
 377      * queues are at even indices. Grouping them together in this way
 378      * simplifies and speeds up task scanning.



 379      *
 380      * All worker thread creation is on-demand, triggered by task
 381      * submissions, replacement of terminated workers, and/or
 382      * compensation for blocked workers. However, all other support
 383      * code is set up to work with other policies.  To ensure that we
 384      * do not hold on to worker or task references that would prevent
 385      * GC, all accesses to workQueues are via indices into the
 386      * queues array (which is one source of some of the messy code
 387      * constructions here). In essence, the queues array serves as
 388      * a weak reference mechanism. Thus for example the stack top
 389      * subfield of ctl stores indices, not references.
 390      *
 391      * Queuing Idle Workers. Unlike HPC work-stealing frameworks, we
 392      * cannot let workers spin indefinitely scanning for tasks when
 393      * none can be found immediately, and we cannot start/resume
 394      * workers unless there appear to be tasks available.  On the
 395      * other hand, we must quickly prod them into action when new
 396      * tasks are submitted or generated. These latencies are mainly a
 397      * function of JVM park/unpark (and underlying OS) performance,
 398      * which can be slow and variable.  In many usages, ramp-up time
 399      * is the main limiting factor in overall performance, which is
 400      * compounded at program start-up by JIT compilation and
 401      * allocation. On the other hand, throughput degrades when too
 402      * many threads poll for too few tasks.
 403      *
 404      * The "ctl" field atomically maintains total and "released"
 405      * worker counts, plus the head of the available worker queue
 406      * (actually stack, represented by the lower 32bit subfield of
 407      * ctl).  Released workers are those known to be scanning for
 408      * and/or running tasks. Unreleased ("available") workers are
 409      * recorded in the ctl stack. These workers are made available for
 410      * signalling by enqueuing in ctl (see method awaitWork).  The
 411      * "queue" is a form of Treiber stack. This is ideal for
 412      * activating threads in most-recently used order, and improves
 413      * performance and locality, outweighing the disadvantages of
 414      * being prone to contention and inability to release a worker
 415      * unless it is topmost on stack. The top stack state holds the








 416      * value of the "phase" field of the worker: its index and status,
 417      * plus a version counter that, in addition to the count subfields
 418      * (also serving as version stamps) provide protection against
 419      * Treiber stack ABA effects.
 420      *
 421      * Creating workers. To create a worker, we pre-increment counts
 422      * (serving as a reservation), and attempt to construct a
 423      * ForkJoinWorkerThread via its factory. On starting, the new
 424      * thread first invokes registerWorker, where it constructs a
 425      * WorkQueue and is assigned an index in the queues array
 426      * (expanding the array if necessary).  Upon any exception across
 427      * these steps, or null return from factory, deregisterWorker
 428      * adjusts counts and records accordingly.  If a null return, the
 429      * pool continues running with fewer than the target number
 430      * workers. If exceptional, the exception is propagated, generally
 431      * to some external caller.











 432      *
 433      * WorkQueue field "phase" is used by both workers and the pool to
 434      * manage and track whether a worker is UNSIGNALLED (possibly
 435      * blocked waiting for a signal).  When a worker is enqueued its
 436      * phase field is set negative. Note that phase field updates lag
 437      * queue CAS releases; seeing a negative phase does not guarantee
 438      * that the worker is available. When queued, the lower 16 bits of
 439      * its phase must hold its pool index. So we place the index there
 440      * upon initialization and never modify these bits.

 441      *
 442      * The ctl field also serves as the basis for memory
 443      * synchronization surrounding activation. This uses a more
 444      * efficient version of a Dekker-like rule that task producers and
 445      * consumers sync with each other by both writing/CASing ctl (even
 446      * if to its current value).  However, rather than CASing ctl to
 447      * its current value in the common case where no action is
 448      * required, we reduce write contention by ensuring that
 449      * signalWork invocations are prefaced with a full-volatile memory
 450      * access (which is usually needed anyway).
 451      *
 452      * Signalling. Signals (in signalWork) cause new or reactivated
 453      * workers to scan for tasks.  Method signalWork and its callers
 454      * try to approximate the unattainable goal of having the right
 455      * number of workers activated for the tasks at hand, but must err
 456      * on the side of too many workers vs too few to avoid stalls.  If
 457      * computations are purely tree structured, it suffices for every
 458      * worker to activate another when it pushes a task into an empty
 459      * queue, resulting in O(log(#threads)) steps to full activation.
 460      * If instead, tasks come in serially from only a single producer,
 461      * each worker taking its first (since the last quiescence) task
 462      * from a queue should signal another if there are more tasks in
 463      * that queue. This is equivalent to, but generally faster than,
 464      * arranging the stealer take two tasks, re-pushing one on its own
 465      * queue, and signalling (because its queue is empty), also
 466      * resulting in logarithmic full activation time. Because we don't
 467      * know about usage patterns (or most commonly, mixtures), we use
 468      * both approaches.  We approximate the second rule by arranging
 469      * that workers in scan() do not repeat signals when repeatedly
 470      * taking tasks from any given queue, by remembering the previous
 471      * one. There are narrow windows in which both rules may apply,
 472      * leading to duplicate or unnecessary signals. Despite such
 473      * limitations, these rules usually avoid slowdowns that otherwise
 474      * occur when too many workers contend to take too few tasks, or
 475      * when producers waste most of their time resignalling.  However,
 476      * contention and overhead effects may still occur during ramp-up,
 477      * ramp-down, and small computations involving only a few workers.
 478      *
 479      * Scanning. Method scan performs top-level scanning for (and
 480      * execution of) tasks.  Scans by different workers and/or at
 481      * different times are unlikely to poll queues in the same
 482      * order. Each scan traverses and tries to poll from each queue in
 483      * a pseudorandom permutation order by starting at a random index,
 484      * and using a constant cyclically exhaustive stride; restarting
 485      * upon contention.  (Non-top-level scans; for example in
 486      * helpJoin, use simpler linear probes because they do not
 487      * systematically contend with top-level scans.)  The pseudorandom
 488      * generator need not have high-quality statistical properties in
 489      * the long term. We use Marsaglia XorShifts, seeded with the Weyl
 490      * sequence from ThreadLocalRandom probes, which are cheap and
 491      * suffice. Scans do not otherwise explicitly take into account
 492      * core affinities, loads, cache localities, etc, However, they do


 493      * exploit temporal locality (which usually approximates these) by
 494      * preferring to re-poll from the same queue after a successful
 495      * poll before trying others (see method topLevelExec).  This
 496      * reduces fairness, which is partially counteracted by using a
 497      * one-shot form of poll (tryPoll) that may lose to other workers.
 498      *
 499      * Deactivation. Method scan returns a sentinel when no tasks are
 500      * found, leading to deactivation (see awaitWork). The count
 501      * fields in ctl allow accurate discovery of quiescent states
 502      * (i.e., when all workers are idle) after deactivation. However,
 503      * this may also race with new (external) submissions, so a
 504      * recheck is also needed to determine quiescence. Upon apparently
 505      * triggering quiescence, awaitWork re-scans and self-signals if
 506      * it may have missed a signal. In other cases, a missed signal
 507      * may transiently lower parallelism because deactivation does not
 508      * necessarily mean that there is no more work, only that that
 509      * there were no tasks not taken by other workers.  But more
 510      * signals are generated (see above) to eventually reactivate if
 511      * needed.
 512      *
 513      * Trimming workers. To release resources after periods of lack of
 514      * use, a worker starting to wait when the pool is quiescent will
 515      * time out and terminate if the pool has remained quiescent for
 516      * period given by field keepAlive.
 517      *
 518      * Shutdown and Termination. A call to shutdownNow invokes
 519      * tryTerminate to atomically set a mode bit. The calling thread,
 520      * as well as every other worker thereafter terminating, helps
 521      * terminate others by cancelling their unprocessed tasks, and
 522      * waking them up. Calls to non-abrupt shutdown() preface this by
 523      * checking isQuiescent before triggering the "STOP" phase of



 524      * termination.
 525      *
 526      * Joining Tasks
 527      * =============
 528      *
 529      * Normally, the first option when joining a task that is not done
 530      * is to try to unfork it from local queue and run it.  Otherwise,
 531      * any of several actions may be taken when one worker is waiting
 532      * to join a task stolen (or always held) by another.  Because we
 533      * are multiplexing many tasks on to a pool of workers, we can't
 534      * always just let them block (as in Thread.join).  We also cannot
 535      * just reassign the joiner's run-time stack with another and
 536      * replace it later, which would be a form of "continuation", that
 537      * even if possible is not necessarily a good idea since we may
 538      * need both an unblocked task and its continuation to progress.
 539      * Instead we combine two tactics:
 540      *
 541      *   Helping: Arranging for the joiner to execute some task that it
 542      *      could be running if the steal had not occurred.
 543      *
 544      *   Compensating: Unless there are already enough live threads,
 545      *      method tryCompensate() may create or re-activate a spare
 546      *      thread to compensate for blocked joiners until they unblock.
 547      *
 548      * A third form (implemented via tryRemove) amounts to helping a
 549      * hypothetical compensator: If we can readily tell that a
 550      * possible action of a compensator is to steal and execute the
 551      * task being joined, the joining thread can do so directly,
 552      * without the need for a compensation thread; although with a
 553      * (rare) possibility of reduced parallelism because of a
 554      * transient gap in the queue array.
 555      *
 556      * Other intermediate forms available for specific task types (for
 557      * example helpAsyncBlocker) often avoid or postpone the need for
 558      * blocking or compensation.
 559      *
 560      * The ManagedBlocker extension API can't use helping so relies
 561      * only on compensation in method awaitBlocker.
 562      *
 563      * The algorithm in helpJoin entails a form of "linear helping".
 564      * Each worker records (in field "source") the id of the queue
 565      * from which it last stole a task.  The scan in method helpJoin
 566      * uses these markers to try to find a worker to help (i.e., steal
 567      * back a task from and execute it) that could hasten completion
 568      * of the actively joined task.  Thus, the joiner executes a task
 569      * that would be on its own local deque if the to-be-joined task
 570      * had not been stolen. This is a conservative variant of the
 571      * approach described in Wagner & Calder "Leapfrogging: a portable
 572      * technique for implementing efficient futures" SIGPLAN Notices,
 573      * 1993 (http://portal.acm.org/citation.cfm?id=155354). It differs
 574      * mainly in that we only record queue ids, not full dependency
 575      * links.  This requires a linear scan of the queues array to
 576      * locate stealers, but isolates cost to when it is needed, rather
 577      * than adding to per-task overhead. Also, searches are limited to
 578      * direct and at most two levels of indirect stealers, after which
 579      * there are rapidly diminishing returns on increased overhead.
 580      * Searches can fail to locate stealers when stalls delay
 581      * recording sources.  Further, even when accurately identified,
 582      * stealers might not ever produce a task that the joiner can in
 583      * turn help with. So, compensation is tried upon failure to find
 584      * tasks to run.
 585      *
 586      * Joining CountedCompleters (see helpComplete) differs from (and
 587      * is generally more efficient than) other cases because task
 588      * eligibility is determined by checking completion chains rather
 589      * than tracking stealers.
 590      *
 591      * Joining under timeouts (ForkJoinTask timed get) uses a
 592      * constrained mixture of helping and compensating in part because
 593      * pools (actually, only the common pool) may not have any
 594      * available threads: If the pool is saturated (all available
 595      * workers are busy), the caller tries to remove and otherwise
 596      * help; else it blocks under compensation so that it may time out
 597      * independently of any tasks.
 598      *
 599      * Compensation does not by default aim to keep exactly the target
 600      * parallelism number of unblocked threads running at any given
 601      * time. Some previous versions of this class employed immediate
 602      * compensations for any blocked join. However, in practice, the
 603      * vast majority of blockages are transient byproducts of GC and
 604      * other JVM or OS activities that are made worse by replacement
 605      * when they cause longer-term oversubscription.  Rather than
 606      * impose arbitrary policies, we allow users to override the
 607      * default of only adding threads upon apparent starvation.  The
 608      * compensation mechanism may also be bounded.  Bounds for the
 609      * commonPool (see COMMON_MAX_SPARES) better enable JVMs to cope
 610      * with programming errors and abuse before running out of
 611      * resources to do so.
 612      *
 613      * Common Pool
 614      * ===========
 615      *
 616      * The static common pool always exists after static
 617      * initialization.  Since it (or any other created pool) need
 618      * never be used, we minimize initial construction overhead and
 619      * footprint to the setup of about a dozen fields.
 620      *
 621      * When external threads submit to the common pool, they can
 622      * perform subtask processing (see helpComplete and related
 623      * methods) upon joins.  This caller-helps policy makes it
 624      * sensible to set common pool parallelism level to one (or more)
 625      * less than the total number of available cores, or even zero for
 626      * pure caller-runs.  We do not need to record whether external
 627      * submissions are to the common pool -- if not, external help
 628      * methods return quickly. These submitters would otherwise be
 629      * blocked waiting for completion, so the extra effort (with
 630      * liberally sprinkled task status checks) in inapplicable cases
 631      * amounts to an odd form of limited spin-wait before blocking in
 632      * ForkJoinTask.join.
 633      *
 634      * As a more appropriate default in managed environments, unless
 635      * overridden by system properties, we use workers of subclass
 636      * InnocuousForkJoinWorkerThread when there is a SecurityManager
 637      * present. These workers have no permissions set, do not belong
 638      * to any user-defined ThreadGroup, and erase all ThreadLocals
 639      * after executing any top-level task.  The associated mechanics
 640      * may be JVM-dependent and must access particular Thread class
 641      * fields to achieve this effect.
 642      *
 643      * Interrupt handling
 644      * ==================
 645      *
 646      * The framework is designed to manage task cancellation
 647      * (ForkJoinTask.cancel) independently from the interrupt status
 648      * of threads running tasks. (See the public ForkJoinTask
 649      * documentation for rationale.)  Interrupts are issued only in
 650      * tryTerminate, when workers should be terminating and tasks
 651      * should be cancelled anyway. Interrupts are cleared only when
 652      * necessary to ensure that calls to LockSupport.park do not loop
 653      * indefinitely (park returns immediately if the current thread is
 654      * interrupted). If so, interruption is reinstated after blocking
 655      * if status could be visible during the scope of any task.  For
 656      * cases in which task bodies are specified or desired to
 657      * interrupt upon cancellation, ForkJoinTask.cancel can be
 658      * overridden to do so (as is done for invoke{Any,All}).
 659      *
 660      * Memory placement
 661      * ================
 662      *
 663      * Performance can be very sensitive to placement of instances of
 664      * ForkJoinPool and WorkQueues and their queue arrays. To reduce
 665      * false-sharing impact, the @Contended annotation isolates the
 666      * ForkJoinPool.ctl field as well as the most heavily written
 667      * WorkQueue fields. These mainly reduce cache traffic by scanners.
 668      * WorkQueue arrays are presized large enough to avoid resizing
 669      * (which transiently reduces throughput) in most tree-like
 670      * computations, although not in some streaming usages. Initial
 671      * sizes are not large enough to avoid secondary contention
 672      * effects (especially for GC cardmarks) when queues are placed
 673      * near each other in memory. This is common, but has different
 674      * impact in different collectors and remains incompletely
 675      * addressed.
 676      *
 677      * Style notes
 678      * ===========
 679      *
 680      * Memory ordering relies mainly on atomic operations (CAS,
 681      * getAndSet, getAndAdd) along with explicit fences.  This can be
 682      * awkward and ugly, but also reflects the need to control
 683      * outcomes across the unusual cases that arise in very racy code
 684      * with very few invariants. All fields are read into locals
 685      * before use, and null-checked if they are references, even if
 686      * they can never be null under current usages.  Array accesses
 687      * using masked indices include checks (that are always true) that
 688      * the array length is non-zero to avoid compilers inserting more
 689      * expensive traps.  This is usually done in a "C"-like style of
 690      * listing declarations at the heads of methods or blocks, and
 691      * using inline assignments on first encounter.  Nearly all
 692      * explicit checks lead to bypass/return, not exception throws,
 693      * because they may legitimately arise during shutdown.
 694      *
 695      * There is a lot of representation-level coupling among classes
 696      * ForkJoinPool, ForkJoinWorkerThread, and ForkJoinTask.  The
 697      * fields of WorkQueue maintain data structures managed by
 698      * ForkJoinPool, so are directly accessed.  There is little point
 699      * trying to reduce this, since any associated future changes in
 700      * representations will need to be accompanied by algorithmic
 701      * changes anyway. Several methods intrinsically sprawl because
 702      * they must accumulate sets of consistent reads of fields held in
 703      * local variables. Some others are artificially broken up to
 704      * reduce producer/consumer imbalances due to dynamic compilation.
 705      * There are also other coding oddities (including several
 706      * unnecessary-looking hoisted null checks) that help some methods
 707      * perform reasonably even when interpreted (not compiled).
 708      *
 709      * The order of declarations in this file is (with a few exceptions):
 710      * (1) Static utility functions
 711      * (2) Nested (static) classes
 712      * (3) Static fields
 713      * (4) Fields, along with constants used when unpacking some of them
 714      * (5) Internal control methods
 715      * (6) Callbacks and other support for ForkJoinTask methods
 716      * (7) Exported methods
 717      * (8) Static block initializing statics in minimally dependent order
 718      *
 719      * Revision notes
 720      * ==============
 721      *
 722      * The main sources of differences of January 2020 ForkJoin
 723      * classes from previous version are:
 724      *
 725      * * ForkJoinTask now uses field "aux" to support blocking joins
 726      *   and/or record exceptions, replacing reliance on builtin
 727      *   monitors and side tables.
 728      * * Scans probe slots (vs compare indices), along with related
 729      *   changes that reduce performance differences across most
 730      *   garbage collectors, and reduce contention.
 731      * * Refactoring for better integration of special task types and
 732      *   other capabilities that had been incrementally tacked on. Plus
 733      *   many minor reworkings to improve consistency.
 734      */
 735 
 736     // Static utilities
 737 
 738     /**
 739      * If there is a security manager, makes sure caller has
 740      * permission to modify threads.
 741      */
 742     private static void checkPermission() {
 743         SecurityManager security = System.getSecurityManager();
 744         if (security != null)
 745             security.checkPermission(modifyThreadPermission);
 746     }
 747 
 748     static AccessControlContext contextWithPermissions(Permission ... perms) {
 749         Permissions permissions = new Permissions();
 750         for (Permission perm : perms)
 751             permissions.add(perm);
 752         return new AccessControlContext(
 753             new ProtectionDomain[] { new ProtectionDomain(null, permissions) });
 754     }
 755 
 756     // Nested classes
 757 
 758     /**
 759      * Factory for creating new {@link ForkJoinWorkerThread}s.
 760      * A {@code ForkJoinWorkerThreadFactory} must be defined and used
 761      * for {@code ForkJoinWorkerThread} subclasses that extend base
 762      * functionality or initialize threads with different contexts.
 763      */
 764     public static interface ForkJoinWorkerThreadFactory {
 765         /**
 766          * Returns a new worker thread operating in the given pool.
 767          * Returning null or throwing an exception may result in tasks
 768          * never being executed.  If this method throws an exception,
 769          * it is relayed to the caller of the method (for example
 770          * {@code execute}) causing attempted thread creation. If this
 771          * method returns null or throws an exception, it is not
 772          * retried until the next attempted creation (for example
 773          * another call to {@code execute}).
 774          *
 775          * @param pool the pool this thread works in
 776          * @return the new worker thread, or {@code null} if the request
 777          *         to create a thread is rejected
 778          * @throws NullPointerException if the pool is null
 779          */
 780         public ForkJoinWorkerThread newThread(ForkJoinPool pool);
 781     }
 782 








 783     /**
 784      * Default ForkJoinWorkerThreadFactory implementation; creates a
 785      * new ForkJoinWorkerThread using the system class loader as the
 786      * thread context class loader.
 787      */
 788     static final class DefaultForkJoinWorkerThreadFactory
 789         implements ForkJoinWorkerThreadFactory {
 790         // ACC for access to the factory
 791         private static final AccessControlContext ACC = contextWithPermissions(
 792             new RuntimePermission("getClassLoader"),
 793             new RuntimePermission("setContextClassLoader"));
 794         public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
 795             return AccessController.doPrivileged(
 796                 new PrivilegedAction<>() {
 797                     public ForkJoinWorkerThread run() {
 798                         return new ForkJoinWorkerThread(null, pool, true, false);
 799                     }},
 800                 ACC);
 801         }
 802     }
 803 
 804     /**
 805      * Factory for CommonPool unless overridden by System property.
 806      * Creates InnocuousForkJoinWorkerThreads if a security manager is
 807      * present at time of invocation.  Support requires that we break
 808      * quite a lot of encapsulation (some via helper methods in
 809      * ThreadLocalRandom) to access and set Thread fields.
 810      */
 811     static final class DefaultCommonPoolForkJoinWorkerThreadFactory
 812         implements ForkJoinWorkerThreadFactory {
 813         private static final AccessControlContext ACC = contextWithPermissions(
 814             modifyThreadPermission,
 815             new RuntimePermission("enableContextClassLoaderOverride"),
 816             new RuntimePermission("modifyThreadGroup"),
 817             new RuntimePermission("getClassLoader"),
 818             new RuntimePermission("setContextClassLoader"));
 819 
 820         public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
 821             return AccessController.doPrivileged(
 822                  new PrivilegedAction<>() {
 823                      public ForkJoinWorkerThread run() {
 824                          return System.getSecurityManager() == null ?
 825                              new ForkJoinWorkerThread(null, pool, true, true):
 826                              new ForkJoinWorkerThread.
 827                              InnocuousForkJoinWorkerThread(pool); }},
 828                  ACC);
 829         }
 830     }
 831 
 832     // Constants shared across ForkJoinPool and WorkQueue
 833 
 834     // Bounds
 835     static final int SWIDTH       = 16;            // width of short
 836     static final int SMASK        = 0xffff;        // short bits == max index
 837     static final int MAX_CAP      = 0x7fff;        // max #workers - 1

 838 
 839     // Masks and units for WorkQueue.phase and ctl sp subfield
 840     static final int UNSIGNALLED  = 1 << 31;       // must be negative
 841     static final int SS_SEQ       = 1 << 16;       // version count

 842 
 843     // Mode bits and sentinels, some also used in WorkQueue fields

 844     static final int FIFO         = 1 << 16;       // fifo queue or access mode
 845     static final int SRC          = 1 << 17;       // set for valid queue ids
 846     static final int INNOCUOUS    = 1 << 18;       // set for Innocuous workers
 847     static final int QUIET        = 1 << 19;       // quiescing phase or source
 848     static final int SHUTDOWN     = 1 << 24;
 849     static final int TERMINATED   = 1 << 25;
 850     static final int STOP         = 1 << 31;       // must be negative
 851     static final int UNCOMPENSATE = 1 << 16;       // tryCompensate return
















 852 
 853     /**
 854      * Initial capacity of work-stealing queue array.  Must be a power
 855      * of two, at least 2. See above.

 856      */
 857     static final int INITIAL_QUEUE_CAPACITY = 1 << 8;
 858 
 859     /**
 860      * Queues supporting work-stealing as well as external task
 861      * submission. See above for descriptions and algorithms.
 862      */

 863     static final class WorkQueue {
 864         volatile int phase;        // versioned, negative if inactive




 865         int stackPred;             // pool stack (ctl) predecessor link
 866         int config;                // index, mode, ORed with SRC after init
 867         int base;                  // index of next slot for poll
 868         ForkJoinTask<?>[] array;   // the queued tasks; power of 2 size

 869         final ForkJoinWorkerThread owner; // owning thread or null if shared
 870 
 871         // segregate fields frequently updated but not read by scans or steals
 872         @jdk.internal.vm.annotation.Contended("w")
 873         int top;                   // index of next slot for push
 874         @jdk.internal.vm.annotation.Contended("w")
 875         volatile int source;       // source queue id, lock, or sentinel
 876         @jdk.internal.vm.annotation.Contended("w")
 877         int nsteals;               // number of steals from other queues
 878 
 879         // Support for atomic operations
 880         private static final VarHandle QA; // for array slots
 881         private static final VarHandle SOURCE;
 882         private static final VarHandle BASE;
 883         static final ForkJoinTask<?> getSlot(ForkJoinTask<?>[] a, int i) {
 884             return (ForkJoinTask<?>)QA.getAcquire(a, i);
 885         }
 886         static final ForkJoinTask<?> getAndClearSlot(ForkJoinTask<?>[] a,
 887                                                      int i) {
 888             return (ForkJoinTask<?>)QA.getAndSet(a, i, null);
 889         }
 890         static final void setSlotVolatile(ForkJoinTask<?>[] a, int i,
 891                                           ForkJoinTask<?> v) {
 892             QA.setVolatile(a, i, v);
 893         }
 894         static final boolean casSlotToNull(ForkJoinTask<?>[] a, int i,
 895                                           ForkJoinTask<?> c) {
 896             return QA.weakCompareAndSet(a, i, c, null);
 897         }
 898         final boolean tryLock() {
 899             return SOURCE.compareAndSet(this, 0, 1);
 900         }
 901         final void setBaseOpaque(int b) {
 902             BASE.setOpaque(this, b);
 903         }
 904 
 905         /**
 906          * Constructor used by ForkJoinWorkerThreads. Most fields
 907          * are initialized upon thread start, in pool.registerWorker.
 908          */
 909         WorkQueue(ForkJoinWorkerThread owner, boolean isInnocuous) {
 910             this.config = (isInnocuous) ? INNOCUOUS : 0;
 911             this.owner = owner;
 912         }
 913 
 914         /**
 915          * Constructor used for external queues.
 916          */
 917         WorkQueue(int config) {
 918             array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY];
 919             this.config = config;
 920             owner = null;
 921             phase = -1;
 922         }
 923 
 924         /**
 925          * Returns an exportable index (used by ForkJoinWorkerThread).
 926          */
 927         final int getPoolIndex() {
 928             return (config & 0xffff) >>> 1; // ignore odd/even tag bit
 929         }
 930 
 931         /**
 932          * Returns the approximate number of tasks in the queue.
 933          */
 934         final int queueSize() {
 935             VarHandle.acquireFence(); // ensure fresh reads by external callers
 936             int n = top - base;
 937             return (n < 0) ? 0 : n;   // ignore transient negative
 938         }
 939 
 940         /**
 941          * Provides a more conservative estimate of whether this queue
 942          * has any tasks than does queueSize.

 943          */
 944         final boolean isEmpty() {
 945             return !((source != 0 && owner == null) || top - base > 0);





 946         }
 947 
 948         /**
 949          * Pushes a task. Call only by owner in unshared queues.
 950          *
 951          * @param task the task. Caller must ensure non-null.
 952          * @param pool (no-op if null)
 953          * @throws RejectedExecutionException if array cannot be resized
 954          */
 955         final void push(ForkJoinTask<?> task, ForkJoinPool pool) {
 956             ForkJoinTask<?>[] a = array;
 957             int s = top++, d = s - base, cap, m; // skip insert if disabled
 958             if (a != null && pool != null && (cap = a.length) > 0) {
 959                 setSlotVolatile(a, (m = cap - 1) & s, task);


 960                 if (d == m)
 961                     growArray();
 962                 if (d == m || a[m & (s - 1)] == null)
 963                     pool.signalWork(); // signal if was empty or resized


 964             }
 965         }
 966 
 967         /**
 968          * Pushes task to a shared queue with lock already held, and unlocks.
 969          *
 970          * @return true if caller should signal work
 971          */
 972         final boolean lockedPush(ForkJoinTask<?> task) {
 973             ForkJoinTask<?>[] a = array;
 974             int s = top++, d = s - base, cap, m;
 975             if (a != null && (cap = a.length) > 0) {
 976                 a[(m = cap - 1) & s] = task;


 977                 if (d == m)
 978                     growArray();
 979                 source = 0; // unlock
 980                 if (d == m || a[m & (s - 1)] == null)
 981                     return true;


 982             }
 983             return false;
 984         }
 985 
 986         /**
 987          * Doubles the capacity of array. Called by owner or with lock
 988          * held after pre-incrementing top, which is reverted on
 989          * allocation failure.
 990          */
 991         final void growArray() {
 992             ForkJoinTask<?>[] oldArray = array, newArray;
 993             int s = top - 1, oldCap, newCap;
 994             if (oldArray != null && (oldCap = oldArray.length) > 0 &&
 995                 (newCap = oldCap << 1) > 0) { // skip if disabled


 996                 try {
 997                     newArray = new ForkJoinTask<?>[newCap];
 998                 } catch (Throwable ex) {
 999                     top = s;
1000                     if (owner == null)
1001                         source = 0; // unlock
1002                     throw new RejectedExecutionException(
1003                         "Queue capacity exceeded");








1004                 }
1005                 int newMask = newCap - 1, oldMask = oldCap - 1;
1006                 for (int k = oldCap; k > 0; --k, --s) {
1007                     ForkJoinTask<?> x;        // poll old, push to new
1008                     if ((x = getAndClearSlot(oldArray, s & oldMask)) == null)
1009                         break;                // others already taken
1010                     newArray[s & newMask] = x;
1011                 }
1012                 VarHandle.releaseFence();     // fill before publish
1013                 array = newArray;

1014             }


1015         }
1016 
1017         // Variants of pop



















1018 
1019         /**
1020          * Pops and returns task, or null if empty. Called only by owner.
1021          */
1022         private ForkJoinTask<?> pop() {
1023             ForkJoinTask<?> t = null;
1024             int s = top, cap; ForkJoinTask<?>[] a;
1025             if ((a = array) != null && (cap = a.length) > 0 && base != s-- &&
1026                 (t = getAndClearSlot(a, (cap - 1) & s)) != null)
1027                 top = s;











1028             return t;
1029         }
1030 
1031         /**
1032          * Pops the given task for owner only if it is at the current top.
1033          */
1034         final boolean tryUnpush(ForkJoinTask<?> task) {
1035             int s = top, cap; ForkJoinTask<?>[] a;
1036             if ((a = array) != null && (cap = a.length) > 0 && base != s-- &&
1037                 casSlotToNull(a, (cap - 1) & s, task)) {
1038                 top = s;
1039                 return true;
1040             }
1041             return false;
1042         }
1043 
1044         /**
1045          * Locking version of tryUnpush.
1046          */
1047         final boolean externalTryUnpush(ForkJoinTask<?> task) {
1048             boolean taken = false;
1049             int s = top, cap, k; ForkJoinTask<?>[] a;
1050             if ((a = array) != null && (cap = a.length) > 0 &&
1051                 a[k = (cap - 1) & (s - 1)] == task && tryLock()) {
1052                 if (top == s && array == a &&
1053                     (taken = casSlotToNull(a, k, task)))
1054                     top = s - 1;
1055                 source = 0; // release lock
1056             }
1057             return taken;
1058         }
1059 
1060         /**
1061          * Deep form of tryUnpush: Traverses from top and removes task if
1062          * present, shifting others to fill gap.
1063          */
1064         final boolean tryRemove(ForkJoinTask<?> task, boolean owned) {
1065             boolean taken = false;
1066             int p = top, cap; ForkJoinTask<?>[] a; ForkJoinTask<?> t;
1067             if ((a = array) != null && task != null && (cap = a.length) > 0) {
1068                 int m = cap - 1, s = p - 1, d = p - base;
1069                 for (int i = s, k; d > 0; --i, --d) {
1070                     if ((t = a[k = i & m]) == task) {
1071                         if (owned || tryLock()) {
1072                             if ((owned || (array == a && top == p)) &&
1073                                 (taken = casSlotToNull(a, k, t))) {
1074                                 for (int j = i; j != s; ) // shift down
1075                                     a[j & m] = getAndClearSlot(a, ++j & m);
1076                                 top = s;

1077                             }
1078                             if (!owned)
1079                                 source = 0;
1080                         }
1081                         break;
1082                     }
1083                 }
1084             }
1085             return taken;
1086         }
1087 
1088         // variants of poll
1089 
1090         /**
1091          * Tries once to poll next task in FIFO order, failing on
1092          * inconsistency or contention.
1093          */
1094         final ForkJoinTask<?> tryPoll() {
1095             int cap, b, k; ForkJoinTask<?>[] a;
1096             if ((a = array) != null && (cap = a.length) > 0) {
1097                 ForkJoinTask<?> t = getSlot(a, k = (cap - 1) & (b = base));
1098                 if (base == b++ && t != null && casSlotToNull(a, k, t)) {
1099                     setBaseOpaque(b);
1100                     return t;
1101                 }
1102             }
1103             return null;
1104         }


1105 
1106         /**
1107          * Takes next task, if one exists, in order specified by mode.


1108          */
1109         final ForkJoinTask<?> nextLocalTask(int cfg) {
1110             ForkJoinTask<?> t = null;
1111             int s = top, cap; ForkJoinTask<?>[] a;
1112             if ((a = array) != null && (cap = a.length) > 0) {
1113                 for (int b, d;;) {
1114                     if ((d = s - (b = base)) <= 0)
1115                         break;
1116                     if (d == 1 || (cfg & FIFO) == 0) {
1117                         if ((t = getAndClearSlot(a, --s & (cap - 1))) != null)
1118                             top = s;
1119                         break;




1120                     }
1121                     if ((t = getAndClearSlot(a, b++ & (cap - 1))) != null) {
1122                         setBaseOpaque(b);
1123                         break;
1124                     }
1125                 }
1126             }
1127             return t;



1128         }
1129 
1130         /**
1131          * Takes next task, if one exists, using configured mode.
1132          */
1133         final ForkJoinTask<?> nextLocalTask() {
1134             return nextLocalTask(config);






















1135         }
1136 
1137         /**
1138          * Returns next task, if one exists, in order specified by mode.
1139          */
1140         final ForkJoinTask<?> peek() {
1141             VarHandle.acquireFence();
1142             int cap; ForkJoinTask<?>[] a;
1143             return ((a = array) != null && (cap = a.length) > 0) ?
1144                 a[(cap - 1) & ((config & FIFO) != 0 ? base : top - 1)] : null;
1145         }
1146 
1147         // specialized execution methods
1148 
1149         /**
1150          * Runs the given (stolen) task if nonnull, as well as
1151          * remaining local tasks and/or others available from the
1152          * given queue.
1153          */
1154         final void topLevelExec(ForkJoinTask<?> task, WorkQueue q) {
1155             int cfg = config, nstolen = 1;
1156             while (task != null) {
1157                 task.doExec();
1158                 if ((task = nextLocalTask(cfg)) == null &&
1159                     q != null && (task = q.tryPoll()) != null)
1160                     ++nstolen;
1161             }
1162             nsteals += nstolen;
1163             source = 0;
1164             if ((cfg & INNOCUOUS) != 0)
1165                 ThreadLocalRandom.eraseThreadLocals(Thread.currentThread());
1166         }
1167 
1168         /**
1169          * Tries to pop and run tasks within the target's computation
1170          * until done, not found, or limit exceeded.
1171          *
1172          * @param task root of CountedCompleter computation
1173          * @param owned true if owned by a ForkJoinWorkerThread
1174          * @param limit max runs, or zero for no limit

1175          * @return task status on exit
1176          */
1177         final int helpComplete(ForkJoinTask<?> task, boolean owned, int limit) {
1178             int status = 0, cap, k, p, s; ForkJoinTask<?>[] a; ForkJoinTask<?> t;
1179             while (task != null && (status = task.status) >= 0 &&
1180                    (a = array) != null && (cap = a.length) > 0 &&
1181                    (t = a[k = (cap - 1) & (s = (p = top) - 1)])
1182                    instanceof CountedCompleter) {
1183                 CountedCompleter<?> f = (CountedCompleter<?>)t;
1184                 boolean taken = false;
1185                 for (;;) {     // exec if root task is a completer of t
1186                     if (f == task) {
1187                         if (owned) {
1188                             if ((taken = casSlotToNull(a, k, t)))
1189                                 top = s;








1190                         }
1191                         else if (tryLock()) {
1192                             if (top == p && array == a &&
1193                                 (taken = casSlotToNull(a, k, t)))
1194                                 top = s;
1195                             source = 0;
1196                         }
1197                         break;
1198                     }
1199                     else if ((f = f.completer) == null)




1200                         break;
1201                 }
1202                 if (!taken)
1203                     break;
1204                 t.doExec();
1205                 if (limit != 0 && --limit == 0)


1206                     break;

1207             }
1208             return status;
1209         }
1210 
1211         /**
1212          * Tries to poll and run AsynchronousCompletionTasks until
1213          * none found or blocker is released.
1214          *
1215          * @param blocker the blocker
1216          */
1217         final void helpAsyncBlocker(ManagedBlocker blocker) {
1218             int cap, b, d, k; ForkJoinTask<?>[] a; ForkJoinTask<?> t;
1219             while (blocker != null && (d = top - (b = base)) > 0 &&
1220                    (a = array) != null && (cap = a.length) > 0 &&
1221                    (((t = getSlot(a, k = (cap - 1) & b)) == null && d > 1) ||
1222                     t instanceof
1223                     CompletableFuture.AsynchronousCompletionTask) &&
1224                    !blocker.isReleasable()) {
1225                 if (t != null && base == b++ && casSlotToNull(a, k, t)) {
1226                     setBaseOpaque(b);




1227                     t.doExec();
1228                 }
1229             }
1230         }
1231 
1232         // misc
1233 
1234         /** AccessControlContext for innocuous workers, created on 1st use. */
1235         private static AccessControlContext INNOCUOUS_ACC;
1236 
1237         /**
1238          * Initializes (upon registration) InnocuousForkJoinWorkerThreads.
1239          */
1240         final void initializeInnocuousWorker() {
1241             AccessControlContext acc; // racy construction OK
1242             if ((acc = INNOCUOUS_ACC) == null)
1243                 INNOCUOUS_ACC = acc = new AccessControlContext(
1244                     new ProtectionDomain[] { new ProtectionDomain(null, null) });
1245             Thread t = Thread.currentThread();
1246             ThreadLocalRandom.setInheritedAccessControlContext(t, acc);
1247             ThreadLocalRandom.eraseThreadLocals(t);
1248         }
1249 
1250         /**
1251          * Returns true if owned by a worker thread and not known to be blocked.
1252          */
1253         final boolean isApparentlyUnblocked() {
1254             Thread wt; Thread.State s;
1255             return ((wt = owner) != null &&
1256                     (s = wt.getState()) != Thread.State.BLOCKED &&
1257                     s != Thread.State.WAITING &&
1258                     s != Thread.State.TIMED_WAITING);
1259         }
1260 




1261         static {
1262             try {
1263                 QA = MethodHandles.arrayElementVarHandle(ForkJoinTask[].class);
1264                 MethodHandles.Lookup l = MethodHandles.lookup();
1265                 SOURCE = l.findVarHandle(WorkQueue.class, "source", int.class);
1266                 BASE = l.findVarHandle(WorkQueue.class, "base", int.class);

1267             } catch (ReflectiveOperationException e) {
1268                 throw new ExceptionInInitializerError(e);
1269             }
1270         }
1271     }
1272 
1273     // static fields (initialized in static initializer below)
1274 
1275     /**
1276      * Creates a new ForkJoinWorkerThread. This factory is used unless
1277      * overridden in ForkJoinPool constructors.
1278      */
1279     public static final ForkJoinWorkerThreadFactory
1280         defaultForkJoinWorkerThreadFactory;
1281 
1282     /**
1283      * Permission required for callers of methods that may start or
1284      * kill threads.
1285      */
1286     static final RuntimePermission modifyThreadPermission;
1287 
1288     /**
1289      * Common (static) pool. Non-null for public use unless a static
1290      * construction exception, but internal usages null-check on use
1291      * to paranoically avoid potential initialization circularities
1292      * as well as to simplify generated code.
1293      */
1294     static final ForkJoinPool common;
1295 
1296     /**
1297      * Common pool parallelism. To allow simpler use and management
1298      * when common pool threads are disabled, we allow the underlying
1299      * common.parallelism field to be zero, but in that case still report
1300      * parallelism as 1 to reflect resulting caller-runs mechanics.
1301      */
1302     static final int COMMON_PARALLELISM;
1303 
1304     /**
1305      * Limit on spare thread construction in tryCompensate.
1306      */
1307     private static final int COMMON_MAX_SPARES;
1308 
1309     /**
1310      * Sequence number for creating worker names
1311      */
1312     private static volatile int poolIds;








1313 
1314     // static configuration constants
1315 
1316     /**
1317      * Default idle timeout value (in milliseconds) for the thread
1318      * triggering quiescence to park waiting for new work
1319      */
1320     private static final long DEFAULT_KEEPALIVE = 60_000L;
1321 
1322     /**
1323      * Undershoot tolerance for idle timeouts
1324      */
1325     private static final long TIMEOUT_SLOP = 20L;
1326 
1327     /**
1328      * The default value for COMMON_MAX_SPARES.  Overridable using the
1329      * "java.util.concurrent.ForkJoinPool.common.maximumSpares" system
1330      * property.  The default value is far in excess of normal
1331      * requirements, but also far short of MAX_CAP and typical OS
1332      * thread limits, so allows JVMs to catch misuse/abuse before
1333      * running out of resources needed to do so.
1334      */
1335     private static final int DEFAULT_COMMON_MAX_SPARES = 256;
1336 






1337     /*
1338      * Bits and masks for field ctl, packed with 4 16 bit subfields:
1339      * RC: Number of released (unqueued) workers minus target parallelism
1340      * TC: Number of total workers minus target parallelism
1341      * SS: version count and status of top waiting thread
1342      * ID: poolIndex of top of Treiber stack of waiters
1343      *
1344      * When convenient, we can extract the lower 32 stack top bits
1345      * (including version bits) as sp=(int)ctl.  The offsets of counts
1346      * by the target parallelism and the positionings of fields makes
1347      * it possible to perform the most common checks via sign tests of
1348      * fields: When ac is negative, there are not enough unqueued
1349      * workers, when tc is negative, there are not enough total
1350      * workers.  When sp is non-zero, there are waiting workers.  To
1351      * deal with possibly negative fields, we use casts in and out of
1352      * "short" and/or signed shifts to maintain signedness.
1353      *
1354      * Because it occupies uppermost bits, we can add one release
1355      * count using getAndAdd of RC_UNIT, rather than CAS, when
1356      * returning from a blocked join.  Other updates entail multiple
1357      * subfields and masking, requiring CAS.
1358      *
1359      * The limits packed in field "bounds" are also offset by the
1360      * parallelism level to make them comparable to the ctl rc and tc
1361      * fields.
1362      */
1363 
1364     // Lower and upper word masks
1365     private static final long SP_MASK    = 0xffffffffL;
1366     private static final long UC_MASK    = ~SP_MASK;
1367 
1368     // Release counts
1369     private static final int  RC_SHIFT   = 48;
1370     private static final long RC_UNIT    = 0x0001L << RC_SHIFT;
1371     private static final long RC_MASK    = 0xffffL << RC_SHIFT;
1372 
1373     // Total counts
1374     private static final int  TC_SHIFT   = 32;
1375     private static final long TC_UNIT    = 0x0001L << TC_SHIFT;
1376     private static final long TC_MASK    = 0xffffL << TC_SHIFT;
1377     private static final long ADD_WORKER = 0x0001L << (TC_SHIFT + 15); // sign
1378 
1379     // Instance fields
1380 

1381     final long keepAlive;                // milliseconds before dropping if idle
1382     volatile long stealCount;            // collects worker nsteals
1383     int scanRover;                       // advances across pollScan calls
1384     volatile int threadIds;              // for worker thread names
1385     final int bounds;                    // min, max threads packed as shorts
1386     volatile int mode;                   // parallelism, runstate, queue mode
1387     WorkQueue[] queues;                  // main registry
1388     final ReentrantLock registrationLock;
1389     Condition termination;               // lazily constructed
1390     final String workerNamePrefix;       // null for common pool
1391     final ForkJoinWorkerThreadFactory factory;
1392     final UncaughtExceptionHandler ueh;  // per-worker UEH
1393     final Predicate<? super ForkJoinPool> saturate;
1394 
1395     @jdk.internal.vm.annotation.Contended("fjpctl") // segregate
1396     volatile long ctl;                   // main pool control
1397 
1398     // Support for atomic operations
1399     private static final VarHandle CTL;
1400     private static final VarHandle MODE;
1401     private static final VarHandle THREADIDS;
1402     private static final VarHandle POOLIDS;
1403     private boolean compareAndSetCtl(long c, long v) {
1404         return CTL.compareAndSet(this, c, v);
1405     }
1406     private long compareAndExchangeCtl(long c, long v) {
1407         return (long)CTL.compareAndExchange(this, c, v);
1408     }
1409     private long getAndAddCtl(long v) {
1410         return (long)CTL.getAndAdd(this, v);
1411     }
1412     private int getAndBitwiseOrMode(int v) {
1413         return (int)MODE.getAndBitwiseOr(this, v);
1414     }
1415     private int getAndAddThreadIds(int x) {
1416         return (int)THREADIDS.getAndAdd(this, x);
1417     }
1418     private static int getAndAddPoolIds(int x) {
1419         return (int)POOLIDS.getAndAdd(x);
1420     }
1421 
1422     // Creating, registering and deregistering workers
1423 
1424     /**
1425      * Tries to construct and start one worker. Assumes that total
1426      * count has already been incremented as a reservation.  Invokes
1427      * deregisterWorker on any failure.
1428      *
1429      * @return true if successful
1430      */
1431     private boolean createWorker() {
1432         ForkJoinWorkerThreadFactory fac = factory;
1433         Throwable ex = null;
1434         ForkJoinWorkerThread wt = null;
1435         try {
1436             if (fac != null && (wt = fac.newThread(this)) != null) {
1437                 wt.start();
1438                 return true;
1439             }
1440         } catch (Throwable rex) {
1441             ex = rex;
1442         }
1443         deregisterWorker(wt, ex);
1444         return false;
1445     }
1446 
1447     /**
1448      * Provides a name for ForkJoinWorkerThread constructor.





1449      */
1450     final String nextWorkerThreadName() {
1451         String prefix = workerNamePrefix;
1452         int tid = getAndAddThreadIds(1) + 1;
1453         if (prefix == null) // commonPool has no prefix
1454             prefix = "ForkJoinPool.commonPool-worker-";
1455         return prefix.concat(Integer.toString(tid));



1456     }
1457 
1458     /**
1459      * Finishes initializing and records owned queue.

1460      *
1461      * @param w caller's WorkQueue

1462      */
1463     final void registerWorker(WorkQueue w) {
1464         ReentrantLock lock = registrationLock;
1465         ThreadLocalRandom.localInit();
1466         int seed = ThreadLocalRandom.getProbe();
1467         if (w != null && lock != null) {
1468             int modebits = (mode & FIFO) | w.config;
1469             w.array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY];
1470             w.stackPred = seed;                         // stash for runWorker
1471             if ((modebits & INNOCUOUS) != 0)
1472                 w.initializeInnocuousWorker();
1473             int id = (seed << 1) | 1;                   // initial index guess
1474             lock.lock();
1475             try {
1476                 WorkQueue[] qs; int n;                  // find queue index
1477                 if ((qs = queues) != null && (n = qs.length) > 0) {
1478                     int k = n, m = n - 1;
1479                     for (; qs[id &= m] != null && k > 0; id -= 2, k -= 2);
1480                     if (k == 0)
1481                         id = n | 1;                     // resize below
1482                     w.phase = w.config = id | modebits; // now publishable









1483 
1484                     if (id < n)
1485                         qs[id] = w;
1486                     else {                              // expand array
1487                         int an = n << 1, am = an - 1;
1488                         WorkQueue[] as = new WorkQueue[an];
1489                         as[id & am] = w;
1490                         for (int j = 1; j < n; j += 2)
1491                             as[j] = qs[j];
1492                         for (int j = 0; j < n; j += 2) {
1493                             WorkQueue q;
1494                             if ((q = qs[j]) != null)    // shared queues may move
1495                                 as[q.config & am] = q;


1496                         }
1497                         VarHandle.releaseFence();       // fill before publish
1498                         queues = as;
1499                     }
1500                 }
1501             } finally {
1502                 lock.unlock();
1503             }

1504         }

1505     }
1506 
1507     /**
1508      * Final callback from terminating worker, as well as upon failure
1509      * to construct or start a worker.  Removes record of worker from
1510      * array, and adjusts counts. If pool is shutting down, tries to
1511      * complete termination.
1512      *
1513      * @param wt the worker thread, or null if construction failed
1514      * @param ex the exception causing failure, or null if none
1515      */
1516     final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) {
1517         ReentrantLock lock = registrationLock;
1518         WorkQueue w = null;
1519         int cfg = 0;
1520         if (wt != null && (w = wt.workQueue) != null && lock != null) {
1521             WorkQueue[] qs; int n, i;
1522             cfg = w.config;
1523             long ns = w.nsteals & 0xffffffffL;
1524             lock.lock();                             // remove index from array
1525             if ((qs = queues) != null && (n = qs.length) > 0 &&
1526                 qs[i = cfg & (n - 1)] == w)
1527                 qs[i] = null;
1528             stealCount += ns;                        // accumulate steals
1529             lock.unlock();
1530             long c = ctl;
1531             if ((cfg & QUIET) == 0) // unless self-signalled, decrement counts
1532                 do {} while (c != (c = compareAndExchangeCtl(
1533                                        c, ((RC_MASK & (c - RC_UNIT)) |





1534                                            (TC_MASK & (c - TC_UNIT)) |
1535                                            (SP_MASK & c)))));
1536             else if ((int)c == 0)                    // was dropped on timeout
1537                 cfg = 0;                             // suppress signal if last
1538             for (ForkJoinTask<?> t; (t = w.pop()) != null; )
1539                 ForkJoinTask.cancelIgnoringExceptions(t); // cancel tasks
1540         }


1541 
1542         if (!tryTerminate(false, false) && w != null && (cfg & SRC) != 0)
1543             signalWork();                            // possibly replace worker
1544         if (ex != null)




1545             ForkJoinTask.rethrow(ex);
1546     }
1547 
1548     /*
1549      * Tries to create or release a worker if too few are running.

1550      */
1551     final void signalWork() {
1552         for (long c = ctl; c < 0L;) {
1553             int sp, i; WorkQueue[] qs; WorkQueue v;
1554             if ((sp = (int)c & ~UNSIGNALLED) == 0) {  // no idle workers
1555                 if ((c & ADD_WORKER) == 0L)           // enough total workers
1556                     break;
1557                 if (c == (c = compareAndExchangeCtl(
1558                               c, ((RC_MASK & (c + RC_UNIT)) |
1559                                   (TC_MASK & (c + TC_UNIT)))))) {
1560                     createWorker();
1561                     break;
1562                 }
1563             }
1564             else if ((qs = queues) == null)
1565                 break;                                // unstarted/terminated
1566             else if (qs.length <= (i = sp & SMASK))
1567                 break;                                // terminated
1568             else if ((v = qs[i]) == null)
1569                 break;                                // terminating
1570             else {


1571                 long nc = (v.stackPred & SP_MASK) | (UC_MASK & (c + RC_UNIT));
1572                 Thread vt = v.owner;
1573                 if (c == (c = compareAndExchangeCtl(c, nc))) {
1574                     v.phase = sp;
1575                     LockSupport.unpark(vt);           // release idle worker

1576                     break;
1577                 }


1578             }
1579         }
1580     }
1581 
1582     /**
1583      * Top-level runloop for workers, called by ForkJoinWorkerThread.run.
1584      * See above for explanation.









1585      *
1586      * @param w caller's WorkQueue (may be null on failed initialization)
1587      */
1588     final void runWorker(WorkQueue w) {
1589         if (w != null) {                        // skip on failed init
1590             w.config |= SRC;                    // mark as valid source
1591             int r = w.stackPred, src = 0;       // use seed from registerWorker
1592             do {
1593                 r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift
1594             } while ((src = scan(w, src, r)) >= 0 ||
1595                      (src = awaitWork(w)) == 0);













1596         }
1597     }
1598 
1599     /**
1600      * Scans for and if found executes top-level tasks: Tries to poll
1601      * each queue starting at a random index with random stride,
1602      * returning source id or retry indicator if contended or
1603      * inconsistent.
1604      *
1605      * @param w caller's WorkQueue
1606      * @param prevSrc the previous queue stolen from in current phase, or 0
1607      * @param r random seed
1608      * @return id of queue if taken, negative if none found, prevSrc for retry
1609      */
1610     private int scan(WorkQueue w, int prevSrc, int r) {
1611         WorkQueue[] qs = queues;
1612         int n = (w == null || qs == null) ? 0 : qs.length;
1613         for (int step = (r >>> 16) | 1, i = n; i > 0; --i, r += step) {
1614             int j, cap, b; WorkQueue q; ForkJoinTask<?>[] a;
1615             if ((q = qs[j = r & (n - 1)]) != null && // poll at qs[j].array[k]
1616                 (a = q.array) != null && (cap = a.length) > 0) {
1617                 int k = (cap - 1) & (b = q.base), nextBase = b + 1;
1618                 int nextIndex = (cap - 1) & nextBase, src = j | SRC;
1619                 ForkJoinTask<?> t = WorkQueue.getSlot(a, k);
1620                 if (q.base != b)                // inconsistent
1621                     return prevSrc;
1622                 else if (t != null && WorkQueue.casSlotToNull(a, k, t)) {
1623                     q.base = nextBase;
1624                     ForkJoinTask<?> next = a[nextIndex];
1625                     if ((w.source = src) != prevSrc && next != null)
1626                         signalWork();           // propagate
1627                     w.topLevelExec(t, q);
1628                     return src;
1629                 }
1630                 else if (a[nextIndex] != null)  // revisit
1631                     return prevSrc;
1632             }
1633         }
1634         return (queues != qs) ? prevSrc: -1;    // possibly resized
1635     }
1636 
1637     /**
1638      * Advances worker phase, pushes onto ctl stack, and awaits signal
1639      * or reports termination.
1640      *
1641      * @return negative if terminated, else 0
1642      */
1643     private int awaitWork(WorkQueue w) {
1644         if (w == null)
1645             return -1;                       // already terminated
1646         int phase = (w.phase + SS_SEQ) & ~UNSIGNALLED;
1647         w.phase = phase | UNSIGNALLED;       // advance phase
1648         long prevCtl = ctl, c;               // enqueue
1649         do {
1650             w.stackPred = (int)prevCtl;
1651             c = ((prevCtl - RC_UNIT) & UC_MASK) | (phase & SP_MASK);
1652         } while (prevCtl != (prevCtl = compareAndExchangeCtl(prevCtl, c)));
1653 
1654         Thread.interrupted();                // clear status
1655         LockSupport.setCurrentBlocker(this); // prepare to block (exit also OK)
1656         long deadline = 0L;                  // nonzero if possibly quiescent
1657         int ac = (int)(c >> RC_SHIFT), md;
1658         if ((md = mode) < 0)                 // pool is terminating
1659             return -1;
1660         else if ((md & SMASK) + ac <= 0) {
1661             boolean checkTermination = (md & SHUTDOWN) != 0;
1662             if ((deadline = System.currentTimeMillis() + keepAlive) == 0L)
1663                 deadline = 1L;               // avoid zero
1664             WorkQueue[] qs = queues;         // check for racing submission
1665             int n = (qs == null) ? 0 : qs.length;
1666             for (int i = 0; i < n; i += 2) {
1667                 WorkQueue q; ForkJoinTask<?>[] a; int cap, b;
1668                 if (ctl != c) {              // already signalled
1669                     checkTermination = false;
1670                     break;
1671                 }
1672                 else if ((q = qs[i]) != null &&
1673                          (a = q.array) != null && (cap = a.length) > 0 &&
1674                          ((b = q.base) != q.top || a[(cap - 1) & b] != null ||
1675                           q.source != 0)) {
1676                     if (compareAndSetCtl(c, prevCtl))
1677                         w.phase = phase;     // self-signal
1678                     checkTermination = false;
1679                     break;
1680                 }
1681             }
1682             if (checkTermination && tryTerminate(false, false))
1683                 return -1;                   // trigger quiescent termination
1684         }
1685 
1686         for (boolean alt = false;;) {        // await activation or termination
1687             if (w.phase >= 0)
1688                 break;
1689             else if (mode < 0)
1690                 return -1;
1691             else if ((c = ctl) == prevCtl)
1692                 Thread.onSpinWait();         // signal in progress
1693             else if (!(alt = !alt))          // check between park calls
1694                 Thread.interrupted();
1695             else if (deadline == 0L)
1696                 LockSupport.park();
1697             else if (deadline - System.currentTimeMillis() > TIMEOUT_SLOP)
1698                 LockSupport.parkUntil(deadline);
1699             else if (((int)c & SMASK) == (w.config & SMASK) &&
1700                      compareAndSetCtl(c, ((UC_MASK & (c - TC_UNIT)) |
1701                                           (prevCtl & SP_MASK)))) {
1702                 w.config |= QUIET;           // sentinel for deregisterWorker
1703                 return -1;                   // drop on timeout
1704             }
1705             else if ((deadline += keepAlive) == 0L)
1706                 deadline = 1L;               // not at head; restart timer
1707         }
1708         return 0;
1709     }
1710 
1711     // Utilities used by ForkJoinTask
1712 
1713     /**
1714      * Returns true if all workers are busy, possibly creating one if allowed
1715      */
1716     final boolean isSaturated() {
1717         int maxTotal = bounds >>> SWIDTH;
1718         for (long c;;) {
1719             if (((int)(c = ctl) & ~UNSIGNALLED) != 0)
1720                 return false;
1721             if ((short)(c >>> TC_SHIFT) >= maxTotal)
1722                 return true;
1723             long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK);
1724             if (compareAndSetCtl(c, nc))
1725                 return !createWorker();
1726         }
1727     }
1728 
1729     /**
1730      * Returns true if can start terminating if enabled, or already terminated

1731      */
1732     final boolean canStop() {
1733         outer: for (long oldSum = 0L;;) { // repeat until stable
1734             int md; WorkQueue[] qs;  long c;
1735             if ((qs = queues) == null || ((md = mode) & STOP) != 0)
1736                 return true;
1737             if ((md & SMASK) + (int)((c = ctl) >> RC_SHIFT) > 0)






























1738                 break;
1739             long checkSum = c;
1740             for (int i = 1; i < qs.length; i += 2) { // scan submitters
1741                 WorkQueue q; ForkJoinTask<?>[] a; int s = 0, cap;
1742                 if ((q = qs[i]) != null && (a = q.array) != null &&
1743                     (cap = a.length) > 0 &&
1744                     ((s = q.top) != q.base || a[(cap - 1) & s] != null ||
1745                      q.source != 0))
1746                     break outer;
1747                 checkSum += (((long)i) << 32) ^ s;
1748             }
1749             if (oldSum == (oldSum = checkSum) && queues == qs)
1750                 return true;
1751         }
1752         return (mode & STOP) != 0; // recheck mode on false return
1753     }
1754 
1755     /**
1756      * Tries to decrement counts (sometimes implicitly) and possibly
1757      * arrange for a compensating worker in preparation for
1758      * blocking. May fail due to interference, in which case -1 is
1759      * returned so caller may retry. A zero return value indicates
1760      * that the caller doesn't need to re-adjust counts when later
1761      * unblocked.
1762      *
1763      * @param c incoming ctl value
1764      * @return UNCOMPENSATE: block then adjust, 0: block, -1 : retry
1765      */
1766     private int tryCompensate(long c) {
1767         Predicate<? super ForkJoinPool> sat;
1768         int b = bounds; // counts are signed; centered at parallelism level == 0
1769         int minActive = (short)(b & SMASK),
1770             maxTotal  = b >>> SWIDTH,
1771             active    = (int)(c >> RC_SHIFT),
1772             total     = (short)(c >>> TC_SHIFT),
1773             sp        = (int)c & ~UNSIGNALLED;
1774         if (total >= 0) {
1775             if (sp != 0) {                        // activate idle worker
1776                 WorkQueue[] qs; int n; WorkQueue v;
1777                 if ((qs = queues) != null && (n = qs.length) > 0 &&
1778                     (v = qs[sp & (n - 1)]) != null) {
1779                     Thread vt = v.owner;
1780                     long nc = ((long)v.stackPred & SP_MASK) | (UC_MASK & c);
1781                     if (compareAndSetCtl(c, nc)) {
1782                         v.phase = sp;
1783                         LockSupport.unpark(vt);
1784                         return UNCOMPENSATE;
1785                     }




1786                 }
1787                 return -1;                        // retry
1788             }
1789             else if (active > minActive) {        // reduce parallelism
1790                 long nc = ((RC_MASK & (c - RC_UNIT)) | (~RC_MASK & c));
1791                 return compareAndSetCtl(c, nc) ? UNCOMPENSATE : -1;
1792             }

1793         }
1794         if (total < maxTotal) {                   // expand pool
1795             long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK);
1796             return (!compareAndSetCtl(c, nc) ? -1 :
1797                     !createWorker() ? 0 : UNCOMPENSATE);
1798         }
1799         else if (!compareAndSetCtl(c, c))         // validate
1800             return -1;
1801         else if ((sat = saturate) != null && sat.test(this))
1802             return 0;
1803         else
1804             throw new RejectedExecutionException(
1805                 "Thread limit exceeded replacing blocked worker");
1806     }
1807 
1808     /**
1809      * Readjusts RC count; called from ForkJoinTask after blocking.
1810      */
1811     final void uncompensate() {
1812         getAndAddCtl(RC_UNIT);
1813     }
1814 
1815     /**
1816      * Helps if possible until the given task is done.  Scans other
1817      * queues for a task produced by one of w's stealers; returning
1818      * compensated blocking sentinel if none are found.
1819      *
1820      * @param task the task
1821      * @param w caller's WorkQueue
1822      * @return task status on exit, or UNCOMPENSATE for compensated blocking
1823      */
1824     final int helpJoin(ForkJoinTask<?> task, WorkQueue w) {
1825         int s = 0;
1826         if (task != null && w != null) {
1827             int wsrc = w.source, wid = w.config & SMASK, r = wid + 2;
1828             boolean scan = true;
1829             long c = 0L;                          // track ctl stability
1830             outer: for (;;) {
1831                 if ((s = task.status) < 0)
1832                     break;
1833                 else if (scan = !scan) {          // previous scan was empty
1834                     if (mode < 0)
1835                         ForkJoinTask.cancelIgnoringExceptions(task);
1836                     else if (c == (c = ctl) && (s = tryCompensate(c)) >= 0)
1837                         break;                    // block
1838                 }
1839                 else {                            // scan for subtasks
1840                     WorkQueue[] qs = queues;
1841                     int n = (qs == null) ? 0 : qs.length, m = n - 1;
1842                     for (int i = n; i > 0; i -= 2, r += 2) {
1843                         int j; WorkQueue q, x, y; ForkJoinTask<?>[] a;
1844                         if ((q = qs[j = r & m]) != null) {
1845                             int sq = q.source & SMASK, cap, b;
1846                             if ((a = q.array) != null && (cap = a.length) > 0) {
1847                                 int k = (cap - 1) & (b = q.base);
1848                                 int nextBase = b + 1, src = j | SRC, sx;
1849                                 ForkJoinTask<?> t = WorkQueue.getSlot(a, k);
1850                                 boolean eligible = sq == wid ||
1851                                     ((x = qs[sq & m]) != null &&   // indirect
1852                                      ((sx = (x.source & SMASK)) == wid ||
1853                                       ((y = qs[sx & m]) != null && // 2-indirect
1854                                        (y.source & SMASK) == wid)));
1855                                 if ((s = task.status) < 0)
1856                                     break outer;
1857                                 else if ((q.source & SMASK) != sq ||
1858                                          q.base != b)
1859                                     scan = true;          // inconsistent
1860                                 else if (t == null)
1861                                     scan |= (a[nextBase & (cap - 1)] != null ||
1862                                              q.top != b); // lagging
1863                                 else if (eligible) {
1864                                     if (WorkQueue.casSlotToNull(a, k, t)) {
1865                                         q.base = nextBase;
1866                                         w.source = src;
1867                                         t.doExec();
1868                                         w.source = wsrc;
1869                                     }
1870                                     scan = true;
1871                                     break;
1872                                 }
1873                             }
1874                         }

1875                     }




1876                 }
1877             }
1878         }
1879         return s;
1880     }
1881 
1882     /**
1883      * Extra helpJoin steps for CountedCompleters.  Scans for and runs
1884      * subtasks of the given root task, returning if none are found.


1885      *
1886      * @param task root of CountedCompleter computation
1887      * @param w caller's WorkQueue
1888      * @param owned true if owned by a ForkJoinWorkerThread
1889      * @return task status on exit
1890      */
1891     final int helpComplete(ForkJoinTask<?> task, WorkQueue w, boolean owned) {
1892         int s = 0;
1893         if (task != null && w != null) {
1894             int r = w.config;
1895             boolean scan = true, locals = true;
1896             long c = 0L;
1897             outer: for (;;) {
1898                 if (locals) {                     // try locals before scanning
1899                     if ((s = w.helpComplete(task, owned, 0)) < 0)
1900                         break;
1901                     locals = false;
1902                 }
1903                 else if ((s = task.status) < 0)
1904                     break;
1905                 else if (scan = !scan) {
1906                     if (c == (c = ctl))
1907                         break;
1908                 }
1909                 else {                            // scan for subtasks
1910                     WorkQueue[] qs = queues;
1911                     int n = (qs == null) ? 0 : qs.length;
1912                     for (int i = n; i > 0; --i, ++r) {
1913                         int j, cap, b; WorkQueue q; ForkJoinTask<?>[] a;
1914                         boolean eligible = false;
1915                         if ((q = qs[j = r & (n - 1)]) != null &&
1916                             (a = q.array) != null && (cap = a.length) > 0) {
1917                             int k = (cap - 1) & (b = q.base), nextBase = b + 1;
1918                             ForkJoinTask<?> t = WorkQueue.getSlot(a, k);
1919                             if (t instanceof CountedCompleter) {
1920                                 CountedCompleter<?> f = (CountedCompleter<?>)t;
1921                                 do {} while (!(eligible = (f == task)) &&
1922                                              (f = f.completer) != null);
1923                             }
1924                             if ((s = task.status) < 0)
1925                                 break outer;
1926                             else if (q.base != b)
1927                                 scan = true;       // inconsistent
1928                             else if (t == null)
1929                                 scan |= (a[nextBase & (cap - 1)] != null ||
1930                                          q.top != b);
1931                             else if (eligible) {
1932                                 if (WorkQueue.casSlotToNull(a, k, t)) {
1933                                     q.setBaseOpaque(nextBase);
1934                                     t.doExec();
1935                                     locals = true;
1936                                 }
1937                                 scan = true;
1938                                 break;
1939                             }



1940                         }
1941                     }















1942                 }
1943             }
1944         }
1945         return s;
1946     }
1947 
1948     /**
1949      * Scans for and returns a polled task, if available.  Used only
1950      * for untracked polls. Begins scan at an index (scanRover)
1951      * advanced on each call, to avoid systematic unfairness.
1952      *
1953      * @param submissionsOnly if true, only scan submission queues
1954      */
1955     private ForkJoinTask<?> pollScan(boolean submissionsOnly) {
1956         VarHandle.acquireFence();
1957         int r = scanRover += 0x61c88647; // Weyl increment; raciness OK
1958         if (submissionsOnly)             // even indices only
1959             r &= ~1;
1960         int step = (submissionsOnly) ? 2 : 1;
1961         WorkQueue[] qs; int n;
1962         while ((qs = queues) != null && (n = qs.length) > 0) {
1963             boolean scan = false;
1964             for (int i = 0; i < n; i += step) {
1965                 int j, cap, b; WorkQueue q; ForkJoinTask<?>[] a;
1966                 if ((q = qs[j = (n - 1) & (r + i)]) != null &&
1967                     (a = q.array) != null && (cap = a.length) > 0) {
1968                     int k = (cap - 1) & (b = q.base), nextBase = b + 1;
1969                     ForkJoinTask<?> t = WorkQueue.getSlot(a, k);
1970                     if (q.base != b)
1971                         scan = true;
1972                     else if (t == null)
1973                         scan |= (q.top != b || a[nextBase & (cap - 1)] != null);
1974                     else if (!WorkQueue.casSlotToNull(a, k, t))
1975                         scan = true;
1976                     else {
1977                         q.setBaseOpaque(nextBase);
1978                         return t;
1979                     }
1980                 }
1981             }
1982             if (!scan && queues == qs)
1983                 break;
1984         }
1985         return null;
1986     }
1987 
1988     /**
1989      * Runs tasks until {@code isQuiescent()}. Rather than blocking
1990      * when tasks cannot be found, rescans until all others cannot
1991      * find tasks either.
1992      *
1993      * @param nanos max wait time (Long.MAX_VALUE if effectively untimed)
1994      * @param interruptible true if return on interrupt
1995      * @return positive if quiescent, negative if interrupted, else 0
1996      */
1997     final int helpQuiescePool(WorkQueue w, long nanos, boolean interruptible) {
1998         if (w == null)
1999             return 0;
2000         long startTime = System.nanoTime(), parkTime = 0L;
2001         int prevSrc = w.source, wsrc = prevSrc, cfg = w.config, r = cfg + 1;
2002         for (boolean active = true, locals = true;;) {
2003             boolean busy = false, scan = false;
2004             if (locals) {  // run local tasks before (re)polling
2005                 locals = false;
2006                 for (ForkJoinTask<?> u; (u = w.nextLocalTask(cfg)) != null;)
2007                     u.doExec();
2008             }
2009             WorkQueue[] qs = queues;
2010             int n = (qs == null) ? 0 : qs.length;
2011             for (int i = n; i > 0; --i, ++r) {
2012                 int j, b, cap; WorkQueue q; ForkJoinTask<?>[] a;
2013                 if ((q = qs[j = (n - 1) & r]) != null && q != w &&
2014                     (a = q.array) != null && (cap = a.length) > 0) {
2015                     int k = (cap - 1) & (b = q.base);
2016                     int nextBase = b + 1, src = j | SRC;
2017                     ForkJoinTask<?> t = WorkQueue.getSlot(a, k);
2018                     if (q.base != b)
2019                         busy = scan = true;
2020                     else if (t != null) {
2021                         busy = scan = true;
2022                         if (!active) {    // increment before taking
2023                             active = true;
2024                             getAndAddCtl(RC_UNIT);





2025                         }
2026                         if (WorkQueue.casSlotToNull(a, k, t)) {
2027                             q.base = nextBase;
2028                             w.source = src;
2029                             t.doExec();
2030                             w.source = wsrc = prevSrc;
2031                             locals = true;
2032                         }
2033                         break;
2034                     }
2035                     else if (!busy) {
2036                         if (q.top != b || a[nextBase & (cap - 1)] != null)
2037                             busy = scan = true;
2038                         else if (q.source != QUIET && q.phase >= 0)
2039                             busy = true;
2040                     }
2041                 }
2042             }
2043             VarHandle.acquireFence();
2044             if (!scan && queues == qs) {
2045                 boolean interrupted;
2046                 if (!busy) {
2047                     w.source = prevSrc;
2048                     if (!active)
2049                         getAndAddCtl(RC_UNIT);
2050                     return 1;
2051                 }
2052                 if (wsrc != QUIET)
2053                     w.source = wsrc = QUIET;
2054                 if (active) {                 // decrement
2055                     active = false;
2056                     parkTime = 0L;
2057                     getAndAddCtl(RC_MASK & -RC_UNIT);
2058                 }
2059                 else if (parkTime == 0L) {
2060                     parkTime = 1L << 10; // initially about 1 usec
2061                     Thread.yield();
2062                 }
2063                 else if ((interrupted = interruptible && Thread.interrupted()) ||
2064                          System.nanoTime() - startTime > nanos) {
2065                     getAndAddCtl(RC_UNIT);
2066                     return interrupted ? -1 : 0;
2067                 }
2068                 else {
2069                     LockSupport.parkNanos(this, parkTime);
2070                     if (parkTime < nanos >>> 8 && parkTime < 1L << 20)
2071                         parkTime <<= 1;  // max sleep approx 1 sec or 1% nanos


2072                 }
2073             }
2074         }
2075     }
2076 
2077     /**
2078      * Helps quiesce from external caller until done, interrupted, or timeout

2079      *
2080      * @param nanos max wait time (Long.MAX_VALUE if effectively untimed)
2081      * @param interruptible true if return on interrupt
2082      * @return positive if quiescent, negative if interrupted, else 0
2083      */
2084     final int externalHelpQuiescePool(long nanos, boolean interruptible) {
2085         for (long startTime = System.nanoTime(), parkTime = 0L;;) {
2086             ForkJoinTask<?> t;
2087             if ((t = pollScan(false)) != null) {
2088                 t.doExec();
2089                 parkTime = 0L;






















2090             }
2091             else if (canStop())
2092                 return 1;
2093             else if (parkTime == 0L) {
2094                 parkTime = 1L << 10;
2095                 Thread.yield();
2096             }
2097             else if ((System.nanoTime() - startTime) > nanos)
2098                 return 0;
2099             else if (interruptible && Thread.interrupted())
2100                 return -1;
2101             else {
2102                 LockSupport.parkNanos(this, parkTime);
2103                 if (parkTime < nanos >>> 8 && parkTime < 1L << 20)
2104                     parkTime <<= 1;
2105             }
2106         }

2107     }
2108 
2109     /**
2110      * Gets and removes a local or stolen task for the given worker.
2111      *
2112      * @return a task, if available
2113      */
2114     final ForkJoinTask<?> nextTaskFor(WorkQueue w) {
2115         ForkJoinTask<?> t;
2116         if (w == null || (t = w.nextLocalTask(w.config)) == null)
2117             t = pollScan(false);
2118         return t;
2119     }
2120 
2121     // External operations
2122 
2123     /**
2124      * Finds and locks a WorkQueue for an external submitter, or
2125      * returns null if shutdown or terminating.


2126      */
2127     final WorkQueue submissionQueue() {
2128         int r;
2129         if ((r = ThreadLocalRandom.getProbe()) == 0) {
2130             ThreadLocalRandom.localInit();           // initialize caller's probe
2131             r = ThreadLocalRandom.getProbe();
2132         }
2133         for (int id = r << 1;;) {                    // even indices only
2134             int md = mode, n, i; WorkQueue q; ReentrantLock lock;
2135             WorkQueue[] qs = queues;
2136             if ((md & SHUTDOWN) != 0 || qs == null || (n = qs.length) <= 0)
2137                 return null;
2138             else if ((q = qs[i = (n - 1) & id]) == null) {
2139                 if ((lock = registrationLock) != null) {
2140                     WorkQueue w = new WorkQueue(id | SRC);
2141                     lock.lock();                    // install under lock
2142                     if (qs[i] == null)
2143                         qs[i] = w;                  // else lost race; discard
2144                     lock.unlock();










2145                 }
2146             }
2147             else if (!q.tryLock())                  // move and restart
2148                 id = (r = ThreadLocalRandom.advanceProbe(r)) << 1;
2149             else
2150                 return q;


2151         }
2152     }
2153 
2154     /**
2155      * Adds the given task to an external submission queue, or throws
2156      * exception if shutdown or terminating.
2157      *
2158      * @param task the task. Caller must ensure non-null.
2159      */
2160     final void externalPush(ForkJoinTask<?> task) {
2161         WorkQueue q;
2162         if ((q = submissionQueue()) == null)
2163             throw new RejectedExecutionException(); // shutdown or disabled
2164         else if (q.lockedPush(task))
2165             signalWork();
2166     }
2167 
2168     /**
2169      * Pushes a possibly-external submission.
2170      */
2171     private <T> ForkJoinTask<T> externalSubmit(ForkJoinTask<T> task) {
2172         Thread t; ForkJoinWorkerThread wt; WorkQueue q;
2173         if (task == null)
2174             throw new NullPointerException();
2175         if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) &&
2176             (q = (wt = (ForkJoinWorkerThread)t).workQueue) != null &&
2177             wt.pool == this)
2178             q.push(task, this);
2179         else
2180             externalPush(task);
2181         return task;
2182     }
2183 
2184     /**
2185      * Returns common pool queue for an external thread that has
2186      * possibly ever submitted a common pool task (nonzero probe), or
2187      * null if none.
2188      */
2189     static WorkQueue commonQueue() {
2190         ForkJoinPool p; WorkQueue[] qs;
2191         int r = ThreadLocalRandom.getProbe(), n;
2192         return ((p = common) != null && (qs = p.queues) != null &&
2193                 (n = qs.length) > 0 && r != 0) ?
2194             qs[(n - 1) & (r << 1)] : null;
2195     }
2196 
2197     /**
2198      * If the given executor is a ForkJoinPool, poll and execute
2199      * AsynchronousCompletionTasks from worker's queue until none are
2200      * available or blocker is released.










2201      */
2202     static void helpAsyncBlocker(Executor e, ManagedBlocker blocker) {
2203         WorkQueue w = null; Thread t; ForkJoinWorkerThread wt;
2204         if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
2205             if ((wt = (ForkJoinWorkerThread)t).pool == e)
2206                 w = wt.workQueue;

2207         }
2208         else if (e == common)
2209             w = commonQueue();
2210         if (w != null)
2211             w.helpAsyncBlocker(blocker);










2212     }
2213 
2214     /**
2215      * Returns a cheap heuristic guide for task partitioning when
2216      * programmers, frameworks, tools, or languages have little or no
2217      * idea about task granularity.  In essence, by offering this
2218      * method, we ask users only about tradeoffs in overhead vs
2219      * expected throughput and its variance, rather than how finely to
2220      * partition tasks.
2221      *
2222      * In a steady state strict (tree-structured) computation, each
2223      * thread makes available for stealing enough tasks for other
2224      * threads to remain active. Inductively, if all threads play by
2225      * the same rules, each thread should make available only a
2226      * constant number of tasks.
2227      *
2228      * The minimum useful constant is just 1. But using a value of 1
2229      * would require immediate replenishment upon each steal to
2230      * maintain enough tasks, which is infeasible.  Further,
2231      * partitionings/granularities of offered tasks should minimize
2232      * steal rates, which in general means that threads nearer the top
2233      * of computation tree should generate more than those nearer the
2234      * bottom. In perfect steady state, each thread is at
2235      * approximately the same level of computation tree. However,
2236      * producing extra tasks amortizes the uncertainty of progress and
2237      * diffusion assumptions.
2238      *
2239      * So, users will want to use values larger (but not much larger)
2240      * than 1 to both smooth over transient shortages and hedge
2241      * against uneven progress; as traded off against the cost of
2242      * extra task overhead. We leave the user to pick a threshold
2243      * value to compare with the results of this call to guide
2244      * decisions, but recommend values such as 3.
2245      *
2246      * When all threads are active, it is on average OK to estimate
2247      * surplus strictly locally. In steady-state, if one thread is
2248      * maintaining say 2 surplus tasks, then so are others. So we can
2249      * just use estimated queue length.  However, this strategy alone
2250      * leads to serious mis-estimates in some non-steady-state
2251      * conditions (ramp-up, ramp-down, other stalls). We can detect
2252      * many of these by further considering the number of "idle"
2253      * threads, that are known to have zero queued tasks, so
2254      * compensate by a factor of (#idle/#active) threads.
2255      */
2256     static int getSurplusQueuedTaskCount() {
2257         Thread t; ForkJoinWorkerThread wt; ForkJoinPool pool; WorkQueue q;
2258         if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) &&
2259             (pool = (wt = (ForkJoinWorkerThread)t).pool) != null &&
2260             (q = wt.workQueue) != null) {
2261             int p = pool.mode & SMASK;
2262             int a = p + (int)(pool.ctl >> RC_SHIFT);
2263             int n = q.top - q.base;
2264             return n - (a > (p >>>= 1) ? 0 :
2265                         a > (p >>>= 1) ? 1 :
2266                         a > (p >>>= 1) ? 2 :
2267                         a > (p >>>= 1) ? 4 :
2268                         8);
2269         }
2270         return 0;
2271     }
2272 
2273     // Termination
2274 
2275     /**
2276      * Possibly initiates and/or completes termination.
2277      *
2278      * @param now if true, unconditionally terminate, else only
2279      * if no work and no active workers
2280      * @param enable if true, terminate when next possible
2281      * @return true if terminating or terminated
2282      */
2283     private boolean tryTerminate(boolean now, boolean enable) {
2284         int md; // try to set SHUTDOWN, then STOP, then help terminate
2285         if (((md = mode) & SHUTDOWN) == 0) {
2286             if (!enable)

2287                 return false;
2288             md = getAndBitwiseOrMode(SHUTDOWN);


























2289         }
2290         if ((md & STOP) == 0) {
2291             if (!now && !canStop())

2292                 return false;
2293             md = getAndBitwiseOrMode(STOP);


2294         }
2295         for (int k = 0; k < 2; ++k) { // twice in case of lagging qs updates
2296             for (ForkJoinTask<?> t; (t = pollScan(false)) != null; )
2297                 ForkJoinTask.cancelIgnoringExceptions(t); // help cancel
2298             WorkQueue[] qs; int n; WorkQueue q; Thread thread;
2299             if ((qs = queues) != null && (n = qs.length) > 0) {
2300                 for (int j = 1; j < n; j += 2) { // unblock other workers
2301                     if ((q = qs[j]) != null && (thread = q.owner) != null &&
2302                         !thread.isInterrupted()) {
2303                         try {
2304                             thread.interrupt();






2305                         } catch (Throwable ignore) {
2306                         }
2307                     }

2308                 }
2309             }
2310             ReentrantLock lock; Condition cond; // signal when no workers
2311             if (((md = mode) & TERMINATED) == 0 &&
2312                 (md & SMASK) + (short)(ctl >>> TC_SHIFT) <= 0 &&
2313                 (getAndBitwiseOrMode(TERMINATED) & TERMINATED) == 0 &&
2314                 (lock = registrationLock) != null) {
2315                 lock.lock();
2316                 if ((cond = termination) != null)
2317                     cond.signalAll();
2318                 lock.unlock();





2319             }
2320         }
2321         return true;
2322     }
2323 
2324     // Exported methods
2325 
2326     // Constructors
2327 
2328     /**
2329      * Creates a {@code ForkJoinPool} with parallelism equal to {@link
2330      * java.lang.Runtime#availableProcessors}, using defaults for all
2331      * other parameters (see {@link #ForkJoinPool(int,
2332      * ForkJoinWorkerThreadFactory, UncaughtExceptionHandler, boolean,
2333      * int, int, int, Predicate, long, TimeUnit)}).
2334      *
2335      * @throws SecurityException if a security manager exists and
2336      *         the caller is not permitted to modify threads
2337      *         because it does not hold {@link
2338      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2339      */
2340     public ForkJoinPool() {
2341         this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
2342              defaultForkJoinWorkerThreadFactory, null, false,
2343              0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS);
2344     }
2345 
2346     /**
2347      * Creates a {@code ForkJoinPool} with the indicated parallelism
2348      * level, using defaults for all other parameters (see {@link
2349      * #ForkJoinPool(int, ForkJoinWorkerThreadFactory,
2350      * UncaughtExceptionHandler, boolean, int, int, int, Predicate,
2351      * long, TimeUnit)}).
2352      *
2353      * @param parallelism the parallelism level
2354      * @throws IllegalArgumentException if parallelism less than or
2355      *         equal to zero, or greater than implementation limit
2356      * @throws SecurityException if a security manager exists and
2357      *         the caller is not permitted to modify threads
2358      *         because it does not hold {@link
2359      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2360      */
2361     public ForkJoinPool(int parallelism) {
2362         this(parallelism, defaultForkJoinWorkerThreadFactory, null, false,
2363              0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS);
2364     }
2365 
2366     /**
2367      * Creates a {@code ForkJoinPool} with the given parameters (using
2368      * defaults for others -- see {@link #ForkJoinPool(int,
2369      * ForkJoinWorkerThreadFactory, UncaughtExceptionHandler, boolean,
2370      * int, int, int, Predicate, long, TimeUnit)}).
2371      *
2372      * @param parallelism the parallelism level. For default value,
2373      * use {@link java.lang.Runtime#availableProcessors}.
2374      * @param factory the factory for creating new threads. For default value,
2375      * use {@link #defaultForkJoinWorkerThreadFactory}.
2376      * @param handler the handler for internal worker threads that
2377      * terminate due to unrecoverable errors encountered while executing
2378      * tasks. For default value, use {@code null}.
2379      * @param asyncMode if true,
2380      * establishes local first-in-first-out scheduling mode for forked
2381      * tasks that are never joined. This mode may be more appropriate
2382      * than default locally stack-based mode in applications in which
2383      * worker threads only process event-style asynchronous tasks.
2384      * For default value, use {@code false}.
2385      * @throws IllegalArgumentException if parallelism less than or
2386      *         equal to zero, or greater than implementation limit
2387      * @throws NullPointerException if the factory is null
2388      * @throws SecurityException if a security manager exists and
2389      *         the caller is not permitted to modify threads
2390      *         because it does not hold {@link
2391      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2392      */
2393     public ForkJoinPool(int parallelism,
2394                         ForkJoinWorkerThreadFactory factory,
2395                         UncaughtExceptionHandler handler,
2396                         boolean asyncMode) {
2397         this(parallelism, factory, handler, asyncMode,
2398              0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS);
2399     }
2400 
2401     /**
2402      * Creates a {@code ForkJoinPool} with the given parameters.
2403      *
2404      * @param parallelism the parallelism level. For default value,
2405      * use {@link java.lang.Runtime#availableProcessors}.
2406      *
2407      * @param factory the factory for creating new threads. For
2408      * default value, use {@link #defaultForkJoinWorkerThreadFactory}.
2409      *
2410      * @param handler the handler for internal worker threads that
2411      * terminate due to unrecoverable errors encountered while
2412      * executing tasks. For default value, use {@code null}.
2413      *
2414      * @param asyncMode if true, establishes local first-in-first-out
2415      * scheduling mode for forked tasks that are never joined. This
2416      * mode may be more appropriate than default locally stack-based
2417      * mode in applications in which worker threads only process
2418      * event-style asynchronous tasks.  For default value, use {@code
2419      * false}.
2420      *
2421      * @param corePoolSize the number of threads to keep in the pool
2422      * (unless timed out after an elapsed keep-alive). Normally (and
2423      * by default) this is the same value as the parallelism level,
2424      * but may be set to a larger value to reduce dynamic overhead if
2425      * tasks regularly block. Using a smaller value (for example
2426      * {@code 0}) has the same effect as the default.
2427      *
2428      * @param maximumPoolSize the maximum number of threads allowed.
2429      * When the maximum is reached, attempts to replace blocked
2430      * threads fail.  (However, because creation and termination of
2431      * different threads may overlap, and may be managed by the given
2432      * thread factory, this value may be transiently exceeded.)  To
2433      * arrange the same value as is used by default for the common
2434      * pool, use {@code 256} plus the {@code parallelism} level. (By
2435      * default, the common pool allows a maximum of 256 spare
2436      * threads.)  Using a value (for example {@code
2437      * Integer.MAX_VALUE}) larger than the implementation's total
2438      * thread limit has the same effect as using this limit (which is
2439      * the default).
2440      *
2441      * @param minimumRunnable the minimum allowed number of core
2442      * threads not blocked by a join or {@link ManagedBlocker}.  To
2443      * ensure progress, when too few unblocked threads exist and
2444      * unexecuted tasks may exist, new threads are constructed, up to
2445      * the given maximumPoolSize.  For the default value, use {@code
2446      * 1}, that ensures liveness.  A larger value might improve
2447      * throughput in the presence of blocked activities, but might
2448      * not, due to increased overhead.  A value of zero may be
2449      * acceptable when submitted tasks cannot have dependencies
2450      * requiring additional threads.
2451      *
2452      * @param saturate if non-null, a predicate invoked upon attempts
2453      * to create more than the maximum total allowed threads.  By
2454      * default, when a thread is about to block on a join or {@link
2455      * ManagedBlocker}, but cannot be replaced because the
2456      * maximumPoolSize would be exceeded, a {@link
2457      * RejectedExecutionException} is thrown.  But if this predicate
2458      * returns {@code true}, then no exception is thrown, so the pool
2459      * continues to operate with fewer than the target number of
2460      * runnable threads, which might not ensure progress.
2461      *
2462      * @param keepAliveTime the elapsed time since last use before
2463      * a thread is terminated (and then later replaced if needed).
2464      * For the default value, use {@code 60, TimeUnit.SECONDS}.
2465      *
2466      * @param unit the time unit for the {@code keepAliveTime} argument
2467      *
2468      * @throws IllegalArgumentException if parallelism is less than or
2469      *         equal to zero, or is greater than implementation limit,
2470      *         or if maximumPoolSize is less than parallelism,
2471      *         of if the keepAliveTime is less than or equal to zero.
2472      * @throws NullPointerException if the factory is null
2473      * @throws SecurityException if a security manager exists and
2474      *         the caller is not permitted to modify threads
2475      *         because it does not hold {@link
2476      *         java.lang.RuntimePermission}{@code ("modifyThread")}
2477      * @since 9
2478      */
2479     public ForkJoinPool(int parallelism,
2480                         ForkJoinWorkerThreadFactory factory,
2481                         UncaughtExceptionHandler handler,
2482                         boolean asyncMode,
2483                         int corePoolSize,
2484                         int maximumPoolSize,
2485                         int minimumRunnable,
2486                         Predicate<? super ForkJoinPool> saturate,
2487                         long keepAliveTime,
2488                         TimeUnit unit) {
2489         checkPermission();
2490         int p = parallelism;
2491         if (p <= 0 || p > MAX_CAP || p > maximumPoolSize || keepAliveTime <= 0L)
2492             throw new IllegalArgumentException();
2493         if (factory == null || unit == null)
2494             throw new NullPointerException();















2495         this.factory = factory;
2496         this.ueh = handler;
2497         this.saturate = saturate;
2498         this.keepAlive = Math.max(unit.toMillis(keepAliveTime), TIMEOUT_SLOP);
2499         int size = 1 << (33 - Integer.numberOfLeadingZeros(p - 1));
2500         int corep = Math.min(Math.max(corePoolSize, p), MAX_CAP);
2501         int maxSpares = Math.min(maximumPoolSize, MAX_CAP) - p;
2502         int minAvail = Math.min(Math.max(minimumRunnable, 0), MAX_CAP);
2503         this.bounds = ((minAvail - p) & SMASK) | (maxSpares << SWIDTH);
2504         this.mode = p | (asyncMode ? FIFO : 0);
2505         this.ctl = ((((long)(-corep) << TC_SHIFT) & TC_MASK) |
2506                     (((long)(-p)     << RC_SHIFT) & RC_MASK));
2507         this.registrationLock = new ReentrantLock();
2508         this.queues = new WorkQueue[size];
2509         String pid = Integer.toString(getAndAddPoolIds(1) + 1);
2510         this.workerNamePrefix = "ForkJoinPool-" + pid + "-worker-";
2511     }
2512 
2513     // helper method for commonPool constructor
2514     private static Object newInstanceFromSystemProperty(String property)
2515         throws ReflectiveOperationException {
2516         String className = System.getProperty(property);
2517         return (className == null)
2518             ? null
2519             : ClassLoader.getSystemClassLoader().loadClass(className)
2520             .getConstructor().newInstance();
2521     }
2522 
2523     /**
2524      * Constructor for common pool using parameters possibly
2525      * overridden by system properties
2526      */
2527     private ForkJoinPool(byte forCommonPoolOnly) {
2528         int parallelism = Runtime.getRuntime().availableProcessors() - 1;
2529         ForkJoinWorkerThreadFactory fac = null;
2530         UncaughtExceptionHandler handler = null;
2531         try {  // ignore exceptions in accessing/parsing properties




2532             fac = (ForkJoinWorkerThreadFactory) newInstanceFromSystemProperty(
2533                 "java.util.concurrent.ForkJoinPool.common.threadFactory");
2534             handler = (UncaughtExceptionHandler) newInstanceFromSystemProperty(
2535                 "java.util.concurrent.ForkJoinPool.common.exceptionHandler");
2536             String pp = System.getProperty
2537                 ("java.util.concurrent.ForkJoinPool.common.parallelism");
2538             if (pp != null)
2539                 parallelism = Integer.parseInt(pp);
2540         } catch (Exception ignore) {
2541         }
2542         int p = this.mode = Math.min(Math.max(parallelism, 0), MAX_CAP);
2543         int size = 1 << (33 - Integer.numberOfLeadingZeros(p > 0 ? p - 1 : 1));
2544         this.factory = (fac != null) ? fac :
2545             new DefaultCommonPoolForkJoinWorkerThreadFactory();



















2546         this.ueh = handler;

2547         this.keepAlive = DEFAULT_KEEPALIVE;
2548         this.saturate = null;
2549         this.workerNamePrefix = null;
2550         this.bounds = ((1 - p) & SMASK) | (COMMON_MAX_SPARES << SWIDTH);
2551         this.ctl = ((((long)(-p) << TC_SHIFT) & TC_MASK) |
2552                     (((long)(-p) << RC_SHIFT) & RC_MASK));
2553         this.queues = new WorkQueue[size];
2554         this.registrationLock = new ReentrantLock();
2555     }
2556 
2557     /**
2558      * Returns the common pool instance. This pool is statically
2559      * constructed; its run state is unaffected by attempts to {@link
2560      * #shutdown} or {@link #shutdownNow}. However this pool and any
2561      * ongoing processing are automatically terminated upon program
2562      * {@link System#exit}.  Any program that relies on asynchronous
2563      * task processing to complete before program termination should
2564      * invoke {@code commonPool().}{@link #awaitQuiescence awaitQuiescence},
2565      * before exit.
2566      *
2567      * @return the common pool instance
2568      * @since 1.8
2569      */
2570     public static ForkJoinPool commonPool() {
2571         // assert common != null : "static init error";
2572         return common;
2573     }
2574 
2575     // Execution methods
2576 
2577     /**
2578      * Performs the given task, returning its result upon completion.
2579      * If the computation encounters an unchecked Exception or Error,
2580      * it is rethrown as the outcome of this invocation.  Rethrown
2581      * exceptions behave in the same way as regular exceptions, but,
2582      * when possible, contain stack traces (as displayed for example
2583      * using {@code ex.printStackTrace()}) of both the current thread
2584      * as well as the thread actually encountering the exception;
2585      * minimally only the latter.
2586      *
2587      * @param task the task
2588      * @param <T> the type of the task's result
2589      * @return the task's result
2590      * @throws NullPointerException if the task is null
2591      * @throws RejectedExecutionException if the task cannot be
2592      *         scheduled for execution
2593      */
2594     public <T> T invoke(ForkJoinTask<T> task) {


2595         externalSubmit(task);
2596         return task.join();
2597     }
2598 
2599     /**
2600      * Arranges for (asynchronous) execution of the given task.
2601      *
2602      * @param task the task
2603      * @throws NullPointerException if the task is null
2604      * @throws RejectedExecutionException if the task cannot be
2605      *         scheduled for execution
2606      */
2607     public void execute(ForkJoinTask<?> task) {
2608         externalSubmit(task);
2609     }
2610 
2611     // AbstractExecutorService methods
2612 
2613     /**
2614      * @throws NullPointerException if the task is null
2615      * @throws RejectedExecutionException if the task cannot be
2616      *         scheduled for execution
2617      */
2618     @Override
2619     @SuppressWarnings("unchecked")
2620     public void execute(Runnable task) {
2621         externalSubmit((task instanceof ForkJoinTask<?>)
2622                        ? (ForkJoinTask<Void>) task // avoid re-wrap
2623                        : new ForkJoinTask.RunnableExecuteAction(task));





2624     }
2625 
2626     /**
2627      * Submits a ForkJoinTask for execution.
2628      *
2629      * @param task the task to submit
2630      * @param <T> the type of the task's result
2631      * @return the task
2632      * @throws NullPointerException if the task is null
2633      * @throws RejectedExecutionException if the task cannot be
2634      *         scheduled for execution
2635      */
2636     public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
2637         return externalSubmit(task);
2638     }
2639 
2640     /**
2641      * @throws NullPointerException if the task is null
2642      * @throws RejectedExecutionException if the task cannot be
2643      *         scheduled for execution
2644      */
2645     @Override
2646     public <T> ForkJoinTask<T> submit(Callable<T> task) {
2647         return externalSubmit(new ForkJoinTask.AdaptedCallable<T>(task));
2648     }
2649 
2650     /**
2651      * @throws NullPointerException if the task is null
2652      * @throws RejectedExecutionException if the task cannot be
2653      *         scheduled for execution
2654      */
2655     @Override
2656     public <T> ForkJoinTask<T> submit(Runnable task, T result) {
2657         return externalSubmit(new ForkJoinTask.AdaptedRunnable<T>(task, result));
2658     }
2659 
2660     /**
2661      * @throws NullPointerException if the task is null
2662      * @throws RejectedExecutionException if the task cannot be
2663      *         scheduled for execution
2664      */
2665     @Override
2666     @SuppressWarnings("unchecked")
2667     public ForkJoinTask<?> submit(Runnable task) {


2668         return externalSubmit((task instanceof ForkJoinTask<?>)
2669             ? (ForkJoinTask<Void>) task // avoid re-wrap
2670             : new ForkJoinTask.AdaptedRunnableAction(task));
2671     }
2672 
2673     /**
2674      * @throws NullPointerException       {@inheritDoc}
2675      * @throws RejectedExecutionException {@inheritDoc}
2676      */
2677     @Override
2678     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {



2679         ArrayList<Future<T>> futures = new ArrayList<>(tasks.size());

2680         try {
2681             for (Callable<T> t : tasks) {
2682                 ForkJoinTask<T> f =
2683                     new ForkJoinTask.AdaptedInterruptibleCallable<T>(t);
2684                 futures.add(f);
2685                 externalSubmit(f);
2686             }
2687             for (int i = futures.size() - 1; i >= 0; --i)
2688                 ((ForkJoinTask<?>)futures.get(i)).quietlyJoin();
2689             return futures;
2690         } catch (Throwable t) {
2691             for (Future<T> e : futures)
2692                 ForkJoinTask.cancelIgnoringExceptions(e);
2693             throw t;
2694         }
2695     }
2696 
2697     @Override
2698     public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
2699                                          long timeout, TimeUnit unit)
2700         throws InterruptedException {
2701         long nanos = unit.toNanos(timeout);
2702         ArrayList<Future<T>> futures = new ArrayList<>(tasks.size());
2703         try {
2704             for (Callable<T> t : tasks) {
2705                 ForkJoinTask<T> f =
2706                     new ForkJoinTask.AdaptedInterruptibleCallable<T>(t);
2707                 futures.add(f);
2708                 externalSubmit(f);
2709             }
2710             long startTime = System.nanoTime(), ns = nanos;
2711             boolean timedOut = (ns < 0L);
2712             for (int i = futures.size() - 1; i >= 0; --i) {
2713                 Future<T> f = futures.get(i);
2714                 if (!f.isDone()) {
2715                     if (timedOut)
2716                         ForkJoinTask.cancelIgnoringExceptions(f);
2717                     else {
2718                         try {
2719                             f.get(ns, TimeUnit.NANOSECONDS);
2720                         } catch (CancellationException | TimeoutException |
2721                                  ExecutionException ok) {
2722                         }
2723                         if ((ns = nanos - (System.nanoTime() - startTime)) < 0L)
2724                             timedOut = true;
2725                     }
2726                 }
2727             }
2728             return futures;
2729         } catch (Throwable t) {
2730             for (Future<T> e : futures)
2731                 ForkJoinTask.cancelIgnoringExceptions(e);
2732             throw t;
2733         }
2734     }
2735 
2736     // Task to hold results from InvokeAnyTasks
2737     static final class InvokeAnyRoot<E> extends ForkJoinTask<E> {
2738         private static final long serialVersionUID = 2838392045355241008L;
2739         @SuppressWarnings("serial") // Conditionally serializable
2740         volatile E result;
2741         final AtomicInteger count;  // in case all throw
2742         final ForkJoinPool pool;    // to check shutdown while collecting
2743         InvokeAnyRoot(int n, ForkJoinPool p) {
2744             pool = p;
2745             count = new AtomicInteger(n);
2746         }
2747         final void tryComplete(Callable<E> c) { // called by InvokeAnyTasks
2748             Throwable ex = null;
2749             boolean failed = (c == null || isCancelled() ||
2750                               (pool != null && pool.mode < 0));
2751             if (!failed && !isDone()) {
2752                 try {
2753                     complete(c.call());
2754                 } catch (Throwable tx) {
2755                     ex = tx;
2756                     failed = true;
2757                 }
2758             }
2759             if ((pool != null && pool.mode < 0) ||
2760                 (failed && count.getAndDecrement() <= 1))
2761                 trySetThrown(ex != null ? ex : new CancellationException());
2762         }
2763         public final boolean exec()         { return false; } // never forked
2764         public final E getRawResult()       { return result; }
2765         public final void setRawResult(E v) { result = v; }
2766     }
2767 
2768     // Variant of AdaptedInterruptibleCallable with results in InvokeAnyRoot
2769     static final class InvokeAnyTask<E> extends ForkJoinTask<E> {
2770         private static final long serialVersionUID = 2838392045355241008L;
2771         final InvokeAnyRoot<E> root;
2772         @SuppressWarnings("serial") // Conditionally serializable
2773         final Callable<E> callable;
2774         transient volatile Thread runner;
2775         InvokeAnyTask(InvokeAnyRoot<E> root, Callable<E> callable) {
2776             this.root = root;
2777             this.callable = callable;
2778         }
2779         public final boolean exec() {
2780             Thread.interrupted();
2781             runner = Thread.currentThread();
2782             root.tryComplete(callable);
2783             runner = null;
2784             Thread.interrupted();
2785             return true;
2786         }
2787         public final boolean cancel(boolean mayInterruptIfRunning) {
2788             Thread t;
2789             boolean stat = super.cancel(false);
2790             if (mayInterruptIfRunning && (t = runner) != null) {
2791                 try {
2792                     t.interrupt();
2793                 } catch (Throwable ignore) {
2794                 }
2795             }
2796             return stat;
2797         }
2798         public final void setRawResult(E v) {} // unused
2799         public final E getRawResult()       { return null; }
2800     }
2801 
2802     @Override
2803     public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
2804         throws InterruptedException, ExecutionException {
2805         int n = tasks.size();
2806         if (n <= 0)
2807             throw new IllegalArgumentException();
2808         InvokeAnyRoot<T> root = new InvokeAnyRoot<T>(n, this);
2809         ArrayList<InvokeAnyTask<T>> fs = new ArrayList<>(n);
2810         try {
2811             for (Callable<T> c : tasks) {
2812                 if (c == null)
2813                     throw new NullPointerException();
2814                 InvokeAnyTask<T> f = new InvokeAnyTask<T>(root, c);
2815                 fs.add(f);
2816                 externalSubmit(f);
2817                 if (root.isDone())
2818                     break;
2819             }
2820             return root.get();
2821         } finally {
2822             for (InvokeAnyTask<T> f : fs)
2823                 ForkJoinTask.cancelIgnoringExceptions(f);
2824         }
2825     }
2826 
2827     @Override
2828     public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
2829                            long timeout, TimeUnit unit)
2830         throws InterruptedException, ExecutionException, TimeoutException {
2831         long nanos = unit.toNanos(timeout);
2832         int n = tasks.size();
2833         if (n <= 0)
2834             throw new IllegalArgumentException();
2835         InvokeAnyRoot<T> root = new InvokeAnyRoot<T>(n, this);
2836         ArrayList<InvokeAnyTask<T>> fs = new ArrayList<>(n);
2837         try {
2838             for (Callable<T> c : tasks) {
2839                 if (c == null)
2840                     throw new NullPointerException();
2841                 InvokeAnyTask<T> f = new InvokeAnyTask<T>(root, c);
2842                 fs.add(f);
2843                 externalSubmit(f);
2844                 if (root.isDone())
2845                     break;
2846             }
2847             return root.get(nanos, TimeUnit.NANOSECONDS);
2848         } finally {
2849             for (InvokeAnyTask<T> f : fs)
2850                 ForkJoinTask.cancelIgnoringExceptions(f);
2851         }
2852     }
2853 
2854     /**
2855      * Returns the factory used for constructing new workers.
2856      *
2857      * @return the factory used for constructing new workers
2858      */
2859     public ForkJoinWorkerThreadFactory getFactory() {
2860         return factory;
2861     }
2862 
2863     /**
2864      * Returns the handler for internal worker threads that terminate
2865      * due to unrecoverable errors encountered while executing tasks.
2866      *
2867      * @return the handler, or {@code null} if none
2868      */
2869     public UncaughtExceptionHandler getUncaughtExceptionHandler() {
2870         return ueh;
2871     }
2872 
2873     /**
2874      * Returns the targeted parallelism level of this pool.
2875      *
2876      * @return the targeted parallelism level of this pool
2877      */
2878     public int getParallelism() {
2879         int par = mode & SMASK;
2880         return (par > 0) ? par : 1;
2881     }
2882 
2883     /**
2884      * Returns the targeted parallelism level of the common pool.
2885      *
2886      * @return the targeted parallelism level of the common pool
2887      * @since 1.8
2888      */
2889     public static int getCommonPoolParallelism() {
2890         return COMMON_PARALLELISM;
2891     }
2892 
2893     /**
2894      * Returns the number of worker threads that have started but not
2895      * yet terminated.  The result returned by this method may differ
2896      * from {@link #getParallelism} when threads are created to
2897      * maintain parallelism when others are cooperatively blocked.
2898      *
2899      * @return the number of worker threads
2900      */
2901     public int getPoolSize() {
2902         return ((mode & SMASK) + (short)(ctl >>> TC_SHIFT));
2903     }
2904 
2905     /**
2906      * Returns {@code true} if this pool uses local first-in-first-out
2907      * scheduling mode for forked tasks that are never joined.
2908      *
2909      * @return {@code true} if this pool uses async mode
2910      */
2911     public boolean getAsyncMode() {
2912         return (mode & FIFO) != 0;
2913     }
2914 
2915     /**
2916      * Returns an estimate of the number of worker threads that are
2917      * not blocked waiting to join tasks or for other managed
2918      * synchronization. This method may overestimate the
2919      * number of running threads.
2920      *
2921      * @return the number of worker threads
2922      */
2923     public int getRunningThreadCount() {

2924         VarHandle.acquireFence();
2925         WorkQueue[] qs; WorkQueue q;
2926         int rc = 0;
2927         if ((qs = queues) != null) {
2928             for (int i = 1; i < qs.length; i += 2) {
2929                 if ((q = qs[i]) != null && q.isApparentlyUnblocked())
2930                     ++rc;
2931             }
2932         }
2933         return rc;
2934     }
2935 
2936     /**
2937      * Returns an estimate of the number of threads that are currently
2938      * stealing or executing tasks. This method may overestimate the
2939      * number of active threads.
2940      *
2941      * @return the number of active threads
2942      */
2943     public int getActiveThreadCount() {
2944         int r = (mode & SMASK) + (int)(ctl >> RC_SHIFT);
2945         return (r <= 0) ? 0 : r; // suppress momentarily negative values
2946     }
2947 
2948     /**
2949      * Returns {@code true} if all worker threads are currently idle.
2950      * An idle worker is one that cannot obtain a task to execute
2951      * because none are available to steal from other threads, and
2952      * there are no pending submissions to the pool. This method is
2953      * conservative; it might not return {@code true} immediately upon
2954      * idleness of all threads, but will eventually become true if
2955      * threads remain inactive.
2956      *
2957      * @return {@code true} if all threads are currently idle
2958      */
2959     public boolean isQuiescent() {
2960         return canStop();























2961     }
2962 
2963     /**
2964      * Returns an estimate of the total number of completed tasks that
2965      * were executed by a thread other than their submitter. The
2966      * reported value underestimates the actual total number of steals
2967      * when the pool is not quiescent. This value may be useful for
2968      * monitoring and tuning fork/join programs: in general, steal
2969      * counts should be high enough to keep threads busy, but low
2970      * enough to avoid overhead and contention across threads.
2971      *
2972      * @return the number of steals
2973      */
2974     public long getStealCount() {
2975         long count = stealCount;
2976         WorkQueue[] qs; WorkQueue q;
2977         if ((qs = queues) != null) {
2978             for (int i = 1; i < qs.length; i += 2) {
2979                 if ((q = qs[i]) != null)
2980                     count += (long)q.nsteals & 0xffffffffL;
2981             }
2982         }
2983         return count;
2984     }
2985 
2986     /**
2987      * Returns an estimate of the total number of tasks currently held
2988      * in queues by worker threads (but not including tasks submitted
2989      * to the pool that have not begun executing). This value is only
2990      * an approximation, obtained by iterating across all threads in
2991      * the pool. This method may be useful for tuning task
2992      * granularities.
2993      *
2994      * @return the number of queued tasks
2995      */
2996     public long getQueuedTaskCount() {

2997         VarHandle.acquireFence();
2998         WorkQueue[] qs; WorkQueue q;
2999         int count = 0;
3000         if ((qs = queues) != null) {
3001             for (int i = 1; i < qs.length; i += 2) {
3002                 if ((q = qs[i]) != null)
3003                     count += q.queueSize();
3004             }
3005         }
3006         return count;
3007     }
3008 
3009     /**
3010      * Returns an estimate of the number of tasks submitted to this
3011      * pool that have not yet begun executing.  This method may take
3012      * time proportional to the number of submissions.
3013      *
3014      * @return the number of queued submissions
3015      */
3016     public int getQueuedSubmissionCount() {

3017         VarHandle.acquireFence();
3018         WorkQueue[] qs; WorkQueue q;
3019         int count = 0;
3020         if ((qs = queues) != null) {
3021             for (int i = 0; i < qs.length; i += 2) {
3022                 if ((q = qs[i]) != null)
3023                     count += q.queueSize();
3024             }
3025         }
3026         return count;
3027     }
3028 
3029     /**
3030      * Returns {@code true} if there are any tasks submitted to this
3031      * pool that have not yet begun executing.
3032      *
3033      * @return {@code true} if there are any queued submissions
3034      */
3035     public boolean hasQueuedSubmissions() {

3036         VarHandle.acquireFence();
3037         WorkQueue[] qs; WorkQueue q;
3038         if ((qs = queues) != null) {
3039             for (int i = 0; i < qs.length; i += 2) {
3040                 if ((q = qs[i]) != null && !q.isEmpty())
3041                     return true;
3042             }
3043         }
3044         return false;
3045     }
3046 
3047     /**
3048      * Removes and returns the next unexecuted submission if one is
3049      * available.  This method may be useful in extensions to this
3050      * class that re-assign work in systems with multiple pools.
3051      *
3052      * @return the next submission, or {@code null} if none
3053      */
3054     protected ForkJoinTask<?> pollSubmission() {
3055         return pollScan(true);
3056     }
3057 
3058     /**
3059      * Removes all available unexecuted submitted and forked tasks
3060      * from scheduling queues and adds them to the given collection,
3061      * without altering their execution status. These may include
3062      * artificially generated or wrapped tasks. This method is
3063      * designed to be invoked only when the pool is known to be
3064      * quiescent. Invocations at other times may not remove all
3065      * tasks. A failure encountered while attempting to add elements
3066      * to collection {@code c} may result in elements being in
3067      * neither, either or both collections when the associated
3068      * exception is thrown.  The behavior of this operation is
3069      * undefined if the specified collection is modified while the
3070      * operation is in progress.
3071      *
3072      * @param c the collection to transfer elements into
3073      * @return the number of elements transferred
3074      */
3075     protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {


3076         int count = 0;
3077         for (ForkJoinTask<?> t; (t = pollScan(false)) != null; ) {



3078             c.add(t);
3079             ++count;
3080         }



3081         return count;
3082     }
3083 
3084     /**
3085      * Returns a string identifying this pool, as well as its state,
3086      * including indications of run state, parallelism level, and
3087      * worker and task counts.
3088      *
3089      * @return a string identifying this pool, as well as its state
3090      */
3091     public String toString() {
3092         // Use a single pass through queues to collect counts
3093         int md = mode; // read volatile fields first
3094         long c = ctl;
3095         long st = stealCount;
3096         long qt = 0L, ss = 0L; int rc = 0;
3097         WorkQueue[] qs; WorkQueue q;
3098         if ((qs = queues) != null) {
3099             for (int i = 0; i < qs.length; ++i) {
3100                 if ((q = qs[i]) != null) {
3101                     int size = q.queueSize();
3102                     if ((i & 1) == 0)
3103                         ss += size;
3104                     else {
3105                         qt += size;
3106                         st += (long)q.nsteals & 0xffffffffL;
3107                         if (q.isApparentlyUnblocked())
3108                             ++rc;
3109                     }
3110                 }
3111             }
3112         }
3113 
3114         int pc = (md & SMASK);
3115         int tc = pc + (short)(c >>> TC_SHIFT);
3116         int ac = pc + (int)(c >> RC_SHIFT);
3117         if (ac < 0) // ignore transient negative
3118             ac = 0;
3119         String level = ((md & TERMINATED) != 0 ? "Terminated" :
3120                         (md & STOP)       != 0 ? "Terminating" :
3121                         (md & SHUTDOWN)   != 0 ? "Shutting down" :
3122                         "Running");
3123         return super.toString() +
3124             "[" + level +
3125             ", parallelism = " + pc +
3126             ", size = " + tc +
3127             ", active = " + ac +
3128             ", running = " + rc +
3129             ", steals = " + st +
3130             ", tasks = " + qt +
3131             ", submissions = " + ss +
3132             "]";
3133     }
3134 
3135     /**
3136      * Possibly initiates an orderly shutdown in which previously
3137      * submitted tasks are executed, but no new tasks will be
3138      * accepted. Invocation has no effect on execution state if this
3139      * is the {@link #commonPool()}, and no additional effect if
3140      * already shut down.  Tasks that are in the process of being
3141      * submitted concurrently during the course of this method may or
3142      * may not be rejected.
3143      *
3144      * @throws SecurityException if a security manager exists and
3145      *         the caller is not permitted to modify threads
3146      *         because it does not hold {@link
3147      *         java.lang.RuntimePermission}{@code ("modifyThread")}
3148      */
3149     public void shutdown() {
3150         checkPermission();
3151         if (this != common)
3152             tryTerminate(false, true);
3153     }
3154 
3155     /**
3156      * Possibly attempts to cancel and/or stop all tasks, and reject
3157      * all subsequently submitted tasks.  Invocation has no effect on
3158      * execution state if this is the {@link #commonPool()}, and no
3159      * additional effect if already shut down. Otherwise, tasks that
3160      * are in the process of being submitted or executed concurrently
3161      * during the course of this method may or may not be
3162      * rejected. This method cancels both existing and unexecuted
3163      * tasks, in order to permit termination in the presence of task
3164      * dependencies. So the method always returns an empty list
3165      * (unlike the case for some other Executors).
3166      *
3167      * @return an empty list
3168      * @throws SecurityException if a security manager exists and
3169      *         the caller is not permitted to modify threads
3170      *         because it does not hold {@link
3171      *         java.lang.RuntimePermission}{@code ("modifyThread")}
3172      */
3173     public List<Runnable> shutdownNow() {
3174         checkPermission();
3175         if (this != common)
3176             tryTerminate(true, true);
3177         return Collections.emptyList();
3178     }
3179 
3180     /**
3181      * Returns {@code true} if all tasks have completed following shut down.
3182      *
3183      * @return {@code true} if all tasks have completed following shut down
3184      */
3185     public boolean isTerminated() {
3186         return (mode & TERMINATED) != 0;
3187     }
3188 
3189     /**
3190      * Returns {@code true} if the process of termination has
3191      * commenced but not yet completed.  This method may be useful for
3192      * debugging. A return of {@code true} reported a sufficient
3193      * period after shutdown may indicate that submitted tasks have
3194      * ignored or suppressed interruption, or are waiting for I/O,
3195      * causing this executor not to properly terminate. (See the
3196      * advisory notes for class {@link ForkJoinTask} stating that
3197      * tasks should not normally entail blocking operations.  But if
3198      * they do, they must abort them on interrupt.)
3199      *
3200      * @return {@code true} if terminating but not yet terminated
3201      */
3202     public boolean isTerminating() {
3203         return (mode & (STOP | TERMINATED)) == STOP;

3204     }
3205 
3206     /**
3207      * Returns {@code true} if this pool has been shut down.
3208      *
3209      * @return {@code true} if this pool has been shut down
3210      */
3211     public boolean isShutdown() {
3212         return (mode & SHUTDOWN) != 0;
3213     }
3214 
3215     /**
3216      * Blocks until all tasks have completed execution after a
3217      * shutdown request, or the timeout occurs, or the current thread
3218      * is interrupted, whichever happens first. Because the {@link
3219      * #commonPool()} never terminates until program shutdown, when
3220      * applied to the common pool, this method is equivalent to {@link
3221      * #awaitQuiescence(long, TimeUnit)} but always returns {@code false}.
3222      *
3223      * @param timeout the maximum time to wait
3224      * @param unit the time unit of the timeout argument
3225      * @return {@code true} if this executor terminated and
3226      *         {@code false} if the timeout elapsed before termination
3227      * @throws InterruptedException if interrupted while waiting
3228      */
3229     public boolean awaitTermination(long timeout, TimeUnit unit)
3230         throws InterruptedException {
3231         ReentrantLock lock; Condition cond;
3232         long nanos = unit.toNanos(timeout);
3233         boolean terminated = false;
3234         if (this == common) {
3235             Thread t; ForkJoinWorkerThread wt; int q;
3236             if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread &&
3237                 (wt = (ForkJoinWorkerThread)t).pool == this)
3238                 q = helpQuiescePool(wt.workQueue, nanos, true);
3239             else
3240                 q = externalHelpQuiescePool(nanos, true);
3241             if (q < 0)
3242                 throw new InterruptedException();
3243         }
3244         else if (!(terminated = ((mode & TERMINATED) != 0)) &&
3245                  (lock = registrationLock) != null) {
3246             lock.lock();
3247             try {
3248                 if ((cond = termination) == null)
3249                     termination = cond = lock.newCondition();
3250                 while (!(terminated = ((mode & TERMINATED) != 0)) && nanos > 0L)
3251                     nanos = cond.awaitNanos(nanos);
3252             } finally {
3253                 lock.unlock();





3254             }
3255         }
3256         return terminated;
3257     }
3258 
3259     /**
3260      * If called by a ForkJoinTask operating in this pool, equivalent
3261      * in effect to {@link ForkJoinTask#helpQuiesce}. Otherwise,
3262      * waits and/or attempts to assist performing tasks until this
3263      * pool {@link #isQuiescent} or the indicated timeout elapses.
3264      *
3265      * @param timeout the maximum time to wait
3266      * @param unit the time unit of the timeout argument
3267      * @return {@code true} if quiescent; {@code false} if the
3268      * timeout elapsed.
3269      */
3270     public boolean awaitQuiescence(long timeout, TimeUnit unit) {
3271         Thread t; ForkJoinWorkerThread wt; int q;
3272         long nanos = unit.toNanos(timeout);
3273         if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread &&
3274             (wt = (ForkJoinWorkerThread)t).pool == this)
3275             q = helpQuiescePool(wt.workQueue, nanos, false);













3276         else
3277             q = externalHelpQuiescePool(nanos, false);
3278         return (q > 0);









3279     }
3280 
3281     /**
3282      * Interface for extending managed parallelism for tasks running
3283      * in {@link ForkJoinPool}s.
3284      *
3285      * <p>A {@code ManagedBlocker} provides two methods.  Method
3286      * {@link #isReleasable} must return {@code true} if blocking is
3287      * not necessary. Method {@link #block} blocks the current thread
3288      * if necessary (perhaps internally invoking {@code isReleasable}
3289      * before actually blocking). These actions are performed by any
3290      * thread invoking {@link
3291      * ForkJoinPool#managedBlock(ManagedBlocker)}.  The unusual
3292      * methods in this API accommodate synchronizers that may, but
3293      * don't usually, block for long periods. Similarly, they allow
3294      * more efficient internal handling of cases in which additional
3295      * workers may be, but usually are not, needed to ensure
3296      * sufficient parallelism.  Toward this end, implementations of
3297      * method {@code isReleasable} must be amenable to repeated
3298      * invocation. Neither method is invoked after a prior invocation
3299      * of {@code isReleasable} or {@code block} returns {@code true}.
3300      *
3301      * <p>For example, here is a ManagedBlocker based on a
3302      * ReentrantLock:
3303      * <pre> {@code
3304      * class ManagedLocker implements ManagedBlocker {
3305      *   final ReentrantLock lock;
3306      *   boolean hasLock = false;
3307      *   ManagedLocker(ReentrantLock lock) { this.lock = lock; }
3308      *   public boolean block() {
3309      *     if (!hasLock)
3310      *       lock.lock();
3311      *     return true;
3312      *   }
3313      *   public boolean isReleasable() {
3314      *     return hasLock || (hasLock = lock.tryLock());
3315      *   }
3316      * }}</pre>
3317      *
3318      * <p>Here is a class that possibly blocks waiting for an
3319      * item on a given queue:
3320      * <pre> {@code
3321      * class QueueTaker<E> implements ManagedBlocker {
3322      *   final BlockingQueue<E> queue;
3323      *   volatile E item = null;
3324      *   QueueTaker(BlockingQueue<E> q) { this.queue = q; }
3325      *   public boolean block() throws InterruptedException {
3326      *     if (item == null)
3327      *       item = queue.take();
3328      *     return true;
3329      *   }
3330      *   public boolean isReleasable() {
3331      *     return item != null || (item = queue.poll()) != null;
3332      *   }
3333      *   public E getItem() { // call after pool.managedBlock completes
3334      *     return item;
3335      *   }
3336      * }}</pre>
3337      */
3338     public static interface ManagedBlocker {
3339         /**
3340          * Possibly blocks the current thread, for example waiting for
3341          * a lock or condition.
3342          *
3343          * @return {@code true} if no additional blocking is necessary
3344          * (i.e., if isReleasable would return true)
3345          * @throws InterruptedException if interrupted while waiting
3346          * (the method is not required to do so, but is allowed to)
3347          */
3348         boolean block() throws InterruptedException;
3349 
3350         /**
3351          * Returns {@code true} if blocking is unnecessary.
3352          * @return {@code true} if blocking is unnecessary
3353          */
3354         boolean isReleasable();
3355     }
3356 
3357     /**
3358      * Runs the given possibly blocking task.  When {@linkplain
3359      * ForkJoinTask#inForkJoinPool() running in a ForkJoinPool}, this
3360      * method possibly arranges for a spare thread to be activated if
3361      * necessary to ensure sufficient parallelism while the current
3362      * thread is blocked in {@link ManagedBlocker#block blocker.block()}.
3363      *
3364      * <p>This method repeatedly calls {@code blocker.isReleasable()} and
3365      * {@code blocker.block()} until either method returns {@code true}.
3366      * Every call to {@code blocker.block()} is preceded by a call to
3367      * {@code blocker.isReleasable()} that returned {@code false}.
3368      *
3369      * <p>If not running in a ForkJoinPool, this method is
3370      * behaviorally equivalent to
3371      * <pre> {@code
3372      * while (!blocker.isReleasable())
3373      *   if (blocker.block())
3374      *     break;}</pre>
3375      *
3376      * If running in a ForkJoinPool, the pool may first be expanded to
3377      * ensure sufficient parallelism available during the call to
3378      * {@code blocker.block()}.
3379      *
3380      * @param blocker the blocker task
3381      * @throws InterruptedException if {@code blocker.block()} did so
3382      */
3383     public static void managedBlock(ManagedBlocker blocker)
3384         throws InterruptedException {
3385         Thread t; ForkJoinPool p;
3386         if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread &&
3387             (p = ((ForkJoinWorkerThread)t).pool) != null)
3388             p.compensatedBlock(blocker);
3389         else
3390             unmanagedBlock(blocker);
3391     }
3392 
3393     /** ManagedBlock for ForkJoinWorkerThreads */
3394     private void compensatedBlock(ManagedBlocker blocker)
3395         throws InterruptedException {
3396         if (blocker == null) throw new NullPointerException();
3397         for (;;) {
3398             int comp; boolean done;
3399             long c = ctl;
3400             if (blocker.isReleasable())
3401                 break;
3402             if ((comp = tryCompensate(c)) >= 0) {
3403                 long post = (comp == 0) ? 0L : RC_UNIT;



3404                 try {
3405                     done = blocker.block();

3406                 } finally {
3407                     getAndAddCtl(post);
3408                 }
3409                 if (done)
3410                     break;
3411             }
3412         }
3413     }





3414 
3415     /** ManagedBlock for external threads */
3416     private static void unmanagedBlock(ManagedBlocker blocker)
3417         throws InterruptedException {
3418         if (blocker == null) throw new NullPointerException();
3419         do {} while (!blocker.isReleasable() && !blocker.block());
















3420     }
3421 
3422     // AbstractExecutorService.newTaskFor overrides rely on
3423     // undocumented fact that ForkJoinTask.adapt returns ForkJoinTasks
3424     // that also implement RunnableFuture.
3425 
3426     @Override
3427     protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
3428         return new ForkJoinTask.AdaptedRunnable<T>(runnable, value);
3429     }
3430 
3431     @Override
3432     protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
3433         return new ForkJoinTask.AdaptedCallable<T>(callable);
3434     }
3435 





3436     static {
3437         try {
3438             MethodHandles.Lookup l = MethodHandles.lookup();
3439             CTL = l.findVarHandle(ForkJoinPool.class, "ctl", long.class);
3440             MODE = l.findVarHandle(ForkJoinPool.class, "mode", int.class);
3441             THREADIDS = l.findVarHandle(ForkJoinPool.class, "threadIds", int.class);
3442             POOLIDS = l.findStaticVarHandle(ForkJoinPool.class, "poolIds", int.class);
3443         } catch (ReflectiveOperationException e) {
3444             throw new ExceptionInInitializerError(e);
3445         }
3446 
3447         // Reduce the risk of rare disastrous classloading in first call to
3448         // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
3449         Class<?> ensureLoaded = LockSupport.class;
3450 
3451         int commonMaxSpares = DEFAULT_COMMON_MAX_SPARES;
3452         try {
3453             String p = System.getProperty
3454                 ("java.util.concurrent.ForkJoinPool.common.maximumSpares");
3455             if (p != null)
3456                 commonMaxSpares = Integer.parseInt(p);
3457         } catch (Exception ignore) {}
3458         COMMON_MAX_SPARES = commonMaxSpares;
3459 
3460         defaultForkJoinWorkerThreadFactory =
3461             new DefaultForkJoinWorkerThreadFactory();
3462         modifyThreadPermission = new RuntimePermission("modifyThread");

3463         common = AccessController.doPrivileged(new PrivilegedAction<>() {
3464             public ForkJoinPool run() {
3465                 return new ForkJoinPool((byte)0); }});
3466 
3467         COMMON_PARALLELISM = Math.max(common.mode & SMASK, 1);



























3468     }
3469 }
--- EOF ---