/*

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

import java.util.Arrays;

import java.util.Collection;

import java.util.Collections;

import java.util.List;

import java.util.Random;

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.TimeUnit;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.concurrent.locks.LockSupport;

import java.util.concurrent.locks.ReentrantLock;

import java.util.concurrent.locks.Condition;

/**

 * 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 subtasks 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). 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 {@code ForkJoinPool} is 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 IO 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 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

 * <em>NOT</em> use these pool execution methods, but instead use the

 * within-computation forms listed in the table.

 *

 * <table BORDER CELLPADDING=3 CELLSPACING=1>

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

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

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

 *  </tr>

 *  <tr>

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

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

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

 *  </tr>

 *  <tr>

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

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

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

 *  </tr>

 * </table>

 *

 * <p><b>Sample Usage.</b> Normally a single {@code ForkJoinPool} is

 * used for all parallel task execution in a program or subsystem.

 * Otherwise, use would not usually outweigh the construction and

 * bookkeeping overhead of creating a large set of threads. For

 * example, a common pool could be used for the {@code SortTasks}

 * illustrated in {@link RecursiveAction}. Because {@code

 * ForkJoinPool} uses threads in {@linkplain java.lang.Thread#isDaemon

 * daemon} mode, there is typically no need to explicitly {@link

 * #shutdown} such a pool upon program exit.

 *

 * <pre>

 * static final ForkJoinPool mainPool = new ForkJoinPool();

 * ...

 * public void sort(long[] array) {

 *   mainPool.invoke(new SortTask(array, 0, array.length));

 * }

 * </pre>

 *

 * <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 provides the central bookkeeping and control for a

     * set of worker threads: Submissions from non-FJ threads enter

     * into a submission queue. 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 worker queues, and

     * lastly to new submissions.

     *

     * 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 a single 64bit volatile

     * variable ("ctl"). This variable is read on the order of 10-100

     * times as often as it is modified (always via CAS). (There is

     * some additional control state, for example variable "shutdown"

     * for which we can cope with uncoordinated updates.)  This

     * streamlines synchronization and control at the expense of messy

     * constructions needed to repack status bits upon updates.

     * Updates tend not to contend with each other except during

     * bursts while submitted tasks begin or end.  In some cases when

     * they do contend, threads can instead do something else

     * (usually, scan for tasks) until contention subsides.

     *

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

     *

     * Recording Workers.  Workers are recorded in the "workers" array

     * that is created upon pool construction and expanded if (rarely)

     * necessary.  This is an array as opposed to some other data

     * structure to support index-based random steals by workers.

     * Updates to the array recording new workers and unrecording

     * terminated ones are protected from each other by a seqLock

     * (scanGuard) but the array is otherwise concurrently readable,

     * and accessed directly by workers. To simplify index-based

     * operations, the array size is always a power of two, and all

     * readers must tolerate null slots. To avoid flailing during

     * start-up, the array is presized to hold twice #parallelism

     * workers (which is unlikely to need further resizing during

     * execution). But to avoid dealing with so many null slots,

     * variable scanGuard includes a mask for the nearest power of two

     * that contains all current workers.  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 workers are

     * via indices into the workers array (which is one source of some

     * of the messy code constructions here). In essence, the workers

     * array serves as a weak reference mechanism. Thus for example

     * the wait queue field of ctl stores worker indices, not worker

     * references.  Access to the workers 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 workers array, as well as queue arrays, check

     * that the array 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.

     *

     * Wait 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.  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 to both wake up waiters (by advancing their

     * count) and avoid ABA effects. Successors are held in worker

     * field "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 both before (in scan()) and after

     * (in tryAwaitWork()) a newly waiting worker is added to the wait

     * queue. During a rescan, the worker might release some other

     * queued worker rather than itself, which has the same net

     * effect. Because enqueued workers may actually be rescanning

     * rather than waiting, we set and clear the "parked" field of

     * ForkJoinWorkerThread to reduce unnecessary calls to unpark.

     * (Use of the parked field requires a secondary recheck to avoid

     * missed signals.)

     *

     * 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 previously had two or fewer tasks, they

     * signal waiting workers (or trigger creation of new ones if

     * fewer than the given parallelism level -- see signalWork).

     * These primary signals are buttressed by signals during rescans

     * as well as those performed when a worker steals a task and

     * notices that there are more tasks too; together these cover the

     * signals needed in cases when more than two tasks are pushed

     * but untaken.

     *

     * 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

     * SHRINK_RATE nanosecs. This will slowly propagate, eventually

     * terminating all workers after long periods of non-use.

     *

     * Submissions. External submissions are maintained in an

     * array-based queue that is structured identically to

     * ForkJoinWorkerThread queues except for the use of

     * submissionLock in method addSubmission. Unlike the case for

     * worker queues, multiple external threads can add new

     * submissions, so adding requires a lock.

     *

     * Compensation. Beyond work-stealing support and lifecycle

     * control, the main responsibility of this framework is to take

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

     *      ForkJoinWorkerThread.joinTask tracks joining->stealing

     *      links to try to find such a task.

     *

     *   Compensating: Unless there are already enough live threads,

     *      method tryPreBlock() may create or re-activate a spare

     *      thread to compensate for blocked joiners until they

     *      unblock.

     *

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

     * only on compensation in method awaitBlocker.

     *

     * 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 and require heuristic

     * guidance, so we rely on multiple retries of each.  Currently,

     * in keeping with on-demand signalling policy, we compensate only

     * if blocking would leave less than one active (non-waiting,

     * non-blocked) worker. Additionally, to avoid some false alarms

     * due to GC, lagging counters, system activity, etc, compensated

     * blocking for joins is only attempted after rechecks stabilize

     * (retries are interspersed with Thread.yield, for good

     * citizenship).  The variable blockedCount, incremented before

     * blocking and decremented after, is sometimes needed to

     * distinguish cases of waiting for work vs blocking on joins or

     * other managed sync. Both cases are equivalent for most pool

     * control, so we can update non-atomically. (Additionally,

     * contention on blockedCount alleviates some contention on ctl).

     *

     * Shutdown and Termination. A call to shutdownNow atomically sets

     * the ctl stop bit and then (non-atomically) sets each workers

     * "terminate" 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 quiesence (i.e., that

     * there is no more work) which is reflected in active counts so

     * long as there are no current blockers, as well as possible

     * re-evaluations during independent changes in blocking or

     * quiescing workers.

     *

     * Style notes: There is a lot of representation-level coupling

     * among classes ForkJoinPool, ForkJoinWorkerThread, and

     * ForkJoinTask.  Most fields of ForkJoinWorkerThread maintain

     * data structures managed by ForkJoinPool, so are directly

     * accessed.  Conversely we allow access to "workers" array by

     * workers, and direct access to ForkJoinTask.status by both

     * ForkJoinPool and ForkJoinWorkerThread.  There is little point

     * trying to reduce this, since any associated future changes in

     * representations will need to be accompanied by algorithmic

     * changes anyway. All together, these low-level implementation

     * choices produce as much as a factor of 4 performance

     * improvement compared to naive implementations, and enable the

     * processing of billions of tasks per second, at the expense of

     * some ugliness.

     *

     * Methods signalWork() and scan() are the main bottlenecks so are

     * especially heavily micro-optimized/mangled.  There are lots of

     * inline assignments (of form "while ((local = field) != 0)")

     * which are usually the simplest way to ensure the required read

     * orderings (which are sometimes critical). This leads to a

     * "C"-like style of listing declarations of these locals at the

     * heads of methods or blocks.  There are several occurrences of

     * the unusual "do {} while (!cas...)"  which is the simplest way

     * to force an update of a CAS'ed variable. There are also other

     * coding oddities that help some methods perform reasonably even

     * when interpreted (not compiled).

     *

     * The order of declarations in this file is: (1) declarations of

     * statics (2) fields (along with constants used when unpacking

     * some of them), listed in an order that tends to reduce

     * contention among them a bit under most JVMs.  (3) internal

     * control methods (4) callbacks and other support for

     * ForkJoinTask and ForkJoinWorkerThread classes, (5) exported

     * methods (plus a few little helpers). (6) static block

     * initializing all statics in a minimally dependent order.

     */

    /**

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

         *

         * @param pool the pool this thread works in

         * @throws NullPointerException if the pool is null

         */

        public ForkJoinWorkerThread newThread(ForkJoinPool pool);

    }

    /**

     * Default ForkJoinWorkerThreadFactory implementation; creates a

     * new ForkJoinWorkerThread.

     */

    static class DefaultForkJoinWorkerThreadFactory

        implements ForkJoinWorkerThreadFactory {

        public ForkJoinWorkerThread newThread(ForkJoinPool pool) {

            return new ForkJoinWorkerThread(pool);

        }

    }

    /**

     * Creates a new ForkJoinWorkerThread. This factory is used unless

     * overridden in ForkJoinPool constructors.

     */

    public static final ForkJoinWorkerThreadFactory

        defaultForkJoinWorkerThreadFactory;

    /**

     * If there is a security manager, makes sure caller has

     * permission to modify threads.

     */

    private static void checkPermission() {

        throw new SecurityException();

    }

    /**

     * Generator for assigning sequence numbers as pool names.

     */

    private static final AtomicInteger poolNumberGenerator;

    /**

     * Generator for initial random seeds for worker victim

     * selection. This is used only to create initial seeds. Random

     * steals use a cheaper xorshift generator per steal attempt. We

     * don't expect much contention on seedGenerator, so just use a

     * plain Random.

     */

    static final Random workerSeedGenerator;

    /**

     * Array holding all worker threads in the pool.  Initialized upon

     * construction. Array size must be a power of two.  Updates and

     * replacements are protected by scanGuard, but the array is

     * always kept in a consistent enough state to be randomly

     * accessed without locking by workers performing work-stealing,

     * as well as other traversal-based methods in this class, so long

     * as reads memory-acquire by first reading ctl. All readers must

     * tolerate that some array slots may be null.

     */

    ForkJoinWorkerThread[] workers;

    /**

     * Initial size for submission queue array. Must be a power of

     * two.  In many applications, these always stay small so we use a

     * small initial cap.

     */

    private static final int INITIAL_QUEUE_CAPACITY = 8;

    /**

     * Maximum size for submission queue array. Must be a power of two

     * less than or equal to 1 << (31 - width of array entry) to

     * ensure lack of index wraparound, but is capped at a lower

     * value to help users trap runaway computations.

     */

    private static final int MAXIMUM_QUEUE_CAPACITY = 1 << 24; // 16M

    /**

     * Array serving as submission queue. Initialized upon construction.

     */

    private ForkJoinTask<?>[] submissionQueue;

    /**

     * Lock protecting submissions array for addSubmission

     */

    private final ReentrantLock submissionLock;

    /**

     * Condition for awaitTermination, using submissionLock for

     * convenience.

     */

    private final Condition termination;

    /**

     * Creation factory for worker threads.

     */

    private final ForkJoinWorkerThreadFactory factory;

    /**

     * The uncaught exception handler used when any worker abruptly

     * terminates.

     */

    final Thread.UncaughtExceptionHandler ueh;

    /**

     * Prefix for assigning names to worker threads

     */

    private final String workerNamePrefix;

    /**

     * Sum of per-thread steal counts, updated only when threads are

     * idle or terminating.

     */

    private volatile long stealCount;

    /**

     * Main pool control -- a long packed with:

     * AC: Number of active running workers minus target parallelism (16 bits)

     * TC: Number of total workers minus target parallelism (16bits)

     * ST: true if pool is terminating (1 bit)

     * EC: the wait count of top waiting thread (15 bits)

     * ID: ~poolIndex of top of Treiber stack of waiting threads (16 bits)

     *

     * When convenient, we can extract the upper 32 bits of counts and

     * the lower 32 bits of queue state, u = (int)(ctl >>> 32) and e =

     * (int)ctl.  The ec field is never accessed alone, but always

     * together with id and st. 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 active workers, when tc is

     * negative, there are not enough total workers, when id is

     * negative, there is at least one waiting worker, and when e is

     * negative, the pool is terminating.  To deal with these possibly

     * negative fields, we use casts in and out of "short" and/or

     * signed shifts to maintain signedness.

     */

    volatile long ctl;

    // bit positions/shifts for fields

    private static final int  AC_SHIFT   = 48;

    private static final int  TC_SHIFT   = 32;

    private static final int  ST_SHIFT   = 31;

    private static final int  EC_SHIFT   = 16;

    // bounds

    private static final int  MAX_ID     = 0x7fff;  // max poolIndex

    private static final int  SMASK      = 0xffff;  // mask short bits

    private static final int  SHORT_SIGN = 1 << 15;

    private static final int  INT_SIGN   = 1 << 31;

    // masks

    private static final long STOP_BIT   = 0x0001L << ST_SHIFT;

    private static final long AC_MASK    = ((long)SMASK) << AC_SHIFT;

    private static final long TC_MASK    = ((long)SMASK) << TC_SHIFT;

    // units for incrementing and decrementing

    private static final long TC_UNIT    = 1L << TC_SHIFT;

    private static final long AC_UNIT    = 1L << AC_SHIFT;

    // masks and units for dealing with u = (int)(ctl >>> 32)

    private static final int  UAC_SHIFT  = AC_SHIFT - 32;

    private static final int  UTC_SHIFT  = TC_SHIFT - 32;

    private static final int  UAC_MASK   = SMASK << UAC_SHIFT;

    private static final int  UTC_MASK   = SMASK << UTC_SHIFT;

    private static final int  UAC_UNIT   = 1 << UAC_SHIFT;

    private static final int  UTC_UNIT   = 1 << UTC_SHIFT;

    // masks and units for dealing with e = (int)ctl

    private static final int  E_MASK     = 0x7fffffff; // no STOP_BIT

    private static final int  EC_UNIT    = 1 << EC_SHIFT;

    /**

     * The target parallelism level.

     */

    final int parallelism;

    /**

     * Index (mod submission queue length) of next element to take

     * from submission queue. Usage is identical to that for

     * per-worker queues -- see ForkJoinWorkerThread internal

     * documentation.

     */

    volatile int queueBase;

    /**

     * Index (mod submission queue length) of next element to add

     * in submission queue. Usage is identical to that for

     * per-worker queues -- see ForkJoinWorkerThread internal

     * documentation.

     */

    int queueTop;

    /**

     * True when shutdown() has been called.

     */

    volatile boolean shutdown;

    /**

     * True if use local fifo, not default lifo, for local polling

     * Read by, and replicated by ForkJoinWorkerThreads

     */

    final boolean locallyFifo;

    /**

     * The number of threads in ForkJoinWorkerThreads.helpQuiescePool.

     * When non-zero, suppresses automatic shutdown when active

     * counts become zero.

     */

    volatile int quiescerCount;

    /**

     * The number of threads blocked in join.

     */

    volatile int blockedCount;

    /**

     * Counter for worker Thread names (unrelated to their poolIndex)

     */

    private volatile int nextWorkerNumber;

    /**

     * The index for the next created worker. Accessed under scanGuard.

     */

    private int nextWorkerIndex;

    /**

     * SeqLock and index masking for updates to workers array.  Locked

     * when SG_UNIT is set. Unlocking clears bit by adding

     * SG_UNIT. Staleness of read-only operations can be checked by

     * comparing scanGuard to value before the reads. The low 16 bits

     * (i.e, anding with SMASK) hold (the smallest power of two

     * covering all worker indices, minus one, and is used to avoid

     * dealing with large numbers of null slots when the workers array

     * is overallocated.

     */

    volatile int scanGuard;

    private static final int SG_UNIT = 1 << 16;

    /**

     * The wakeup interval (in nanoseconds) for a worker waiting for a

     * task when the pool is quiescent to instead try to shrink the

     * number of workers.  The exact value does not matter too

     * much. It must be short enough to release resources during

     * sustained periods of idleness, but not so short that threads

     * are continually re-created.

     */

    private static final long SHRINK_RATE =

        4L * 1000L * 1000L * 1000L; // 4 seconds

    /**

     * Top-level loop for worker threads: On each step: if the

     * previous step swept through all queues and found no tasks, or

     * there are excess threads, then possibly blocks. Otherwise,

     * scans for and, if found, executes a task. Returns when pool

     * and/or worker terminate.

     *

     * @param w the worker

     */

    final void work(ForkJoinWorkerThread w) {

        boolean swept = false;                // true on empty scans

        long c;

        while (!w.terminate && (int)(c = ctl) >= 0) {

            int a;                            // active count

            if (!swept && (a = (int)(c >> AC_SHIFT)) <= 0)

                swept = scan(w, a);

            else if (tryAwaitWork(w, c))

                swept = false;

        }

    }

    // Signalling

    /**

     * Wakes up or creates a worker.

     */

    final void signalWork() {

        /*

         * The while condition is true if: (there is are too few total

         * workers OR there is at least one waiter) AND (there are too

         * few active workers OR the pool is terminating).  The value

         * of e distinguishes the remaining cases: zero (no waiters)

         * for create, negative if terminating (in which case do

         * nothing), else release a waiter. The secondary checks for

         * release (non-null array etc) can fail if the pool begins

         * terminating after the test, and don't impose any added cost

         * because JVMs must perform null and bounds checks anyway.

         */

        long c; int e, u;

        while ((((e = (int)(c = ctl)) | (u = (int)(c >>> 32))) &

                (INT_SIGN|SHORT_SIGN)) == (INT_SIGN|SHORT_SIGN) && e >= 0) {

            if (e > 0) {                         // release a waiting worker

                int i; ForkJoinWorkerThread w; ForkJoinWorkerThread[] ws;

                if ((ws = workers) == null ||

                    (i = ~e & SMASK) >= ws.length ||

                    (w = ws[i]) == null)

                    break;

                long nc = (((long)(w.nextWait & E_MASK)) |

                           ((long)(u + UAC_UNIT) << 32));

                if (w.eventCount == e &&

                    UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc)) {

                    w.eventCount = (e + EC_UNIT) & E_MASK;

                    if (w.parked)

                        UNSAFE.unpark(w);

                    break;

                }

            }

            else if (UNSAFE.compareAndSwapLong

                     (this, ctlOffset, c,

                      (long)(((u + UTC_UNIT) & UTC_MASK) |

                             ((u + UAC_UNIT) & UAC_MASK)) << 32)) {

                addWorker();

                break;

            }

        }

    }

    /**

     * Variant of signalWork to help release waiters on rescans.

     * Tries once to release a waiter if active count < 0.

     *

     * @return false if failed due to contention, else true

     */

    private boolean tryReleaseWaiter() {

        long c; int e, i; ForkJoinWorkerThread w; ForkJoinWorkerThread[] ws;

        if ((e = (int)(c = ctl)) > 0 &&

            (int)(c >> AC_SHIFT) < 0 &&

            (ws = workers) != null &&

            (i = ~e & SMASK) < ws.length &&

            (w = ws[i]) != null) {

            long nc = ((long)(w.nextWait & E_MASK) |

                       ((c + AC_UNIT) & (AC_MASK|TC_MASK)));

            if (w.eventCount != e ||

                !UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc))

                return false;

            w.eventCount = (e + EC_UNIT) & E_MASK;

            if (w.parked)

                UNSAFE.unpark(w);

        }

        return true;

    }

    // Scanning for tasks

    /**

     * Scans for and, if found, executes one task. Scans start at a

     * random index of workers array, and randomly select the first

     * (2*#workers)-1 probes, and then, if all empty, resort to 2

     * circular sweeps, which is necessary to check quiescence. and

     * taking a submission only if no stealable tasks were found.  The

     * steal code inside the loop is a specialized form of

     * ForkJoinWorkerThread.deqTask, followed bookkeeping to support

     * helpJoinTask and signal propagation. The code for submission

     * queues is almost identical. On each steal, the worker completes

     * not only the task, but also all local tasks that this task may

     * have generated. On detecting staleness or contention when

     * trying to take a task, this method returns without finishing

     * sweep, which allows global state rechecks before retry.

     *

     * @param w the worker

     * @param a the number of active workers

     * @return true if swept all queues without finding a task

     */

    private boolean scan(ForkJoinWorkerThread w, int a) {

        int g = scanGuard; // mask 0 avoids useless scans if only one active

        int m = (parallelism == 1 - a && blockedCount == 0) ? 0 : g & SMASK;

        ForkJoinWorkerThread[] ws = workers;

        if (ws == null || ws.length <= m)         // staleness check

            return false;

        for (int r = w.seed, k = r, j = -(m + m); j <= m + m; ++j) {

            ForkJoinTask<?> t; ForkJoinTask<?>[] q; int b, i;

            ForkJoinWorkerThread v = ws[k & m];

            if (v != null && (b = v.queueBase) != v.queueTop &&

                (q = v.queue) != null && (i = (q.length - 1) & b) >= 0) {

                long u = (i << ASHIFT) + ABASE;

                if ((t = q[i]) != null && v.queueBase == b &&

                    UNSAFE.compareAndSwapObject(q, u, t, null)) {

                    int d = (v.queueBase = b + 1) - v.queueTop;

                    v.stealHint = w.poolIndex;

                    if (d != 0)

                        signalWork();             // propagate if nonempty

                    w.execTask(t);

                }

                r ^= r << 13; r ^= r >>> 17; w.seed = r ^ (r << 5);

                return false;                     // store next seed

            }

            else if (j < 0) {                     // xorshift

                r ^= r << 13; r ^= r >>> 17; k = r ^= r << 5;

            }

            else

                ++k;

        }

        if (scanGuard != g)                       // staleness check

            return false;

        else {                                    // try to take submission

            ForkJoinTask<?> t; ForkJoinTask<?>[] q; int b, i;

            if ((b = queueBase) != queueTop &&

                (q = submissionQueue) != null &&

                (i = (q.length - 1) & b) >= 0) {

                long u = (i << ASHIFT) + ABASE;

                if ((t = q[i]) != null && queueBase == b &&

                    UNSAFE.compareAndSwapObject(q, u, t, null)) {

                    queueBase = b + 1;

                    w.execTask(t);

                }

                return false;

            }

            return true;                         // all queues empty

        }

    }

    /**

     * Tries to enqueue worker w in wait queue and await change in

     * worker's eventCount.  If the pool is quiescent and there is

     * more than one worker, possibly terminates worker upon exit.

     * Otherwise, before blocking, rescans queues to avoid missed

     * signals.  Upon finding work, releases at least one worker

     * (which may be the current worker). Rescans restart upon

     * detected staleness or failure to release due to

     * contention. Note the unusual conventions about Thread.interrupt

     * here and elsewhere: Because interrupts are used solely to alert

     * threads to check termination, which is checked here anyway, 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.

     *

     * @param w the calling worker

     * @param c the ctl value on entry

     * @return true if waited or another thread was released upon enq

     */

    private boolean tryAwaitWork(ForkJoinWorkerThread w, long c) {

        int v = w.eventCount;

        w.nextWait = (int)c;                      // w's successor record

        long nc = (long)(v & E_MASK) | ((c - AC_UNIT) & (AC_MASK|TC_MASK));

        if (ctl != c || !UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc)) {

            long d = ctl; // return true if lost to a deq, to force scan

            return (int)d != (int)c && ((d - c) & AC_MASK) >= 0L;

        }

        for (int sc = w.stealCount; sc != 0;) {   // accumulate stealCount

            long s = stealCount;

            if (UNSAFE.compareAndSwapLong(this, stealCountOffset, s, s + sc))

                sc = w.stealCount = 0;

            else if (w.eventCount != v)

                return true;                      // update next time

        }

        if ((!shutdown || !tryTerminate(false)) &&

            (int)c != 0 && parallelism + (int)(nc >> AC_SHIFT) == 0 &&

            blockedCount == 0 && quiescerCount == 0)

            idleAwaitWork(w, nc, c, v);           // quiescent

        for (boolean rescanned = false;;) {

            if (w.eventCount != v)

                return true;

            if (!rescanned) {

                int g = scanGuard, m = g & SMASK;

                ForkJoinWorkerThread[] ws = workers;

                if (ws != null && m < ws.length) {

                    rescanned = true;

                    for (int i = 0; i <= m; ++i) {

                        ForkJoinWorkerThread u = ws[i];

                        if (u != null) {

                            if (u.queueBase != u.queueTop &&

                                !tryReleaseWaiter())

                                rescanned = false; // contended

                            if (w.eventCount != v)

                                return true;

                        }

                    }

                }

                if (scanGuard != g ||              // stale

                    (queueBase != queueTop && !tryReleaseWaiter()))

                    rescanned = false;

                if (!rescanned)

                    Thread.yield();                // reduce contention

                else

                    Thread.interrupted();          // clear before park

            }

            else {

                w.parked = true;                   // must recheck

                if (w.eventCount != v) {

                    w.parked = false;

                    return true;

                }

                LockSupport.park(this);

                rescanned = w.parked = false;

            }

        }

    }

    /**

     * If inactivating worker w has caused pool to become

     * quiescent, check for pool termination, and wait for event

     * for up to SHRINK_RATE nanosecs (rescans are unnecessary in

     * this case because quiescence reflects consensus about lack

     * of work). On timeout, if ctl has not changed, terminate the

     * worker. Upon its termination (see deregisterWorker), it may

     * wake up another worker to possibly repeat this process.

     *

     * @param w the calling worker

     * @param currentCtl the ctl value after enqueuing w

     * @param prevCtl the ctl value if w terminated

     * @param v the eventCount w awaits change

     */

    private void idleAwaitWork(ForkJoinWorkerThread w, long currentCtl,

                               long prevCtl, int v) {

        if (w.eventCount == v) {

            if (shutdown)

                tryTerminate(false);

            ForkJoinTask.helpExpungeStaleExceptions(); // help clean weak refs

            while (ctl == currentCtl) {

                long startTime = System.nanoTime();

                w.parked = true;

                if (w.eventCount == v)             // must recheck

                    LockSupport.parkNanos(this, SHRINK_RATE);

                w.parked = false;

                if (w.eventCount != v)

                    break;

                else if (System.nanoTime() - startTime <

                         SHRINK_RATE - (SHRINK_RATE / 10)) // timing slop

                    Thread.interrupted();          // spurious wakeup

                else if (UNSAFE.compareAndSwapLong(this, ctlOffset,

                                                   currentCtl, prevCtl)) {

                    w.terminate = true;            // restore previous

                    w.eventCount = ((int)currentCtl + EC_UNIT) & E_MASK;

                    break;

                }

            }

        }

    }

    // Submissions

    /**

     * Enqueues the given task in the submissionQueue.  Same idea as

     * ForkJoinWorkerThread.pushTask except for use of submissionLock.

     *

     * @param t the task

     */

    private void addSubmission(ForkJoinTask<?> t) {

        final ReentrantLock lock = this.submissionLock;

        lock.lock();

        try {

            ForkJoinTask<?>[] q; int s, m;

            if ((q = submissionQueue) != null) {    // ignore if queue removed

                long u = (((s = queueTop) & (m = q.length-1)) << ASHIFT)+ABASE;

                UNSAFE.putOrderedObject(q, u, t);

                queueTop = s + 1;

                if (s - queueBase == m)

                    growSubmissionQueue();

            }

        } finally {

            lock.unlock();

        }

        signalWork();

    }

    //  (pollSubmission is defined below with exported methods)

    /**

     * Creates or doubles submissionQueue array.

     * Basically identical to ForkJoinWorkerThread version.

     */

    private void growSubmissionQueue() {

        ForkJoinTask<?>[] oldQ = submissionQueue;

        int size = oldQ != null ? oldQ.length << 1 : INITIAL_QUEUE_CAPACITY;

        if (size > MAXIMUM_QUEUE_CAPACITY)

            throw new RejectedExecutionException("Queue capacity exceeded");

        if (size < INITIAL_QUEUE_CAPACITY)

            size = INITIAL_QUEUE_CAPACITY;

        ForkJoinTask<?>[] q = submissionQueue = new ForkJoinTask<?>[size];

        int mask = size - 1;

        int top = queueTop;

        int oldMask;

        if (oldQ != null && (oldMask = oldQ.length - 1) >= 0) {

            for (int b = queueBase; b != top; ++b) {

                long u = ((b & oldMask) << ASHIFT) + ABASE;

                Object x = UNSAFE.getObjectVolatile(oldQ, u);

                if (x != null && UNSAFE.compareAndSwapObject(oldQ, u, x, null))

                    UNSAFE.putObjectVolatile

                        (q, ((b & mask) << ASHIFT) + ABASE, x);

            }

        }

    }

    // Blocking support

    /**

     * Tries to increment blockedCount, decrement active count

     * (sometimes implicitly) and possibly release or create a

     * compensating worker in preparation for blocking. Fails

     * on contention or termination.

     *

     * @return true if the caller can block, else should recheck and retry

     */

    private boolean tryPreBlock() {

        int b = blockedCount;

        if (UNSAFE.compareAndSwapInt(this, blockedCountOffset, b, b + 1)) {

            int pc = parallelism;

            do {

                ForkJoinWorkerThread[] ws; ForkJoinWorkerThread w;

                int e, ac, tc, rc, i;

                long c = ctl;

                int u = (int)(c >>> 32);

                if ((e = (int)c) < 0) {

                                                 // skip -- terminating

                }

                else if ((ac = (u >> UAC_SHIFT)) <= 0 && e != 0 &&

                         (ws = workers) != null &&

                         (i = ~e & SMASK) < ws.length &&

                         (w = ws[i]) != null) {

                    long nc = ((long)(w.nextWait & E_MASK) |

                               (c & (AC_MASK|TC_MASK)));

                    if (w.eventCount == e &&

                        UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc)) {

                        w.eventCount = (e + EC_UNIT) & E_MASK;

                        if (w.parked)

                            UNSAFE.unpark(w);

                        return true;             // release an idle worker

                    }

                }

                else if ((tc = (short)(u >>> UTC_SHIFT)) >= 0 && ac + pc > 1) {

                    long nc = ((c - AC_UNIT) & AC_MASK) | (c & ~AC_MASK);

                    if (UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc))