/*

 * 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.util.ArrayList;

import java.util.Arrays;

import java.util.Collection;

import java.util.Collections;

import java.util.List;

import java.util.concurrent.AbstractExecutorService;

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Future;

import java.util.concurrent.RejectedExecutionException;

import java.util.concurrent.RunnableFuture;

import java.util.concurrent.ThreadLocalRandom;

import java.util.concurrent.TimeUnit;

/**

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

 * </ul>

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

     *

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

     * See also "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 (atomic, volatile etc) issues.  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.  So, both a successful pop and poll

     * mainly entail a CAS of a slot from non-null to null.  Because

     * we rely on CASes of references, we do not need tag bits on base

     * or top.  They are simple ints as used in any circular

     * array-based queue (see for example ArrayDeque).  Updates to the

     * indices must still be ordered in a way that guarantees that top

     * == base means the queue is empty, but otherwise may err on the

     * side of possibly making the queue appear nonempty when a push,

     * pop, or poll have not fully committed. Note that this means

     * that 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) completes.

     * However, in the aggregate, we ensure at least probabilistic

     * non-blockingness.  If an attempted steal fails, a thief always

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

     *

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

     * mode considers affinities, loads, cache localities, etc, so

     * rarely provide the best possible performance on a given

     * machine, but portably provide good throughput by averaging over

     * these factors.  (Further, even if we did try to use such

     * information, we do not usually have a basis for exploiting it.

     * For example, some sets of tasks profit from cache affinities,

     * but others are harmed by cache pollution effects.)

     *

     * WorkQueues are also used in a similar way for tasks submitted

     * to the pool. We cannot mix these tasks in the same queues used

     * for work-stealing (this would contaminate lifo/fifo

     * processing). 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).  However, because most

     * shared/external queue operations are more expensive than

     * internal, and because, at steady state, external submitters

     * will compete for CPU with workers, ForkJoinTask.join and

     * related methods disable them from repeatedly helping to process

     * tasks if all workers are active.  Insertion of tasks in shared

     * mode requires a lock (mainly to protect in the case of

     * resizing) but we use only a simple spinlock (using bits in

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

     *

     * Management

     * ==========

     *

     * The main throughput advantages of work-stealing stem from

     * decentralized control -- workers mostly take tasks from

     * themselves or each other. We cannot negate this in the

     * implementation of other management responsibilities. The main

     * tactic for avoiding bottlenecks is packing nearly all

     * essentially atomic control state into two volatile variables

     * that are by far most often read (not written) as status and

     * consistency checks.

     *

     * Field "ctl" contains 64 bits holding all the 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

     * fields.

     *

     * Field "plock" is a form of sequence lock with a saturating

     * shutdown bit (similarly for per-queue "qlocks"), mainly

     * protecting updates to the workQueues array, as well as to

     * enable shutdown.  When used as a lock, it is normally only very

     * briefly held, so is nearly always available after at most a

     * brief spin, but we use a monitor-based backup strategy to

     * block when needed.

     *

     * Recording WorkQueues.  WorkQueues are recorded in the

     * "workQueues" array that is created upon first use and expanded

     * if necessary.  Updates to the array while recording new workers

     * and unrecording terminated ones are protected from each other

     * by a lock but the array is otherwise concurrently readable, and

     * accessed directly.  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 wait queue

     * field of ctl stores indices, not references.  Access to the

     * workQueues in associated methods (for example signalWork) must

     * both index-check and null-check the IDs. All such accesses

     * ignore bad IDs by returning out early from what they are doing,

     * since this can only be associated with termination, in which

     * case it is OK to give up.  All uses of the workQueues array

     * also check that it is non-null (even if previously

     * non-null). This allows nulling during termination, which is

     * currently not necessary, but remains an option for

     * resource-revocation-based shutdown schemes. It also helps

     * reduce JIT issuance of uncommon-trap code, which tends to

     * unnecessarily complicate control flow in some methods.

     *

     * Event Queuing. 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 (this is

     * compounded at program start-up by JIT compilation and

     * allocation). So we try to streamline this as much as possible.

     * We park/unpark workers after placing in an event wait queue

     * when they cannot find work. This "queue" is actually a simple

     * Treiber stack, headed by the "id" field of ctl, plus a 15bit

     * counter value (that reflects the number of times a worker has

     * been inactivated) to avoid ABA effects (we need only as many

     * version numbers as worker threads). Successors are held in

     * field WorkQueue.nextWait.  Queuing deals with several intrinsic

     * races, mainly that a task-producing thread can miss seeing (and

     * signalling) another thread that gave up looking for work but

     * has not yet entered the wait queue. We solve this by requiring

     * a full sweep of all workers (via repeated calls to method

     * scan()) both before and after a newly waiting worker is added

     * to the wait queue.  Because enqueued workers may actually be

     * rescanning rather than waiting, we set and clear the "parker"

     * field of WorkQueues to reduce unnecessary calls to unpark.

     * (This requires a secondary recheck to avoid missed signals.)

     * Note the unusual conventions about Thread.interrupts

     * surrounding parking and other blocking: Because interrupts are

     * used solely to alert threads to check termination, which is

     * checked anyway upon blocking, we clear status (using

     * Thread.interrupted) before any call to park, so that park does

     * not immediately return due to status being set via some other

     * unrelated call to interrupt in user code.

     *

     * Signalling.  We create or wake up workers only when there

     * appears to be at least one task they might be able to find and

     * execute.  When a submission is added or another worker adds a

     * task to a queue that has fewer than two tasks, they signal

     * waiting workers (or trigger creation of new ones if fewer than

     * the given parallelism level -- signalWork).  These primary

     * signals are buttressed by others whenever other threads remove

     * a task from a queue and notice that there are other tasks there

     * as well.  So in general, pools will be over-signalled. On most

     * platforms, signalling (unpark) overhead time is noticeably

     * long, and the time between signalling a thread and it actually

     * making progress can be very noticeably long, so it is worth

     * offloading these delays from critical paths as much as

     * possible. Additionally, workers spin-down gradually, by staying

     * alive so long as they see the ctl state changing.  Similar

     * stability-sensing techniques are also used before blocking in

     * awaitJoin and helpComplete.

     *

     * 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 if the pool has remained quiescent for a

     * given period -- a short period if there are more threads than

     * parallelism, longer as the number of threads decreases. This

     * will slowly propagate, eventually terminating all workers after

     * periods of non-use.

     *

     * Shutdown and Termination. A call to shutdownNow atomically sets

     * a plock bit and then (non-atomically) sets each worker's

     * qlock status, cancels all unprocessed tasks, and wakes up

     * all waiting workers.  Detecting whether termination should

     * commence after a non-abrupt shutdown() call requires more work

     * and bookkeeping. We need consensus about quiescence (i.e., that

     * there is no more work). The active count provides a primary

     * indication but non-abrupt shutdown still requires a rechecking

     * scan for any workers that are inactive but not queued.

     *

     * 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

     * 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 sometimes 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 (although at the

     * expense of larger run-time stacks, but the tradeoff is

     * typically worthwhile).

     *

     * The ManagedBlocker extension API can't use helping so relies

     * only on compensation in method awaitBlocker.

     *

     * The algorithm in tryHelpStealer entails a form of "linear"

     * helping: Each worker records (in field currentSteal) the most

     * recent task it stole from some other worker. Plus, it records

     * (in field currentJoin) the task it is currently actively

     * joining. Method tryHelpStealer 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.

     * In essence, the joiner executes a task that would be on its own

     * local deque had the to-be-joined task not been stolen. This may

     * be seen as a conservative variant of the approach in Wagner &

     * Calder "Leapfrogging: a portable technique for implementing

     * efficient futures" SIGPLAN Notices, 1993

     * (http://portal.acm.org/citation.cfm?id=155354). It differs in

     * that: (1) We only maintain dependency links across workers upon

     * steals, rather than use per-task bookkeeping.  This sometimes

     * requires a linear scan of workQueues array to locate stealers,

     * but often doesn't because stealers leave hints (that may become

     * stale/wrong) of where to locate them.  It is only a hint

     * because a worker might have had multiple steals and the hint

     * records only one of them (usually the most current).  Hinting

     * isolates cost to when it is needed, rather than adding to

     * per-task overhead.  (2) It is "shallow", ignoring nesting and

     * potentially cyclic mutual steals.  (3) It is intentionally

     * racy: field currentJoin is updated only while actively joining,

     * which means that we miss links in the chain during long-lived

     * tasks, GC stalls etc (which is OK since blocking in such cases

     * is usually a good idea).  (4) We bound the number of attempts

     * to find work (see MAX_HELP) and fall back to suspending the

     * worker and if necessary replacing it with another.

     *

     * Helping actions for CountedCompleters are much simpler: Method

     * helpComplete can take and execute any task with the same root

     * as the task being waited on. However, this still entails some

     * traversal of completer chains, so is less efficient than using

     * CountedCompleters without explicit joins.

     *

     * It is impossible to keep exactly the target parallelism number

     * of threads running at any given time.  Determining the

     * existence of conservatively safe helping targets, the

     * availability of already-created spares, and the apparent need

     * to create new spares are all racy, so we rely on multiple

     * retries of each.  Compensation in the apparent absence of

     * helping opportunities is challenging to control on JVMs, where

     * GC and other activities can stall progress of tasks that in

     * turn stall out many other dependent tasks, without us being

     * able to determine whether they will ever require compensation.

     * Even though work-stealing otherwise encounters little

     * degradation in the presence of more threads than cores,

     * aggressively adding new threads in such cases entails risk of

     * unwanted positive feedback control loops in which more threads

     * cause more dependent stalls (as well as delayed progress of

     * unblocked threads to the point that we know they are available)

     * leading to more situations requiring more threads, and so

     * on. This aspect of control can be seen as an (analytically

     * intractable) game with an opponent that may choose the worst

     * (for us) active thread to stall at any time.  We take several

     * precautions to bound losses (and thus bound gains), mainly in

     * methods tryCompensate and awaitJoin.

     *

     * 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, with no nested

     * allocation. Most bootstrapping occurs within method

     * fullExternalPush during the first submission to the pool.

     *

     * When external threads submit to the common pool, they can

     * perform subtask processing (see externalHelpJoin and related

     * methods).  This caller-helps policy makes it sensible to set

     * common pool parallelism level to one (or more) less than the