/*

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

import java.util.concurrent.atomic.*;

import java.util.*;

/**

 * An {@link ExecutorService} that executes each submitted task using

 * one of possibly several pooled threads, normally configured

 * using {@link Executors} factory methods.

 *

 * <p>Thread pools address two different problems: they usually

 * provide improved performance when executing large numbers of

 * asynchronous tasks, due to reduced per-task invocation overhead,

 * and they provide a means of bounding and managing the resources,

 * including threads, consumed when executing a collection of tasks.

 * Each {@code ThreadPoolExecutor} also maintains some basic

 * statistics, such as the number of completed tasks.

 *

 * <p>To be useful across a wide range of contexts, this class

 * provides many adjustable parameters and extensibility

 * hooks. However, programmers are urged to use the more convenient

 * {@link Executors} factory methods {@link

 * Executors#newCachedThreadPool} (unbounded thread pool, with

 * automatic thread reclamation), {@link Executors#newFixedThreadPool}

 * (fixed size thread pool) and {@link

 * Executors#newSingleThreadExecutor} (single background thread), that

 * preconfigure settings for the most common usage

 * scenarios. Otherwise, use the following guide when manually

 * configuring and tuning this class:

 *

 * <dl>

 *

 * <dt>Core and maximum pool sizes</dt>

 *

 * <dd>A {@code ThreadPoolExecutor} will automatically adjust the

 * pool size (see {@link #getPoolSize})

 * according to the bounds set by

 * corePoolSize (see {@link #getCorePoolSize}) and

 * maximumPoolSize (see {@link #getMaximumPoolSize}).

 *

 * When a new task is submitted in method {@link #execute}, and fewer

 * than corePoolSize threads are running, a new thread is created to

 * handle the request, even if other worker threads are idle.  If

 * there are more than corePoolSize but less than maximumPoolSize

 * threads running, a new thread will be created only if the queue is

 * full.  By setting corePoolSize and maximumPoolSize the same, you

 * create a fixed-size thread pool. By setting maximumPoolSize to an

 * essentially unbounded value such as {@code Integer.MAX_VALUE}, you

 * allow the pool to accommodate an arbitrary number of concurrent

 * tasks. Most typically, core and maximum pool sizes are set only

 * upon construction, but they may also be changed dynamically using

 * {@link #setCorePoolSize} and {@link #setMaximumPoolSize}. </dd>

 *

 * <dt>On-demand construction</dt>

 *

 * <dd> By default, even core threads are initially created and

 * started only when new tasks arrive, but this can be overridden

 * dynamically using method {@link #prestartCoreThread} or {@link

 * #prestartAllCoreThreads}.  You probably want to prestart threads if

 * you construct the pool with a non-empty queue. </dd>

 *

 * <dt>Creating new threads</dt>

 *

 * <dd>New threads are created using a {@link ThreadFactory}.  If not

 * otherwise specified, a {@link Executors#defaultThreadFactory} is

 * used, that creates threads to all be in the same {@link

 * ThreadGroup} and with the same {@code NORM_PRIORITY} priority and

 * non-daemon status. By supplying a different ThreadFactory, you can

 * alter the thread's name, thread group, priority, daemon status,

 * etc. If a {@code ThreadFactory} fails to create a thread when asked

 * by returning null from {@code newThread}, the executor will

 * continue, but might not be able to execute any tasks. Threads

 * should possess the "modifyThread" {@code RuntimePermission}. If

 * worker threads or other threads using the pool do not possess this

 * permission, service may be degraded: configuration changes may not

 * take effect in a timely manner, and a shutdown pool may remain in a

 * state in which termination is possible but not completed.</dd>

 *

 * <dt>Keep-alive times</dt>

 *

 * <dd>If the pool currently has more than corePoolSize threads,

 * excess threads will be terminated if they have been idle for more

 * than the keepAliveTime (see {@link #getKeepAliveTime}). This

 * provides a means of reducing resource consumption when the pool is

 * not being actively used. If the pool becomes more active later, new

 * threads will be constructed. This parameter can also be changed

 * dynamically using method {@link #setKeepAliveTime}. Using a value

 * of {@code Long.MAX_VALUE} {@link TimeUnit#NANOSECONDS} effectively

 * disables idle threads from ever terminating prior to shut down. By

 * default, the keep-alive policy applies only when there are more

 * than corePoolSizeThreads. But method {@link

 * #allowCoreThreadTimeOut(boolean)} can be used to apply this

 * time-out policy to core threads as well, so long as the

 * keepAliveTime value is non-zero. </dd>

 *

 * <dt>Queuing</dt>

 *

 * <dd>Any {@link BlockingQueue} may be used to transfer and hold

 * submitted tasks.  The use of this queue interacts with pool sizing:

 *

 * <ul>

 *

 * <li> If fewer than corePoolSize threads are running, the Executor

 * always prefers adding a new thread

 * rather than queuing.</li>

 *

 * <li> If corePoolSize or more threads are running, the Executor

 * always prefers queuing a request rather than adding a new

 * thread.</li>

 *

 * <li> If a request cannot be queued, a new thread is created unless

 * this would exceed maximumPoolSize, in which case, the task will be

 * rejected.</li>

 *

 * </ul>

 *

 * There are three general strategies for queuing:

 * <ol>

 *

 * <li> <em> Direct handoffs.</em> A good default choice for a work

 * queue is a {@link SynchronousQueue} that hands off tasks to threads

 * without otherwise holding them. Here, an attempt to queue a task

 * will fail if no threads are immediately available to run it, so a

 * new thread will be constructed. This policy avoids lockups when

 * handling sets of requests that might have internal dependencies.

 * Direct handoffs generally require unbounded maximumPoolSizes to

 * avoid rejection of new submitted tasks. This in turn admits the

 * possibility of unbounded thread growth when commands continue to

 * arrive on average faster than they can be processed.  </li>

 *

 * <li><em> Unbounded queues.</em> Using an unbounded queue (for

 * example a {@link LinkedBlockingQueue} without a predefined

 * capacity) will cause new tasks to wait in the queue when all

 * corePoolSize threads are busy. Thus, no more than corePoolSize

 * threads will ever be created. (And the value of the maximumPoolSize

 * therefore doesn't have any effect.)  This may be appropriate when

 * each task is completely independent of others, so tasks cannot

 * affect each others execution; for example, in a web page server.

 * While this style of queuing can be useful in smoothing out

 * transient bursts of requests, it admits the possibility of

 * unbounded work queue growth when commands continue to arrive on

 * average faster than they can be processed.  </li>

 *

 * <li><em>Bounded queues.</em> A bounded queue (for example, an

 * {@link ArrayBlockingQueue}) helps prevent resource exhaustion when

 * used with finite maximumPoolSizes, but can be more difficult to

 * tune and control.  Queue sizes and maximum pool sizes may be traded

 * off for each other: Using large queues and small pools minimizes

 * CPU usage, OS resources, and context-switching overhead, but can

 * lead to artificially low throughput.  If tasks frequently block (for

 * example if they are I/O bound), a system may be able to schedule

 * time for more threads than you otherwise allow. Use of small queues

 * generally requires larger pool sizes, which keeps CPUs busier but

 * may encounter unacceptable scheduling overhead, which also

 * decreases throughput.  </li>

 *

 * </ol>

 *

 * </dd>

 *

 * <dt>Rejected tasks</dt>

 *

 * <dd> New tasks submitted in method {@link #execute} will be

 * <em>rejected</em> when the Executor has been shut down, and also

 * when the Executor uses finite bounds for both maximum threads and

 * work queue capacity, and is saturated.  In either case, the {@code

 * execute} method invokes the {@link

 * RejectedExecutionHandler#rejectedExecution} method of its {@link

 * RejectedExecutionHandler}.  Four predefined handler policies are

 * provided:

 *

 * <ol>

 *

 * <li> In the default {@link ThreadPoolExecutor.AbortPolicy}, the

 * handler throws a runtime {@link RejectedExecutionException} upon

 * rejection. </li>

 *

 * <li> In {@link ThreadPoolExecutor.CallerRunsPolicy}, the thread

 * that invokes {@code execute} itself runs the task. This provides a

 * simple feedback control mechanism that will slow down the rate that

 * new tasks are submitted. </li>

 *

 * <li> In {@link ThreadPoolExecutor.DiscardPolicy}, a task that

 * cannot be executed is simply dropped.  </li>

 *

 * <li>In {@link ThreadPoolExecutor.DiscardOldestPolicy}, if the

 * executor is not shut down, the task at the head of the work queue

 * is dropped, and then execution is retried (which can fail again,

 * causing this to be repeated.) </li>

 *

 * </ol>

 *

 * It is possible to define and use other kinds of {@link

 * RejectedExecutionHandler} classes. Doing so requires some care

 * especially when policies are designed to work only under particular

 * capacity or queuing policies. </dd>

 *

 * <dt>Hook methods</dt>

 *

 * <dd>This class provides {@code protected} overridable {@link

 * #beforeExecute} and {@link #afterExecute} methods that are called

 * before and after execution of each task.  These can be used to

 * manipulate the execution environment; for example, reinitializing

 * ThreadLocals, gathering statistics, or adding log

 * entries. Additionally, method {@link #terminated} can be overridden

 * to perform any special processing that needs to be done once the

 * Executor has fully terminated.

 *

 * <p>If hook or callback methods throw exceptions, internal worker

 * threads may in turn fail and abruptly terminate.</dd>

 *

 * <dt>Queue maintenance</dt>

 *

 * <dd> Method {@link #getQueue} allows access to the work queue for

 * purposes of monitoring and debugging.  Use of this method for any

 * other purpose is strongly discouraged.  Two supplied methods,

 * {@link #remove} and {@link #purge} are available to assist in

 * storage reclamation when large numbers of queued tasks become

 * cancelled.</dd>

 *

 * <dt>Finalization</dt>

 *

 * <dd> A pool that is no longer referenced in a program <em>AND</em>

 * has no remaining threads will be {@code shutdown} automatically. If

 * you would like to ensure that unreferenced pools are reclaimed even

 * if users forget to call {@link #shutdown}, then you must arrange

 * that unused threads eventually die, by setting appropriate

 * keep-alive times, using a lower bound of zero core threads and/or

 * setting {@link #allowCoreThreadTimeOut(boolean)}.  </dd>

 *

 * </dl>

 *

 * <p> <b>Extension example</b>. Most extensions of this class

 * override one or more of the protected hook methods. For example,

 * here is a subclass that adds a simple pause/resume feature:

 *

 *  <pre> {@code

 * class PausableThreadPoolExecutor extends ThreadPoolExecutor {

 *   private boolean isPaused;

 *   private ReentrantLock pauseLock = new ReentrantLock();

 *   private Condition unpaused = pauseLock.newCondition();

 *

 *   public PausableThreadPoolExecutor(...) { super(...); }

 *

 *   protected void beforeExecute(Thread t, Runnable r) {

 *     super.beforeExecute(t, r);

 *     pauseLock.lock();

 *     try {

 *       while (isPaused) unpaused.await();

 *     } catch (InterruptedException ie) {

 *       t.interrupt();

 *     } finally {

 *       pauseLock.unlock();

 *     }

 *   }

 *

 *   public void pause() {

 *     pauseLock.lock();

 *     try {

 *       isPaused = true;

 *     } finally {

 *       pauseLock.unlock();

 *     }

 *   }

 *

 *   public void resume() {

 *     pauseLock.lock();

 *     try {

 *       isPaused = false;

 *       unpaused.signalAll();

 *     } finally {

 *       pauseLock.unlock();

 *     }

 *   }

 * }}</pre>

 *

 * @since 1.5

 * @author Doug Lea

 */

public class ThreadPoolExecutor extends AbstractExecutorService {

    /**

     * The main pool control state, ctl, is an atomic integer packing

     * two conceptual fields

     *   workerCount, indicating the effective number of threads

     *   runState,    indicating whether running, shutting down etc

     *

     * In order to pack them into one int, we limit workerCount to

     * (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2

     * billion) otherwise representable. If this is ever an issue in

     * the future, the variable can be changed to be an AtomicLong,

     * and the shift/mask constants below adjusted. But until the need

     * arises, this code is a bit faster and simpler using an int.

     *

     * The workerCount is the number of workers that have been

     * permitted to start and not permitted to stop.  The value may be

     * transiently different from the actual number of live threads,

     * for example when a ThreadFactory fails to create a thread when

     * asked, and when exiting threads are still performing

     * bookkeeping before terminating. The user-visible pool size is

     * reported as the current size of the workers set.

     *

     * The runState provides the main lifecyle control, taking on values:

     *

     *   RUNNING:  Accept new tasks and process queued tasks

     *   SHUTDOWN: Don't accept new tasks, but process queued tasks

     *   STOP:     Don't accept new tasks, don't process queued tasks,

     *             and interrupt in-progress tasks

     *   TIDYING:  All tasks have terminated, workerCount is zero,

     *             the thread transitioning to state TIDYING

     *             will run the terminated() hook method

     *   TERMINATED: terminated() has completed

     *

     * The numerical order among these values matters, to allow

     * ordered comparisons. The runState monotonically increases over

     * time, but need not hit each state. The transitions are:

     *

     * RUNNING -> SHUTDOWN

     *    On invocation of shutdown(), perhaps implicitly in finalize()

     * (RUNNING or SHUTDOWN) -> STOP

     *    On invocation of shutdownNow()

     * SHUTDOWN -> TIDYING

     *    When both queue and pool are empty

     * STOP -> TIDYING

     *    When pool is empty

     * TIDYING -> TERMINATED

     *    When the terminated() hook method has completed

     *

     * Threads waiting in awaitTermination() will return when the

     * state reaches TERMINATED.

     *

     * Detecting the transition from SHUTDOWN to TIDYING is less

     * straightforward than you'd like because the queue may become

     * empty after non-empty and vice versa during SHUTDOWN state, but

     * we can only terminate if, after seeing that it is empty, we see

     * that workerCount is 0 (which sometimes entails a recheck -- see

     * below).

     */

    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

    private static final int COUNT_BITS = Integer.SIZE - 3;

    private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

    // runState is stored in the high-order bits

    private static final int RUNNING    = -1 << COUNT_BITS;

    private static final int SHUTDOWN   =  0 << COUNT_BITS;

    private static final int STOP       =  1 << COUNT_BITS;

    private static final int TIDYING    =  2 << COUNT_BITS;

    private static final int TERMINATED =  3 << COUNT_BITS;

    // Packing and unpacking ctl

    private static int runStateOf(int c)     { return c & ~CAPACITY; }

    private static int workerCountOf(int c)  { return c & CAPACITY; }

    private static int ctlOf(int rs, int wc) { return rs | wc; }

    /*

     * Bit field accessors that don't require unpacking ctl.

     * These depend on the bit layout and on workerCount being never negative.

     */

    private static boolean runStateLessThan(int c, int s) {

        return c < s;

    }

    private static boolean runStateAtLeast(int c, int s) {

        return c >= s;

    }

    private static boolean isRunning(int c) {

        return c < SHUTDOWN;

    }

    /**

     * Attempt to CAS-increment the workerCount field of ctl.

     */

    private boolean compareAndIncrementWorkerCount(int expect) {

        return ctl.compareAndSet(expect, expect + 1);

    }

    /**

     * Attempt to CAS-decrement the workerCount field of ctl.

     */

    private boolean compareAndDecrementWorkerCount(int expect) {

        return ctl.compareAndSet(expect, expect - 1);

    }

    /**

     * Decrements the workerCount field of ctl. This is called only on

     * abrupt termination of a thread (see processWorkerExit). Other

     * decrements are performed within getTask.

     */

    private void decrementWorkerCount() {

        do {} while (! compareAndDecrementWorkerCount(ctl.get()));

    }

    /**

     * The queue used for holding tasks and handing off to worker

     * threads.  We do not require that workQueue.poll() returning

     * null necessarily means that workQueue.isEmpty(), so rely

     * solely on isEmpty to see if the queue is empty (which we must

     * do for example when deciding whether to transition from

     * SHUTDOWN to TIDYING).  This accommodates special-purpose

     * queues such as DelayQueues for which poll() is allowed to

     * return null even if it may later return non-null when delays

     * expire.

     */

    private final BlockingQueue<Runnable> workQueue;

    /**

     * Lock held on access to workers set and related bookkeeping.

     * While we could use a concurrent set of some sort, it turns out

     * to be generally preferable to use a lock. Among the reasons is

     * that this serializes interruptIdleWorkers, which avoids

     * unnecessary interrupt storms, especially during shutdown.

     * Otherwise exiting threads would concurrently interrupt those

     * that have not yet interrupted. It also simplifies some of the

     * associated statistics bookkeeping of largestPoolSize etc. We

     * also hold mainLock on shutdown and shutdownNow, for the sake of

     * ensuring workers set is stable while separately checking

     * permission to interrupt and actually interrupting.

     */

    private final ReentrantLock mainLock = new ReentrantLock();

    /**

     * Set containing all worker threads in pool. Accessed only when

     * holding mainLock.

     */

    private final HashSet<Worker> workers = new HashSet<Worker>();

    /**

     * Wait condition to support awaitTermination

     */

    private final Condition termination = mainLock.newCondition();

    /**

     * Tracks largest attained pool size. Accessed only under

     * mainLock.

     */

    private int largestPoolSize;

    /**

     * Counter for completed tasks. Updated only on termination of

     * worker threads. Accessed only under mainLock.

     */

    private long completedTaskCount;

    /*

     * All user control parameters are declared as volatiles so that

     * ongoing actions are based on freshest values, but without need

     * for locking, since no internal invariants depend on them

     * changing synchronously with respect to other actions.

     */

    /**

     * Factory for new threads. All threads are created using this

     * factory (via method addWorker).  All callers must be prepared

     * for addWorker to fail, which may reflect a system or user's

     * policy limiting the number of threads.  Even though it is not

     * treated as an error, failure to create threads may result in

     * new tasks being rejected or existing ones remaining stuck in

     * the queue. On the other hand, no special precautions exist to

     * handle OutOfMemoryErrors that might be thrown while trying to

     * create threads, since there is generally no recourse from

     * within this class.

     */

    private volatile ThreadFactory threadFactory;

    /**

     * Handler called when saturated or shutdown in execute.

     */

    private volatile RejectedExecutionHandler handler;

    /**

     * Timeout in nanoseconds for idle threads waiting for work.

     * Threads use this timeout when there are more than corePoolSize

     * present or if allowCoreThreadTimeOut. Otherwise they wait

     * forever for new work.

     */

    private volatile long keepAliveTime;

    /**

     * If false (default), core threads stay alive even when idle.

     * If true, core threads use keepAliveTime to time out waiting

     * for work.

     */

    private volatile boolean allowCoreThreadTimeOut;

    /**

     * Core pool size is the minimum number of workers to keep alive

     * (and not allow to time out etc) unless allowCoreThreadTimeOut

     * is set, in which case the minimum is zero.

     */

    private volatile int corePoolSize;

    /**