/*

 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

 *

 * This code is free software; you can redistribute it and/or modify it

 * under the terms of the GNU General Public License version 2 only, as

 * published by the Free Software Foundation.  Oracle designates this

 * particular file as subject to the "Classpath" exception as provided

 * by Oracle in the LICENSE file that accompanied this code.

 *

 * This code is distributed in the hope that it will be useful, but WITHOUT

 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

 * version 2 for more details (a copy is included in the LICENSE file that

 * accompanied this code).

 *

 * You should have received a copy of the GNU General Public License version

 * 2 along with this work; if not, write to the Free Software Foundation,

 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

 *

 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

 * or visit www.oracle.com if you need additional information or have any

 * questions.

 */

/*

 * This file is available under and governed by the GNU General Public

 * License version 2 only, as published by the Free Software Foundation.

 * However, the following notice accompanied the original version of this

 * file:

 *

 * Written by Doug Lea with assistance from members of JCP JSR-166

 * Expert Group and released to the public domain, as explained at

 * http://creativecommons.org/publicdomain/zero/1.0/

 */

package java.util.concurrent;

import java.lang.Thread.UncaughtExceptionHandler;

import java.lang.invoke.MethodHandles;

import java.lang.invoke.VarHandle;

import java.security.AccessControlContext;

import java.security.Permissions;

import java.security.ProtectionDomain;

import java.util.ArrayList;

import java.util.Collection;

import java.util.Collections;

import java.util.List;

import java.util.function.Predicate;

import java.util.concurrent.locks.LockSupport;

/**

 * 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.

 *

 * <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>

 *    <td ALIGN=CENTER> <b>Call from non-fork/join clients</b></td>

 *    <td ALIGN=CENTER> <b>Call from within fork/join computations</b></td>

 *  </tr>

 *  <tr>

 *    <td> <b>Arrange async execution</b></td>

 *    <td> {@link #execute(ForkJoinTask)}</td>

 *    <td> {@link ForkJoinTask#fork}</td>

 *  </tr>

 *  <tr>

 *    <td> <b>Await and obtain result</b></td>

 *    <td> {@link #invoke(ForkJoinTask)}</td>

 *    <td> {@link ForkJoinTask#invoke}</td>

 *  </tr>

 *  <tr>

 *    <td> <b>Arrange exec and obtain Future</b></td>

 *    <td> {@link #submit(ForkJoinTask)}</td>

 *    <td> {@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td>

 *  </tr>

 * </table>

 *

 * <p>The common pool is by default constructed with default

 * parameters, but these 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}

 * <li>{@code java.util.concurrent.ForkJoinPool.common.exceptionHandler}

 * - the class name of a {@link UncaughtExceptionHandler}

 * <li>{@code java.util.concurrent.ForkJoinPool.common.maximumSpares}

 * - the maximum number of allowed extra threads to maintain target

 * parallelism (default 256).

 * </ul>

 * If a {@link SecurityManager} is present and no factory is

 * specified, then the default pool uses a factory supplying

 * threads that have no {@link Permissions} enabled.

 * The system class loader is used to load these classes.

 * Upon any error in establishing these settings, default parameters

 * are used. It is possible to disable or limit the use of threads in

 * the common pool by setting the parallelism property to zero, and/or

 * using a factory that may return {@code null}. However doing so may

 * cause unjoined tasks to never be executed.

 *

 * <p><b>Implementation notes</b>: This implementation restricts the

 * maximum number of running threads to 32767. Attempts to create

 * pools with greater than the maximum number result in

 * {@code IllegalArgumentException}.

 *

 * <p>This implementation rejects submitted tasks (that is, by throwing

 * {@link RejectedExecutionException}) only when the pool is shut down

 * or internal resources have been exhausted.

 *

 * @since 1.7

 * @author Doug Lea

 */

public class ForkJoinPool extends AbstractExecutorService {

    /*

     * Implementation Overview

     *

     * This class and its nested classes provide the main

     * functionality and control for a set of worker threads:

     * Submissions from non-FJ threads enter into submission queues.

     * Workers take these tasks and typically split them into subtasks

     * that may be stolen by other workers.  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 setRelease).

     *

     * 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.

     *