/*

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

import java.util.HashSet;

import java.util.Iterator;

import java.util.List;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.concurrent.locks.AbstractQueuedSynchronizer;

import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.ReentrantLock;

/**

 * 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(Runnable)},

 * if fewer than corePoolSize threads are running, a new thread is

 * created to handle the request, even if other worker threads are

 * idle.  Else if fewer than maximumPoolSize threads are running, a

 * new thread will be created to handle the request 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(TimeUnit)}).

 * 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(long,

 * TimeUnit)}.  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 corePoolSize threads, 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>If corePoolSize or more threads are running, the Executor

 * always prefers queuing a request rather than adding a new

 * thread.

 *

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

 *

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

 *

 * </ol>

 *

 * </dd>

 *

 * <dt>Rejected tasks</dt>

 *

 * <dd>New tasks submitted in method {@link #execute(Runnable)} 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(Runnable, ThreadPoolExecutor)}

 * 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>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>In {@link ThreadPoolExecutor.DiscardPolicy}, a task that

 * cannot be executed is simply dropped.

 *

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

 *

 * </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(Thread, Runnable)} and

 * {@link #afterExecute(Runnable, Throwable)} 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, callback, or BlockingQueue methods throw exceptions,

 * internal worker threads may in turn fail, abruptly terminate, and

 * possibly be replaced.</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(Runnable)} and {@link #purge} are available to

 * assist in storage reclamation when large numbers of queued tasks

 * become cancelled.</dd>

 *

 * <dt>Reclamation</dt>

 *

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

 * has no remaining threads may be reclaimed (garbage collected)

 * without being explicitly shutdown. You can configure a pool to

 * allow all unused threads to 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 lifecycle 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()

     * (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 COUNT_MASK = (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 & ~COUNT_MASK; }

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

    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;

    }

    /**

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

     */

    private boolean compareAndIncrementWorkerCount(int expect) {

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

    }

    /**

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

        ctl.addAndGet(-1);

    }

    /**

     * 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<>();

    /**

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

     *

     * We go further and preserve pool invariants even in the face of

     * errors such as OutOfMemoryError, that might be thrown while

     * trying to create threads.  Such errors are rather common due to

     * the need to allocate a native stack in Thread.start, and users

     * will want to perform clean pool shutdown to clean up.  There

     * will likely be enough memory available for the cleanup code to

     * complete without encountering yet another OutOfMemoryError.

     */

    private volatile ThreadFactory threadFactory;

    /**

     * Handler called when saturated or shutdown in execute.