/*

 * 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.security.AccessControlContext;

import java.security.Permissions;

import java.security.ProtectionDomain;

import java.util.ArrayList;

import java.util.Arrays;

import java.util.Collection;

import java.util.Collections;

import java.util.List;

import java.util.concurrent.locks.ReentrantLock;

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.

 *

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

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

 */

@sun.misc.Contended

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; for example most

     * versions of poll pre-screen the CAS by rechecking that the base

     * has not changed since reading the slot, and most methods only

     * attempt the CAS if base appears not to be equal to top.

     *

     * 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 is why we normally use method pollAt and its

     * variants that try once at the apparent base index, else

     * consider alternative actions, rather than method poll, which

     * retries.)

     *

     * 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 (or in the case of CountedCompleters,

     * others with the same root task).  Insertion of tasks in shared

     * mode requires a lock but we use only a simple spinlock (using

     * field qlock), because submitters encountering a busy queue move

     * on to try 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).  The qlock is also used during termination

     * detection, in which case it is forced to a negative

     * non-lockable value.

     *

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

     * volatile variables that are by far most often read (not

     * written) as status and consistency checks. (Also, field

     * "config" holds unchanging configuration state.)

     *

     * Field "ctl" contains 64 bits holding information needed to

     * atomically decide to add, inactivate, enqueue (on an event

     * queue), dequeue, and/or re-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 "runState" holds lifetime status, atomically and

     * monotonically setting STARTED, SHUTDOWN, STOP, and finally

     * TERMINATED bits.

     *

     * Field "auxState" is a ReentrantLock subclass that also

     * opportunistically holds some other bookkeeping fields accessed

     * only when locked.  It is mainly used to lock (infrequent)

     * updates to workQueues.  The auxState instance is itself lazily

     * constructed (see tryInitialize), requiring a double-check-style

     * bootstrapping use of field runState, and locking a private

     * static.

     *

     * Field "workQueues" holds references to WorkQueues.  It is

     * updated (only during worker creation and termination) under the

     * lock, but is otherwise concurrently readable, and accessed

     * directly. We also ensure that reads of the array reference

     * itself never become too stale (for example, re-reading before

     * each scan). 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

     * to activate workers 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 active and total worker

     * counts as well as a queue to place waiting threads so they can

     * be located for signalling. Active counts also play the role of

     * quiescence indicators, so are decremented when workers believe

     * that there are no more tasks to execute. The "queue" is

     * actually a form of Treiber stack.  A stack is ideal for

     * activating threads in most-recently used order. This improves

     * performance and locality, outweighing the disadvantages of

     * being prone to contention and inability to release a worker

     * unless it is topmost on stack.  We block/unblock workers after

     * pushing on the idle worker stack (represented by the lower

     * 32bit subfield of ctl) when they cannot find work.  The top

     * stack state holds the value of the "scanState" 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 total

     * count (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 scanState 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 inactivated,

     * its scanState field is set, and is prevented from executing

     * tasks, even though it must scan once for them to avoid queuing

     * races. Note that scanState updates lag queue CAS releases so

     * usage requires care. 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. (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". (4) For internal task producers we

     * rely on the fact that even if no other workers awaken, the

     * producer itself will eventually see the task and execute it.

     *

     * 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 scan() performs top-level scanning for tasks.

     * Each scan traverses (and tries to poll from) each queue in

     * pseudorandom permutation order by randomly selecting an origin

     * index and a step value.  (The pseudorandom generator need not

     * have high-quality statistical properties in the long term, but

     * just within computations; We use 64bit and 32bit Marsaglia

     * XorShifts, which are cheap and suffice here.)  Scanning also

     * employs contention reduction: When scanning workers fail a CAS

     * polling for work, they soon restart with a different

     * pseudorandom scan order (thus likely retrying at different

     * intervals). This improves throughput when many threads are

     * trying to take tasks from few queues.  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 (up

     * to POLL_LIMIT times) from the same queue after a successful

     * poll before trying others.  Restricted forms of scanning occur

     * in methods helpComplete and findNonEmptyStealQueue, and take

     * similar but simpler forms.

     *

     * Deactivation and waiting. Queuing encounters several intrinsic

     * races; most notably that an inactivating scanning worker can

     * miss seeing a task produced during a scan.  So when a worker

     * cannot find a task to steal, it inactivates and enqueues, and

     * then rescans to ensure that it didn't miss one, reactivating

     * upon seeing one with probability approximately proportional to

     * probability of a miss.  (In most cases, the worker will be

     * signalled before self-signalling, avoiding cascades of multiple

     * signals for the same task).

     *

     * Workers block (in method awaitWork) using park/unpark;

     * advertising the need for signallers to unpark by setting their

     * "parker" fields.

     *

     * 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 awaitWork) if the pool has remained

     * quiescent for period given by IDLE_TIMEOUT_MS, increasing the

     * period as the number of threads decreases, eventually removing

     * all workers.

     *

     * 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 setting their (qlock) status,

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