1 /*

     2  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     3  *

     4  * This code is free software; you can redistribute it and/or modify it

     5  * under the terms of the GNU General Public License version 2 only, as

     6  * published by the Free Software Foundation.  Sun designates this

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

     8  * by Sun in the LICENSE file that accompanied this code.

     9  *

    10  * This code is distributed in the hope that it will be useful, but WITHOUT

    11  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    12  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    13  * version 2 for more details (a copy is included in the LICENSE file that

    14  * accompanied this code).

    15  *

    16  * You should have received a copy of the GNU General Public License version

    17  * 2 along with this work; if not, write to the Free Software Foundation,

    18  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    19  *

    20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,

    21  * CA 95054 USA or visit www.sun.com if you need additional information or

    22  * have any questions.

    23  */

    24 

    25 /*

    26  * This file is available under and governed by the GNU General Public

    27  * License version 2 only, as published by the Free Software Foundation.

    28  * However, the following notice accompanied the original version of this

    29  * file:

    30  *

    31  * Written by Doug Lea with assistance from members of JCP JSR-166

    32  * Expert Group and released to the public domain, as explained at

    33  * http://creativecommons.org/licenses/publicdomain

    34  */

    35 

    36 package java.util.concurrent;

    37 import java.util.concurrent.locks.*;

    38 import java.util.concurrent.atomic.*;

    39 import java.util.*;

    40 

    41 /**

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

    43  * one of possibly several pooled threads, normally configured

    44  * using {@link Executors} factory methods.

    45  *

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

    47  * provide improved performance when executing large numbers of

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

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

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

    51  * Each {@code ThreadPoolExecutor} also maintains some basic

    52  * statistics, such as the number of completed tasks.

    53  *

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

    55  * provides many adjustable parameters and extensibility

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

    57  * {@link Executors} factory methods {@link

    58  * Executors#newCachedThreadPool} (unbounded thread pool, with

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

    60  * (fixed size thread pool) and {@link

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

    62  * preconfigure settings for the most common usage

    63  * scenarios. Otherwise, use the following guide when manually

    64  * configuring and tuning this class:

    65  *

    66  * <dl>

    67  *

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

    69  *

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

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

    72  * according to the bounds set by

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

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

    75  *

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

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

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

    79  * there are more than corePoolSize but less than maximumPoolSize

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

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

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

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

    84  * allow the pool to accommodate an arbitrary number of concurrent

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

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

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

    88  *

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

    90  *

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

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

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

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

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

    96  *

    97  * <dt>Creating new threads</dt>

    98  *

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

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

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

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

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

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

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

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

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

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

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

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

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

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

   113  *

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

   115  *

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

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

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

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

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

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

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

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

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

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

   126  * than corePoolSizeThreads. But method {@link

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

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

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

   130  *

   131  * <dt>Queuing</dt>

   132  *

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

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

   135  *

   136  * <ul>

   137  *

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

   139  * always prefers adding a new thread

   140  * rather than queuing.</li>

   141  *

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

   143  * always prefers queuing a request rather than adding a new

   144  * thread.</li>

   145  *

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

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

   148  * rejected.</li>

   149  *

   150  * </ul>

   151  *

   152  * There are three general strategies for queuing:

   153  * <ol>

   154  *

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

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

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

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

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

   160  * handling sets of requests that might have internal dependencies.

   161  * Direct handoffs generally require unbounded maximumPoolSizes to

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

   163  * possibility of unbounded thread growth when commands continue to

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

   165  *

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

   167  * example a {@link LinkedBlockingQueue} without a predefined

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

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

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

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

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

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

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

   175  * transient bursts of requests, it admits the possibility of

   176  * unbounded work queue growth when commands continue to arrive on

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

   178  *

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

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

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

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

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

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

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

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

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

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

   189  * may encounter unacceptable scheduling overhead, which also

   190  * decreases throughput.  </li>

   191  *

   192  * </ol>

   193  *

   194  * </dd>

   195  *

   196  * <dt>Rejected tasks</dt>

   197  *

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

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

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

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

   202  * execute} method invokes the {@link

   203  * RejectedExecutionHandler#rejectedExecution} method of its {@link

   204  * RejectedExecutionHandler}.  Four predefined handler policies are

   205  * provided:

   206  *

   207  * <ol>

   208  *

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

   210  * handler throws a runtime {@link RejectedExecutionException} upon

   211  * rejection. </li>

   212  *

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

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

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

   216  * new tasks are submitted. </li>

   217  *

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

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

   220  *

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

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

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

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

   225  *

   226  * </ol>

   227  *

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

   229  * RejectedExecutionHandler} classes. Doing so requires some care

   230  * especially when policies are designed to work only under particular

   231  * capacity or queuing policies. </dd>

   232  *

   233  * <dt>Hook methods</dt>

   234  *

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

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

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

   238  * manipulate the execution environment; for example, reinitializing

   239  * ThreadLocals, gathering statistics, or adding log

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

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

   242  * Executor has fully terminated.

   243  *

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

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

   246  *

   247  * <dt>Queue maintenance</dt>

   248  *

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

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

   251  * other purpose is strongly discouraged.  Two supplied methods,

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

   253  * storage reclamation when large numbers of queued tasks become

   254  * cancelled.</dd>

   255  *

   256  * <dt>Finalization</dt>

   257  *

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

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

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

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

   262  * that unused threads eventually die, by setting appropriate

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

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

   265  *

   266  * </dl>

   267  *

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

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

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

   271  *

   272  *  <pre> {@code

   273  * class PausableThreadPoolExecutor extends ThreadPoolExecutor {

   274  *   private boolean isPaused;

   275  *   private ReentrantLock pauseLock = new ReentrantLock();

   276  *   private Condition unpaused = pauseLock.newCondition();

   277  *

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

   279  *

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

   281  *     super.beforeExecute(t, r);

   282  *     pauseLock.lock();

   283  *     try {

   284  *       while (isPaused) unpaused.await();

   285  *     } catch (InterruptedException ie) {

   286  *       t.interrupt();

   287  *     } finally {

   288  *       pauseLock.unlock();

   289  *     }

   290  *   }

   291  *

   292  *   public void pause() {

   293  *     pauseLock.lock();

   294  *     try {

   295  *       isPaused = true;

   296  *     } finally {

   297  *       pauseLock.unlock();

   298  *     }

   299  *   }

   300  *

   301  *   public void resume() {

   302  *     pauseLock.lock();

   303  *     try {

   304  *       isPaused = false;

   305  *       unpaused.signalAll();

   306  *     } finally {

   307  *       pauseLock.unlock();

   308  *     }

   309  *   }

   310  * }}</pre>

   311  *

   312  * @since 1.5

   313  * @author Doug Lea

   314  */

   315 public class ThreadPoolExecutor extends AbstractExecutorService {

   316     /**

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

   318      * two conceptual fields

   319      *   workerCount, indicating the effective number of threads

   320      *   runState,    indicating whether running, shutting down etc

   321      *

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

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

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

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

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

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

   328      *

   329      * The workerCount is the number of workers that have been

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

   331      * transiently different from the actual number of live threads,

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

   333      * asked, and when exiting threads are still performing

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

   335      * reported as the current size of the workers set.

   336      *

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

   338      *

   339      *   RUNNING:  Accept new tasks and process queued tasks

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

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

   342      *             and interrupt in-progress tasks

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

   344      *             the thread transitioning to state TIDYING

   345      *             will run the terminated() hook method

   346      *   TERMINATED: terminated() has completed

   347      *

   348      * The numerical order among these values matters, to allow

   349      * ordered comparisons. The runState monotonically increases over

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

   351      *

   352      * RUNNING -> SHUTDOWN

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

   354      * (RUNNING or SHUTDOWN) -> STOP

   355      *    On invocation of shutdownNow()

   356      * SHUTDOWN -> TIDYING

   357      *    When both queue and pool are empty

   358      * STOP -> TIDYING

   359      *    When pool is empty

   360      * TIDYING -> TERMINATED

   361      *    When the terminated() hook method has completed

   362      *

   363      * Threads waiting in awaitTermination() will return when the

   364      * state reaches TERMINATED.

   365      *

   366      * Detecting the transition from SHUTDOWN to TIDYING is less

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

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

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

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

   371      * below).

   372      */

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

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

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

   376 

   377     // runState is stored in the high-order bits

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

   379     private static final int SHUTDOWN   =  0 << COUNT_BITS;

   380     private static final int STOP       =  1 << COUNT_BITS;

   381     private static final int TIDYING    =  2 << COUNT_BITS;

   382     private static final int TERMINATED =  3 << COUNT_BITS;

   383 

   384     // Packing and unpacking ctl

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

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

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

   388 

   389     /*

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

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

   392      */

   393 

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

   395         return c < s;

   396     }

   397 

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

   399         return c >= s;

   400     }

   401 

   402     private static boolean isRunning(int c) {

   403         return c < SHUTDOWN;

   404     }

   405 

   406     /**

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

   408      */

   409     private boolean compareAndIncrementWorkerCount(int expect) {

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

   411     }

   412 

   413     /**

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

   415      */

   416     private boolean compareAndDecrementWorkerCount(int expect) {

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

   418     }

   419 

   420     /**

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

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

   423      * decrements are performed within getTask.

   424      */

   425     private void decrementWorkerCount() {

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

   427     }

   428 

   429     /**

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

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

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

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

   434      * do for example when deciding whether to transition from

   435      * SHUTDOWN to TIDYING).  This accommodates special-purpose

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

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

   438      * expire.

   439      */

   440     private final BlockingQueue<Runnable> workQueue;

   441 

   442     /**

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

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

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

   446      * that this serializes interruptIdleWorkers, which avoids

   447      * unnecessary interrupt storms, especially during shutdown.

   448      * Otherwise exiting threads would concurrently interrupt those

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

   450      * associated statistics bookkeeping of largestPoolSize etc. We

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

   452      * ensuring workers set is stable while separately checking

   453      * permission to interrupt and actually interrupting.

   454      */

   455     private final ReentrantLock mainLock = new ReentrantLock();

   456 

   457     /**

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

   459      * holding mainLock.

   460      */

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

   462 

   463     /**

   464      * Wait condition to support awaitTermination

   465      */

   466     private final Condition termination = mainLock.newCondition();

   467 

   468     /**

   469      * Tracks largest attained pool size. Accessed only under

   470      * mainLock.

   471      */

   472     private int largestPoolSize;

   473 

   474     /**

   475      * Counter for completed tasks. Updated only on termination of

   476      * worker threads. Accessed only under mainLock.

   477      */

   478     private long completedTaskCount;

   479 

   480     /*

   481      * All user control parameters are declared as volatiles so that

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

   483      * for locking, since no internal invariants depend on them

   484      * changing synchronously with respect to other actions.

   485      */

   486 

   487     /**

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

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

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

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

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

   493      * new tasks being rejected or existing ones remaining stuck in

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

   495      * handle OutOfMemoryErrors that might be thrown while trying to

   496      * create threads, since there is generally no recourse from

   497      * within this class.

   498      */

   499     private volatile ThreadFactory threadFactory;

   500 

   501     /**

   502      * Handler called when saturated or shutdown in execute.

   503      */

   504     private volatile RejectedExecutionHandler handler;

   505 

   506     /**

   507      * Timeout in nanoseconds for idle threads waiting for work.

   508      * Threads use this timeout when there are more than corePoolSize

   509      * present or if allowCoreThreadTimeOut. Otherwise they wait

   510      * forever for new work.

   511      */

   512     private volatile long keepAliveTime;

   513 

   514     /**

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

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

   517      * for work.

   518      */

   519     private volatile boolean allowCoreThreadTimeOut;

   520 

   521     /**

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

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

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

   525      */

   526     private volatile int corePoolSize;

   527 

   528     /**

   529      * Maximum pool size. Note that the actual maximum is internally

   530      * bounded by CAPACITY.

   531      */

   532     private volatile int maximumPoolSize;

   533 

   534     /**

   535      * The default rejected execution handler

   536      */

   537     private static final RejectedExecutionHandler defaultHandler =

   538         new AbortPolicy();

   539 

   540     /**

   541      * Permission required for callers of shutdown and shutdownNow.

   542      * We additionally require (see checkShutdownAccess) that callers

   543      * have permission to actually interrupt threads in the worker set

   544      * (as governed by Thread.interrupt, which relies on

   545      * ThreadGroup.checkAccess, which in turn relies on

   546      * SecurityManager.checkAccess). Shutdowns are attempted only if

   547      * these checks pass.

   548      *

   549      * All actual invocations of Thread.interrupt (see

   550      * interruptIdleWorkers and interruptWorkers) ignore

   551      * SecurityExceptions, meaning that the attempted interrupts

   552      * silently fail. In the case of shutdown, they should not fail

   553      * unless the SecurityManager has inconsistent policies, sometimes

   554      * allowing access to a thread and sometimes not. In such cases,

   555      * failure to actually interrupt threads may disable or delay full

   556      * termination. Other uses of interruptIdleWorkers are advisory,

   557      * and failure to actually interrupt will merely delay response to

   558      * configuration changes so is not handled exceptionally.

   559      */

   560     private static final RuntimePermission shutdownPerm =

   561         new RuntimePermission("modifyThread");

   562 

   563     /**

   564      * Class Worker mainly maintains interrupt control state for

   565      * threads running tasks, along with other minor bookkeeping.

   566      * This class opportunistically extends AbstractQueuedSynchronizer

   567      * to simplify acquiring and releasing a lock surrounding each

   568      * task execution.  This protects against interrupts that are

   569      * intended to wake up a worker thread waiting for a task from

   570      * instead interrupting a task being run.  We implement a simple

   571      * non-reentrant mutual exclusion lock rather than use ReentrantLock

   572      * because we do not want worker tasks to be able to reacquire the

   573      * lock when they invoke pool control methods like setCorePoolSize.

   574      */

   575     private final class Worker

   576         extends AbstractQueuedSynchronizer

   577         implements Runnable

   578     {

   579         /**

   580          * This class will never be serialized, but we provide a

   581          * serialVersionUID to suppress a javac warning.

   582          */

   583         private static final long serialVersionUID = 6138294804551838833L;

   584 

   585         /** Thread this worker is running in.  Null if factory fails. */

   586         final Thread thread;

   587         /** Initial task to run.  Possibly null. */

   588         Runnable firstTask;

   589         /** Per-thread task counter */

   590         volatile long completedTasks;

   591 

   592         /**

   593          * Creates with given first task and thread from ThreadFactory.

   594          * @param firstTask the first task (null if none)

   595          */

   596         Worker(Runnable firstTask) {

   597             this.firstTask = firstTask;

   598             this.thread = getThreadFactory().newThread(this);

   599         }

   600 

   601         /** Delegates main run loop to outer runWorker  */

   602         public void run() {

   603             runWorker(this);

   604         }

   605 

   606         // Lock methods

   607         //

   608         // The value 0 represents the unlocked state.

   609         // The value 1 represents the locked state.

   610 

   611         protected boolean isHeldExclusively() {

   612             return getState() == 1;

   613         }

   614 

   615         protected boolean tryAcquire(int unused) {

   616             if (compareAndSetState(0, 1)) {

   617                 setExclusiveOwnerThread(Thread.currentThread());

   618                 return true;

   619             }

   620             return false;

   621         }

   622 

   623         protected boolean tryRelease(int unused) {

   624             setExclusiveOwnerThread(null);

   625             setState(0);

   626             return true;

   627         }

   628 

   629         public void lock()        { acquire(1); }

   630         public boolean tryLock()  { return tryAcquire(1); }

   631         public void unlock()      { release(1); }

   632         public boolean isLocked() { return isHeldExclusively(); }

   633     }

   634 

   635     /*

   636      * Methods for setting control state

   637      */

   638 

   639     /**

   640      * Transitions runState to given target, or leaves it alone if

   641      * already at least the given target.

   642      *

   643      * @param targetState the desired state, either SHUTDOWN or STOP

   644      *        (but not TIDYING or TERMINATED -- use tryTerminate for that)

   645      */

   646     private void advanceRunState(int targetState) {

   647         for (;;) {

   648             int c = ctl.get();

   649             if (runStateAtLeast(c, targetState) ||

   650                 ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))

   651                 break;

   652         }

   653     }

   654 

   655     /**

   656      * Transitions to TERMINATED state if either (SHUTDOWN and pool

   657      * and queue empty) or (STOP and pool empty).  If otherwise

   658      * eligible to terminate but workerCount is nonzero, interrupts an

   659      * idle worker to ensure that shutdown signals propagate. This

   660      * method must be called following any action that might make

   661      * termination possible -- reducing worker count or removing tasks

   662      * from the queue during shutdown. The method is non-private to

   663      * allow access from ScheduledThreadPoolExecutor.

   664      */

   665     final void tryTerminate() {

   666         for (;;) {

   667             int c = ctl.get();

   668             if (isRunning(c) ||

   669                 runStateAtLeast(c, TIDYING) ||

   670                 (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))

   671                 return;

   672             if (workerCountOf(c) != 0) { // Eligible to terminate

   673                 interruptIdleWorkers(ONLY_ONE);

   674                 return;

   675             }

   676 

   677             final ReentrantLock mainLock = this.mainLock;

   678             mainLock.lock();

   679             try {

   680                 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {

   681                     try {

   682                         terminated();

   683                     } finally {

   684                         ctl.set(ctlOf(TERMINATED, 0));

   685                         termination.signalAll();

   686                     }

   687                     return;

   688                 }

   689             } finally {

   690                 mainLock.unlock();

   691             }

   692             // else retry on failed CAS

   693         }

   694     }

   695 

   696     /*

   697      * Methods for controlling interrupts to worker threads.

   698      */

   699 

   700     /**

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

   702      * permission to shut down threads in general (see shutdownPerm).

   703      * If this passes, additionally makes sure the caller is allowed

   704      * to interrupt each worker thread. This might not be true even if

   705      * first check passed, if the SecurityManager treats some threads

   706      * specially.

   707      */

   708     private void checkShutdownAccess() {

   709         SecurityManager security = System.getSecurityManager();

   710         if (security != null) {

   711             security.checkPermission(shutdownPerm);

   712             final ReentrantLock mainLock = this.mainLock;

   713             mainLock.lock();

   714             try {

   715                 for (Worker w : workers)

   716                     security.checkAccess(w.thread);

   717             } finally {

   718                 mainLock.unlock();

   719             }

   720         }

   721     }

   722 

   723     /**

   724      * Interrupts all threads, even if active. Ignores SecurityExceptions

   725      * (in which case some threads may remain uninterrupted).

   726      */

   727     private void interruptWorkers() {

   728         final ReentrantLock mainLock = this.mainLock;

   729         mainLock.lock();

   730         try {

   731             for (Worker w : workers) {

   732                 try {

   733                     w.thread.interrupt();

   734                 } catch (SecurityException ignore) {

   735                 }

   736             }

   737         } finally {

   738             mainLock.unlock();

   739         }

   740     }

   741 

   742     /**

   743      * Interrupts threads that might be waiting for tasks (as

   744      * indicated by not being locked) so they can check for

   745      * termination or configuration changes. Ignores

   746      * SecurityExceptions (in which case some threads may remain

   747      * uninterrupted).

   748      *

   749      * @param onlyOne If true, interrupt at most one worker. This is

   750      * called only from tryTerminate when termination is otherwise

   751      * enabled but there are still other workers.  In this case, at

   752      * most one waiting worker is interrupted to propagate shutdown

   753      * signals in case all threads are currently waiting.

   754      * Interrupting any arbitrary thread ensures that newly arriving

   755      * workers since shutdown began will also eventually exit.

   756      * To guarantee eventual termination, it suffices to always

   757      * interrupt only one idle worker, but shutdown() interrupts all

   758      * idle workers so that redundant workers exit promptly, not

   759      * waiting for a straggler task to finish.

   760      */

   761     private void interruptIdleWorkers(boolean onlyOne) {

   762         final ReentrantLock mainLock = this.mainLock;

   763         mainLock.lock();

   764         try {

   765             for (Worker w : workers) {

   766                 Thread t = w.thread;

   767                 if (!t.isInterrupted() && w.tryLock()) {

   768                     try {

   769                         t.interrupt();

   770                     } catch (SecurityException ignore) {

   771                     } finally {

   772                         w.unlock();

   773                     }

   774                 }

   775                 if (onlyOne)

   776                     break;

   777             }

   778         } finally {

   779             mainLock.unlock();

   780         }

   781     }

   782 

   783     /**

   784      * Common form of interruptIdleWorkers, to avoid having to

   785      * remember what the boolean argument means.

   786      */

   787     private void interruptIdleWorkers() {

   788         interruptIdleWorkers(false);

   789     }

   790 

   791     private static final boolean ONLY_ONE = true;

   792 

   793     /**

   794      * Ensures that unless the pool is stopping, the current thread

   795      * does not have its interrupt set. This requires a double-check

   796      * of state in case the interrupt was cleared concurrently with a

   797      * shutdownNow -- if so, the interrupt is re-enabled.

   798      */

   799     private void clearInterruptsForTaskRun() {

   800         if (runStateLessThan(ctl.get(), STOP) &&

   801             Thread.interrupted() &&

   802             runStateAtLeast(ctl.get(), STOP))

   803             Thread.currentThread().interrupt();

   804     }

   805 

   806     /*

   807      * Misc utilities, most of which are also exported to

   808      * ScheduledThreadPoolExecutor

   809      */

   810 

   811     /**

   812      * Invokes the rejected execution handler for the given command.

   813      * Package-protected for use by ScheduledThreadPoolExecutor.

   814      */

   815     final void reject(Runnable command) {

   816         handler.rejectedExecution(command, this);

   817     }

   818 

   819     /**

   820      * Performs any further cleanup following run state transition on

   821      * invocation of shutdown.  A no-op here, but used by

   822      * ScheduledThreadPoolExecutor to cancel delayed tasks.

   823      */

   824     void onShutdown() {

   825     }

   826 

   827     /**

   828      * State check needed by ScheduledThreadPoolExecutor to

   829      * enable running tasks during shutdown.

   830      *

   831      * @param shutdownOK true if should return true if SHUTDOWN

   832      */

   833     final boolean isRunningOrShutdown(boolean shutdownOK) {

   834         int rs = runStateOf(ctl.get());

   835         return rs == RUNNING || (rs == SHUTDOWN && shutdownOK);

   836     }

   837 

   838     /**

   839      * Drains the task queue into a new list, normally using

   840      * drainTo. But if the queue is a DelayQueue or any other kind of

   841      * queue for which poll or drainTo may fail to remove some

   842      * elements, it deletes them one by one.

   843      */

   844     private List<Runnable> drainQueue() {

   845         BlockingQueue<Runnable> q = workQueue;

   846         List<Runnable> taskList = new ArrayList<Runnable>();

   847         q.drainTo(taskList);

   848         if (!q.isEmpty()) {

   849             for (Runnable r : q.toArray(new Runnable[0])) {

   850                 if (q.remove(r))

   851                     taskList.add(r);

   852             }

   853         }

   854         return taskList;

   855     }

   856 

   857     /*

   858      * Methods for creating, running and cleaning up after workers

   859      */

   860 

   861     /**

   862      * Checks if a new worker can be added with respect to current

   863      * pool state and the given bound (either core or maximum). If so,

   864      * the worker count is adjusted accordingly, and, if possible, a

   865      * new worker is created and started running firstTask as its

   866      * first task. This method returns false if the pool is stopped or

   867      * eligible to shut down. It also returns false if the thread

   868      * factory fails to create a thread when asked, which requires a

   869      * backout of workerCount, and a recheck for termination, in case

   870      * the existence of this worker was holding up termination.

   871      *

   872      * @param firstTask the task the new thread should run first (or

   873      * null if none). Workers are created with an initial first task

   874      * (in method execute()) to bypass queuing when there are fewer

   875      * than corePoolSize threads (in which case we always start one),

   876      * or when the queue is full (in which case we must bypass queue).

   877      * Initially idle threads are usually created via

   878      * prestartCoreThread or to replace other dying workers.

   879      *

   880      * @param core if true use corePoolSize as bound, else

   881      * maximumPoolSize. (A boolean indicator is used here rather than a

   882      * value to ensure reads of fresh values after checking other pool

   883      * state).

   884      * @return true if successful

   885      */

   886     private boolean addWorker(Runnable firstTask, boolean core) {

   887         retry:

   888         for (;;) {

   889             int c = ctl.get();

   890             int rs = runStateOf(c);

   891 

   892             // Check if queue empty only if necessary.

   893             if (rs >= SHUTDOWN &&

   894                 ! (rs == SHUTDOWN &&

   895                    firstTask == null &&

   896                    ! workQueue.isEmpty()))

   897                 return false;

   898 

   899             for (;;) {

   900                 int wc = workerCountOf(c);

   901                 if (wc >= CAPACITY ||

   902                     wc >= (core ? corePoolSize : maximumPoolSize))

   903                     return false;

   904                 if (compareAndIncrementWorkerCount(c))

   905                     break retry;

   906                 c = ctl.get();  // Re-read ctl

   907                 if (runStateOf(c) != rs)

   908                     continue retry;

   909                 // else CAS failed due to workerCount change; retry inner loop

   910             }

   911         }

   912 

   913         Worker w = new Worker(firstTask);

   914         Thread t = w.thread;

   915 

   916         final ReentrantLock mainLock = this.mainLock;

   917         mainLock.lock();

   918         try {

   919             // Recheck while holding lock.

   920             // Back out on ThreadFactory failure or if

   921             // shut down before lock acquired.

   922             int c = ctl.get();

   923             int rs = runStateOf(c);

   924 

   925             if (t == null ||

   926                 (rs >= SHUTDOWN &&

   927                  ! (rs == SHUTDOWN &&

   928                     firstTask == null))) {

   929                 decrementWorkerCount();

   930                 tryTerminate();

   931                 return false;

   932             }

   933 

   934             workers.add(w);

   935 

   936             int s = workers.size();

   937             if (s > largestPoolSize)

   938                 largestPoolSize = s;

   939         } finally {

   940             mainLock.unlock();

   941         }

   942 

   943         t.start();

   944         // It is possible (but unlikely) for a thread to have been

   945         // added to workers, but not yet started, during transition to

   946         // STOP, which could result in a rare missed interrupt,

   947         // because Thread.interrupt is not guaranteed to have any effect

   948         // on a non-yet-started Thread (see Thread#interrupt).

   949         if (runStateOf(ctl.get()) == STOP && ! t.isInterrupted())

   950             t.interrupt();

   951 

   952         return true;

   953     }

   954 

   955     /**

   956      * Performs cleanup and bookkeeping for a dying worker. Called

   957      * only from worker threads. Unless completedAbruptly is set,

   958      * assumes that workerCount has already been adjusted to account

   959      * for exit.  This method removes thread from worker set, and

   960      * possibly terminates the pool or replaces the worker if either

   961      * it exited due to user task exception or if fewer than

   962      * corePoolSize workers are running or queue is non-empty but

   963      * there are no workers.

   964      *

   965      * @param w the worker

   966      * @param completedAbruptly if the worker died due to user exception

   967      */

   968     private void processWorkerExit(Worker w, boolean completedAbruptly) {

   969         if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted

   970             decrementWorkerCount();

   971 

   972         final ReentrantLock mainLock = this.mainLock;

   973         mainLock.lock();

   974         try {

   975             completedTaskCount += w.completedTasks;

   976             workers.remove(w);

   977         } finally {

   978             mainLock.unlock();

   979         }

   980 

   981         tryTerminate();

   982 

   983         int c = ctl.get();

   984         if (runStateLessThan(c, STOP)) {

   985             if (!completedAbruptly) {

   986                 int min = allowCoreThreadTimeOut ? 0 : corePoolSize;

   987                 if (min == 0 && ! workQueue.isEmpty())

   988                     min = 1;

   989                 if (workerCountOf(c) >= min)

   990                     return; // replacement not needed

   991             }

   992             addWorker(null, false);

   993         }

   994     }

   995 

   996     /**

   997      * Performs blocking or timed wait for a task, depending on

   998      * current configuration settings, or returns null if this worker

   999      * must exit because of any of:

  1000      * 1. There are more than maximumPoolSize workers (due to

  1001      *    a call to setMaximumPoolSize).

  1002      * 2. The pool is stopped.

  1003      * 3. The pool is shutdown and the queue is empty.

  1004      * 4. This worker timed out waiting for a task, and timed-out

  1005      *    workers are subject to termination (that is,

  1006      *    {@code allowCoreThreadTimeOut || workerCount > corePoolSize})

  1007      *    both before and after the timed wait.

  1008      *

  1009      * @return task, or null if the worker must exit, in which case

  1010      *         workerCount is decremented

  1011      */

  1012     private Runnable getTask() {

  1013         boolean timedOut = false; // Did the last poll() time out?

  1014 

  1015         retry:

  1016         for (;;) {

  1017             int c = ctl.get();

  1018             int rs = runStateOf(c);

  1019 

  1020             // Check if queue empty only if necessary.

  1021             if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {

  1022                 decrementWorkerCount();

  1023                 return null;

  1024             }

  1025 

  1026             boolean timed;      // Are workers subject to culling?

  1027 

  1028             for (;;) {

  1029                 int wc = workerCountOf(c);

  1030                 timed = allowCoreThreadTimeOut || wc > corePoolSize;

  1031 

  1032                 if (wc <= maximumPoolSize && ! (timedOut && timed))

  1033                     break;

  1034                 if (compareAndDecrementWorkerCount(c))

  1035                     return null;

  1036                 c = ctl.get();  // Re-read ctl

  1037                 if (runStateOf(c) != rs)

  1038                     continue retry;

  1039                 // else CAS failed due to workerCount change; retry inner loop

  1040             }

  1041 

  1042             try {

  1043                 Runnable r = timed ?

  1044                     workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :

  1045                     workQueue.take();

  1046                 if (r != null)

  1047                     return r;

  1048                 timedOut = true;

  1049             } catch (InterruptedException retry) {

  1050                 timedOut = false;

  1051             }

  1052         }

  1053     }

  1054 

  1055     /**

  1056      * Main worker run loop.  Repeatedly gets tasks from queue and

  1057      * executes them, while coping with a number of issues:

  1058      *

  1059      * 1. We may start out with an initial task, in which case we

  1060      * don't need to get the first one. Otherwise, as long as pool is

  1061      * running, we get tasks from getTask. If it returns null then the

  1062      * worker exits due to changed pool state or configuration

  1063      * parameters.  Other exits result from exception throws in

  1064      * external code, in which case completedAbruptly holds, which

  1065      * usually leads processWorkerExit to replace this thread.

  1066      *

  1067      * 2. Before running any task, the lock is acquired to prevent

  1068      * other pool interrupts while the task is executing, and

  1069      * clearInterruptsForTaskRun called to ensure that unless pool is

  1070      * stopping, this thread does not have its interrupt set.

  1071      *

  1072      * 3. Each task run is preceded by a call to beforeExecute, which

  1073      * might throw an exception, in which case we cause thread to die

  1074      * (breaking loop with completedAbruptly true) without processing

  1075      * the task.

  1076      *

  1077      * 4. Assuming beforeExecute completes normally, we run the task,

  1078      * gathering any of its thrown exceptions to send to

  1079      * afterExecute. We separately handle RuntimeException, Error

  1080      * (both of which the specs guarantee that we trap) and arbitrary

  1081      * Throwables.  Because we cannot rethrow Throwables within

  1082      * Runnable.run, we wrap them within Errors on the way out (to the

  1083      * thread's UncaughtExceptionHandler).  Any thrown exception also

  1084      * conservatively causes thread to die.

  1085      *

  1086      * 5. After task.run completes, we call afterExecute, which may

  1087      * also throw an exception, which will also cause thread to

  1088      * die. According to JLS Sec 14.20, this exception is the one that

  1089      * will be in effect even if task.run throws.

  1090      *

  1091      * The net effect of the exception mechanics is that afterExecute

  1092      * and the thread's UncaughtExceptionHandler have as accurate

  1093      * information as we can provide about any problems encountered by

  1094      * user code.

  1095      *

  1096      * @param w the worker

  1097      */

  1098     final void runWorker(Worker w) {

  1099         Runnable task = w.firstTask;

  1100         w.firstTask = null;

  1101         boolean completedAbruptly = true;

  1102         try {

  1103             while (task != null || (task = getTask()) != null) {

  1104                 w.lock();

  1105                 clearInterruptsForTaskRun();

  1106                 try {

  1107                     beforeExecute(w.thread, task);

  1108                     Throwable thrown = null;

  1109                     try {

  1110                         task.run();

  1111                     } catch (RuntimeException x) {

  1112                         thrown = x; throw x;

  1113                     } catch (Error x) {

  1114                         thrown = x; throw x;

  1115                     } catch (Throwable x) {

  1116                         thrown = x; throw new Error(x);

  1117                     } finally {

  1118                         afterExecute(task, thrown);

  1119                     }

  1120                 } finally {

  1121                     task = null;

  1122                     w.completedTasks++;

  1123                     w.unlock();

  1124                 }

  1125             }

  1126             completedAbruptly = false;

  1127         } finally {

  1128             processWorkerExit(w, completedAbruptly);

  1129         }

  1130     }

  1131 

  1132     // Public constructors and methods

  1133 

  1134     /**

  1135      * Creates a new {@code ThreadPoolExecutor} with the given initial