/*

 * 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

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

 *

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

 *

 */

/*

 * 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/licenses/publicdomain

 */

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.concurrent.locks.Condition;

import java.util.concurrent.locks.LockSupport;

import java.util.concurrent.locks.ReentrantLock;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.concurrent.atomic.AtomicLong;

/**

 * An {@link ExecutorService} for running {@link ForkJoinTask}s.

 * A {@code ForkJoinPool} provides the entry point for submissions

 * from non-{@code ForkJoinTask}s, 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). A {@code ForkJoinPool} may also be used for mixed

 * execution of some plain {@code Runnable}- or {@code Callable}-

 * based activities along with {@code ForkJoinTask}s. When setting

 * {@linkplain #setAsyncMode async mode}, a {@code ForkJoinPool} may

 * also be appropriate for use with fine-grained tasks of any form

 * that are never joined. Otherwise, other {@code ExecutorService}

 * implementations are typically more appropriate choices.

 *

 * <p>A {@code ForkJoinPool} is constructed with a given target

 * parallelism level; by default, equal to the number of available

 * processors. Unless configured otherwise via {@link

 * #setMaintainsParallelism}, the pool attempts to maintain this

 * number of 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 performed in the face of blocked IO or other unmanaged

 * synchronization. The nested {@link ManagedBlocker} interface

 * enables extension of the kinds of synchronization accommodated.

 * The target parallelism level may also be changed dynamically

 * ({@link #setParallelism}). The total number of threads may be

 * limited using method {@link #setMaximumPoolSize}, in which case it

 * may become possible for the activities of a pool to stall due to

 * the lack of available threads to process new tasks.

 *

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

 *

 * @since 1.7

 * @author Doug Lea

 */

public class ForkJoinPool extends AbstractExecutorService {

    /*

     * See the extended comments interspersed below for design,

     * rationale, and walkthroughs.

     */

    /** Mask for packing and unpacking shorts */

    private static final int  shortMask = 0xffff;

    /** Max pool size -- must be a power of two minus 1 */

    private static final int MAX_THREADS =  0x7FFF;

    /**

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

            try {

                return new ForkJoinWorkerThread(pool);

            } catch (OutOfMemoryError oom)  {

                return null;

            }

        }

    }

    /**

     * Creates a new ForkJoinWorkerThread. This factory is used unless

     * overridden in ForkJoinPool constructors.

     */

    public static final ForkJoinWorkerThreadFactory

        defaultForkJoinWorkerThreadFactory =

        new DefaultForkJoinWorkerThreadFactory();

    /**

     * Permission required for callers of methods that may start or

     * kill threads.

     */

    private static final RuntimePermission modifyThreadPermission =

        new RuntimePermission("modifyThread");

    /**

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

     * permission to modify threads.

     */

    private static void checkPermission() {

        SecurityManager security = System.getSecurityManager();

        if (security != null)

            security.checkPermission(modifyThreadPermission);

    }

    /**

     * Generator for assigning sequence numbers as pool names.

     */

    private static final AtomicInteger poolNumberGenerator =

        new AtomicInteger();

    /**

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

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

     * replacements are protected by workerLock, but it is always kept

     * in a consistent enough state to be randomly accessed without

     * locking by workers performing work-stealing.

     */

    volatile ForkJoinWorkerThread[] workers;

    /**

     * Lock protecting access to workers.

     */

    private final ReentrantLock workerLock;

    /**

     * Condition for awaitTermination.

     */

    private final Condition termination;

    /**

     * The uncaught exception handler used when any worker

     * abruptly terminates

     */

    private Thread.UncaughtExceptionHandler ueh;

    /**

     * Creation factory for worker threads.

     */

    private final ForkJoinWorkerThreadFactory factory;

    /**

     * Head of stack of threads that were created to maintain

     * parallelism when other threads blocked, but have since

     * suspended when the parallelism level rose.

     */

    private volatile WaitQueueNode spareStack;

    /**

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

     * idle or terminating.

     */

    private final AtomicLong stealCount;

    /**

     * Queue for external submissions.

     */

    private final LinkedTransferQueue<ForkJoinTask<?>> submissionQueue;

    /**

     * Head of Treiber stack for barrier sync. See below for explanation.

     */

    private volatile WaitQueueNode syncStack;

    /**

     * The count for event barrier

     */

    private volatile long eventCount;

    /**

     * Pool number, just for assigning useful names to worker threads

     */

    private final int poolNumber;

    /**

     * The maximum allowed pool size

     */

    private volatile int maxPoolSize;

    /**

     * The desired parallelism level, updated only under workerLock.

     */

    private volatile int parallelism;

    /**

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

     */

    private volatile boolean locallyFifo;

    /**

     * Holds number of total (i.e., created and not yet terminated)

     * and running (i.e., not blocked on joins or other managed sync)

     * threads, packed into one int to ensure consistent snapshot when

     * making decisions about creating and suspending spare

     * threads. Updated only by CAS.  Note: CASes in

     * updateRunningCount and preJoin assume that running active count

     * is in low word, so need to be modified if this changes.

     */

    private volatile int workerCounts;

    private static int totalCountOf(int s)           { return s >>> 16;  }

    private static int runningCountOf(int s)         { return s & shortMask; }

    private static int workerCountsFor(int t, int r) { return (t << 16) + r; }

    /**

     * Adds delta (which may be negative) to running count.  This must

     * be called before (with negative arg) and after (with positive)

     * any managed synchronization (i.e., mainly, joins).

     *

     * @param delta the number to add

     */

    final void updateRunningCount(int delta) {

        int s;

        do {} while (!casWorkerCounts(s = workerCounts, s + delta));

    }

    /**

     * Adds delta (which may be negative) to both total and running

     * count.  This must be called upon creation and termination of

     * worker threads.

     *

     * @param delta the number to add

     */

    private void updateWorkerCount(int delta) {

        int d = delta + (delta << 16); // add to both lo and hi parts

        int s;

        do {} while (!casWorkerCounts(s = workerCounts, s + d));

    }

    /**

     * Lifecycle control. High word contains runState, low word

     * contains the number of workers that are (probably) executing

     * tasks. This value is atomically incremented before a worker

     * gets a task to run, and decremented when worker has no tasks

     * and cannot find any. These two fields are bundled together to

     * support correct termination triggering.  Note: activeCount

     * CAS'es cheat by assuming active count is in low word, so need

     * to be modified if this changes

     */

    private volatile int runControl;

    // RunState values. Order among values matters

    private static final int RUNNING     = 0;

    private static final int SHUTDOWN    = 1;

    private static final int TERMINATING = 2;

    private static final int TERMINATED  = 3;

    private static int runStateOf(int c)             { return c >>> 16; }

    private static int activeCountOf(int c)          { return c & shortMask; }

    private static int runControlFor(int r, int a)   { return (r << 16) + a; }

    /**

     * Tries incrementing active count; fails on contention.

     * Called by workers before/during executing tasks.

     *

     * @return true on success

     */

    final boolean tryIncrementActiveCount() {

        int c = runControl;

        return casRunControl(c, c+1);

    }

    /**

     * Tries decrementing active count; fails on contention.

     * Possibly triggers termination on success.

     * Called by workers when they can't find tasks.

     *

     * @return true on success

     */

    final boolean tryDecrementActiveCount() {

        int c = runControl;

        int nextc = c - 1;

        if (!casRunControl(c, nextc))

            return false;

        if (canTerminateOnShutdown(nextc))

            terminateOnShutdown();

        return true;

    }

    /**

     * Returns {@code true} if argument represents zero active count

     * and nonzero runstate, which is the triggering condition for

     * terminating on shutdown.

     */

    private static boolean canTerminateOnShutdown(int c) {

        // i.e. least bit is nonzero runState bit

        return ((c & -c) >>> 16) != 0;

    }

    /**

     * Transition run state to at least the given state. Return true

     * if not already at least given state.

     */

    private boolean transitionRunStateTo(int state) {

        for (;;) {

            int c = runControl;

            if (runStateOf(c) >= state)

                return false;

            if (casRunControl(c, runControlFor(state, activeCountOf(c))))

                return true;

        }

    }

    /**

     * Controls whether to add spares to maintain parallelism

     */

    private volatile boolean maintainsParallelism;

    // Constructors

    /**

     * Creates a {@code ForkJoinPool} with parallelism equal to {@link

     * java.lang.Runtime#availableProcessors}, and using the {@linkplain

     * #defaultForkJoinWorkerThreadFactory default thread factory}.

     *

     * @throws SecurityException if a security manager exists and

     *         the caller is not permitted to modify threads

     *         because it does not hold {@link

     *         java.lang.RuntimePermission}{@code ("modifyThread")}

     */

    public ForkJoinPool() {

        this(Runtime.getRuntime().availableProcessors(),

             defaultForkJoinWorkerThreadFactory);

    }

    /**

     * Creates a {@code ForkJoinPool} with the indicated parallelism

     * level and using the {@linkplain

     * #defaultForkJoinWorkerThreadFactory default thread factory}.

     *

     * @param parallelism the parallelism level

     * @throws IllegalArgumentException if parallelism less than or

     *         equal to zero, or greater than implementation limit

     * @throws SecurityException if a security manager exists and

     *         the caller is not permitted to modify threads

     *         because it does not hold {@link

     *         java.lang.RuntimePermission}{@code ("modifyThread")}

     */

    public ForkJoinPool(int parallelism) {

        this(parallelism, defaultForkJoinWorkerThreadFactory);

    }

    /**

     * Creates a {@code ForkJoinPool} with parallelism equal to {@link

     * java.lang.Runtime#availableProcessors}, and using the given

     * thread factory.

     *

     * @param factory the factory for creating new threads

     * @throws NullPointerException if the factory is null

     * @throws SecurityException if a security manager exists and

     *         the caller is not permitted to modify threads

     *         because it does not hold {@link

     *         java.lang.RuntimePermission}{@code ("modifyThread")}

     */

    public ForkJoinPool(ForkJoinWorkerThreadFactory factory) {

        this(Runtime.getRuntime().availableProcessors(), factory);

    }

    /**

     * Creates a {@code ForkJoinPool} with the given parallelism and

     * thread factory.

     *

     * @param parallelism the parallelism level

     * @param factory the factory for creating new threads

     * @throws IllegalArgumentException if parallelism less than or

     *         equal to zero, or greater than implementation limit

     * @throws NullPointerException if the factory is null

     * @throws SecurityException if a security manager exists and

     *         the caller is not permitted to modify threads

     *         because it does not hold {@link

     *         java.lang.RuntimePermission}{@code ("modifyThread")}

     */

    public ForkJoinPool(int parallelism, ForkJoinWorkerThreadFactory factory) {

        if (parallelism <= 0 || parallelism > MAX_THREADS)

            throw new IllegalArgumentException();

        if (factory == null)

            throw new NullPointerException();

        checkPermission();

        this.factory = factory;

        this.parallelism = parallelism;

        this.maxPoolSize = MAX_THREADS;

        this.maintainsParallelism = true;

        this.poolNumber = poolNumberGenerator.incrementAndGet();

        this.workerLock = new ReentrantLock();

        this.termination = workerLock.newCondition();

        this.stealCount = new AtomicLong();

        this.submissionQueue = new LinkedTransferQueue<ForkJoinTask<?>>();

        // worker array and workers are lazily constructed

    }

    /**

     * Creates a new worker thread using factory.

     *

     * @param index the index to assign worker

     * @return new worker, or null if factory failed

     */

    private ForkJoinWorkerThread createWorker(int index) {

        Thread.UncaughtExceptionHandler h = ueh;

        ForkJoinWorkerThread w = factory.newThread(this);

        if (w != null) {

            w.poolIndex = index;

            w.setDaemon(true);

            w.setAsyncMode(locallyFifo);

            w.setName("ForkJoinPool-" + poolNumber + "-worker-" + index);

            if (h != null)

                w.setUncaughtExceptionHandler(h);

        }

        return w;

    }

    /**

     * Returns a good size for worker array given pool size.

     * Currently requires size to be a power of two.

     */

    private static int arraySizeFor(int poolSize) {

        if (poolSize <= 1)

            return 1;

        // See Hackers Delight, sec 3.2

        int c = poolSize >= MAX_THREADS ? MAX_THREADS : (poolSize - 1);

        c |= c >>>  1;

        c |= c >>>  2;

        c |= c >>>  4;

        c |= c >>>  8;

        c |= c >>> 16;

        return c + 1;

    }

    /**

     * Creates or resizes array if necessary to hold newLength.

     * Call only under exclusion.

     *

     * @return the array

     */

    private ForkJoinWorkerThread[] ensureWorkerArrayCapacity(int newLength) {

        ForkJoinWorkerThread[] ws = workers;

        if (ws == null)

            return workers = new ForkJoinWorkerThread[arraySizeFor(newLength)];

        else if (newLength > ws.length)

            return workers = Arrays.copyOf(ws, arraySizeFor(newLength));

        else

            return ws;

    }

    /**