alluxio-master-2.x/core/common/src/main/java/alluxio/concurrent/jsr/ForkJoinPool.java
/*
 * Written by Doug Lea with assistance from members of JCP JSR-166 Expert Group and released to the
 * public domain, as explained at http://creativecommons.org/publicdomain/zero/1.0/
 */
package alluxio.concurrent.jsr;


import java.lang.Thread.UncaughtExceptionHandler;
import java.security.AccessControlContext;
import java.security.AccessController;
import java.security.Permission;
import java.security.Permissions;
import java.security.PrivilegedAction;
import java.security.ProtectionDomain;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RunnableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Predicate;

/**
 * An {@link ExecutorService} for running {@link ForkJoinTask}s. A {@code ForkJoinPool} provides the
 * entry point for submissions from non-{@code ForkJoinTask} clients, as well as management and
 * monitoring operations.
 *
 * <p>
 * A {@code ForkJoinPool} differs from other kinds of {@link ExecutorService} mainly by virtue of
 * employing <em>work-stealing</em>: all threads in the pool attempt to find and execute tasks
 * submitted to the pool and/or created by other active tasks (eventually blocking waiting for work
 * if none exist). This enables efficient processing when most tasks spawn other subtasks (as do
 * most {@code ForkJoinTask}s), as well as when many small tasks are submitted to the pool from
 * external clients. Especially when setting <em>asyncMode</em> to true in constructors, {@code
 * ForkJoinPool}s may also be appropriate for use with event-style tasks that are never joined. All
 * worker threads are initialized with {@link Thread#isDaemon} set {@code true}.
 *
 * <p>
 * A static {@link #commonPool()} is available and appropriate for most applications. The common
 * pool is used by any ForkJoinTask that is not explicitly submitted to a specified pool. Using the
 * common pool normally reduces resource usage (its threads are slowly reclaimed during periods of
 * non-use, and reinstated upon subsequent use).
 *
 * <p>
 * For applications that require separate or custom pools, a {@code
 * ForkJoinPool} may be constructed with a given target parallelism level; by default, equal to the
 * number of available processors. The pool attempts to maintain enough active (or available)
 * threads by dynamically adding, suspending, or resuming internal worker threads, even if some
 * tasks are stalled waiting to join others. However, no such adjustments are guaranteed in the face
 * of blocked I/O or other unmanaged synchronization. The nested {@link ManagedBlocker} interface
 * enables extension of the kinds of synchronization accommodated. The default policies may be
 * overridden using a constructor with parameters corresponding to those documented in class
 * {@link ThreadPoolExecutor}.
 *
 * <p>
 * In addition to execution and lifecycle control methods, this class provides status check methods
 * (for example {@link #getStealCount}) that are intended to aid in developing, tuning, and
 * monitoring fork/join applications. Also, method {@link #toString} returns indications of pool
 * state in a convenient form for informal monitoring.
 *
 * <p>
 * As is the case with other ExecutorServices, there are three main task execution methods
 * summarized in the following table. These are designed to be used primarily by clients not already
 * engaged in fork/join computations in the current pool. The main forms of these methods accept
 * instances of {@code ForkJoinTask}, but overloaded forms also allow mixed execution of plain
 * {@code
 * Runnable}- or {@code Callable}- based activities as well. However, tasks that are already
 * executing in a pool should normally instead use the within-computation forms listed in the table
 * unless using async event-style tasks that are not usually joined, in which case there is little
 * difference among choice of methods.
 *
 * <table BORDER CELLPADDING=3 CELLSPACING=1> <caption>Summary of task execution methods</caption>
 * <tr>
 * <td></td>
 * <th scope="col">Call from non-fork/join clients</th>
 * <th scope="col">Call from within fork/join computations</th>
 * </tr>
 * <tr>
 * <th scope="row" style="text-align:left">Arrange async execution</th>
 * <td>{@link #execute(ForkJoinTask)}</td>
 * <td>{@link ForkJoinTask#fork}</td>
 * </tr>
 * <tr>
 * <th scope="row" style="text-align:left">Await and obtain result</th>
 * <td>{@link #invoke(ForkJoinTask)}</td>
 * <td>{@link ForkJoinTask#invoke}</td>
 * </tr>
 * <tr>
 * <th scope="row" style="text-align:left">Arrange exec and obtain Future</th>
 * <td>{@link #submit(ForkJoinTask)}</td>
 * <td>{@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td>
 * </tr>
 * </table>
 *
 * <p>
 * The parameters used to construct the common pool may be controlled by setting the following
 * {@linkplain System#getProperty system properties}:
 * <ul>
 * <li>{@code java.util.concurrent.ForkJoinPool.common.parallelism} - the parallelism level, a
 * non-negative integer
 * <li>{@code java.util.concurrent.ForkJoinPool.common.threadFactory} - the class name of a
 * {@link ForkJoinWorkerThreadFactory}. The {@linkplain ClassLoader#getSystemClassLoader() system
 * class loader} is used to load this class.
 * <li>{@code java.util.concurrent.ForkJoinPool.common.exceptionHandler} - the class name of a
 * {@link UncaughtExceptionHandler}. The {@linkplain ClassLoader#getSystemClassLoader() system class
 * loader} is used to load this class.
 * <li>{@code java.util.concurrent.ForkJoinPool.common.maximumSpares} - the maximum number of
 * allowed extra threads to maintain target parallelism (default 256).
 * </ul>
 * If no thread factory is supplied via a system property, then the common pool uses a factory that
 * uses the system class loader as the {@linkplain Thread#getContextClassLoader() thread context
 * class loader}. In addition, if a {@link SecurityManager} is present, then the common pool uses a
 * factory supplying threads that have no {@link Permissions} enabled.
 *
 * Upon any error in establishing these settings, default parameters are used. It is possible to
 * disable or limit the use of threads in the common pool by setting the parallelism property to
 * zero, and/or using a factory that may return {@code null}. However doing so may cause unjoined
 * tasks to never be executed.
 *
 * <p>
 * <b>Implementation Note:</b> This implementation restricts the maximum number of running threads
 * to 32767. Attempts to create pools with greater than the maximum number result in
 * {@code IllegalArgumentException}.
 *
 * <p>
 * This implementation rejects submitted tasks (that is, by throwing
 * {@link RejectedExecutionException}) only when the pool is shut down or internal resources have
 * been exhausted.
 *
 * @since 1.7
 * @author Doug Lea
 */
public class ForkJoinPool extends AbstractExecutorService {
  // CVS rev. 1.344
  /*
   * Implementation Overview
   *
   * This class and its nested classes provide the main functionality and control for a set of
   * worker threads: Submissions from non-FJ threads enter into submission queues. Workers take
   * these tasks and typically split them into subtasks that may be stolen by other workers.
   * Preference rules give first priority to processing tasks from their own queues (LIFO or FIFO,
   * depending on mode), then to randomized FIFO steals of tasks in other queues. This framework
   * began as vehicle for supporting tree-structured parallelism using work-stealing. Over time, its
   * scalability advantages led to extensions and changes to better support more diverse usage
   * contexts. Because most internal methods and nested classes are interrelated, their main
   * rationale and descriptions are presented here; individual methods and nested classes contain
   * only brief comments about details.
   *
   * WorkQueues ==========
   *
   * Most operations occur within work-stealing queues (in nested class WorkQueue). These are
   * special forms of Deques that support only three of the four possible end-operations -- push,
   * pop, and poll (aka steal), under the further constraints that push and pop are called only from
   * the owning thread (or, as extended here, under a lock), while poll may be called from other
   * threads. (If you are unfamiliar with them, you probably want to read Herlihy and Shavit's book
   * "The Art of Multiprocessor programming", chapter 16 describing these in more detail before
   * proceeding.) The main work-stealing queue design is roughly similar to those in the papers
   * "Dynamic Circular Work-Stealing Deque" by Chase and Lev, SPAA 2005
   * (http://research.sun.com/scalable/pubs/index.html) and "Idempotent work stealing" by Michael,
   * Saraswat, and Vechev, PPoPP 2009 (http://portal.acm.org/citation.cfm?id=1504186). The main
   * differences ultimately stem from GC requirements that we null out taken slots as soon as we
   * can, to maintain as small a footprint as possible even in programs generating huge numbers of
   * tasks. To accomplish this, we shift the CAS arbitrating pop vs poll (steal) from being on the
   * indices ("base" and "top") to the slots themselves.
   *
   * Adding tasks then takes the form of a classic array push(task) in a circular buffer:
   * q.array[q.top++ % length] = task;
   *
   * (The actual code needs to null-check and size-check the array, uses masking, not mod, for
   * indexing a power-of-two-sized array, properly fences accesses, and possibly signals waiting
   * workers to start scanning -- see below.) Both a successful pop and poll mainly entail a CAS of
   * a slot from non-null to null.
   *
   * The pop operation (always performed by owner) is: if ((the task at top slot is not null) and
   * (CAS slot to null)) decrement top and return task;
   *
   * And the poll operation (usually by a stealer) is if ((the task at base slot is not null) and
   * (CAS slot to null)) increment base and return task;
   *
   * There are several variants of each of these. In particular, almost all uses of poll occur
   * within scan operations that also interleave contention tracking (with associated code sprawl.)
   *
   * Memory ordering. See "Correct and Efficient Work-Stealing for Weak Memory Models" by Le, Pop,
   * Cohen, and Nardelli, PPoPP 2013 (http://www.di.ens.fr/~zappa/readings/ppopp13.pdf) for an
   * analysis of memory ordering requirements in work-stealing algorithms similar to (but different
   * than) the one used here. Extracting tasks in array slots via (fully fenced) CAS provides
   * primary synchronization. The base and top indices imprecisely guide where to extract from. We
   * do not always require strict orderings of array and index updates, so sometimes let them be
   * subject to compiler and processor reorderings. However, the volatile "base" index also serves
   * as a basis for memory ordering: Slot accesses are preceded by a read of base, ensuring
   * happens-before ordering with respect to stealers (so the slots themselves can be read via plain
   * array reads.) The only other memory orderings relied on are maintained in the course of
   * signalling and activation (see below). A check that base == top indicates (momentary)
   * emptiness, but otherwise may err on the side of possibly making the queue appear nonempty when
   * a push, pop, or poll have not fully committed, or making it appear empty when an update of top
   * has not yet been visibly written. (Method isEmpty() checks the case of a partially completed
   * removal of the last element.) Because of this, the poll operation, considered individually, is
   * not wait-free. One thief cannot successfully continue until another in-progress one (or, if
   * previously empty, a push) visibly completes. However, in the aggregate, we ensure at least
   * probabilistic non-blockingness. If an attempted steal fails, a scanning thief chooses a
   * different random victim target to try next. So, in order for one thief to progress, it suffices
   * for any in-progress poll or new push on any empty queue to complete.
   *
   * This approach also enables support of a user mode in which local task processing is in FIFO,
   * not LIFO order, simply by using poll rather than pop. This can be useful in message-passing
   * frameworks in which tasks are never joined.
   *
   * WorkQueues are also used in a similar way for tasks submitted to the pool. We cannot mix these
   * tasks in the same queues used by workers. Instead, we randomly associate submission queues with
   * submitting threads, using a form of hashing. The ThreadLocalRandom probe value serves as a hash
   * code for choosing existing queues, and may be randomly repositioned upon contention with other
   * submitters. In essence, submitters act like workers except that they are restricted to
   * executing local tasks that they submitted. Insertion of tasks in shared mode requires a lock
   * but we use only a simple spinlock (using field phase), because submitters encountering a busy
   * queue move to a different position to use or create other queues -- they block only when
   * creating and registering new queues. Because it is used only as a spinlock, unlocking requires
   * only a "releasing" store (using putOrderedInt).
   *
   * Management ==========
   *
   * The main throughput advantages of work-stealing stem from decentralized control -- workers
   * mostly take tasks from themselves or each other, at rates that can exceed a billion per second.
   * The pool itself creates, activates (enables scanning for and running tasks), deactivates,
   * blocks, and terminates threads, all with minimal central information. There are only a few
   * properties that we can globally track or maintain, so we pack them into a small number of
   * variables, often maintaining atomicity without blocking or locking. Nearly all essentially
   * atomic control state is held in a few volatile variables that are by far most often read (not
   * written) as status and consistency checks. We pack as much information into them as we can.
   *
   * Field "ctl" contains 64 bits holding information needed to atomically decide to add, enqueue
   * (on an event queue), and dequeue (and release)-activate workers. To enable this packing, we
   * restrict maximum parallelism to (1<<15)-1 (which is far in excess of normal operating range) to
   * allow ids, counts, and their negations (used for thresholding) to fit into 16bit subfields.
   *
   * Field "mode" holds configuration parameters as well as lifetime status, atomically and
   * monotonically setting SHUTDOWN, STOP, and finally TERMINATED bits.
   *
   * Field "workQueues" holds references to WorkQueues. It is updated (only during worker creation
   * and termination) under lock (using field workerNamePrefix as lock), but is otherwise
   * concurrently readable, and accessed directly. We also ensure that uses of the array reference
   * itself never become too stale in case of resizing. To simplify index-based operations, the
   * array size is always a power of two, and all readers must tolerate null slots. Worker queues
   * are at odd indices. Shared (submission) queues are at even indices, up to a maximum of 64
   * slots, to limit growth even if array needs to expand to add more workers. Grouping them
   * together in this way simplifies and speeds up task scanning.
   *
   * All worker thread creation is on-demand, triggered by task submissions, replacement of
   * terminated workers, and/or compensation for blocked workers. However, all other support code is
   * set up to work with other policies. To ensure that we do not hold on to worker references that
   * would prevent GC, all accesses to workQueues are via indices into the workQueues array (which
   * is one source of some of the messy code constructions here). In essence, the workQueues array
   * serves as a weak reference mechanism. Thus for example the stack top subfield of ctl stores
   * indices, not references.
   *
   * Queuing Idle Workers. Unlike HPC work-stealing frameworks, we cannot let workers spin
   * indefinitely scanning for tasks when none can be found immediately, and we cannot start/resume
   * workers unless there appear to be tasks available. On the other hand, we must quickly prod them
   * into action when new tasks are submitted or generated. In many usages, ramp-up time is the main
   * limiting factor in overall performance, which is compounded at program start-up by JIT
   * compilation and allocation. So we streamline this as much as possible.
   *
   * The "ctl" field atomically maintains total worker and "released" worker counts, plus the head
   * of the available worker queue (actually stack, represented by the lower 32bit subfield of ctl).
   * Released workers are those known to be scanning for and/or running tasks. Unreleased
   * ("available") workers are recorded in the ctl stack. These workers are made available for
   * signalling by enqueuing in ctl (see method runWorker). The "queue" is a form of Treiber stack.
   * This is ideal for activating threads in most-recently used order, and improves performance and
   * locality, outweighing the disadvantages of being prone to contention and inability to release a
   * worker unless it is topmost on stack. To avoid missed signal problems inherent in any
   * wait/signal design, available workers rescan for (and if found run) tasks after enqueuing.
   * Normally their release status will be updated while doing so, but the released worker ctl count
   * may underestimate the number of active threads. (However, it is still possible to determine
   * quiescence via a validation traversal -- see isQuiescent). After an unsuccessful rescan,
   * available workers are blocked until signalled (see signalWork). The top stack state holds the
   * value of the "phase" field of the worker: its index and status, plus a version counter that, in
   * addition to the count subfields (also serving as version stamps) provide protection against
   * Treiber stack ABA effects.
   *
   * Creating workers. To create a worker, we pre-increment counts (serving as a reservation), and
   * attempt to construct a ForkJoinWorkerThread via its factory. Upon construction, the new thread
   * invokes registerWorker, where it constructs a WorkQueue and is assigned an index in the
   * workQueues array (expanding the array if necessary). The thread is then started. Upon any
   * exception across these steps, or null return from factory, deregisterWorker adjusts counts and
   * records accordingly. If a null return, the pool continues running with fewer than the target
   * number workers. If exceptional, the exception is propagated, generally to some external caller.
   * Worker index assignment avoids the bias in scanning that would occur if entries were
   * sequentially packed starting at the front of the workQueues array. We treat the array as a
   * simple power-of-two hash table, expanding as needed. The seedIndex increment ensures no
   * collisions until a resize is needed or a worker is deregistered and replaced, and thereafter
   * keeps probability of collision low. We cannot use ThreadLocalRandom.getProbe() for similar
   * purposes here because the thread has not started yet, but do so for creating submission queues
   * for existing external threads (see externalPush).
   *
   * WorkQueue field "phase" is used by both workers and the pool to manage and track whether a
   * worker is UNSIGNALLED (possibly blocked waiting for a signal). When a worker is enqueued its
   * phase field is set. Note that phase field updates lag queue CAS releases so usage requires care
   * -- seeing a negative phase does not guarantee that the worker is available. When queued, the
   * lower 16 bits of scanState must hold its pool index. So we place the index there upon
   * initialization (see registerWorker) and otherwise keep it there or restore it when necessary.
   *
   * The ctl field also serves as the basis for memory synchronization surrounding activation. This
   * uses a more efficient version of a Dekker-like rule that task producers and consumers sync with
   * each other by both writing/CASing ctl (even if to its current value). This would be extremely
   * costly. So we relax it in several ways: (1) Producers only signal when their queue is empty.
   * Other workers propagate this signal (in method scan) when they find tasks; to further reduce
   * flailing, each worker signals only one other per activation. (2) Workers only enqueue after
   * scanning (see below) and not finding any tasks. (3) Rather than CASing ctl to its current value
   * in the common case where no action is required, we reduce write contention by equivalently
   * prefacing signalWork when called by an external task producer using a memory access with
   * full-volatile semantics or a "fullFence".
   *
   * Almost always, too many signals are issued. A task producer cannot in general tell if some
   * existing worker is in the midst of finishing one task (or already scanning) and ready to take
   * another without being signalled. So the producer might instead activate a different worker that
   * does not find any work, and then inactivates. This scarcely matters in steady-state
   * computations involving all workers, but can create contention and bookkeeping bottlenecks
   * during ramp-up, ramp-down, and small computations involving only a few workers.
   *
   * Scanning. Method runWorker performs top-level scanning for tasks. Each scan traverses and tries
   * to poll from each queue starting at a random index and circularly stepping. Scans are not
   * performed in ideal random permutation order, to reduce cacheline contention. The pseudorandom
   * generator need not have high-quality statistical properties in the long term, but just within
   * computations; We use Marsaglia XorShifts (often via ThreadLocalRandom.nextSecondarySeed), which
   * are cheap and suffice. Scanning also employs contention reduction: When scanning workers fail
   * to extract an apparently existing task, they soon restart at a different pseudorandom index.
   * This improves throughput when many threads are trying to take tasks from few queues, which can
   * be common in some usages. Scans do not otherwise explicitly take into account core affinities,
   * loads, cache localities, etc, However, they do exploit temporal locality (which usually
   * approximates these) by preferring to re-poll (at most #workers times) from the same queue after
   * a successful poll before trying others.
   *
   * Trimming workers. To release resources after periods of lack of use, a worker starting to wait
   * when the pool is quiescent will time out and terminate (see method scan) if the pool has
   * remained quiescent for period given by field keepAlive.
   *
   * Shutdown and Termination. A call to shutdownNow invokes tryTerminate to atomically set a
   * runState bit. The calling thread, as well as every other worker thereafter terminating, helps
   * terminate others by cancelling their unprocessed tasks, and waking them up, doing so repeatedly
   * until stable. Calls to non-abrupt shutdown() preface this by checking whether termination
   * should commence by sweeping through queues (until stable) to ensure lack of in-flight
   * submissions and workers about to process them before triggering the "STOP" phase of
   * termination.
   *
   * Joining Tasks =============
   *
   * Any of several actions may be taken when one worker is waiting to join a task stolen (or always
   * held) by another. Because we are multiplexing many tasks on to a pool of workers, we can't
   * always just let them block (as in Thread.join). We also cannot just reassign the joiner's
   * run-time stack with another and replace it later, which would be a form of "continuation", that
   * even if possible is not necessarily a good idea since we may need both an unblocked task and
   * its continuation to progress. Instead we combine two tactics:
   *
   * Helping: Arranging for the joiner to execute some task that it would be running if the steal
   * had not occurred.
   *
   * Compensating: Unless there are already enough live threads, method tryCompensate() may create
   * or re-activate a spare thread to compensate for blocked joiners until they unblock.
   *
   * A third form (implemented in tryRemoveAndExec) amounts to helping a hypothetical compensator:
   * If we can readily tell that a possible action of a compensator is to steal and execute the task
   * being joined, the joining thread can do so directly, without the need for a compensation
   * thread.
   *
   * The ManagedBlocker extension API can't use helping so relies only on compensation in method
   * awaitBlocker.
   *
   * The algorithm in awaitJoin entails a form of "linear helping". Each worker records (in field
   * source) the id of the queue from which it last stole a task. The scan in method awaitJoin uses
   * these markers to try to find a worker to help (i.e., steal back a task from and execute it)
   * that could hasten completion of the actively joined task. Thus, the joiner executes a task that
   * would be on its own local deque if the to-be-joined task had not been stolen. This is a
   * conservative variant of the approach described in Wagner & Calder "Leapfrogging: a portable
   * technique for implementing efficient futures" SIGPLAN Notices, 1993
   * (http://portal.acm.org/citation.cfm?id=155354). It differs mainly in that we only record queue
   * ids, not full dependency links. This requires a linear scan of the workQueues array to locate
   * stealers, but isolates cost to when it is needed, rather than adding to per-task overhead.
   * Searches can fail to locate stealers GC stalls and the like delay recording sources. Further,
   * even when accurately identified, stealers might not ever produce a task that the joiner can in
   * turn help with. So, compensation is tried upon failure to find tasks to run.
   *
   * Compensation does not by default aim to keep exactly the target parallelism number of unblocked
   * threads running at any given time. Some previous versions of this class employed immediate
   * compensations for any blocked join. However, in practice, the vast majority of blockages are
   * transient byproducts of GC and other JVM or OS activities that are made worse by replacement.
   * Rather than impose arbitrary policies, we allow users to override the default of only adding
   * threads upon apparent starvation. The compensation mechanism may also be bounded. Bounds for
   * the commonPool (see COMMON_MAX_SPARES) better enable JVMs to cope with programming errors and
   * abuse before running out of resources to do so.
   *
   * Common Pool ===========
   *
   * The static common pool always exists after static initialization. Since it (or any other
   * created pool) need never be used, we minimize initial construction overhead and footprint to
   * the setup of about a dozen fields.
   *
   * When external threads submit to the common pool, they can perform subtask processing (see
   * externalHelpComplete and related methods) upon joins. This caller-helps policy makes it
   * sensible to set common pool parallelism level to one (or more) less than the total number of
   * available cores, or even zero for pure caller-runs. We do not need to record whether external
   * submissions are to the common pool -- if not, external help methods return quickly. These
   * submitters would otherwise be blocked waiting for completion, so the extra effort (with
   * liberally sprinkled task status checks) in inapplicable cases amounts to an odd form of limited
   * spin-wait before blocking in ForkJoinTask.join.
   *
   * As a more appropriate default in managed environments, unless overridden by system properties,
   * we use workers of subclass InnocuousForkJoinWorkerThread when there is a SecurityManager
   * present. These workers have no permissions set, do not belong to any user-defined ThreadGroup,
   * and erase all ThreadLocals after executing any top-level task (see
   * WorkQueue.afterTopLevelExec). The associated mechanics (mainly in ForkJoinWorkerThread) may be
   * JVM-dependent and must access particular Thread class fields to achieve this effect.
   *
   * Style notes ===========
   *
   * Memory ordering relies mainly on Unsafe intrinsics that carry the further responsibility of
   * explicitly performing null- and bounds- checks otherwise carried out implicitly by JVMs. This
   * can be awkward and ugly, but also reflects the need to control outcomes across the unusual
   * cases that arise in very racy code with very few invariants. All fields are read into locals
   * before use, and null-checked if they are references. This is usually done in a "C"-like style
   * of listing declarations at the heads of methods or blocks, and using inline assignments on
   * first encounter. Nearly all explicit checks lead to bypass/return, not exception throws,
   * because they may legitimately arise due to cancellation/revocation during shutdown.
   *
   * There is a lot of representation-level coupling among classes ForkJoinPool,
   * ForkJoinWorkerThread, and ForkJoinTask. The fields of WorkQueue maintain data structures
   * managed by ForkJoinPool, so are directly accessed. There is little point trying to reduce this,
   * since any associated future changes in representations will need to be accompanied by
   * algorithmic changes anyway. Several methods intrinsically sprawl because they must accumulate
   * sets of consistent reads of fields held in local variables. There are also other coding
   * oddities (including several unnecessary-looking hoisted null checks) that help some methods
   * perform reasonably even when interpreted (not compiled).
   *
   * The order of declarations in this file is (with a few exceptions): (1) Static utility functions
   * (2) Nested (static) classes (3) Static fields (4) Fields, along with constants used when
   * unpacking some of them (5) Internal control methods (6) Callbacks and other support for
   * ForkJoinTask methods (7) Exported methods (8) Static block initializing statics in minimally
   * dependent order
   */

  // Static utilities

  /**
   * Creates a new ForkJoinWorkerThread. This factory is used unless overridden in ForkJoinPool
   * constructors.
   */
  public static final ForkJoinWorkerThreadFactory defaultForkJoinWorkerThreadFactory;

  // Nested classes
  // Bounds
  static final int SWIDTH = 16; // width of short
  static final int SMASK = 0xffff; // short bits == max index
  static final int MAX_CAP = 0x7fff; // max #workers - 1

  // Constants shared across ForkJoinPool and WorkQueue
  static final int SQMASK = 0x007e; // max 64 (even) slots
  // Masks and units for WorkQueue.phase and ctl sp subfield
  static final int UNSIGNALLED = 1 << 31; // must be negative
  static final int SS_SEQ = 1 << 16; // version count
  static final int QLOCK = 1; // must be 1
  // Mode bits and sentinels, some also used in WorkQueue id and.source fields
  static final int OWNED = 1; // queue has owner thread
  static final int FIFO = 1 << 16; // fifo queue or access mode
  static final int SHUTDOWN = 1 << 18;
  static final int TERMINATED = 1 << 19;
  static final int STOP = 1 << 31; // must be negative
  static final int QUIET = 1 << 30; // not scanning or working
  static final int DORMANT = QUIET | UNSIGNALLED;
  /**
   * The maximum number of local polls from the same queue before checking others. This is a
   * safeguard against infinitely unfair looping under unbounded user task recursion, and must be
   * larger than plausible cases of intentional bounded task recursion.
   */
  static final int POLL_LIMIT = 1 << 10;
  /**
   * Permission required for callers of methods that may start or kill threads.
   */
  static final RuntimePermission modifyThreadPermission;
  /**
   * Common (static) pool. Non-null for public use unless a static construction exception, but
   * internal usages null-check on use to paranoically avoid potential initialization circularities
   * as well as to simplify generated code.
   */
  static final ForkJoinPool common;
  /**
   * Common pool parallelism. To allow simpler use and management when common pool threads are
   * disabled, we allow the underlying common.parallelism field to be zero, but in that case still
   * report parallelism as 1 to reflect resulting caller-runs mechanics.
   */
  static final int COMMON_PARALLELISM;
  /**
   * Limit on spare thread construction in tryCompensate.
   */
  private static final int COMMON_MAX_SPARES;

  // static fields (initialized in static initializer below)
  /**
   * Default idle timeout value (in milliseconds) for the thread triggering quiescence to park
   * waiting for new work
   */
  private static final long DEFAULT_KEEPALIVE = 60_000L;
  /**
   * Undershoot tolerance for idle timeouts
   */
  private static final long TIMEOUT_SLOP = 20L;
  /**
   * The default value for COMMON_MAX_SPARES. Overridable using the
   * "java.util.concurrent.ForkJoinPool.common.maximumSpares" system property. The default value is
   * far in excess of normal requirements, but also far short of MAX_CAP and typical OS thread
   * limits, so allows JVMs to catch misuse/abuse before running out of resources needed to do so.
   */
  private static final int DEFAULT_COMMON_MAX_SPARES = 256;
  /**
   * Increment for seed generators. See class ThreadLocal for explanation.
   */
  private static final int SEED_INCREMENT = 0x9e3779b9;
  // Lower and upper word masks
  private static final long SP_MASK = 0xffffffffL;
  private static final long UC_MASK = ~SP_MASK;
  // Release counts
  private static final int RC_SHIFT = 48;

  // static configuration constants
  private static final long RC_UNIT = 0x0001L << RC_SHIFT;
  private static final long RC_MASK = 0xffffL << RC_SHIFT;
  // Total counts
  private static final int TC_SHIFT = 32;
  private static final long TC_UNIT = 0x0001L << TC_SHIFT;

  /*
   * Bits and masks for field ctl, packed with 4 16 bit subfields: RC: Number of released (unqueued)
   * workers minus target parallelism TC: Number of total workers minus target parallelism SS:
   * version count and status of top waiting thread ID: poolIndex of top of Treiber stack of waiters
   *
   * When convenient, we can extract the lower 32 stack top bits (including version bits) as
   * sp=(int)ctl. The offsets of counts by the target parallelism and the positionings of fields
   * makes it possible to perform the most common checks via sign tests of fields: When ac is
   * negative, there are not enough unqueued workers, when tc is negative, there are not enough
   * total workers. When sp is non-zero, there are waiting workers. To deal with possibly negative
   * fields, we use casts in and out of "short" and/or signed shifts to maintain signedness.
   *
   * Because it occupies uppermost bits, we can add one release count using getAndAddLong of
   * RC_UNIT, rather than CAS, when returning from a blocked join. Other updates entail multiple
   * subfields and masking, requiring CAS.
   *
   * The limits packed in field "bounds" are also offset by the parallelism level to make them
   * comparable to the ctl rc and tc fields.
   */
  private static final long TC_MASK = 0xffffL << TC_SHIFT;
  private static final long ADD_WORKER = 0x0001L << (TC_SHIFT + 15); // sign
  // Unsafe mechanics
  private static final sun.misc.Unsafe U = UnsafeAccess.unsafe;
  private static final long CTL;
  private static final long MODE;
  private static final int ABASE;
  private static final int ASHIFT;
  /**
   * Sequence number for creating workerNamePrefix.
   */
  private static int poolNumberSequence;

  static {
    try {
      CTL = U.objectFieldOffset(ForkJoinPool.class.getDeclaredField("ctl"));
      MODE = U.objectFieldOffset(ForkJoinPool.class.getDeclaredField("mode"));
      ABASE = U.arrayBaseOffset(ForkJoinTask[].class);
      int scale = U.arrayIndexScale(ForkJoinTask[].class);
      if ((scale & (scale - 1)) != 0)
        throw new ExceptionInInitializerError("array index scale not a power of two");
      ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
    } catch (Exception e) {
      throw new ExceptionInInitializerError(e);
    }

    // Reduce the risk of rare disastrous classloading in first call to
    // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773
    @SuppressWarnings("unused")
    Class<?> ensureLoaded = LockSupport.class;

    int commonMaxSpares = DEFAULT_COMMON_MAX_SPARES;
    try {
      String p = System.getProperty("java.util.concurrent.ForkJoinPool.common.maximumSpares");
      if (p != null)
        commonMaxSpares = Integer.parseInt(p);
    } catch (Exception ignore) {
    }
    COMMON_MAX_SPARES = commonMaxSpares;

    defaultForkJoinWorkerThreadFactory = new DefaultForkJoinWorkerThreadFactory();
    modifyThreadPermission = new RuntimePermission("modifyThread");

    common = AccessController.doPrivileged(new PrivilegedAction<ForkJoinPool>() {
      public ForkJoinPool run() {
        return new ForkJoinPool((byte) 0);
      }
    });

    COMMON_PARALLELISM = Math.max(common.mode & SMASK, 1);
  }

  // Instance fields

  final long keepAlive; // milliseconds before dropping if idle
  final int bounds; // min, max threads packed as shorts
  final String workerNamePrefix; // for worker thread string; sync lock
  final ForkJoinWorkerThreadFactory factory;
  final UncaughtExceptionHandler ueh; // per-worker UEH
  final Predicate<? super ForkJoinPool> saturate;
  // Segregate ctl field, For now using padding vs @Contended
  volatile long pad00, pad01, pad02, pad03, pad04, pad05, pad06, pad07;
  volatile long pad08, pad09, pad0a, pad0b, pad0c, pad0d, pad0e, pad0f;
  volatile long ctl; // main pool control
  volatile long pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17;
  volatile long pad18, pad19, pad1a, pad1b, pad1c, pad1d, pad1e;
  volatile long stealCount; // collects worker nsteals
  int indexSeed; // next worker index
  volatile int mode; // parallelism, runstate, queue mode
  WorkQueue[] workQueues; // main registry

  // Creating, registering and deregistering workers

  /**
   * Creates a {@code ForkJoinPool} with parallelism equal to
   * {@link java.lang.Runtime#availableProcessors}, using defaults for all other parameters (see
   * {@link #ForkJoinPool(int, ForkJoinWorkerThreadFactory, Thread.UncaughtExceptionHandler, boolean, int, int, int, Predicate, long, TimeUnit)}).
   *
   * @throws SecurityException if a security manager exists and the caller is not permitted to
   *         modify threads because it does not hold
   *         {@link java.lang.RuntimePermission}{@code ("modifyThread")}
   */
  public ForkJoinPool() {
    this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
        defaultForkJoinWorkerThreadFactory, null, false, 0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE,
        TimeUnit.MILLISECONDS);
  }

  /**
   * Creates a {@code ForkJoinPool} with the indicated parallelism level, using defaults for all
   * other parameters (see
   * {@link #ForkJoinPool(int, ForkJoinWorkerThreadFactory, Thread.UncaughtExceptionHandler, boolean, int, int, int, Predicate, long, TimeUnit)}).
   *
   * @param parallelism the parallelism level
   * @throws IllegalArgumentException if parallelism less than or equal to zero, or greater than
   *         implementation limit
   * @throws SecurityException if a security manager exists and the caller is not permitted to
   *         modify threads because it does not hold
   *         {@link java.lang.RuntimePermission}{@code ("modifyThread")}
   */
  public ForkJoinPool(int parallelism) {
    this(parallelism, defaultForkJoinWorkerThreadFactory, null, false, 0, MAX_CAP, 1, null,
        DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS);
  }

  /**
   * Creates a {@code ForkJoinPool} with the given parameters (using defaults for others -- see
   * {@link #ForkJoinPool(int, ForkJoinWorkerThreadFactory, Thread.UncaughtExceptionHandler, boolean, int, int, int, Predicate, long, TimeUnit)}).
   *
   * @param parallelism the parallelism level. For default value, use
   *        {@link java.lang.Runtime#availableProcessors}.
   * @param factory the factory for creating new threads. For default value, use
   *        {@link #defaultForkJoinWorkerThreadFactory}.
   * @param handler the handler for internal worker threads that terminate due to unrecoverable
   *        errors encountered while executing tasks. For default value, use {@code null}.
   * @param asyncMode if true, establishes local first-in-first-out scheduling mode for forked tasks
   *        that are never joined. This mode may be more appropriate than default locally
   *        stack-based mode in applications in which worker threads only process event-style
   *        asynchronous tasks. For default value, use {@code false}.
   * @throws IllegalArgumentException if parallelism less than or equal to zero, or greater than
   *         implementation limit
   * @throws NullPointerException if the factory is null
   * @throws SecurityException if a security manager exists and the caller is not permitted to
   *         modify threads because it does not hold
   *         {@link java.lang.RuntimePermission}{@code ("modifyThread")}
   */
  public ForkJoinPool(int parallelism, ForkJoinWorkerThreadFactory factory,
      UncaughtExceptionHandler handler, boolean asyncMode) {
    this(parallelism, factory, handler, asyncMode, 0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE,
        TimeUnit.MILLISECONDS);
  }

  /**
   * Creates a {@code ForkJoinPool} with the given parameters.
   *
   * @param parallelism the parallelism level. For default value, use
   *        {@link java.lang.Runtime#availableProcessors}.
   *
   * @param factory the factory for creating new threads. For default value, use
   *        {@link #defaultForkJoinWorkerThreadFactory}.
   *
   * @param handler the handler for internal worker threads that terminate due to unrecoverable
   *        errors encountered while executing tasks. For default value, use {@code null}.
   *
   * @param asyncMode if true, establishes local first-in-first-out scheduling mode for forked tasks
   *        that are never joined. This mode may be more appropriate than default locally
   *        stack-based mode in applications in which worker threads only process event-style
   *        asynchronous tasks. For default value, use {@code
   * false}.
   *
   * @param corePoolSize the number of threads to keep in the pool (unless timed out after an
   *        elapsed keep-alive). Normally (and by default) this is the same value as the parallelism
   *        level, but may be set to a larger value to reduce dynamic overhead if tasks regularly
   *        block. Using a smaller value (for example {@code 0}) has the same effect as the default.
   *
   * @param maximumPoolSize the maximum number of threads allowed. When the maximum is reached,
   *        attempts to replace blocked threads fail. (However, because creation and termination of
   *        different threads may overlap, and may be managed by the given thread factory, this
   *        value may be transiently exceeded.) To arrange the same value as is used by default for
   *        the common pool, use {@code 256} plus the {@code parallelism} level. (By default, the
   *        common pool allows a maximum of 256 spare threads.) Using a value (for example {@code
   * Integer.MAX_VALUE}) larger than the implementation's total thread limit has the same effect as
   *        using this limit (which is the default).
   *
   * @param minimumRunnable the minimum allowed number of core threads not blocked by a join or
   *        {@link ManagedBlocker}. To ensure progress, when too few unblocked threads exist and
   *        unexecuted tasks may exist, new threads are constructed, up to the given
   *        maximumPoolSize. For the default value, use {@code
   * 1}  , that ensures liveness. A larger value might improve throughput in the presence of
   *        blocked activities, but might not, due to increased overhead. A value of zero may be
   *        acceptable when submitted tasks cannot have dependencies requiring additional threads.
   *
   * @param saturate if non-null, a predicate invoked upon attempts to create more than the maximum
   *        total allowed threads. By default, when a thread is about to block on a join or
   *        {@link ManagedBlocker}, but cannot be replaced because the maximumPoolSize would be
   *        exceeded, a {@link RejectedExecutionException} is thrown. But if this predicate returns
   *        {@code true}, then no exception is thrown, so the pool continues to operate with fewer
   *        than the target number of runnable threads, which might not ensure progress.
   *
   * @param keepAliveTime the elapsed time since last use before a thread is terminated (and then
   *        later replaced if needed). For the default value, use {@code 60, TimeUnit.SECONDS}.
   *
   * @param unit the time unit for the {@code keepAliveTime} argument
   *
   * @throws IllegalArgumentException if parallelism is less than or equal to zero, or is greater
   *         than implementation limit, or if maximumPoolSize is less than parallelism, of if the
   *         keepAliveTime is less than or equal to zero.
   * @throws NullPointerException if the factory is null
   * @throws SecurityException if a security manager exists and the caller is not permitted to
   *         modify threads because it does not hold
   *         {@link java.lang.RuntimePermission}{@code ("modifyThread")}
   * @since 9
   */
  public ForkJoinPool(int parallelism, ForkJoinWorkerThreadFactory factory,
      UncaughtExceptionHandler handler, boolean asyncMode, int corePoolSize, int maximumPoolSize,
      int minimumRunnable, Predicate<? super ForkJoinPool> saturate, long keepAliveTime,
      TimeUnit unit) {
    // check, encode, pack parameters
    if (parallelism <= 0 || parallelism > MAX_CAP || maximumPoolSize < parallelism
        || keepAliveTime <= 0L)
      throw new IllegalArgumentException();
    Objects.requireNonNull(factory);
    long ms = Math.max(unit.toMillis(keepAliveTime), TIMEOUT_SLOP);

    int corep = Math.min(Math.max(corePoolSize, parallelism), MAX_CAP);
    long c = ((((long) (-corep) << TC_SHIFT) & TC_MASK)
        | (((long) (-parallelism) << RC_SHIFT) & RC_MASK));
    int m = parallelism | (asyncMode ? FIFO : 0);
    int maxSpares = Math.min(maximumPoolSize, MAX_CAP) - parallelism;
    int minAvail = Math.min(Math.max(minimumRunnable, 0), MAX_CAP);
    int b = ((minAvail - parallelism) & SMASK) | (maxSpares << SWIDTH);
    int n = (parallelism > 1) ? parallelism - 1 : 1; // at least 2 slots
    n |= n >>> 1;
    n |= n >>> 2;
    n |= n >>> 4;
    n |= n >>> 8;
    n |= n >>> 16;
    n = (n + 1) << 1; // power of two, including space for submission queues

    this.workerNamePrefix = "ForkJoinPool-" + nextPoolId() + "-worker-";
    this.workQueues = new WorkQueue[n];
    this.factory = factory;
    this.ueh = handler;
    this.saturate = saturate;
    this.keepAlive = ms;
    this.bounds = b;
    this.mode = m;
    this.ctl = c;
    checkPermission();
  }

  /**
   * Constructor for common pool using parameters possibly overridden by system properties
   */
  private ForkJoinPool(byte forCommonPoolOnly) {
    int parallelism = -1;
    ForkJoinWorkerThreadFactory fac = null;
    UncaughtExceptionHandler handler = null;
    try { // ignore exceptions in accessing/parsing properties
      String pp = System.getProperty("java.util.concurrent.ForkJoinPool.common.parallelism");
      if (pp != null)
        parallelism = Integer.parseInt(pp);
      fac = (ForkJoinWorkerThreadFactory) newInstanceFromSystemProperty(
          "java.util.concurrent.ForkJoinPool.common.threadFactory");
      handler = (UncaughtExceptionHandler) newInstanceFromSystemProperty(
          "java.util.concurrent.ForkJoinPool.common.exceptionHandler");
    } catch (Exception ignore) {
    }

    if (fac == null) {
      if (System.getSecurityManager() == null) {
        fac = defaultForkJoinWorkerThreadFactory;
      } else { // use security-managed default
        fac = new InnocuousForkJoinWorkerThreadFactory();
      }
    }
    if (parallelism < 0 && // default 1 less than #cores
        (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)
      parallelism = 1;
    if (parallelism > MAX_CAP)
      parallelism = MAX_CAP;

    long c = ((((long) (-parallelism) << TC_SHIFT) & TC_MASK)
        | (((long) (-parallelism) << RC_SHIFT) & RC_MASK));
    int b = ((1 - parallelism) & SMASK) | (COMMON_MAX_SPARES << SWIDTH);
    int n = (parallelism > 1) ? parallelism - 1 : 1;
    n |= n >>> 1;
    n |= n >>> 2;
    n |= n >>> 4;
    n |= n >>> 8;
    n |= n >>> 16;
    n = (n + 1) << 1;

    this.workerNamePrefix = "ForkJoinPool.commonPool-worker-";
    this.workQueues = new WorkQueue[n];
    this.factory = fac;
    this.ueh = handler;
    this.saturate = null;
    this.keepAlive = DEFAULT_KEEPALIVE;
    this.bounds = b;
    this.mode = parallelism;
    this.ctl = c;
  }

  /**
   * If there is a security manager, makes sure caller has permission to modify threads.
   */
  private static void checkPermission() {
    SecurityManager security = System.getSecurityManager();
    if (security != null)
      security.checkPermission(modifyThreadPermission);
  }

  static AccessControlContext contextWithPermissions(Permission... perms) {
    Permissions permissions = new Permissions();
    for (Permission perm : perms)
      permissions.add(perm);
    return new AccessControlContext(
        new ProtectionDomain[] {new ProtectionDomain(null, permissions)});
  }

  /**
   * Returns the next sequence number. We don't expect this to ever contend, so use simple builtin
   * sync.
   */
  private static final synchronized int nextPoolId() {
    return ++poolNumberSequence;
  }

  /**
   * Returns common pool queue for an external thread.
   */
  static WorkQueue commonSubmitterQueue() {
    ForkJoinPool p = common;
    int r = TLRandom.getProbe();
    WorkQueue[] ws;
    int n;
    return (p != null && (ws = p.workQueues) != null && (n = ws.length) > 0)
        ? ws[(n - 1) & r & SQMASK]
        : null;
  }

  /**
   * Returns a cheap heuristic guide for task partitioning when programmers, frameworks, tools, or
   * languages have little or no idea about task granularity. In essence, by offering this method,
   * we ask users only about tradeoffs in overhead vs expected throughput and its variance, rather
   * than how finely to partition tasks.
   *
   * In a steady state strict (tree-structured) computation, each thread makes available for
   * stealing enough tasks for other threads to remain active. Inductively, if all threads play by
   * the same rules, each thread should make available only a constant number of tasks.
   *
   * The minimum useful constant is just 1. But using a value of 1 would require immediate
   * replenishment upon each steal to maintain enough tasks, which is infeasible. Further,
   * partitionings/granularities of offered tasks should minimize steal rates, which in general
   * means that threads nearer the top of computation tree should generate more than those nearer
   * the bottom. In perfect steady state, each thread is at approximately the same level of
   * computation tree. However, producing extra tasks amortizes the uncertainty of progress and
   * diffusion assumptions.
   *
   * So, users will want to use values larger (but not much larger) than 1 to both smooth over
   * transient shortages and hedge against uneven progress; as traded off against the cost of extra
   * task overhead. We leave the user to pick a threshold value to compare with the results of this
   * call to guide decisions, but recommend values such as 3.
   *
   * When all threads are active, it is on average OK to estimate surplus strictly locally. In
   * steady-state, if one thread is maintaining say 2 surplus tasks, then so are others. So we can
   * just use estimated queue length. However, this strategy alone leads to serious mis-estimates in
   * some non-steady-state conditions (ramp-up, ramp-down, other stalls). We can detect many of
   * these by further considering the number of "idle" threads, that are known to have zero queued
   * tasks, so compensate by a factor of (#idle/#active) threads.
   */
  static int getSurplusQueuedTaskCount() {
    Thread t;
    ForkJoinWorkerThread wt;
    ForkJoinPool pool;
    WorkQueue q;
    if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
        && (pool = (wt = (ForkJoinWorkerThread) t).pool) != null && (q = wt.workQueue) != null) {
      int p = pool.mode & SMASK;
      int a = p + (int) (pool.ctl >> RC_SHIFT);
      int n = q.top - q.base;
      return n
          - (a > (p >>>= 1) ? 0 : a > (p >>>= 1) ? 1 : a > (p >>>= 1) ? 2 : a > (p >>>= 1) ? 4 : 8);
    }
    return 0;
  }

  private static Object newInstanceFromSystemProperty(String property) throws Exception {
    String className = System.getProperty(property);
    return (className == null) ? null
        : ClassLoader.getSystemClassLoader().loadClass(className).getConstructor().newInstance();
  }

  // External operations

  /**
   * Returns the common pool instance. This pool is statically constructed; its run state is
   * unaffected by attempts to {@link #shutdown} or {@link #shutdownNow}. However this pool and any
   * ongoing processing are automatically terminated upon program {@link System#exit}. Any program
   * that relies on asynchronous task processing to complete before program termination should
   * invoke {@code commonPool().}{@link #awaitQuiescence awaitQuiescence}, before exit.
   *
   * @return the common pool instance
   * @since 1.8
   */
  public static ForkJoinPool commonPool() {
    // assert common != null : "static init error";
    return common;
  }

  /**
   * Returns the targeted parallelism level of the common pool.
   *
   * @return the targeted parallelism level of the common pool
   * @since 1.8
   */
  public static int getCommonPoolParallelism() {
    return COMMON_PARALLELISM;
  }

  /**
   * Waits and/or attempts to assist performing tasks indefinitely until the {@link #commonPool()}
   * {@link #isQuiescent}.
   */
  static void quiesceCommonPool() {
    common.awaitQuiescence(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
  }

  /**
   * Runs the given possibly blocking task. When {@linkplain ForkJoinTask#inForkJoinPool() running
   * in a ForkJoinPool}, this method possibly arranges for a spare thread to be activated if
   * necessary to ensure sufficient parallelism while the current thread is blocked in
   * {@link ManagedBlocker#block blocker.block()}.
   *
   * <p>
   * This method repeatedly calls {@code blocker.isReleasable()} and {@code blocker.block()} until
   * either method returns {@code true}. Every call to {@code blocker.block()} is preceded by a call
   * to {@code blocker.isReleasable()} that returned {@code false}.
   *
   * <p>
   * If not running in a ForkJoinPool, this method is behaviorally equivalent to
   *
   * <pre>
   *  {@code
   * while (!blocker.isReleasable())
   *   if (blocker.block())
   *     break;}
   * </pre>
   *
   * If running in a ForkJoinPool, the pool may first be expanded to ensure sufficient parallelism
   * available during the call to {@code blocker.block()}.
   *
   * @param blocker the blocker task
   * @throws InterruptedException if {@code blocker.block()} did so
   */
  public static void managedBlock(ManagedBlocker blocker) throws InterruptedException {
    ForkJoinPool p;
    ForkJoinWorkerThread wt;
    WorkQueue w;
    Thread t = Thread.currentThread();
    if ((t instanceof ForkJoinWorkerThread) && (p = (wt = (ForkJoinWorkerThread) t).pool) != null
        && (w = wt.workQueue) != null) {
      int block;
      while (!blocker.isReleasable()) {
        if ((block = p.tryCompensate(w)) != 0) {
          try {
            do {
            } while (!blocker.isReleasable() && !blocker.block());
          } finally {
            U.getAndAddLong(p, CTL, (block > 0) ? RC_UNIT : 0L);
          }
          break;
        }
      }
    } else {
      do {
      } while (!blocker.isReleasable() && !blocker.block());
    }
  }

  /**
   * If the given executor is a ForkJoinPool, poll and execute AsynchronousCompletionTasks from
   * worker's queue until none are available or blocker is released.
   */
  static void helpAsyncBlocker(Executor e, ManagedBlocker blocker) {
    if (blocker != null && (e instanceof ForkJoinPool)) {
      WorkQueue w;
      ForkJoinWorkerThread wt;
      WorkQueue[] ws;
      int r, n;
      ForkJoinPool p = (ForkJoinPool) e;
      Thread thread = Thread.currentThread();
      if (thread instanceof ForkJoinWorkerThread && (wt = (ForkJoinWorkerThread) thread).pool == p)
        w = wt.workQueue;
      else if ((r = TLRandom.getProbe()) != 0 && (ws = p.workQueues) != null && (n = ws.length) > 0)
        w = ws[(n - 1) & r & SQMASK];
      else
        w = null;
      if (w != null) {
        for (;;) {
          int b = w.base, s = w.top, d, al;
          ForkJoinTask<?>[] a;
          if ((a = w.array) != null && (d = b - s) < 0 && (al = a.length) > 0) {
            int index = (al - 1) & b;
            long offset = ((long) index << ASHIFT) + ABASE;
            ForkJoinTask<?> t = (ForkJoinTask<?>) U.getObjectVolatile(a, offset);
            if (blocker.isReleasable())
              break;
            else if (b++ == w.base) {
              if (t == null) {
                if (d == -1)
                  break;
              } else if (!(t instanceof CompletableFuture.AsynchronousCompletionTask))
                break;
              else if (U.compareAndSwapObject(a, offset, t, null)) {
                w.base = b;
                t.doExec();
              }
            }
          } else
            break;
        }
      }
    }
  }

  /**
   * Tries to construct and start one worker. Assumes that total count has already been incremented
   * as a reservation. Invokes deregisterWorker on any failure.
   *
   * @return true if successful
   */
  private boolean createWorker() {
    ForkJoinWorkerThreadFactory fac = factory;
    Throwable ex = null;
    ForkJoinWorkerThread wt = null;
    try {
      if (fac != null && (wt = fac.newThread(this)) != null) {
        wt.start();
        return true;
      }
    } catch (Throwable rex) {
      ex = rex;
    }
    deregisterWorker(wt, ex);
    return false;
  }

  /**
   * Tries to add one worker, incrementing ctl counts before doing so, relying on createWorker to
   * back out on failure.
   *
   * @param c incoming ctl value, with total count negative and no idle workers. On CAS failure, c
   *        is refreshed and retried if this holds (otherwise, a new worker is not needed).
   */
  private void tryAddWorker(long c) {
    do {
      long nc = ((RC_MASK & (c + RC_UNIT)) | (TC_MASK & (c + TC_UNIT)));
      if (ctl == c && U.compareAndSwapLong(this, CTL, c, nc)) {
        createWorker();
        break;
      }
    } while (((c = ctl) & ADD_WORKER) != 0L && (int) c == 0);
  }

  // Termination

  /**
   * Callback from ForkJoinWorkerThread constructor to establish and record its WorkQueue.
   *
   * @param wt the worker thread
   * @return the worker's queue
   */
  final WorkQueue registerWorker(ForkJoinWorkerThread wt) {
    UncaughtExceptionHandler handler;
    wt.setDaemon(true); // configure thread
    if ((handler = ueh) != null)
      wt.setUncaughtExceptionHandler(handler);
    WorkQueue w = new WorkQueue(this, wt);
    int tid = 0; // for thread name
    int fifo = mode & FIFO;
    String prefix = workerNamePrefix;
    if (prefix != null) {
      synchronized (prefix) {
        WorkQueue[] ws = workQueues;
        int n;
        int s = indexSeed += SEED_INCREMENT;
        if (ws != null && (n = ws.length) > 1) {
          int m = n - 1;
          tid = s & m;
          int i = m & ((s << 1) | 1); // odd-numbered indices
          for (int probes = n >>> 1;;) { // find empty slot
            WorkQueue q;
            if ((q = ws[i]) == null || q.phase == QUIET)
              break;
            else if (--probes == 0) {
              i = n | 1; // resize below
              break;
            } else
              i = (i + 2) & m;
          }

          int id = i | fifo | (s & ~(SMASK | FIFO | DORMANT));
          w.phase = w.id = id; // now publishable

          if (i < n)
            ws[i] = w;
          else { // expand array
            int an = n << 1;
            WorkQueue[] as = new WorkQueue[an];
            as[i] = w;
            int am = an - 1;
            for (int j = 0; j < n; ++j) {
              WorkQueue v; // copy external queue
              if ((v = ws[j]) != null) // position may change
                as[v.id & am & SQMASK] = v;
              if (++j >= n)
                break;
              as[j] = ws[j]; // copy worker
            }
            workQueues = as;
          }
        }
      }
      wt.setName(prefix.concat(Integer.toString(tid)));
    }
    return w;
  }

  // Exported methods

  // Constructors

  /**
   * Final callback from terminating worker, as well as upon failure to construct or start a worker.
   * Removes record of worker from array, and adjusts counts. If pool is shutting down, tries to
   * complete termination.
   *
   * @param wt the worker thread, or null if construction failed
   * @param ex the exception causing failure, or null if none
   */
  final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) {
    WorkQueue w = null;
    int phase = 0;
    if (wt != null && (w = wt.workQueue) != null) {
      Object lock = workerNamePrefix;
      long ns = (long) w.nsteals & 0xffffffffL;
      int idx = w.id & SMASK;
      if (lock != null) {
        WorkQueue[] ws; // remove index from array
        synchronized (lock) {
          if ((ws = workQueues) != null && ws.length > idx && ws[idx] == w)
            ws[idx] = null;
          stealCount += ns;
        }
      }
      phase = w.phase;
    }
    if (phase != QUIET) { // else pre-adjusted
      long c; // decrement counts
      do {
      } while (!U.compareAndSwapLong(this, CTL, c = ctl,
          ((RC_MASK & (c - RC_UNIT)) | (TC_MASK & (c - TC_UNIT)) | (SP_MASK & c))));
    }
    if (w != null)
      w.cancelAll(); // cancel remaining tasks

    if (!tryTerminate(false, false) && // possibly replace worker
        w != null && w.array != null) // avoid repeated failures
      signalWork();

    if (ex == null) // help clean on way out
      ForkJoinTask.helpExpungeStaleExceptions();
    else // rethrow
      ForkJoinTask.rethrow(ex);
  }

  /**
   * Tries to create or release a worker if too few are running.
   */
  final void signalWork() {
    for (;;) {
      long c;
      int sp;
      WorkQueue[] ws;
      int i;
      WorkQueue v;
      if ((c = ctl) >= 0L) // enough workers
        break;
      else if ((sp = (int) c) == 0) { // no idle workers
        if ((c & ADD_WORKER) != 0L) // too few workers
          tryAddWorker(c);
        break;
      } else if ((ws = workQueues) == null)
        break; // unstarted/terminated
      else if (ws.length <= (i = sp & SMASK))
        break; // terminated
      else if ((v = ws[i]) == null)
        break; // terminating
      else {
        int np = sp & ~UNSIGNALLED;
        int vp = v.phase;
        long nc = (v.stackPred & SP_MASK) | (UC_MASK & (c + RC_UNIT));
        Thread vt = v.owner;
        if (sp == vp && U.compareAndSwapLong(this, CTL, c, nc)) {
          v.phase = np;
          if (v.source < 0)
            LockSupport.unpark(vt);
          break;
        }
      }
    }
  }

  /**
   * Tries to decrement counts (sometimes implicitly) and possibly arrange for a compensating worker
   * in preparation for blocking: If not all core workers yet exist, creates one, else if any are
   * unreleased (possibly including caller) releases one, else if fewer than the minimum allowed
   * number of workers running, checks to see that they are all active, and if so creates an extra
   * worker unless over maximum limit and policy is to saturate. Most of these steps can fail due to
   * interference, in which case 0 is returned so caller will retry. A negative return value
   * indicates that the caller doesn't need to re-adjust counts when later unblocked.
   *
   * @return 1: block then adjust, -1: block without adjust, 0 : retry
   */
  private int tryCompensate(WorkQueue w) {
    int t, n, sp;
    long c = ctl;
    WorkQueue[] ws = workQueues;
    if ((t = (short) (c >>> TC_SHIFT)) >= 0) {
      if (ws == null || (n = ws.length) <= 0 || w == null)
        return 0; // disabled
      else if ((sp = (int) c) != 0) { // replace or release
        WorkQueue v = ws[sp & (n - 1)];
        int wp = w.phase;
        long uc = UC_MASK & ((wp < 0) ? c + RC_UNIT : c);
        int np = sp & ~UNSIGNALLED;
        if (v != null) {
          int vp = v.phase;
          Thread vt = v.owner;
          long nc = ((long) v.stackPred & SP_MASK) | uc;
          if (vp == sp && U.compareAndSwapLong(this, CTL, c, nc)) {
            v.phase = np;
            if (v.source < 0)
              LockSupport.unpark(vt);
            return (wp < 0) ? -1 : 1;
          }
        }
        return 0;
      } else if ((int) (c >> RC_SHIFT) - // reduce parallelism
          (short) (bounds & SMASK) > 0) {
        long nc = ((RC_MASK & (c - RC_UNIT)) | (~RC_MASK & c));
        return U.compareAndSwapLong(this, CTL, c, nc) ? 1 : 0;
      } else { // validate
        int md = mode, pc = md & SMASK, tc = pc + t, bc = 0;
        boolean unstable = false;
        for (int i = 1; i < n; i += 2) {
          WorkQueue q;
          Thread wt;
          Thread.State ts;
          if ((q = ws[i]) != null) {
            if (q.source == 0) {
              unstable = true;
              break;
            } else {
              --tc;
              if ((wt = q.owner) != null
                  && ((ts = wt.getState()) == Thread.State.BLOCKED || ts == Thread.State.WAITING))
                ++bc; // worker is blocking
            }
          }
        }
        if (unstable || tc != 0 || ctl != c)
          return 0; // inconsistent
        else if (t + pc >= MAX_CAP || t >= (bounds >>> SWIDTH)) {
          Predicate<? super ForkJoinPool> sat;
          if ((sat = saturate) != null && sat.test(this))
            return -1;
          else if (bc < pc) { // lagging
            Thread.yield(); // for retry spins
            return 0;
          } else
            throw new RejectedExecutionException("Thread limit exceeded replacing blocked worker");
        }
      }
    }

    long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK); // expand pool
    return U.compareAndSwapLong(this, CTL, c, nc) && createWorker() ? 1 : 0;
  }

  /**
   * Top-level runloop for workers, called by ForkJoinWorkerThread.run. See above for explanation.
   */
  final void runWorker(WorkQueue w) {
    WorkQueue[] ws;
    w.growArray(); // allocate queue
    int r = w.id ^ TLRandom.nextSecondarySeed();
    if (r == 0) // initial nonzero seed
      r = 1;
    int lastSignalId = 0; // avoid unneeded signals
    while ((ws = workQueues) != null) {
      boolean nonempty = false; // scan
      for (int n = ws.length, j = n, m = n - 1; j > 0; --j) {
        WorkQueue q;
        int i, b, al;
        ForkJoinTask<?>[] a;
        if ((i = r & m) >= 0 && i < n && // always true
            (q = ws[i]) != null && (b = q.base) - q.top < 0 && (a = q.array) != null
            && (al = a.length) > 0) {
          int qid = q.id; // (never zero)
          int index = (al - 1) & b;
          long offset = ((long) index << ASHIFT) + ABASE;
          ForkJoinTask<?> t = (ForkJoinTask<?>) U.getObjectVolatile(a, offset);
          if (t != null && b++ == q.base && U.compareAndSwapObject(a, offset, t, null)) {
            if ((q.base = b) - q.top < 0 && qid != lastSignalId)
              signalWork(); // propagate signal
            w.source = lastSignalId = qid;
            t.doExec();
            if ((w.id & FIFO) != 0) // run remaining locals
              w.localPollAndExec(POLL_LIMIT);
            else
              w.localPopAndExec(POLL_LIMIT);
            ForkJoinWorkerThread thread = w.owner;
            ++w.nsteals;
            w.source = 0; // now idle
            if (thread != null)
              thread.afterTopLevelExec();
          }
          nonempty = true;
        } else if (nonempty)
          break;
        else
          ++r;
      }

      if (nonempty) { // move (xorshift)
        r ^= r << 13;
        r ^= r >>> 17;
        r ^= r << 5;
      } else {
        int phase;
        lastSignalId = 0; // clear for next scan
        if ((phase = w.phase) >= 0) { // enqueue
          int np = w.phase = (phase + SS_SEQ) | UNSIGNALLED;
          long c, nc;
          do {
            w.stackPred = (int) (c = ctl);
            nc = ((c - RC_UNIT) & UC_MASK) | (SP_MASK & np);
          } while (!U.compareAndSwapLong(this, CTL, c, nc));
        } else { // already queued
          int pred = w.stackPred;
          w.source = DORMANT; // enable signal
          for (int steps = 0;;) {
            int md, rc;
            long c;
            if (w.phase >= 0) {
              w.source = 0;
              break;
            } else if ((md = mode) < 0) // shutting down
              return;
            else if ((rc = ((md & SMASK) + // possibly quiescent
                (int) ((c = ctl) >> RC_SHIFT))) <= 0 && (md & SHUTDOWN) != 0
                && tryTerminate(false, false))
              return; // help terminate
            else if ((++steps & 1) == 0)
              Thread.interrupted(); // clear between parks
            else if (rc <= 0 && pred != 0 && phase == (int) c) {
              long d = keepAlive + System.currentTimeMillis();
              LockSupport.parkUntil(this, d);
              if (ctl == c && d - System.currentTimeMillis() <= TIMEOUT_SLOP) {
                long nc = ((UC_MASK & (c - TC_UNIT)) | (SP_MASK & pred));
                if (U.compareAndSwapLong(this, CTL, c, nc)) {
                  w.phase = QUIET;
                  return; // drop on timeout
                }
              }
            } else
              LockSupport.park(this);
          }
        }
      }
    }
  }

  /**
   * Helps and/or blocks until the given task is done or timeout. First tries locally helping, then
   * scans other queues for a task produced by one of w's stealers; compensating and blocking if
   * none are found (rescanning if tryCompensate fails).
   *
   * @param w caller
   * @param task the task
   * @param deadline for timed waits, if nonzero
   * @return task status on exit
   */
  final int awaitJoin(WorkQueue w, ForkJoinTask<?> task, long deadline) {
    int s = 0;
    if (w != null && task != null && (!(task instanceof CountedCompleter)
        || (s = w.localHelpCC((CountedCompleter<?>) task, 0)) >= 0)) {
      w.tryRemoveAndExec(task);
      int src = w.source, id = w.id;
      s = task.status;
      while (s >= 0) {
        WorkQueue[] ws;
        boolean nonempty = false;
        int r = TLRandom.nextSecondarySeed() | 1; // odd indices
        if ((ws = workQueues) != null) { // scan for matching id
          for (int n = ws.length, m = n - 1, j = -n; j < n; j += 2) {
            WorkQueue q;
            int i, b, al;
            ForkJoinTask<?>[] a;
            if ((i = (r + j) & m) >= 0 && i < n && (q = ws[i]) != null && q.source == id
                && (b = q.base) - q.top < 0 && (a = q.array) != null && (al = a.length) > 0) {
              int qid = q.id;
              int index = (al - 1) & b;
              long offset = ((long) index << ASHIFT) + ABASE;
              ForkJoinTask<?> t = (ForkJoinTask<?>) U.getObjectVolatile(a, offset);
              if (t != null && b++ == q.base && id == q.source
                  && U.compareAndSwapObject(a, offset, t, null)) {
                q.base = b;
                w.source = qid;
                t.doExec();
                w.source = src;
              }
              nonempty = true;
              break;
            }
          }
        }
        if ((s = task.status) < 0)
          break;
        else if (!nonempty) {
          long ms, ns;
          int block;
          if (deadline == 0L)
            ms = 0L; // untimed
          else if ((ns = deadline - System.nanoTime()) <= 0L)
            break; // timeout
          else if ((ms = TimeUnit.NANOSECONDS.toMillis(ns)) <= 0L)
            ms = 1L; // avoid 0 for timed wait
          if ((block = tryCompensate(w)) != 0) {
            task.internalWait(ms);
            U.getAndAddLong(this, CTL, (block > 0) ? RC_UNIT : 0L);
          }
          s = task.status;
        }
      }
    }
    return s;
  }

  /**
   * Runs tasks until {@code isQuiescent()}. Rather than blocking when tasks cannot be found,
   * rescans until all others cannot find tasks either.
   */
  final void helpQuiescePool(WorkQueue w) {
    int prevSrc = w.source, fifo = w.id & FIFO;
    for (int source = prevSrc, released = -1;;) { // -1 until known
      WorkQueue[] ws;
      if (fifo != 0)
        w.localPollAndExec(0);
      else
        w.localPopAndExec(0);
      if (released == -1 && w.phase >= 0)
        released = 1;
      boolean quiet = true, empty = true;
      int r = TLRandom.nextSecondarySeed();
      if ((ws = workQueues) != null) {
        for (int n = ws.length, j = n, m = n - 1; j > 0; --j) {
          WorkQueue q;
          int i, b, al;
          ForkJoinTask<?>[] a;
          if ((i = (r - j) & m) >= 0 && i < n && (q = ws[i]) != null) {
            if ((b = q.base) - q.top < 0 && (a = q.array) != null && (al = a.length) > 0) {
              int qid = q.id;
              if (released == 0) { // increment
                released = 1;
                U.getAndAddLong(this, CTL, RC_UNIT);
              }
              int index = (al - 1) & b;
              long offset = ((long) index << ASHIFT) + ABASE;
              ForkJoinTask<?> t = (ForkJoinTask<?>) U.getObjectVolatile(a, offset);
              if (t != null && b++ == q.base && U.compareAndSwapObject(a, offset, t, null)) {
                q.base = b;
                w.source = source = q.id;
                t.doExec();
                w.source = source = prevSrc;
              }
              quiet = empty = false;
              break;
            } else if ((q.source & QUIET) == 0)
              quiet = false;
          }
        }
      }
      if (quiet) {
        if (released == 0)
          U.getAndAddLong(this, CTL, RC_UNIT);
        w.source = prevSrc;
        break;
      } else if (empty) {
        if (source != QUIET)
          w.source = source = QUIET;
        if (released == 1) { // decrement
          released = 0;
          U.getAndAddLong(this, CTL, RC_MASK & -RC_UNIT);
        }
      }
    }
  }

  /**
   * Scans for and returns a polled task, if available. Used only for untracked polls.
   *
   * @param submissionsOnly if true, only scan submission queues
   */
  private ForkJoinTask<?> pollScan(boolean submissionsOnly) {
    WorkQueue[] ws;
    int n;
    rescan: while ((mode & STOP) == 0 && (ws = workQueues) != null && (n = ws.length) > 0) {
      int m = n - 1;
      int r = TLRandom.nextSecondarySeed();
      int h = r >>> 16;
      int origin, step;
      if (submissionsOnly) {
        origin = (r & ~1) & m; // even indices and steps
        step = (h & ~1) | 2;
      } else {
        origin = r & m;
        step = h | 1;
      }
      for (int k = origin, oldSum = 0, checkSum = 0;;) {
        WorkQueue q;
        int b, al;
        ForkJoinTask<?>[] a;
        if ((q = ws[k]) != null) {
          checkSum += b = q.base;
          if (b - q.top < 0 && (a = q.array) != null && (al = a.length) > 0) {
            int index = (al - 1) & b;
            long offset = ((long) index << ASHIFT) + ABASE;
            ForkJoinTask<?> t = (ForkJoinTask<?>) U.getObjectVolatile(a, offset);
            if (t != null && b++ == q.base && U.compareAndSwapObject(a, offset, t, null)) {
              q.base = b;
              return t;
            } else
              break; // restart
          }
        }
        if ((k = (k + step) & m) == origin) {
          if (oldSum == (oldSum = checkSum))
            break rescan;
          checkSum = 0;
        }
      }
    }
    return null;
  }

  // Execution methods

  /**
   * Gets and removes a local or stolen task for the given worker.
   *
   * @return a task, if available
   */
  final ForkJoinTask<?> nextTaskFor(WorkQueue w) {
    ForkJoinTask<?> t;
    if (w != null && (t = (w.id & FIFO) != 0 ? w.poll() : w.pop()) != null)
      return t;
    else
      return pollScan(false);
  }

  /**
   * Adds the given task to a submission queue at submitter's current queue, creating one if null or
   * contended.
   *
   * @param task the task. Caller must ensure non-null.
   */
  final void externalPush(ForkJoinTask<?> task) {
    int r; // initialize caller's probe
    if ((r = TLRandom.getProbe()) == 0) {
      TLRandom.localInit();
      r = TLRandom.getProbe();
    }
    for (;;) {
      int md = mode, n;
      WorkQueue[] ws = workQueues;
      if ((md & SHUTDOWN) != 0 || ws == null || (n = ws.length) <= 0)
        throw new RejectedExecutionException();
      else {
        WorkQueue q;
        boolean push = false, grow = false;
        if ((q = ws[(n - 1) & r & SQMASK]) == null) {
          Object lock = workerNamePrefix;
          int qid = (r | QUIET) & ~(FIFO | OWNED);
          q = new WorkQueue(this, null);
          q.id = qid;
          q.source = QUIET;
          q.phase = QLOCK; // lock queue
          if (lock != null) {
            synchronized (lock) { // lock pool to install
              int i;
              if ((ws = workQueues) != null && (n = ws.length) > 0
                  && ws[i = qid & (n - 1) & SQMASK] == null) {
                ws[i] = q;
                push = grow = true;
              }
            }
          }
        } else if (q.tryLockSharedQueue()) {
          int b = q.base, s = q.top, al, d;
          ForkJoinTask<?>[] a;
          if ((a = q.array) != null && (al = a.length) > 0 && al - 1 + (d = b - s) > 0) {
            a[(al - 1) & s] = task;
            q.top = s + 1; // relaxed writes OK here
            q.phase = 0;
            if (d < 0 && q.base - s < -1)
              break; // no signal needed
          } else
            grow = true;
          push = true;
        }
        if (push) {
          if (grow) {
            try {
              q.growArray();
              int s = q.top, al;
              ForkJoinTask<?>[] a;
              if ((a = q.array) != null && (al = a.length) > 0) {
                a[(al - 1) & s] = task;
                q.top = s + 1;
              }
            } finally {
              q.phase = 0;
            }
          }
          signalWork();
          break;
        } else // move if busy
          r = TLRandom.advanceProbe(r);
      }
    }
  }

  // AbstractExecutorService methods

  /**
   * Pushes a possibly-external submission.
   */
  private <T> ForkJoinTask<T> externalSubmit(ForkJoinTask<T> task) {
    Thread t;
    ForkJoinWorkerThread w;
    WorkQueue q;
    Objects.requireNonNull(task);
    if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
        && (w = (ForkJoinWorkerThread) t).pool == this && (q = w.workQueue) != null)
      q.push(task);
    else
      externalPush(task);
    return task;
  }

  /**
   * Performs tryUnpush for an external submitter.
   */
  final boolean tryExternalUnpush(ForkJoinTask<?> task) {
    int r = TLRandom.getProbe();
    WorkQueue[] ws;
    WorkQueue w;
    int n;
    return ((ws = workQueues) != null && (n = ws.length) > 0
        && (w = ws[(n - 1) & r & SQMASK]) != null && w.trySharedUnpush(task));
  }

  /**
   * Performs helpComplete for an external submitter.
   */
  final int externalHelpComplete(CountedCompleter<?> task, int maxTasks) {
    int r = TLRandom.getProbe();
    WorkQueue[] ws;
    WorkQueue w;
    int n;
    return ((ws = workQueues) != null && (n = ws.length) > 0
        && (w = ws[(n - 1) & r & SQMASK]) != null) ? w.sharedHelpCC(task, maxTasks) : 0;
  }

  /**
   * Tries to steal and run tasks within the target's computation. The maxTasks argument supports
   * external usages; internal calls use zero, allowing unbounded steps (external calls trap
   * non-positive values).
   *
   * @param w caller
   * @param maxTasks if non-zero, the maximum number of other tasks to run
   * @return task status on exit
   */
  final int helpComplete(WorkQueue w, CountedCompleter<?> task, int maxTasks) {
    return (w == null) ? 0 : w.localHelpCC(task, maxTasks);
  }

  /**
   * Possibly initiates and/or completes termination.
   *
   * @param now if true, unconditionally terminate, else only if no work and no active workers
   * @param enable if true, terminate when next possible
   * @return true if terminating or terminated
   */
  private boolean tryTerminate(boolean now, boolean enable) {
    int md; // 3 phases: try to set SHUTDOWN, then STOP, then TERMINATED

    while (((md = mode) & SHUTDOWN) == 0) {
      if (!enable || this == common) // cannot shutdown
        return false;
      else
        U.compareAndSwapInt(this, MODE, md, md | SHUTDOWN);
    }

    while (((md = mode) & STOP) == 0) { // try to initiate termination
      if (!now) { // check if quiescent & empty
        for (long oldSum = 0L;;) { // repeat until stable
          boolean running = false;
          long checkSum = ctl;
          WorkQueue[] ws = workQueues;
          if ((md & SMASK) + (int) (checkSum >> RC_SHIFT) > 0)
            running = true;
          else if (ws != null) {
            WorkQueue w;
            for (int i = 0; i < ws.length; ++i) {
              if ((w = ws[i]) != null) {
                int s = w.source, p = w.phase;
                int d = w.id, b = w.base;
                if (b != w.top || ((d & 1) == 1 && (s >= 0 || p >= 0))) {
                  running = true;
                  break; // working, scanning, or have work
                }
                checkSum += (((long) s << 48) + ((long) p << 32) + ((long) b << 16) + (long) d);
              }
            }
          }
          if (((md = mode) & STOP) != 0)
            break; // already triggered
          else if (running)
            return false;
          else if (workQueues == ws && oldSum == (oldSum = checkSum))
            break;
        }
      }
      if ((md & STOP) == 0)
        U.compareAndSwapInt(this, MODE, md, md | STOP);
    }

    while (((md = mode) & TERMINATED) == 0) { // help terminate others
      for (long oldSum = 0L;;) { // repeat until stable
        WorkQueue[] ws;
        WorkQueue w;
        long checkSum = ctl;
        if ((ws = workQueues) != null) {
          for (int i = 0; i < ws.length; ++i) {
            if ((w = ws[i]) != null) {
              ForkJoinWorkerThread wt = w.owner;
              w.cancelAll(); // clear queues
              if (wt != null) {
                try { // unblock join or park
                  wt.interrupt();
                } catch (Throwable ignore) {
                }
              }
              checkSum += ((long) w.phase << 32) + w.base;
            }
          }
        }
        if (((md = mode) & TERMINATED) != 0 || (workQueues == ws && oldSum == (oldSum = checkSum)))
          break;
      }
      if ((md & TERMINATED) != 0)
        break;
      else if ((md & SMASK) + (short) (ctl >>> TC_SHIFT) > 0)
        break;
      else if (U.compareAndSwapInt(this, MODE, md, md | TERMINATED)) {
        synchronized (this) {
          notifyAll(); // for awaitTermination
        }
        break;
      }
    }
    return true;
  }

  /**
   * Performs the given task, returning its result upon completion. If the computation encounters an
   * unchecked Exception or Error, it is rethrown as the outcome of this invocation. Rethrown
   * exceptions behave in the same way as regular exceptions, but, when possible, contain stack
   * traces (as displayed for example using {@code ex.printStackTrace()}) of both the current thread
   * as well as the thread actually encountering the exception; minimally only the latter.
   *
   * @param task the task
   * @param <T> the type of the task's result
   * @return the task's result
   * @throws NullPointerException if the task is null
   * @throws RejectedExecutionException if the task cannot be scheduled for execution
   */
  public <T> T invoke(ForkJoinTask<T> task) {
    externalSubmit(Objects.requireNonNull(task));
    return task.join();
  }

  /**
   * Arranges for (asynchronous) execution of the given task.
   *
   * @param task the task
   * @throws NullPointerException if the task is null
   * @throws RejectedExecutionException if the task cannot be scheduled for execution
   */
  public void execute(ForkJoinTask<?> task) {
    externalSubmit(task);
  }

  /**
   * @throws NullPointerException if the task is null
   * @throws RejectedExecutionException if the task cannot be scheduled for execution
   */
  public void execute(Runnable task) {
    Objects.requireNonNull(task);
    ForkJoinTask<?> job;
    if (task instanceof ForkJoinTask<?>) // avoid re-wrap
      job = (ForkJoinTask<?>) task;
    else
      job = new ForkJoinTask.RunnableExecuteAction(task);
    externalSubmit(job);
  }

  /**
   * Submits a ForkJoinTask for execution.
   *
   * @param task the task to submit
   * @param <T> the type of the task's result
   * @return the task
   * @throws NullPointerException if the task is null
   * @throws RejectedExecutionException if the task cannot be scheduled for execution
   */
  public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
    return externalSubmit(task);
  }

  /**
   * @throws NullPointerException if the task is null
   * @throws RejectedExecutionException if the task cannot be scheduled for execution
   */
  public <T> ForkJoinTask<T> submit(Callable<T> task) {
    return externalSubmit(new ForkJoinTask.AdaptedCallable<T>(task));
  }

  /**
   * @throws NullPointerException if the task is null
   * @throws RejectedExecutionException if the task cannot be scheduled for execution
   */
  public <T> ForkJoinTask<T> submit(Runnable task, T result) {
    return externalSubmit(new ForkJoinTask.AdaptedRunnable<T>(task, result));
  }

  /**
   * @throws NullPointerException if the task is null
   * @throws RejectedExecutionException if the task cannot be scheduled for execution
   */
  @SuppressWarnings("unchecked")
  public ForkJoinTask<?> submit(Runnable task) {
    Objects.requireNonNull(task);
    return externalSubmit((task instanceof ForkJoinTask<?>) ? (ForkJoinTask<Void>) task // avoid
                                                                                        // re-wrap
        : new ForkJoinTask.AdaptedRunnableAction(task));
  }

  /**
   * @throws NullPointerException {@inheritDoc}
   * @throws RejectedExecutionException {@inheritDoc}
   */
  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
    // In previous versions of this class, this method constructed
    // a task to run ForkJoinTask.invokeAll, but now external
    // invocation of multiple tasks is at least as efficient.
    ArrayList<Future<T>> futures = new ArrayList<>(tasks.size());

    try {
      for (Callable<T> t : tasks) {
        ForkJoinTask<T> f = new ForkJoinTask.AdaptedCallable<T>(t);
        futures.add(f);
        externalSubmit(f);
      }
      for (int i = 0, size = futures.size(); i < size; i++)
        ((ForkJoinTask<?>) futures.get(i)).quietlyJoin();
      return futures;
    } catch (Throwable t) {
      for (int i = 0, size = futures.size(); i < size; i++)
        futures.get(i).cancel(false);
      throw t;
    }
  }

  /**
   * Returns the factory used for constructing new workers.
   *
   * @return the factory used for constructing new workers
   */
  public ForkJoinWorkerThreadFactory getFactory() {
    return factory;
  }

  /**
   * Returns the handler for internal worker threads that terminate due to unrecoverable errors
   * encountered while executing tasks.
   *
   * @return the handler, or {@code null} if none
   */
  public UncaughtExceptionHandler getUncaughtExceptionHandler() {
    return ueh;
  }

  /**
   * Returns the targeted parallelism level of this pool.
   *
   * @return the targeted parallelism level of this pool
   */
  public int getParallelism() {
    int par = mode & SMASK;
    return (par > 0) ? par : 1;
  }

  /**
   * Returns the number of worker threads that have started but not yet terminated. The result
   * returned by this method may differ from {@link #getParallelism} when threads are created to
   * maintain parallelism when others are cooperatively blocked.
   *
   * @return the number of worker threads
   */
  public int getPoolSize() {
    return ((mode & SMASK) + (short) (ctl >>> TC_SHIFT));
  }

  /**
   * Returns {@code true} if this pool uses local first-in-first-out scheduling mode for forked
   * tasks that are never joined.
   *
   * @return {@code true} if this pool uses async mode
   */
  public boolean getAsyncMode() {
    return (mode & FIFO) != 0;
  }

  /**
   * Returns an estimate of the number of worker threads that are not blocked waiting to join tasks
   * or for other managed synchronization. This method may overestimate the number of running
   * threads.
   *
   * @return the number of worker threads
   */
  public int getRunningThreadCount() {
    int rc = 0;
    WorkQueue[] ws;
    WorkQueue w;
    if ((ws = workQueues) != null) {
      for (int i = 1; i < ws.length; i += 2) {
        if ((w = ws[i]) != null && w.isApparentlyUnblocked())
          ++rc;
      }
    }
    return rc;
  }

  /**
   * Returns an estimate of the number of threads that are currently stealing or executing tasks.
   * This method may overestimate the number of active threads.
   *
   * @return the number of active threads
   */
  public int getActiveThreadCount() {
    int r = (mode & SMASK) + (int) (ctl >> RC_SHIFT);
    return (r <= 0) ? 0 : r; // suppress momentarily negative values
  }

  /**
   * Returns {@code true} if all worker threads are currently idle. An idle worker is one that
   * cannot obtain a task to execute because none are available to steal from other threads, and
   * there are no pending submissions to the pool. This method is conservative; it might not return
   * {@code true} immediately upon idleness of all threads, but will eventually become true if
   * threads remain inactive.
   *
   * @return {@code true} if all threads are currently idle
   */
  public boolean isQuiescent() {
    for (;;) {
      long c = ctl;
      int md = mode, pc = md & SMASK;
      int tc = pc + (short) (c >>> TC_SHIFT);
      int rc = pc + (int) (c >> RC_SHIFT);
      if ((md & (STOP | TERMINATED)) != 0)
        return true;
      else if (rc > 0)
        return false;
      else {
        WorkQueue[] ws;
        WorkQueue v;
        if ((ws = workQueues) != null) {
          for (int i = 1; i < ws.length; i += 2) {
            if ((v = ws[i]) != null) {
              if ((v.source & QUIET) == 0)
                return false;
              --tc;
            }
          }
        }
        if (tc == 0 && ctl == c)
          return true;
      }
    }
  }

  /**
   * Returns an estimate of the total number of tasks stolen from one thread's work queue by
   * another. The reported value underestimates the actual total number of steals when the pool is
   * not quiescent. This value may be useful for monitoring and tuning fork/join programs: in
   * general, steal counts should be high enough to keep threads busy, but low enough to avoid
   * overhead and contention across threads.
   *
   * @return the number of steals
   */
  public long getStealCount() {
    long count = stealCount;
    WorkQueue[] ws;
    WorkQueue w;
    if ((ws = workQueues) != null) {
      for (int i = 1; i < ws.length; i += 2) {
        if ((w = ws[i]) != null)
          count += (long) w.nsteals & 0xffffffffL;
      }
    }
    return count;
  }

  /**
   * Returns an estimate of the total number of tasks currently held in queues by worker threads
   * (but not including tasks submitted to the pool that have not begun executing). This value is
   * only an approximation, obtained by iterating across all threads in the pool. This method may be
   * useful for tuning task granularities.
   *
   * @return the number of queued tasks
   */
  public long getQueuedTaskCount() {
    long count = 0;
    WorkQueue[] ws;
    WorkQueue w;
    if ((ws = workQueues) != null) {
      for (int i = 1; i < ws.length; i += 2) {
        if ((w = ws[i]) != null)
          count += w.queueSize();
      }
    }
    return count;
  }

  /**
   * Returns an estimate of the number of tasks submitted to this pool that have not yet begun
   * executing. This method may take time proportional to the number of submissions.
   *
   * @return the number of queued submissions
   */
  public int getQueuedSubmissionCount() {
    int count = 0;
    WorkQueue[] ws;
    WorkQueue w;
    if ((ws = workQueues) != null) {
      for (int i = 0; i < ws.length; i += 2) {
        if ((w = ws[i]) != null)
          count += w.queueSize();
      }
    }
    return count;
  }

  /**
   * Returns {@code true} if there are any tasks submitted to this pool that have not yet begun
   * executing.
   *
   * @return {@code true} if there are any queued submissions
   */
  public boolean hasQueuedSubmissions() {
    WorkQueue[] ws;
    WorkQueue w;
    if ((ws = workQueues) != null) {
      for (int i = 0; i < ws.length; i += 2) {
        if ((w = ws[i]) != null && !w.isEmpty())
          return true;
      }
    }
    return false;
  }

  /**
   * Removes and returns the next unexecuted submission if one is available. This method may be
   * useful in extensions to this class that re-assign work in systems with multiple pools.
   *
   * @return the next submission, or {@code null} if none
   */
  protected ForkJoinTask<?> pollSubmission() {
    return pollScan(true);
  }

  /**
   * Removes all available unexecuted submitted and forked tasks from scheduling queues and adds
   * them to the given collection, without altering their execution status. These may include
   * artificially generated or wrapped tasks. This method is designed to be invoked only when the
   * pool is known to be quiescent. Invocations at other times may not remove all tasks. A failure
   * encountered while attempting to add elements to collection {@code c} may result in elements
   * being in neither, either or both collections when the associated exception is thrown. The
   * behavior of this operation is undefined if the specified collection is modified while the
   * operation is in progress.
   *
   * @param c the collection to transfer elements into
   * @return the number of elements transferred
   */
  protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
    int count = 0;
    WorkQueue[] ws;
    WorkQueue w;
    ForkJoinTask<?> t;
    if ((ws = workQueues) != null) {
      for (int i = 0; i < ws.length; ++i) {
        if ((w = ws[i]) != null) {
          while ((t = w.poll()) != null) {
            c.add(t);
            ++count;
          }
        }
      }
    }
    return count;
  }

  /**
   * Returns a string identifying this pool, as well as its state, including indications of run
   * state, parallelism level, and worker and task counts.
   *
   * @return a string identifying this pool, as well as its state
   */
  public String toString() {
    // Use a single pass through workQueues to collect counts
    long qt = 0L, qs = 0L;
    int rc = 0;
    long st = stealCount;
    WorkQueue[] ws;
    WorkQueue w;
    if ((ws = workQueues) != null) {
      for (int i = 0; i < ws.length; ++i) {
        if ((w = ws[i]) != null) {
          int size = w.queueSize();
          if ((i & 1) == 0)
            qs += size;
          else {
            qt += size;
            st += (long) w.nsteals & 0xffffffffL;
            if (w.isApparentlyUnblocked())
              ++rc;
          }
        }
      }
    }

    int md = mode;
    int pc = (md & SMASK);
    long c = ctl;
    int tc = pc + (short) (c >>> TC_SHIFT);
    int ac = pc + (int) (c >> RC_SHIFT);
    if (ac < 0) // ignore transient negative
      ac = 0;
    String level = ((md & TERMINATED) != 0 ? "Terminated"
        : (md & STOP) != 0 ? "Terminating" : (md & SHUTDOWN) != 0 ? "Shutting down" : "Running");
    return super.toString() + "[" + level + ", parallelism = " + pc + ", size = " + tc
        + ", active = " + ac + ", running = " + rc + ", steals = " + st + ", tasks = " + qt
        + ", submissions = " + qs + "]";
  }

  /**
   * Possibly initiates an orderly shutdown in which previously submitted tasks are executed, but no
   * new tasks will be accepted. Invocation has no effect on execution state if this is the
   * {@link #commonPool()}, and no additional effect if already shut down. Tasks that are in the
   * process of being submitted concurrently during the course of this method may or may not be
   * rejected.
   *
   * @throws SecurityException if a security manager exists and the caller is not permitted to
   *         modify threads because it does not hold
   *         {@link java.lang.RuntimePermission}{@code ("modifyThread")}
   */
  public void shutdown() {
    checkPermission();
    tryTerminate(false, true);
  }

  /**
   * Possibly attempts to cancel and/or stop all tasks, and reject all subsequently submitted tasks.
   * Invocation has no effect on execution state if this is the {@link #commonPool()}, and no
   * additional effect if already shut down. Otherwise, tasks that are in the process of being
   * submitted or executed concurrently during the course of this method may or may not be rejected.
   * This method cancels both existing and unexecuted tasks, in order to permit termination in the
   * presence of task dependencies. So the method always returns an empty list (unlike the case for
   * some other Executors).
   *
   * @return an empty list
   * @throws SecurityException if a security manager exists and the caller is not permitted to
   *         modify threads because it does not hold
   *         {@link java.lang.RuntimePermission}{@code ("modifyThread")}
   */
  public List<Runnable> shutdownNow() {
    checkPermission();
    tryTerminate(true, true);
    return Collections.emptyList();
  }

  /**
   * Returns {@code true} if all tasks have completed following shut down.
   *
   * @return {@code true} if all tasks have completed following shut down
   */
  public boolean isTerminated() {
    return (mode & TERMINATED) != 0;
  }

  /**
   * Returns {@code true} if the process of termination has commenced but not yet completed. This
   * method may be useful for debugging. A return of {@code true} reported a sufficient period after
   * shutdown may indicate that submitted tasks have ignored or suppressed interruption, or are
   * waiting for I/O, causing this executor not to properly terminate. (See the advisory notes for
   * class {@link ForkJoinTask} stating that tasks should not normally entail blocking operations.
   * But if they do, they must abort them on interrupt.)
   *
   * @return {@code true} if terminating but not yet terminated
   */
  public boolean isTerminating() {
    int md = mode;
    return (md & STOP) != 0 && (md & TERMINATED) == 0;
  }

  /**
   * Returns {@code true} if this pool has been shut down.
   *
   * @return {@code true} if this pool has been shut down
   */
  public boolean isShutdown() {
    return (mode & SHUTDOWN) != 0;
  }

  // AbstractExecutorService overrides. These rely on undocumented
  // fact that ForkJoinTask.adapt returns ForkJoinTasks that also
  // implement RunnableFuture.

  /**
   * Blocks until all tasks have completed execution after a shutdown request, or the timeout
   * occurs, or the current thread is interrupted, whichever happens first. Because the
   * {@link #commonPool()} never terminates until program shutdown, when applied to the common pool,
   * this method is equivalent to {@link #awaitQuiescence(long, TimeUnit)} but always returns
   * {@code false}.
   *
   * @param timeout the maximum time to wait
   * @param unit the time unit of the timeout argument
   * @return {@code true} if this executor terminated and {@code false} if the timeout elapsed
   *         before termination
   * @throws InterruptedException if interrupted while waiting
   */
  public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
    if (Thread.interrupted())
      throw new InterruptedException();
    if (this == common) {
      awaitQuiescence(timeout, unit);
      return false;
    }
    long nanos = unit.toNanos(timeout);
    if (isTerminated())
      return true;
    if (nanos <= 0L)
      return false;
    long deadline = System.nanoTime() + nanos;
    synchronized (this) {
      for (;;) {
        if (isTerminated())
          return true;
        if (nanos <= 0L)
          return false;
        long millis = TimeUnit.NANOSECONDS.toMillis(nanos);
        wait(millis > 0L ? millis : 1L);
        nanos = deadline - System.nanoTime();
      }
    }
  }

  /**
   * If called by a ForkJoinTask operating in this pool, equivalent in effect to
   * {@link ForkJoinTask#helpQuiesce}. Otherwise, waits and/or attempts to assist performing tasks
   * until this pool {@link #isQuiescent} or the indicated timeout elapses.
   *
   * @param timeout the maximum time to wait
   * @param unit the time unit of the timeout argument
   * @return {@code true} if quiescent; {@code false} if the timeout elapsed.
   */
  public boolean awaitQuiescence(long timeout, TimeUnit unit) {
    long nanos = unit.toNanos(timeout);
    ForkJoinWorkerThread wt;
    Thread thread = Thread.currentThread();
    if ((thread instanceof ForkJoinWorkerThread)
        && (wt = (ForkJoinWorkerThread) thread).pool == this) {
      helpQuiescePool(wt.workQueue);
      return true;
    } else {
      for (long startTime = System.nanoTime();;) {
        ForkJoinTask<?> t;
        if ((t = pollScan(false)) != null)
          t.doExec();
        else if (isQuiescent())
          return true;
        else if ((System.nanoTime() - startTime) > nanos)
          return false;
        else
          Thread.yield(); // cannot block
      }
    }
  }

  protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new ForkJoinTask.AdaptedRunnable<T>(runnable, value);
  }

  protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
    return new ForkJoinTask.AdaptedCallable<T>(callable);
  }

  /**
   * Factory for creating new {@link ForkJoinWorkerThread}s. A {@code ForkJoinWorkerThreadFactory}
   * must be defined and used for {@code ForkJoinWorkerThread} subclasses that extend base
   * functionality or initialize threads with different contexts.
   */
  public static interface ForkJoinWorkerThreadFactory {
    /**
     * Returns a new worker thread operating in the given pool. Returning null or throwing an
     * exception may result in tasks never being executed. If this method throws an exception, it is
     * relayed to the caller of the method (for example {@code execute}) causing attempted thread
     * creation. If this method returns null or throws an exception, it is not retried until the
     * next attempted creation (for example another call to {@code execute}).
     *
     * @param pool the pool this thread works in
     * @return the new worker thread, or {@code null} if the request to create a thread is rejected
     * @throws NullPointerException if the pool is null
     */
    public ForkJoinWorkerThread newThread(ForkJoinPool pool);
  }
  /**
   * Interface for extending managed parallelism for tasks running in {@link ForkJoinPool}s.
   *
   * <p>
   * A {@code ManagedBlocker} provides two methods. Method {@link #isReleasable} must return
   * {@code true} if blocking is not necessary. Method {@link #block} blocks the current thread if
   * necessary (perhaps internally invoking {@code isReleasable} before actually blocking). These
   * actions are performed by any thread invoking {@link ForkJoinPool#managedBlock(ManagedBlocker)}.
   * The unusual methods in this API accommodate synchronizers that may, but don't usually, block
   * for long periods. Similarly, they allow more efficient internal handling of cases in which
   * additional workers may be, but usually are not, needed to ensure sufficient parallelism. Toward
   * this end, implementations of method {@code isReleasable} must be amenable to repeated
   * invocation.
   *
   * <p>
   * For example, here is a ManagedBlocker based on a ReentrantLock:
   *
   * <pre>
   * {
   *   &#64;code
   *   class ManagedLocker implements ManagedBlocker {
   *     final ReentrantLock lock;
   *     boolean hasLock = false;
   *
   *     ManagedLocker(ReentrantLock lock) {
   *       this.lock = lock;
   *     }
   *
   *     public boolean block() {
   *       if (!hasLock)
   *         lock.lock();
   *       return true;
   *     }
   *
   *     public boolean isReleasable() {
   *       return hasLock || (hasLock = lock.tryLock());
   *     }
   *   }
   * }
   * </pre>
   *
   * <p>
   * Here is a class that possibly blocks waiting for an item on a given queue:
   *
   * <pre>
   * {
   *   &#64;code
   *   class QueueTaker<E> implements ManagedBlocker {
   *     final BlockingQueue<E> queue;
   *     volatile E item = null;
   *
   *     QueueTaker(BlockingQueue<E> q) {
   *       this.queue = q;
   *     }
   *
   *     public boolean block() throws InterruptedException {
   *       if (item == null)
   *         item = queue.take();
   *       return true;
   *     }
   *
   *     public boolean isReleasable() {
   *       return item != null || (item = queue.poll()) != null;
   *     }
   *
   *     public E getItem() { // call after pool.managedBlock completes
   *       return item;
   *     }
   *   }
   * }
   * </pre>
   */
  public static interface ManagedBlocker {
    /**
     * Possibly blocks the current thread, for example waiting for a lock or condition.
     *
     * @return {@code true} if no additional blocking is necessary (i.e., if isReleasable would
     *         return true)
     * @throws InterruptedException if interrupted while waiting (the method is not required to do
     *         so, but is allowed to)
     */
    boolean block() throws InterruptedException;

    /**
     * Returns {@code true} if blocking is unnecessary.
     *
     * @return {@code true} if blocking is unnecessary
     */
    boolean isReleasable();
  }

  /**
   * Default ForkJoinWorkerThreadFactory implementation; creates a new ForkJoinWorkerThread using
   * the system class loader as the thread context class loader.
   */
  private static final class DefaultForkJoinWorkerThreadFactory
      implements ForkJoinWorkerThreadFactory {
    private static final AccessControlContext ACC = contextWithPermissions(
        // new RuntimePermission("setContextClassLoader"), // java9-concurrent-backport changed
        new RuntimePermission("getClassLoader"));

    public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
      return AccessController.doPrivileged(new PrivilegedAction<ForkJoinWorkerThread>() {
        public ForkJoinWorkerThread run() {
          return new ForkJoinWorkerThread(pool, ClassLoader.getSystemClassLoader());
        }
      }, ACC);
    }
  }

  /**
   * Default ForkJoinWorkerThreadFactory implementation; creates a new ForkJoinWorkerThread using
   * the system class loader as the thread context class loader.
   */
  public static final class AlluxioForkJoinWorkerThreadFactory
          implements ForkJoinWorkerThreadFactory {
    private static final AccessControlContext ACC = contextWithPermissions(
            // new RuntimePermission("setContextClassLoader"), // java9-concurrent-backport changed
            new RuntimePermission("getClassLoader"));
    // ForkJoinWorkerThread index counter.
    private static final AtomicLong sThreadIndex = new AtomicLong(0);
    // Thread properties.
    private final String mNameFormat;
    private final boolean mIsDaemon;

    /**
     * Creates a new thread-factory for {@link ForkJoinPool}.
     *
     * @param threadNameFormat thread name format
     * @param isDaemon is daemon
     */
    public AlluxioForkJoinWorkerThreadFactory(String threadNameFormat, boolean isDaemon){
      mNameFormat = threadNameFormat;
      mIsDaemon = isDaemon;
    }

    public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
      return AccessController.doPrivileged(new PrivilegedAction<ForkJoinWorkerThread>() {
        public ForkJoinWorkerThread run() {
          ForkJoinWorkerThread th =
              new ForkJoinWorkerThread(pool, ClassLoader.getSystemClassLoader());
          th.setName(String.format(mNameFormat, sThreadIndex.getAndIncrement()));
          th.setDaemon(mIsDaemon);
          return th;
        }
      }, ACC);
    }
  }

  /**
   * Queues supporting work-stealing as well as external task submission. See above for descriptions
   * and algorithms. Performance on most platforms is very sensitive to placement of instances of
   * both WorkQueues and their arrays -- we absolutely do not want multiple WorkQueue instances or
   * multiple queue arrays sharing cache lines.
   */
  // For now, using manual padding.
  static final class WorkQueue {

    /**
     * Capacity of work-stealing queue array upon initialization. Must be a power of two; at least
     * 4, but should be larger to reduce or eliminate cacheline sharing among queues. Currently, it
     * is much larger, as a partial workaround for the fact that JVMs often place arrays in
     * locations that share GC bookkeeping (especially cardmarks) such that per-write accesses
     * encounter serious memory contention.
     */
    static final int INITIAL_QUEUE_CAPACITY = 1 << 13;

    /**
     * Maximum size for queue arrays. Must be a power of two less than or equal to 1 << (31 - width
     * of array entry) to ensure lack of wraparound of index calculations, but defined to a value a
     * bit less than this to help users trap runaway programs before saturating systems.
     */
    static final int MAXIMUM_QUEUE_CAPACITY = 1 << 26; // 64M
    // Unsafe mechanics. Note that some are (and must be) the same as in FJP
    private static final sun.misc.Unsafe U = UnsafeAccess.unsafe;
    private static final long PHASE;
    private static final int ABASE;
    private static final int ASHIFT;

    static {
      try {
        PHASE = U.objectFieldOffset(WorkQueue.class.getDeclaredField("phase"));
        ABASE = U.arrayBaseOffset(ForkJoinTask[].class);
        int scale = U.arrayIndexScale(ForkJoinTask[].class);
        if ((scale & (scale - 1)) != 0)
          throw new ExceptionInInitializerError("array index scale not a power of two");
        ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
      } catch (Exception e) {
        throw new ExceptionInInitializerError(e);
      }
    }

    final ForkJoinPool pool; // the containing pool (may be null)
    final ForkJoinWorkerThread owner; // owning thread or null if shared
    // Instance fields
    volatile long pad00, pad01, pad02, pad03, pad04, pad05, pad06, pad07;
    volatile long pad08, pad09, pad0a, pad0b, pad0c, pad0d, pad0e, pad0f;
    volatile int phase; // versioned, negative: queued, 1: locked
    int stackPred; // pool stack (ctl) predecessor link
    int nsteals; // number of steals
    int id; // index, mode, tag
    volatile int source; // source queue id, or sentinel
    volatile int base; // index of next slot for poll
    int top; // index of next slot for push
    ForkJoinTask<?>[] array; // the elements (initially unallocated)
    volatile Object pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17;
    volatile Object pad18, pad19, pad1a, pad1b, pad1c, pad1d, pad1e, pad1f;

    WorkQueue(ForkJoinPool pool, ForkJoinWorkerThread owner) {
      this.pool = pool;
      this.owner = owner;
      // Place indices in the center of array (that is not yet allocated)
      base = top = INITIAL_QUEUE_CAPACITY >>> 1;
    }

    /**
     * Returns an exportable index (used by ForkJoinWorkerThread).
     */
    final int getPoolIndex() {
      return (id & 0xffff) >>> 1; // ignore odd/even tag bit
    }

    /**
     * Returns the approximate number of tasks in the queue.
     */
    final int queueSize() {
      int n = base - top; // read base first
      return (n >= 0) ? 0 : -n; // ignore transient negative
    }

    /**
     * Provides a more accurate estimate of whether this queue has any tasks than does queueSize, by
     * checking whether a near-empty queue has at least one unclaimed task.
     */
    final boolean isEmpty() {
      ForkJoinTask<?>[] a;
      int n, al, b;
      return ((n = (b = base) - top) >= 0 || // possibly one task
          (n == -1 && ((a = array) == null || (al = a.length) == 0 || a[(al - 1) & b] == null)));
    }

    /**
     * Pushes a task. Call only by owner in unshared queues.
     *
     * @param task the task. Caller must ensure non-null.
     * @throws RejectedExecutionException if array cannot be resized
     */
    final void push(ForkJoinTask<?> task) {
      int s = top;
      ForkJoinTask<?>[] a;
      int al, d;
      if ((a = array) != null && (al = a.length) > 0) {
        int index = (al - 1) & s;
        long offset = ((long) index << ASHIFT) + ABASE;
        ForkJoinPool p = pool;
        top = s + 1;
        U.putOrderedObject(a, offset, task);
        if ((d = base - s) == 0 && p != null) {
          U.fullFence();
          p.signalWork();
        } else if (d + al == 1)
          growArray();
      }
    }

    /**
     * Initializes or doubles the capacity of array. Call either by owner or with lock held -- it is
     * OK for base, but not top, to move while resizings are in progress.
     */
    final ForkJoinTask<?>[] growArray() {
      ForkJoinTask<?>[] oldA = array;
      int oldSize = oldA != null ? oldA.length : 0;
      int size = oldSize > 0 ? oldSize << 1 : INITIAL_QUEUE_CAPACITY;
      if (size < INITIAL_QUEUE_CAPACITY || size > MAXIMUM_QUEUE_CAPACITY)
        throw new RejectedExecutionException("Queue capacity exceeded");
      int oldMask, t, b;
      ForkJoinTask<?>[] a = array = new ForkJoinTask<?>[size];
      if (oldA != null && (oldMask = oldSize - 1) > 0 && (t = top) - (b = base) > 0) {
        int mask = size - 1;
        do { // emulate poll from old array, push to new array
          int index = b & oldMask;
          long offset = ((long) index << ASHIFT) + ABASE;
          ForkJoinTask<?> x = (ForkJoinTask<?>) U.getObjectVolatile(oldA, offset);
          if (x != null && U.compareAndSwapObject(oldA, offset, x, null))
            a[b & mask] = x;
        } while (++b != t);
        U.storeFence();
      }
      return a;
    }

    /**
     * Takes next task, if one exists, in LIFO order. Call only by owner in unshared queues.
     */
    final ForkJoinTask<?> pop() {
      int b = base, s = top, al;
      ForkJoinTask<?>[] a;
      if ((a = array) != null && b != s && (al = a.length) > 0) {
        int index = (al - 1) & --s;
        long offset = ((long) index << ASHIFT) + ABASE;
        ForkJoinTask<?> t = (ForkJoinTask<?>) U.getObject(a, offset);
        if (t != null && U.compareAndSwapObject(a, offset, t, null)) {
          top = s;
          U.storeFence();
          return t;
        }
      }
      return null;
    }

    // Specialized execution methods

    /**
     * Takes next task, if one exists, in FIFO order.
     */
    final ForkJoinTask<?> poll() {
      for (;;) {
        int b = base, s = top, d, al;
        ForkJoinTask<?>[] a;
        if ((a = array) != null && (d = b - s) < 0 && (al = a.length) > 0) {
          int index = (al - 1) & b;
          long offset = ((long) index << ASHIFT) + ABASE;
          ForkJoinTask<?> t = (ForkJoinTask<?>) U.getObjectVolatile(a, offset);
          if (b++ == base) {
            if (t != null) {
              if (U.compareAndSwapObject(a, offset, t, null)) {
                base = b;
                return t;
              }
            } else if (d == -1)
              break; // now empty
          }
        } else
          break;
      }
      return null;
    }

    /**
     * Takes next task, if one exists, in order specified by mode.
     */
    final ForkJoinTask<?> nextLocalTask() {
      return ((id & FIFO) != 0) ? poll() : pop();
    }

    /**
     * Returns next task, if one exists, in order specified by mode.
     */
    final ForkJoinTask<?> peek() {
      int al;
      ForkJoinTask<?>[] a;
      return ((a = array) != null && (al = a.length) > 0)
          ? a[(al - 1) & ((id & FIFO) != 0 ? base : top - 1)]
          : null;
    }

    /**
     * Pops the given task only if it is at the current top.
     */
    final boolean tryUnpush(ForkJoinTask<?> task) {
      int b = base, s = top, al;
      ForkJoinTask<?>[] a;
      if ((a = array) != null && b != s && (al = a.length) > 0) {
        int index = (al - 1) & --s;
        long offset = ((long) index << ASHIFT) + ABASE;
        if (U.compareAndSwapObject(a, offset, task, null)) {
          top = s;
          U.storeFence();
          return true;
        }
      }
      return false;
    }

    // Operations on shared queues

    /**
     * Removes and cancels all known tasks, ignoring any exceptions.
     */
    final void cancelAll() {
      for (ForkJoinTask<?> t; (t = poll()) != null;)
        ForkJoinTask.cancelIgnoringExceptions(t);
    }

    /**
     * Pops and executes up to limit consecutive tasks or until empty.
     *
     * @param limit max runs, or zero for no limit
     */
    final void localPopAndExec(int limit) {
      for (;;) {
        int b = base, s = top, al;
        ForkJoinTask<?>[] a;
        if ((a = array) != null && b != s && (al = a.length) > 0) {
          int index = (al - 1) & --s;
          long offset = ((long) index << ASHIFT) + ABASE;
          ForkJoinTask<?> t = (ForkJoinTask<?>) U.getAndSetObject(a, offset, null);
          if (t != null) {
            top = s;
            U.storeFence();
            t.doExec();
            if (limit != 0 && --limit == 0)
              break;
          } else
            break;
        } else
          break;
      }
    }

    /**
     * Polls and executes up to limit consecutive tasks or until empty.
     *
     * @param limit, or zero for no limit
     */
    final void localPollAndExec(int limit) {
      for (int polls = 0;;) {
        int b = base, s = top, d, al;
        ForkJoinTask<?>[] a;
        if ((a = array) != null && (d = b - s) < 0 && (al = a.length) > 0) {
          int index = (al - 1) & b++;
          long offset = ((long) index << ASHIFT) + ABASE;
          ForkJoinTask<?> t = (ForkJoinTask<?>) U.getAndSetObject(a, offset, null);
          if (t != null) {
            base = b;
            t.doExec();
            if (limit != 0 && ++polls == limit)
              break;
          } else if (d == -1)
            break; // now empty
          else
            polls = 0; // stolen; reset
        } else
          break;
      }
    }

    /**
     * If present, removes task from queue and executes it.
     */
    final void tryRemoveAndExec(ForkJoinTask<?> task) {
      ForkJoinTask<?>[] wa;
      int s, wal;
      if (base - (s = top) < 0 && // traverse from top
          (wa = array) != null && (wal = wa.length) > 0) {
        for (int m = wal - 1, ns = s - 1, i = ns;; --i) {
          int index = i & m;
          long offset = (index << ASHIFT) + ABASE;
          ForkJoinTask<?> t = (ForkJoinTask<?>) U.getObject(wa, offset);
          if (t == null)
            break;
          else if (t == task) {
            if (U.compareAndSwapObject(wa, offset, t, null)) {
              top = ns; // safely shift down
              for (int j = i; j != ns; ++j) {
                ForkJoinTask<?> f;
                int pindex = (j + 1) & m;
                long pOffset = (pindex << ASHIFT) + ABASE;
                f = (ForkJoinTask<?>) U.getObject(wa, pOffset);
                U.putObjectVolatile(wa, pOffset, null);

                int jindex = j & m;
                long jOffset = (jindex << ASHIFT) + ABASE;
                U.putOrderedObject(wa, jOffset, f);
              }
              U.storeFence();
              t.doExec();
            }
            break;
          }
        }
      }
    }

    /**
     * Tries to steal and run tasks within the target's computation until done, not found, or limit
     * exceeded.
     *
     * @param task root of CountedCompleter computation
     * @param limit max runs, or zero for no limit
     * @return task status on exit
     */
    final int localHelpCC(CountedCompleter<?> task, int limit) {
      int status = 0;
      if (task != null && (status = task.status) >= 0) {
        for (;;) {
          boolean help = false;
          int b = base, s = top, al;
          ForkJoinTask<?>[] a;
          if ((a = array) != null && b != s && (al = a.length) > 0) {
            int index = (al - 1) & (s - 1);
            long offset = ((long) index << ASHIFT) + ABASE;
            ForkJoinTask<?> o = (ForkJoinTask<?>) U.getObject(a, offset);
            if (o instanceof CountedCompleter) {
              CountedCompleter<?> t = (CountedCompleter<?>) o;
              for (CountedCompleter<?> f = t;;) {
                if (f != task) {
                  if ((f = f.completer) == null) // try parent
                    break;
                } else {
                  if (U.compareAndSwapObject(a, offset, t, null)) {
                    top = s - 1;
                    U.storeFence();
                    t.doExec();
                    help = true;
                  }
                  break;
                }
              }
            }
          }
          if ((status = task.status) < 0 || !help || (limit != 0 && --limit == 0))
            break;
        }
      }
      return status;
    }

    /**
     * Tries to lock shared queue by CASing phase field.
     */
    final boolean tryLockSharedQueue() {
      return U.compareAndSwapInt(this, PHASE, 0, QLOCK);
    }

    /**
     * Shared version of tryUnpush.
     */
    final boolean trySharedUnpush(ForkJoinTask<?> task) {
      boolean popped = false;
      int s = top - 1, al;
      ForkJoinTask<?>[] a;
      if ((a = array) != null && (al = a.length) > 0) {
        int index = (al - 1) & s;
        long offset = ((long) index << ASHIFT) + ABASE;
        ForkJoinTask<?> t = (ForkJoinTask<?>) U.getObject(a, offset);
        if (t == task && U.compareAndSwapInt(this, PHASE, 0, QLOCK)) {
          if (top == s + 1 && array == a && U.compareAndSwapObject(a, offset, task, null)) {
            popped = true;
            top = s;
          }
          U.putOrderedInt(this, PHASE, 0);
        }
      }
      return popped;
    }

    /**
     * Shared version of localHelpCC.
     */
    final int sharedHelpCC(CountedCompleter<?> task, int limit) {
      int status = 0;
      if (task != null && (status = task.status) >= 0) {
        for (;;) {
          boolean help = false;
          int b = base, s = top, al;
          ForkJoinTask<?>[] a;
          if ((a = array) != null && b != s && (al = a.length) > 0) {
            int index = (al - 1) & (s - 1);
            long offset = ((long) index << ASHIFT) + ABASE;
            ForkJoinTask<?> o = (ForkJoinTask<?>) U.getObject(a, offset);
            if (o instanceof CountedCompleter) {
              CountedCompleter<?> t = (CountedCompleter<?>) o;
              for (CountedCompleter<?> f = t;;) {
                if (f != task) {
                  if ((f = f.completer) == null)
                    break;
                } else {
                  if (U.compareAndSwapInt(this, PHASE, 0, QLOCK)) {
                    if (top == s && array == a && U.compareAndSwapObject(a, offset, t, null)) {
                      help = true;
                      top = s - 1;
                    }
                    U.putOrderedInt(this, PHASE, 0);
                    if (help)
                      t.doExec();
                  }
                  break;
                }
              }
            }
          }
          if ((status = task.status) < 0 || !help || (limit != 0 && --limit == 0))
            break;
        }
      }
      return status;
    }

    /**
     * Returns true if owned and not known to be blocked.
     */
    final boolean isApparentlyUnblocked() {
      Thread wt;
      Thread.State s;
      return ((wt = owner) != null && (s = wt.getState()) != Thread.State.BLOCKED
          && s != Thread.State.WAITING && s != Thread.State.TIMED_WAITING);
    }
  }

  /**
   * Factory for innocuous worker threads.
   */
  private static final class InnocuousForkJoinWorkerThreadFactory
      implements ForkJoinWorkerThreadFactory {

    /**
     * An ACC to restrict permissions for the factory itself. The constructed workers have no
     * permissions set.
     */
    private static final AccessControlContext ACC = contextWithPermissions(modifyThreadPermission,
        new RuntimePermission("enableContextClassLoaderOverride"),
        new RuntimePermission("modifyThreadGroup"), new RuntimePermission("getClassLoader"),
        new RuntimePermission("setContextClassLoader"));

    public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
      return AccessController.doPrivileged(new PrivilegedAction<ForkJoinWorkerThread>() {
        public ForkJoinWorkerThread run() {
          return new ForkJoinWorkerThread.InnocuousForkJoinWorkerThread(pool);
        }
      }, ACC);
    }
  }
}
