0001 /*
0002 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
0003 *
0004 * This code is free software; you can redistribute it and/or modify it
0005 * under the terms of the GNU General Public License version 2 only, as
0006 * published by the Free Software Foundation. Sun designates this
0007 * particular file as subject to the "Classpath" exception as provided
0008 * by Sun in the LICENSE file that accompanied this code.
0009 *
0010 * This code is distributed in the hope that it will be useful, but WITHOUT
0011 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
0012 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
0013 * version 2 for more details (a copy is included in the LICENSE file that
0014 * accompanied this code).
0015 *
0016 * You should have received a copy of the GNU General Public License version
0017 * 2 along with this work; if not, write to the Free Software Foundation,
0018 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
0019 *
0020 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
0021 * CA 95054 USA or visit www.sun.com if you need additional information or
0022 * have any questions.
0023 */
0024
0025 /*
0026 * This file is available under and governed by the GNU General Public
0027 * License version 2 only, as published by the Free Software Foundation.
0028 * However, the following notice accompanied the original version of this
0029 * file:
0030 *
0031 * Written by Doug Lea with assistance from members of JCP JSR-166
0032 * Expert Group and released to the public domain, as explained at
0033 * http://creativecommons.org/licenses/publicdomain
0034 */
0035
0036 package java.util.concurrent;
0037
0038 import java.util.concurrent.locks.*;
0039 import java.util.concurrent.atomic.*;
0040 import java.util.*;
0041
0042 /**
0043 * An {@link ExecutorService} that executes each submitted task using
0044 * one of possibly several pooled threads, normally configured
0045 * using {@link Executors} factory methods.
0046 *
0047 * <p>Thread pools address two different problems: they usually
0048 * provide improved performance when executing large numbers of
0049 * asynchronous tasks, due to reduced per-task invocation overhead,
0050 * and they provide a means of bounding and managing the resources,
0051 * including threads, consumed when executing a collection of tasks.
0052 * Each {@code ThreadPoolExecutor} also maintains some basic
0053 * statistics, such as the number of completed tasks.
0054 *
0055 * <p>To be useful across a wide range of contexts, this class
0056 * provides many adjustable parameters and extensibility
0057 * hooks. However, programmers are urged to use the more convenient
0058 * {@link Executors} factory methods {@link
0059 * Executors#newCachedThreadPool} (unbounded thread pool, with
0060 * automatic thread reclamation), {@link Executors#newFixedThreadPool}
0061 * (fixed size thread pool) and {@link
0062 * Executors#newSingleThreadExecutor} (single background thread), that
0063 * preconfigure settings for the most common usage
0064 * scenarios. Otherwise, use the following guide when manually
0065 * configuring and tuning this class:
0066 *
0067 * <dl>
0068 *
0069 * <dt>Core and maximum pool sizes</dt>
0070 *
0071 * <dd>A {@code ThreadPoolExecutor} will automatically adjust the
0072 * pool size (see {@link #getPoolSize})
0073 * according to the bounds set by
0074 * corePoolSize (see {@link #getCorePoolSize}) and
0075 * maximumPoolSize (see {@link #getMaximumPoolSize}).
0076 *
0077 * When a new task is submitted in method {@link #execute}, and fewer
0078 * than corePoolSize threads are running, a new thread is created to
0079 * handle the request, even if other worker threads are idle. If
0080 * there are more than corePoolSize but less than maximumPoolSize
0081 * threads running, a new thread will be created only if the queue is
0082 * full. By setting corePoolSize and maximumPoolSize the same, you
0083 * create a fixed-size thread pool. By setting maximumPoolSize to an
0084 * essentially unbounded value such as {@code Integer.MAX_VALUE}, you
0085 * allow the pool to accommodate an arbitrary number of concurrent
0086 * tasks. Most typically, core and maximum pool sizes are set only
0087 * upon construction, but they may also be changed dynamically using
0088 * {@link #setCorePoolSize} and {@link #setMaximumPoolSize}. </dd>
0089 *
0090 * <dt>On-demand construction</dt>
0091 *
0092 * <dd> By default, even core threads are initially created and
0093 * started only when new tasks arrive, but this can be overridden
0094 * dynamically using method {@link #prestartCoreThread} or {@link
0095 * #prestartAllCoreThreads}. You probably want to prestart threads if
0096 * you construct the pool with a non-empty queue. </dd>
0097 *
0098 * <dt>Creating new threads</dt>
0099 *
0100 * <dd>New threads are created using a {@link ThreadFactory}. If not
0101 * otherwise specified, a {@link Executors#defaultThreadFactory} is
0102 * used, that creates threads to all be in the same {@link
0103 * ThreadGroup} and with the same {@code NORM_PRIORITY} priority and
0104 * non-daemon status. By supplying a different ThreadFactory, you can
0105 * alter the thread's name, thread group, priority, daemon status,
0106 * etc. If a {@code ThreadFactory} fails to create a thread when asked
0107 * by returning null from {@code newThread}, the executor will
0108 * continue, but might not be able to execute any tasks. Threads
0109 * should possess the "modifyThread" {@code RuntimePermission}. If
0110 * worker threads or other threads using the pool do not possess this
0111 * permission, service may be degraded: configuration changes may not
0112 * take effect in a timely manner, and a shutdown pool may remain in a
0113 * state in which termination is possible but not completed.</dd>
0114 *
0115 * <dt>Keep-alive times</dt>
0116 *
0117 * <dd>If the pool currently has more than corePoolSize threads,
0118 * excess threads will be terminated if they have been idle for more
0119 * than the keepAliveTime (see {@link #getKeepAliveTime}). This
0120 * provides a means of reducing resource consumption when the pool is
0121 * not being actively used. If the pool becomes more active later, new
0122 * threads will be constructed. This parameter can also be changed
0123 * dynamically using method {@link #setKeepAliveTime}. Using a value
0124 * of {@code Long.MAX_VALUE} {@link TimeUnit#NANOSECONDS} effectively
0125 * disables idle threads from ever terminating prior to shut down. By
0126 * default, the keep-alive policy applies only when there are more
0127 * than corePoolSizeThreads. But method {@link
0128 * #allowCoreThreadTimeOut(boolean)} can be used to apply this
0129 * time-out policy to core threads as well, so long as the
0130 * keepAliveTime value is non-zero. </dd>
0131 *
0132 * <dt>Queuing</dt>
0133 *
0134 * <dd>Any {@link BlockingQueue} may be used to transfer and hold
0135 * submitted tasks. The use of this queue interacts with pool sizing:
0136 *
0137 * <ul>
0138 *
0139 * <li> If fewer than corePoolSize threads are running, the Executor
0140 * always prefers adding a new thread
0141 * rather than queuing.</li>
0142 *
0143 * <li> If corePoolSize or more threads are running, the Executor
0144 * always prefers queuing a request rather than adding a new
0145 * thread.</li>
0146 *
0147 * <li> If a request cannot be queued, a new thread is created unless
0148 * this would exceed maximumPoolSize, in which case, the task will be
0149 * rejected.</li>
0150 *
0151 * </ul>
0152 *
0153 * There are three general strategies for queuing:
0154 * <ol>
0155 *
0156 * <li> <em> Direct handoffs.</em> A good default choice for a work
0157 * queue is a {@link SynchronousQueue} that hands off tasks to threads
0158 * without otherwise holding them. Here, an attempt to queue a task
0159 * will fail if no threads are immediately available to run it, so a
0160 * new thread will be constructed. This policy avoids lockups when
0161 * handling sets of requests that might have internal dependencies.
0162 * Direct handoffs generally require unbounded maximumPoolSizes to
0163 * avoid rejection of new submitted tasks. This in turn admits the
0164 * possibility of unbounded thread growth when commands continue to
0165 * arrive on average faster than they can be processed. </li>
0166 *
0167 * <li><em> Unbounded queues.</em> Using an unbounded queue (for
0168 * example a {@link LinkedBlockingQueue} without a predefined
0169 * capacity) will cause new tasks to wait in the queue when all
0170 * corePoolSize threads are busy. Thus, no more than corePoolSize
0171 * threads will ever be created. (And the value of the maximumPoolSize
0172 * therefore doesn't have any effect.) This may be appropriate when
0173 * each task is completely independent of others, so tasks cannot
0174 * affect each others execution; for example, in a web page server.
0175 * While this style of queuing can be useful in smoothing out
0176 * transient bursts of requests, it admits the possibility of
0177 * unbounded work queue growth when commands continue to arrive on
0178 * average faster than they can be processed. </li>
0179 *
0180 * <li><em>Bounded queues.</em> A bounded queue (for example, an
0181 * {@link ArrayBlockingQueue}) helps prevent resource exhaustion when
0182 * used with finite maximumPoolSizes, but can be more difficult to
0183 * tune and control. Queue sizes and maximum pool sizes may be traded
0184 * off for each other: Using large queues and small pools minimizes
0185 * CPU usage, OS resources, and context-switching overhead, but can
0186 * lead to artificially low throughput. If tasks frequently block (for
0187 * example if they are I/O bound), a system may be able to schedule
0188 * time for more threads than you otherwise allow. Use of small queues
0189 * generally requires larger pool sizes, which keeps CPUs busier but
0190 * may encounter unacceptable scheduling overhead, which also
0191 * decreases throughput. </li>
0192 *
0193 * </ol>
0194 *
0195 * </dd>
0196 *
0197 * <dt>Rejected tasks</dt>
0198 *
0199 * <dd> New tasks submitted in method {@link #execute} will be
0200 * <em>rejected</em> when the Executor has been shut down, and also
0201 * when the Executor uses finite bounds for both maximum threads and
0202 * work queue capacity, and is saturated. In either case, the {@code
0203 * execute} method invokes the {@link
0204 * RejectedExecutionHandler#rejectedExecution} method of its {@link
0205 * RejectedExecutionHandler}. Four predefined handler policies are
0206 * provided:
0207 *
0208 * <ol>
0209 *
0210 * <li> In the default {@link ThreadPoolExecutor.AbortPolicy}, the
0211 * handler throws a runtime {@link RejectedExecutionException} upon
0212 * rejection. </li>
0213 *
0214 * <li> In {@link ThreadPoolExecutor.CallerRunsPolicy}, the thread
0215 * that invokes {@code execute} itself runs the task. This provides a
0216 * simple feedback control mechanism that will slow down the rate that
0217 * new tasks are submitted. </li>
0218 *
0219 * <li> In {@link ThreadPoolExecutor.DiscardPolicy}, a task that
0220 * cannot be executed is simply dropped. </li>
0221 *
0222 * <li>In {@link ThreadPoolExecutor.DiscardOldestPolicy}, if the
0223 * executor is not shut down, the task at the head of the work queue
0224 * is dropped, and then execution is retried (which can fail again,
0225 * causing this to be repeated.) </li>
0226 *
0227 * </ol>
0228 *
0229 * It is possible to define and use other kinds of {@link
0230 * RejectedExecutionHandler} classes. Doing so requires some care
0231 * especially when policies are designed to work only under particular
0232 * capacity or queuing policies. </dd>
0233 *
0234 * <dt>Hook methods</dt>
0235 *
0236 * <dd>This class provides {@code protected} overridable {@link
0237 * #beforeExecute} and {@link #afterExecute} methods that are called
0238 * before and after execution of each task. These can be used to
0239 * manipulate the execution environment; for example, reinitializing
0240 * ThreadLocals, gathering statistics, or adding log
0241 * entries. Additionally, method {@link #terminated} can be overridden
0242 * to perform any special processing that needs to be done once the
0243 * Executor has fully terminated.
0244 *
0245 * <p>If hook or callback methods throw exceptions, internal worker
0246 * threads may in turn fail and abruptly terminate.</dd>
0247 *
0248 * <dt>Queue maintenance</dt>
0249 *
0250 * <dd> Method {@link #getQueue} allows access to the work queue for
0251 * purposes of monitoring and debugging. Use of this method for any
0252 * other purpose is strongly discouraged. Two supplied methods,
0253 * {@link #remove} and {@link #purge} are available to assist in
0254 * storage reclamation when large numbers of queued tasks become
0255 * cancelled.</dd>
0256 *
0257 * <dt>Finalization</dt>
0258 *
0259 * <dd> A pool that is no longer referenced in a program <em>AND</em>
0260 * has no remaining threads will be {@code shutdown} automatically. If
0261 * you would like to ensure that unreferenced pools are reclaimed even
0262 * if users forget to call {@link #shutdown}, then you must arrange
0263 * that unused threads eventually die, by setting appropriate
0264 * keep-alive times, using a lower bound of zero core threads and/or
0265 * setting {@link #allowCoreThreadTimeOut(boolean)}. </dd>
0266 *
0267 * </dl>
0268 *
0269 * <p> <b>Extension example</b>. Most extensions of this class
0270 * override one or more of the protected hook methods. For example,
0271 * here is a subclass that adds a simple pause/resume feature:
0272 *
0273 * <pre> {@code
0274 * class PausableThreadPoolExecutor extends ThreadPoolExecutor {
0275 * private boolean isPaused;
0276 * private ReentrantLock pauseLock = new ReentrantLock();
0277 * private Condition unpaused = pauseLock.newCondition();
0278 *
0279 * public PausableThreadPoolExecutor(...) { super(...); }
0280 *
0281 * protected void beforeExecute(Thread t, Runnable r) {
0282 * super.beforeExecute(t, r);
0283 * pauseLock.lock();
0284 * try {
0285 * while (isPaused) unpaused.await();
0286 * } catch (InterruptedException ie) {
0287 * t.interrupt();
0288 * } finally {
0289 * pauseLock.unlock();
0290 * }
0291 * }
0292 *
0293 * public void pause() {
0294 * pauseLock.lock();
0295 * try {
0296 * isPaused = true;
0297 * } finally {
0298 * pauseLock.unlock();
0299 * }
0300 * }
0301 *
0302 * public void resume() {
0303 * pauseLock.lock();
0304 * try {
0305 * isPaused = false;
0306 * unpaused.signalAll();
0307 * } finally {
0308 * pauseLock.unlock();
0309 * }
0310 * }
0311 * }}</pre>
0312 *
0313 * @since 1.5
0314 * @author Doug Lea
0315 */
0316 public class ThreadPoolExecutor extends AbstractExecutorService {
0317 /**
0318 * The main pool control state, ctl, is an atomic integer packing
0319 * two conceptual fields
0320 * workerCount, indicating the effective number of threads
0321 * runState, indicating whether running, shutting down etc
0322 *
0323 * In order to pack them into one int, we limit workerCount to
0324 * (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
0325 * billion) otherwise representable. If this is ever an issue in
0326 * the future, the variable can be changed to be an AtomicLong,
0327 * and the shift/mask constants below adjusted. But until the need
0328 * arises, this code is a bit faster and simpler using an int.
0329 *
0330 * The workerCount is the number of workers that have been
0331 * permitted to start and not permitted to stop. The value may be
0332 * transiently different from the actual number of live threads,
0333 * for example when a ThreadFactory fails to create a thread when
0334 * asked, and when exiting threads are still performing
0335 * bookkeeping before terminating. The user-visible pool size is
0336 * reported as the current size of the workers set.
0337 *
0338 * The runState provides the main lifecyle control, taking on values:
0339 *
0340 * RUNNING: Accept new tasks and process queued tasks
0341 * SHUTDOWN: Don't accept new tasks, but process queued tasks
0342 * STOP: Don't accept new tasks, don't process queued tasks,
0343 * and interrupt in-progress tasks
0344 * TIDYING: All tasks have terminated, workerCount is zero,
0345 * the thread transitioning to state TIDYING
0346 * will run the terminated() hook method
0347 * TERMINATED: terminated() has completed
0348 *
0349 * The numerical order among these values matters, to allow
0350 * ordered comparisons. The runState monotonically increases over
0351 * time, but need not hit each state. The transitions are:
0352 *
0353 * RUNNING -> SHUTDOWN
0354 * On invocation of shutdown(), perhaps implicitly in finalize()
0355 * (RUNNING or SHUTDOWN) -> STOP
0356 * On invocation of shutdownNow()
0357 * SHUTDOWN -> TIDYING
0358 * When both queue and pool are empty
0359 * STOP -> TIDYING
0360 * When pool is empty
0361 * TIDYING -> TERMINATED
0362 * When the terminated() hook method has completed
0363 *
0364 * Threads waiting in awaitTermination() will return when the
0365 * state reaches TERMINATED.
0366 *
0367 * Detecting the transition from SHUTDOWN to TIDYING is less
0368 * straightforward than you'd like because the queue may become
0369 * empty after non-empty and vice versa during SHUTDOWN state, but
0370 * we can only terminate if, after seeing that it is empty, we see
0371 * that workerCount is 0 (which sometimes entails a recheck -- see
0372 * below).
0373 */
0374 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING,
0375 0));
0376 private static final int COUNT_BITS = Integer.SIZE - 3;
0377 private static final int CAPACITY = (1 << COUNT_BITS) - 1;
0378
0379 // runState is stored in the high-order bits
0380 private static final int RUNNING = -1 << COUNT_BITS;
0381 private static final int SHUTDOWN = 0 << COUNT_BITS;
0382 private static final int STOP = 1 << COUNT_BITS;
0383 private static final int TIDYING = 2 << COUNT_BITS;
0384 private static final int TERMINATED = 3 << COUNT_BITS;
0385
0386 // Packing and unpacking ctl
0387 private static int runStateOf(int c) {
0388 return c & ~CAPACITY;
0389 }
0390
0391 private static int workerCountOf(int c) {
0392 return c & CAPACITY;
0393 }
0394
0395 private static int ctlOf(int rs, int wc) {
0396 return rs | wc;
0397 }
0398
0399 /*
0400 * Bit field accessors that don't require unpacking ctl.
0401 * These depend on the bit layout and on workerCount being never negative.
0402 */
0403
0404 private static boolean runStateLessThan(int c, int s) {
0405 return c < s;
0406 }
0407
0408 private static boolean runStateAtLeast(int c, int s) {
0409 return c >= s;
0410 }
0411
0412 private static boolean isRunning(int c) {
0413 return c < SHUTDOWN;
0414 }
0415
0416 /**
0417 * Attempt to CAS-increment the workerCount field of ctl.
0418 */
0419 private boolean compareAndIncrementWorkerCount(int expect) {
0420 return ctl.compareAndSet(expect, expect + 1);
0421 }
0422
0423 /**
0424 * Attempt to CAS-decrement the workerCount field of ctl.
0425 */
0426 private boolean compareAndDecrementWorkerCount(int expect) {
0427 return ctl.compareAndSet(expect, expect - 1);
0428 }
0429
0430 /**
0431 * Decrements the workerCount field of ctl. This is called only on
0432 * abrupt termination of a thread (see processWorkerExit). Other
0433 * decrements are performed within getTask.
0434 */
0435 private void decrementWorkerCount() {
0436 do {
0437 } while (!compareAndDecrementWorkerCount(ctl.get()));
0438 }
0439
0440 /**
0441 * The queue used for holding tasks and handing off to worker
0442 * threads. We do not require that workQueue.poll() returning
0443 * null necessarily means that workQueue.isEmpty(), so rely
0444 * solely on isEmpty to see if the queue is empty (which we must
0445 * do for example when deciding whether to transition from
0446 * SHUTDOWN to TIDYING). This accommodates special-purpose
0447 * queues such as DelayQueues for which poll() is allowed to
0448 * return null even if it may later return non-null when delays
0449 * expire.
0450 */
0451 private final BlockingQueue<Runnable> workQueue;
0452
0453 /**
0454 * Lock held on access to workers set and related bookkeeping.
0455 * While we could use a concurrent set of some sort, it turns out
0456 * to be generally preferable to use a lock. Among the reasons is
0457 * that this serializes interruptIdleWorkers, which avoids
0458 * unnecessary interrupt storms, especially during shutdown.
0459 * Otherwise exiting threads would concurrently interrupt those
0460 * that have not yet interrupted. It also simplifies some of the
0461 * associated statistics bookkeeping of largestPoolSize etc. We
0462 * also hold mainLock on shutdown and shutdownNow, for the sake of
0463 * ensuring workers set is stable while separately checking
0464 * permission to interrupt and actually interrupting.
0465 */
0466 private final ReentrantLock mainLock = new ReentrantLock();
0467
0468 /**
0469 * Set containing all worker threads in pool. Accessed only when
0470 * holding mainLock.
0471 */
0472 private final HashSet<Worker> workers = new HashSet<Worker>();
0473
0474 /**
0475 * Wait condition to support awaitTermination
0476 */
0477 private final Condition termination = mainLock.newCondition();
0478
0479 /**
0480 * Tracks largest attained pool size. Accessed only under
0481 * mainLock.
0482 */
0483 private int largestPoolSize;
0484
0485 /**
0486 * Counter for completed tasks. Updated only on termination of
0487 * worker threads. Accessed only under mainLock.
0488 */
0489 private long completedTaskCount;
0490
0491 /*
0492 * All user control parameters are declared as volatiles so that
0493 * ongoing actions are based on freshest values, but without need
0494 * for locking, since no internal invariants depend on them
0495 * changing synchronously with respect to other actions.
0496 */
0497
0498 /**
0499 * Factory for new threads. All threads are created using this
0500 * factory (via method addWorker). All callers must be prepared
0501 * for addWorker to fail, which may reflect a system or user's
0502 * policy limiting the number of threads. Even though it is not
0503 * treated as an error, failure to create threads may result in
0504 * new tasks being rejected or existing ones remaining stuck in
0505 * the queue. On the other hand, no special precautions exist to
0506 * handle OutOfMemoryErrors that might be thrown while trying to
0507 * create threads, since there is generally no recourse from
0508 * within this class.
0509 */
0510 private volatile ThreadFactory threadFactory;
0511
0512 /**
0513 * Handler called when saturated or shutdown in execute.
0514 */
0515 private volatile RejectedExecutionHandler handler;
0516
0517 /**
0518 * Timeout in nanoseconds for idle threads waiting for work.
0519 * Threads use this timeout when there are more than corePoolSize
0520 * present or if allowCoreThreadTimeOut. Otherwise they wait
0521 * forever for new work.
0522 */
0523 private volatile long keepAliveTime;
0524
0525 /**
0526 * If false (default), core threads stay alive even when idle.
0527 * If true, core threads use keepAliveTime to time out waiting
0528 * for work.
0529 */
0530 private volatile boolean allowCoreThreadTimeOut;
0531
0532 /**
0533 * Core pool size is the minimum number of workers to keep alive
0534 * (and not allow to time out etc) unless allowCoreThreadTimeOut
0535 * is set, in which case the minimum is zero.
0536 */
0537 private volatile int corePoolSize;
0538
0539 /**
0540 * Maximum pool size. Note that the actual maximum is internally
0541 * bounded by CAPACITY.
0542 */
0543 private volatile int maximumPoolSize;
0544
0545 /**
0546 * The default rejected execution handler
0547 */
0548 private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();
0549
0550 /**
0551 * Permission required for callers of shutdown and shutdownNow.
0552 * We additionally require (see checkShutdownAccess) that callers
0553 * have permission to actually interrupt threads in the worker set
0554 * (as governed by Thread.interrupt, which relies on
0555 * ThreadGroup.checkAccess, which in turn relies on
0556 * SecurityManager.checkAccess). Shutdowns are attempted only if
0557 * these checks pass.
0558 *
0559 * All actual invocations of Thread.interrupt (see
0560 * interruptIdleWorkers and interruptWorkers) ignore
0561 * SecurityExceptions, meaning that the attempted interrupts
0562 * silently fail. In the case of shutdown, they should not fail
0563 * unless the SecurityManager has inconsistent policies, sometimes
0564 * allowing access to a thread and sometimes not. In such cases,
0565 * failure to actually interrupt threads may disable or delay full
0566 * termination. Other uses of interruptIdleWorkers are advisory,
0567 * and failure to actually interrupt will merely delay response to
0568 * configuration changes so is not handled exceptionally.
0569 */
0570 private static final RuntimePermission shutdownPerm = new RuntimePermission(
0571 "modifyThread");
0572
0573 /**
0574 * Class Worker mainly maintains interrupt control state for
0575 * threads running tasks, along with other minor bookkeeping.
0576 * This class opportunistically extends AbstractQueuedSynchronizer
0577 * to simplify acquiring and releasing a lock surrounding each
0578 * task execution. This protects against interrupts that are
0579 * intended to wake up a worker thread waiting for a task from
0580 * instead interrupting a task being run. We implement a simple
0581 * non-reentrant mutual exclusion lock rather than use ReentrantLock
0582 * because we do not want worker tasks to be able to reacquire the
0583 * lock when they invoke pool control methods like setCorePoolSize.
0584 */
0585 private final class Worker extends AbstractQueuedSynchronizer
0586 implements Runnable {
0587 /**
0588 * This class will never be serialized, but we provide a
0589 * serialVersionUID to suppress a javac warning.
0590 */
0591 private static final long serialVersionUID = 6138294804551838833L;
0592
0593 /** Thread this worker is running in. Null if factory fails. */
0594 final Thread thread;
0595 /** Initial task to run. Possibly null. */
0596 Runnable firstTask;
0597 /** Per-thread task counter */
0598 volatile long completedTasks;
0599
0600 /**
0601 * Creates with given first task and thread from ThreadFactory.
0602 * @param firstTask the first task (null if none)
0603 */
0604 Worker(Runnable firstTask) {
0605 this .firstTask = firstTask;
0606 this .thread = getThreadFactory().newThread(this );
0607 }
0608
0609 /** Delegates main run loop to outer runWorker */
0610 public void run() {
0611 runWorker(this );
0612 }
0613
0614 // Lock methods
0615 //
0616 // The value 0 represents the unlocked state.
0617 // The value 1 represents the locked state.
0618
0619 protected boolean isHeldExclusively() {
0620 return getState() == 1;
0621 }
0622
0623 protected boolean tryAcquire(int unused) {
0624 if (compareAndSetState(0, 1)) {
0625 setExclusiveOwnerThread(Thread.currentThread());
0626 return true;
0627 }
0628 return false;
0629 }
0630
0631 protected boolean tryRelease(int unused) {
0632 setExclusiveOwnerThread(null);
0633 setState(0);
0634 return true;
0635 }
0636
0637 public void lock() {
0638 acquire(1);
0639 }
0640
0641 public boolean tryLock() {
0642 return tryAcquire(1);
0643 }
0644
0645 public void unlock() {
0646 release(1);
0647 }
0648
0649 public boolean isLocked() {
0650 return isHeldExclusively();
0651 }
0652 }
0653
0654 /*
0655 * Methods for setting control state
0656 */
0657
0658 /**
0659 * Transitions runState to given target, or leaves it alone if
0660 * already at least the given target.
0661 *
0662 * @param targetState the desired state, either SHUTDOWN or STOP
0663 * (but not TIDYING or TERMINATED -- use tryTerminate for that)
0664 */
0665 private void advanceRunState(int targetState) {
0666 for (;;) {
0667 int c = ctl.get();
0668 if (runStateAtLeast(c, targetState)
0669 || ctl.compareAndSet(c, ctlOf(targetState,
0670 workerCountOf(c))))
0671 break;
0672 }
0673 }
0674
0675 /**
0676 * Transitions to TERMINATED state if either (SHUTDOWN and pool
0677 * and queue empty) or (STOP and pool empty). If otherwise
0678 * eligible to terminate but workerCount is nonzero, interrupts an
0679 * idle worker to ensure that shutdown signals propagate. This
0680 * method must be called following any action that might make
0681 * termination possible -- reducing worker count or removing tasks
0682 * from the queue during shutdown. The method is non-private to
0683 * allow access from ScheduledThreadPoolExecutor.
0684 */
0685 final void tryTerminate() {
0686 for (;;) {
0687 int c = ctl.get();
0688 if (isRunning(c)
0689 || runStateAtLeast(c, TIDYING)
0690 || (runStateOf(c) == SHUTDOWN && !workQueue
0691 .isEmpty()))
0692 return;
0693 if (workerCountOf(c) != 0) { // Eligible to terminate
0694 interruptIdleWorkers(ONLY_ONE);
0695 return;
0696 }
0697
0698 final ReentrantLock mainLock = this .mainLock;
0699 mainLock.lock();
0700 try {
0701 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
0702 try {
0703 terminated();
0704 } finally {
0705 ctl.set(ctlOf(TERMINATED, 0));
0706 termination.signalAll();
0707 }
0708 return;
0709 }
0710 } finally {
0711 mainLock.unlock();
0712 }
0713 // else retry on failed CAS
0714 }
0715 }
0716
0717 /*
0718 * Methods for controlling interrupts to worker threads.
0719 */
0720
0721 /**
0722 * If there is a security manager, makes sure caller has
0723 * permission to shut down threads in general (see shutdownPerm).
0724 * If this passes, additionally makes sure the caller is allowed
0725 * to interrupt each worker thread. This might not be true even if
0726 * first check passed, if the SecurityManager treats some threads
0727 * specially.
0728 */
0729 private void checkShutdownAccess() {
0730 SecurityManager security = System.getSecurityManager();
0731 if (security != null) {
0732 security.checkPermission(shutdownPerm);
0733 final ReentrantLock mainLock = this .mainLock;
0734 mainLock.lock();
0735 try {
0736 for (Worker w : workers)
0737 security.checkAccess(w.thread);
0738 } finally {
0739 mainLock.unlock();
0740 }
0741 }
0742 }
0743
0744 /**
0745 * Interrupts all threads, even if active. Ignores SecurityExceptions
0746 * (in which case some threads may remain uninterrupted).
0747 */
0748 private void interruptWorkers() {
0749 final ReentrantLock mainLock = this .mainLock;
0750 mainLock.lock();
0751 try {
0752 for (Worker w : workers) {
0753 try {
0754 w.thread.interrupt();
0755 } catch (SecurityException ignore) {
0756 }
0757 }
0758 } finally {
0759 mainLock.unlock();
0760 }
0761 }
0762
0763 /**
0764 * Interrupts threads that might be waiting for tasks (as
0765 * indicated by not being locked) so they can check for
0766 * termination or configuration changes. Ignores
0767 * SecurityExceptions (in which case some threads may remain
0768 * uninterrupted).
0769 *
0770 * @param onlyOne If true, interrupt at most one worker. This is
0771 * called only from tryTerminate when termination is otherwise
0772 * enabled but there are still other workers. In this case, at
0773 * most one waiting worker is interrupted to propagate shutdown
0774 * signals in case all threads are currently waiting.
0775 * Interrupting any arbitrary thread ensures that newly arriving
0776 * workers since shutdown began will also eventually exit.
0777 * To guarantee eventual termination, it suffices to always
0778 * interrupt only one idle worker, but shutdown() interrupts all
0779 * idle workers so that redundant workers exit promptly, not
0780 * waiting for a straggler task to finish.
0781 */
0782 private void interruptIdleWorkers(boolean onlyOne) {
0783 final ReentrantLock mainLock = this .mainLock;
0784 mainLock.lock();
0785 try {
0786 for (Worker w : workers) {
0787 Thread t = w.thread;
0788 if (!t.isInterrupted() && w.tryLock()) {
0789 try {
0790 t.interrupt();
0791 } catch (SecurityException ignore) {
0792 } finally {
0793 w.unlock();
0794 }
0795 }
0796 if (onlyOne)
0797 break;
0798 }
0799 } finally {
0800 mainLock.unlock();
0801 }
0802 }
0803
0804 /**
0805 * Common form of interruptIdleWorkers, to avoid having to
0806 * remember what the boolean argument means.
0807 */
0808 private void interruptIdleWorkers() {
0809 interruptIdleWorkers(false);
0810 }
0811
0812 private static final boolean ONLY_ONE = true;
0813
0814 /**
0815 * Ensures that unless the pool is stopping, the current thread
0816 * does not have its interrupt set. This requires a double-check
0817 * of state in case the interrupt was cleared concurrently with a
0818 * shutdownNow -- if so, the interrupt is re-enabled.
0819 */
0820 private void clearInterruptsForTaskRun() {
0821 if (runStateLessThan(ctl.get(), STOP) && Thread.interrupted()
0822 && runStateAtLeast(ctl.get(), STOP))
0823 Thread.currentThread().interrupt();
0824 }
0825
0826 /*
0827 * Misc utilities, most of which are also exported to
0828 * ScheduledThreadPoolExecutor
0829 */
0830
0831 /**
0832 * Invokes the rejected execution handler for the given command.
0833 * Package-protected for use by ScheduledThreadPoolExecutor.
0834 */
0835 final void reject(Runnable command) {
0836 handler.rejectedExecution(command, this );
0837 }
0838
0839 /**
0840 * Performs any further cleanup following run state transition on
0841 * invocation of shutdown. A no-op here, but used by
0842 * ScheduledThreadPoolExecutor to cancel delayed tasks.
0843 */
0844 void onShutdown() {
0845 }
0846
0847 /**
0848 * State check needed by ScheduledThreadPoolExecutor to
0849 * enable running tasks during shutdown.
0850 *
0851 * @param shutdownOK true if should return true if SHUTDOWN
0852 */
0853 final boolean isRunningOrShutdown(boolean shutdownOK) {
0854 int rs = runStateOf(ctl.get());
0855 return rs == RUNNING || (rs == SHUTDOWN && shutdownOK);
0856 }
0857
0858 /**
0859 * Drains the task queue into a new list, normally using
0860 * drainTo. But if the queue is a DelayQueue or any other kind of
0861 * queue for which poll or drainTo may fail to remove some
0862 * elements, it deletes them one by one.
0863 */
0864 private List<Runnable> drainQueue() {
0865 BlockingQueue<Runnable> q = workQueue;
0866 List<Runnable> taskList = new ArrayList<Runnable>();
0867 q.drainTo(taskList);
0868 if (!q.isEmpty()) {
0869 for (Runnable r : q.toArray(new Runnable[0])) {
0870 if (q.remove(r))
0871 taskList.add(r);
0872 }
0873 }
0874 return taskList;
0875 }
0876
0877 /*
0878 * Methods for creating, running and cleaning up after workers
0879 */
0880
0881 /**
0882 * Checks if a new worker can be added with respect to current
0883 * pool state and the given bound (either core or maximum). If so,
0884 * the worker count is adjusted accordingly, and, if possible, a
0885 * new worker is created and started running firstTask as its
0886 * first task. This method returns false if the pool is stopped or
0887 * eligible to shut down. It also returns false if the thread
0888 * factory fails to create a thread when asked, which requires a
0889 * backout of workerCount, and a recheck for termination, in case
0890 * the existence of this worker was holding up termination.
0891 *
0892 * @param firstTask the task the new thread should run first (or
0893 * null if none). Workers are created with an initial first task
0894 * (in method execute()) to bypass queuing when there are fewer
0895 * than corePoolSize threads (in which case we always start one),
0896 * or when the queue is full (in which case we must bypass queue).
0897 * Initially idle threads are usually created via
0898 * prestartCoreThread or to replace other dying workers.
0899 *
0900 * @param core if true use corePoolSize as bound, else
0901 * maximumPoolSize. (A boolean indicator is used here rather than a
0902 * value to ensure reads of fresh values after checking other pool
0903 * state).
0904 * @return true if successful
0905 */
0906 private boolean addWorker(Runnable firstTask, boolean core) {
0907 retry: for (;;) {
0908 int c = ctl.get();
0909 int rs = runStateOf(c);
0910
0911 // Check if queue empty only if necessary.
0912 if (rs >= SHUTDOWN
0913 && !(rs == SHUTDOWN && firstTask == null && !workQueue
0914 .isEmpty()))
0915 return false;
0916
0917 for (;;) {
0918 int wc = workerCountOf(c);
0919 if (wc >= CAPACITY
0920 || wc >= (core ? corePoolSize : maximumPoolSize))
0921 return false;
0922 if (compareAndIncrementWorkerCount(c))
0923 break retry;
0924 c = ctl.get(); // Re-read ctl
0925 if (runStateOf(c) != rs)
0926 continue retry;
0927 // else CAS failed due to workerCount change; retry inner loop
0928 }
0929 }
0930
0931 Worker w = new Worker(firstTask);
0932 Thread t = w.thread;
0933
0934 final ReentrantLock mainLock = this .mainLock;
0935 mainLock.lock();
0936 try {
0937 // Recheck while holding lock.
0938 // Back out on ThreadFactory failure or if
0939 // shut down before lock acquired.
0940 int c = ctl.get();
0941 int rs = runStateOf(c);
0942
0943 if (t == null
0944 || (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null))) {
0945 decrementWorkerCount();
0946 tryTerminate();
0947 return false;
0948 }
0949
0950 workers.add(w);
0951
0952 int s = workers.size();
0953 if (s > largestPoolSize)
0954 largestPoolSize = s;
0955 } finally {
0956 mainLock.unlock();
0957 }
0958
0959 t.start();
0960 // It is possible (but unlikely) for a thread to have been
0961 // added to workers, but not yet started, during transition to
0962 // STOP, which could result in a rare missed interrupt,
0963 // because Thread.interrupt is not guaranteed to have any effect
0964 // on a non-yet-started Thread (see Thread#interrupt).
0965 if (runStateOf(ctl.get()) == STOP && !t.isInterrupted())
0966 t.interrupt();
0967
0968 return true;
0969 }
0970
0971 /**
0972 * Performs cleanup and bookkeeping for a dying worker. Called
0973 * only from worker threads. Unless completedAbruptly is set,
0974 * assumes that workerCount has already been adjusted to account
0975 * for exit. This method removes thread from worker set, and
0976 * possibly terminates the pool or replaces the worker if either
0977 * it exited due to user task exception or if fewer than
0978 * corePoolSize workers are running or queue is non-empty but
0979 * there are no workers.
0980 *
0981 * @param w the worker
0982 * @param completedAbruptly if the worker died due to user exception
0983 */
0984 private void processWorkerExit(Worker w, boolean completedAbruptly) {
0985 if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
0986 decrementWorkerCount();
0987
0988 final ReentrantLock mainLock = this .mainLock;
0989 mainLock.lock();
0990 try {
0991 completedTaskCount += w.completedTasks;
0992 workers.remove(w);
0993 } finally {
0994 mainLock.unlock();
0995 }
0996
0997 tryTerminate();
0998
0999 int c = ctl.get();
1000 if (runStateLessThan(c, STOP)) {
1001 if (!completedAbruptly) {
1002 int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
1003 if (min == 0 && !workQueue.isEmpty())
1004 min = 1;
1005 if (workerCountOf(c) >= min)
1006 return; // replacement not needed
1007 }
1008 addWorker(null, false);
1009 }
1010 }
1011
1012 /**
1013 * Performs blocking or timed wait for a task, depending on
1014 * current configuration settings, or returns null if this worker
1015 * must exit because of any of:
1016 * 1. There are more than maximumPoolSize workers (due to
1017 * a call to setMaximumPoolSize).
1018 * 2. The pool is stopped.
1019 * 3. The pool is shutdown and the queue is empty.
1020 * 4. This worker timed out waiting for a task, and timed-out
1021 * workers are subject to termination (that is,
1022 * {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
1023 * both before and after the timed wait.
1024 *
1025 * @return task, or null if the worker must exit, in which case
1026 * workerCount is decremented
1027 */
1028 private Runnable getTask() {
1029 boolean timedOut = false; // Did the last poll() time out?
1030
1031 retry: for (;;) {
1032 int c = ctl.get();
1033 int rs = runStateOf(c);
1034
1035 // Check if queue empty only if necessary.
1036 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
1037 decrementWorkerCount();
1038 return null;
1039 }
1040
1041 boolean timed; // Are workers subject to culling?
1042
1043 for (;;) {
1044 int wc = workerCountOf(c);
1045 timed = allowCoreThreadTimeOut || wc > corePoolSize;
1046
1047 if (wc <= maximumPoolSize && !(timedOut && timed))
1048 break;
1049 if (compareAndDecrementWorkerCount(c))
1050 return null;
1051 c = ctl.get(); // Re-read ctl
1052 if (runStateOf(c) != rs)
1053 continue retry;
1054 // else CAS failed due to workerCount change; retry inner loop
1055 }
1056
1057 try {
1058 Runnable r = timed ? workQueue.poll(keepAliveTime,
1059 TimeUnit.NANOSECONDS) : workQueue.take();
1060 if (r != null)
1061 return r;
1062 timedOut = true;
1063 } catch (InterruptedException retry) {
1064 timedOut = false;
1065 }
1066 }
1067 }
1068
1069 /**
1070 * Main worker run loop. Repeatedly gets tasks from queue and
1071 * executes them, while coping with a number of issues:
1072 *
1073 * 1. We may start out with an initial task, in which case we
1074 * don't need to get the first one. Otherwise, as long as pool is
1075 * running, we get tasks from getTask. If it returns null then the
1076 * worker exits due to changed pool state or configuration
1077 * parameters. Other exits result from exception throws in
1078 * external code, in which case completedAbruptly holds, which
1079 * usually leads processWorkerExit to replace this thread.
1080 *
1081 * 2. Before running any task, the lock is acquired to prevent
1082 * other pool interrupts while the task is executing, and
1083 * clearInterruptsForTaskRun called to ensure that unless pool is
1084 * stopping, this thread does not have its interrupt set.
1085 *
1086 * 3. Each task run is preceded by a call to beforeExecute, which
1087 * might throw an exception, in which case we cause thread to die
1088 * (breaking loop with completedAbruptly true) without processing
1089 * the task.
1090 *
1091 * 4. Assuming beforeExecute completes normally, we run the task,
1092 * gathering any of its thrown exceptions to send to
1093 * afterExecute. We separately handle RuntimeException, Error
1094 * (both of which the specs guarantee that we trap) and arbitrary
1095 * Throwables. Because we cannot rethrow Throwables within
1096 * Runnable.run, we wrap them within Errors on the way out (to the
1097 * thread's UncaughtExceptionHandler). Any thrown exception also
1098 * conservatively causes thread to die.
1099 *
1100 * 5. After task.run completes, we call afterExecute, which may
1101 * also throw an exception, which will also cause thread to
1102 * die. According to JLS Sec 14.20, this exception is the one that
1103 * will be in effect even if task.run throws.
1104 *
1105 * The net effect of the exception mechanics is that afterExecute
1106 * and the thread's UncaughtExceptionHandler have as accurate
1107 * information as we can provide about any problems encountered by
1108 * user code.
1109 *
1110 * @param w the worker
1111 */
1112 final void runWorker(Worker w) {
1113 Runnable task = w.firstTask;
1114 w.firstTask = null;
1115 boolean completedAbruptly = true;
1116 try {
1117 while (task != null || (task = getTask()) != null) {
1118 w.lock();
1119 clearInterruptsForTaskRun();
1120 try {
1121 beforeExecute(w.thread, task);
1122 Throwable thrown = null;
1123 try {
1124 task.run();
1125 } catch (RuntimeException x) {
1126 thrown = x;
1127 throw x;
1128 } catch (Error x) {
1129 thrown = x;
1130 throw x;
1131 } catch (Throwable x) {
1132 thrown = x;
1133 throw new Error(x);
1134 } finally {
1135 afterExecute(task, thrown);
1136 }
1137 } finally {
1138 task = null;
1139 w.completedTasks++;
1140 w.unlock();
1141 }
1142 }
1143 completedAbruptly = false;
1144 } finally {
1145 processWorkerExit(w, completedAbruptly);
1146 }
1147 }
1148
1149 // Public constructors and methods
1150
1151 /**
1152 * Creates a new {@code ThreadPoolExecutor} with the given initial
1153 * parameters and default thread factory and rejected execution handler.
1154 * It may be more convenient to use one of the {@link Executors} factory
1155 * methods instead of this general purpose constructor.
1156 *
1157 * @param corePoolSize the number of threads to keep in the pool, even
1158 * if they are idle, unless {@code allowCoreThreadTimeOut} is set
1159 * @param maximumPoolSize the maximum number of threads to allow in the
1160 * pool
1161 * @param keepAliveTime when the number of threads is greater than
1162 * the core, this is the maximum time that excess idle threads
1163 * will wait for new tasks before terminating.
1164 * @param unit the time unit for the {@code keepAliveTime} argument
1165 * @param workQueue the queue to use for holding tasks before they are
1166 * executed. This queue will hold only the {@code Runnable}
1167 * tasks submitted by the {@code execute} method.
1168 * @throws IllegalArgumentException if one of the following holds:<br>
1169 * {@code corePoolSize < 0}<br>
1170 * {@code keepAliveTime < 0}<br>
1171 * {@code maximumPoolSize <= 0}<br>
1172 * {@code maximumPoolSize < corePoolSize}
1173 * @throws NullPointerException if {@code workQueue} is null
1174 */
1175 public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
1176 long keepAliveTime, TimeUnit unit,
1177 BlockingQueue<Runnable> workQueue) {
1178 this (corePoolSize, maximumPoolSize, keepAliveTime, unit,
1179 workQueue, Executors.defaultThreadFactory(),
1180 defaultHandler);
1181 }
1182
1183 /**
1184 * Creates a new {@code ThreadPoolExecutor} with the given initial
1185 * parameters and default rejected execution handler.
1186 *
1187 * @param corePoolSize the number of threads to keep in the pool, even
1188 * if they are idle, unless {@code allowCoreThreadTimeOut} is set
1189 * @param maximumPoolSize the maximum number of threads to allow in the
1190 * pool
1191 * @param keepAliveTime when the number of threads is greater than
1192 * the core, this is the maximum time that excess idle threads
1193 * will wait for new tasks before terminating.
1194 * @param unit the time unit for the {@code keepAliveTime} argument
1195 * @param workQueue the queue to use for holding tasks before they are
1196 * executed. This queue will hold only the {@code Runnable}
1197 * tasks submitted by the {@code execute} method.
1198 * @param threadFactory the factory to use when the executor
1199 * creates a new thread
1200 * @throws IllegalArgumentException if one of the following holds:<br>
1201 * {@code corePoolSize < 0}<br>
1202 * {@code keepAliveTime < 0}<br>
1203 * {@code maximumPoolSize <= 0}<br>
1204 * {@code maximumPoolSize < corePoolSize}
1205 * @throws NullPointerException if {@code workQueue}
1206 * or {@code threadFactory} is null
1207 */
1208 public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
1209 long keepAliveTime, TimeUnit unit,
1210 BlockingQueue<Runnable> workQueue,
1211 ThreadFactory threadFactory) {
1212 this (corePoolSize, maximumPoolSize, keepAliveTime, unit,
1213 workQueue, threadFactory, defaultHandler);
1214 }
1215
1216 /**
1217 * Creates a new {@code ThreadPoolExecutor} with the given initial
1218 * parameters and default thread factory.
1219 *
1220 * @param corePoolSize the number of threads to keep in the pool, even
1221 * if they are idle, unless {@code allowCoreThreadTimeOut} is set
1222 * @param maximumPoolSize the maximum number of threads to allow in the
1223 * pool
1224 * @param keepAliveTime when the number of threads is greater than
1225 * the core, this is the maximum time that excess idle threads
1226 * will wait for new tasks before terminating.
1227 * @param unit the time unit for the {@code keepAliveTime} argument
1228 * @param workQueue the queue to use for holding tasks before they are
1229 * executed. This queue will hold only the {@code Runnable}
1230 * tasks submitted by the {@code execute} method.
1231 * @param handler the handler to use when execution is blocked
1232 * because the thread bounds and queue capacities are reached
1233 * @throws IllegalArgumentException if one of the following holds:<br>
1234 * {@code corePoolSize < 0}<br>
1235 * {@code keepAliveTime < 0}<br>
1236 * {@code maximumPoolSize <= 0}<br>
1237 * {@code maximumPoolSize < corePoolSize}
1238 * @throws NullPointerException if {@code workQueue}
1239 * or {@code handler} is null
1240 */
1241 public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
1242 long keepAliveTime, TimeUnit unit,
1243 BlockingQueue<Runnable> workQueue,
1244 RejectedExecutionHandler handler) {
1245 this (corePoolSize, maximumPoolSize, keepAliveTime, unit,
1246 workQueue, Executors.defaultThreadFactory(), handler);
1247 }
1248
1249 /**
1250 * Creates a new {@code ThreadPoolExecutor} with the given initial
1251 * parameters.
1252 *
1253 * @param corePoolSize the number of threads to keep in the pool, even
1254 * if they are idle, unless {@code allowCoreThreadTimeOut} is set
1255 * @param maximumPoolSize the maximum number of threads to allow in the
1256 * pool
1257 * @param keepAliveTime when the number of threads is greater than
1258 * the core, this is the maximum time that excess idle threads
1259 * will wait for new tasks before terminating.
1260 * @param unit the time unit for the {@code keepAliveTime} argument
1261 * @param workQueue the queue to use for holding tasks before they are
1262 * executed. This queue will hold only the {@code Runnable}
1263 * tasks submitted by the {@code execute} method.
1264 * @param threadFactory the factory to use when the executor
1265 * creates a new thread
1266 * @param handler the handler to use when execution is blocked
1267 * because the thread bounds and queue capacities are reached
1268 * @throws IllegalArgumentException if one of the following holds:<br>
1269 * {@code corePoolSize < 0}<br>
1270 * {@code keepAliveTime < 0}<br>
1271 * {@code maximumPoolSize <= 0}<br>
1272 * {@code maximumPoolSize < corePoolSize}
1273 * @throws NullPointerException if {@code workQueue}
1274 * or {@code threadFactory} or {@code handler} is null
1275 */
1276 public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
1277 long keepAliveTime, TimeUnit unit,
1278 BlockingQueue<Runnable> workQueue,
1279 ThreadFactory threadFactory,
1280 RejectedExecutionHandler handler) {
1281 if (corePoolSize < 0 || maximumPoolSize <= 0
1282 || maximumPoolSize < corePoolSize || keepAliveTime < 0)
1283 throw new IllegalArgumentException();
1284 if (workQueue == null || threadFactory == null
1285 || handler == null)
1286 throw new NullPointerException();
1287 this .corePoolSize = corePoolSize;
1288 this .maximumPoolSize = maximumPoolSize;
1289 this .workQueue = workQueue;
1290 this .keepAliveTime = unit.toNanos(keepAliveTime);
1291 this .threadFactory = threadFactory;
1292 this .handler = handler;
1293 }
1294
1295 /**
1296 * Executes the given task sometime in the future. The task
1297 * may execute in a new thread or in an existing pooled thread.
1298 *
1299 * If the task cannot be submitted for execution, either because this
1300 * executor has been shutdown or because its capacity has been reached,
1301 * the task is handled by the current {@code RejectedExecutionHandler}.
1302 *
1303 * @param command the task to execute
1304 * @throws RejectedExecutionException at discretion of
1305 * {@code RejectedExecutionHandler}, if the task
1306 * cannot be accepted for execution
1307 * @throws NullPointerException if {@code command} is null
1308 */
1309 public void execute(Runnable command) {
1310 if (command == null)
1311 throw new NullPointerException();
1312 /*
1313 * Proceed in 3 steps:
1314 *
1315 * 1. If fewer than corePoolSize threads are running, try to
1316 * start a new thread with the given command as its first
1317 * task. The call to addWorker atomically checks runState and
1318 * workerCount, and so prevents false alarms that would add
1319 * threads when it shouldn't, by returning false.
1320 *
1321 * 2. If a task can be successfully queued, then we still need
1322 * to double-check whether we should have added a thread
1323 * (because existing ones died since last checking) or that
1324 * the pool shut down since entry into this method. So we
1325 * recheck state and if necessary roll back the enqueuing if
1326 * stopped, or start a new thread if there are none.
1327 *
1328 * 3. If we cannot queue task, then we try to add a new
1329 * thread. If it fails, we know we are shut down or saturated
1330 * and so reject the task.
1331 */
1332 int c = ctl.get();
1333 if (workerCountOf(c) < corePoolSize) {
1334 if (addWorker(command, true))
1335 return;
1336 c = ctl.get();
1337 }
1338 if (isRunning(c) && workQueue.offer(command)) {
1339 int recheck = ctl.get();
1340 if (!isRunning(recheck) && remove(command))
1341 reject(command);
1342 else if (workerCountOf(recheck) == 0)
1343 addWorker(null, false);
1344 } else if (!addWorker(command, false))
1345 reject(command);
1346 }
1347
1348 /**
1349 * Initiates an orderly shutdown in which previously submitted
1350 * tasks are executed, but no new tasks will be accepted.
1351 * Invocation has no additional effect if already shut down.
1352 *
1353 * @throws SecurityException {@inheritDoc}
1354 */
1355 public void shutdown() {
1356 final ReentrantLock mainLock = this .mainLock;
1357 mainLock.lock();
1358 try {
1359 checkShutdownAccess();
1360 advanceRunState(SHUTDOWN);
1361 interruptIdleWorkers();
1362 onShutdown(); // hook for ScheduledThreadPoolExecutor
1363 } finally {
1364 mainLock.unlock();
1365 }
1366 tryTerminate();
1367 }
1368
1369 /**
1370 * Attempts to stop all actively executing tasks, halts the
1371 * processing of waiting tasks, and returns a list of the tasks
1372 * that were awaiting execution. These tasks are drained (removed)
1373 * from the task queue upon return from this method.
1374 *
1375 * <p>There are no guarantees beyond best-effort attempts to stop
1376 * processing actively executing tasks. This implementation
1377 * cancels tasks via {@link Thread#interrupt}, so any task that
1378 * fails to respond to interrupts may never terminate.
1379 *
1380 * @throws SecurityException {@inheritDoc}
1381 */
1382 public List<Runnable> shutdownNow() {
1383 List<Runnable> tasks;
1384 final ReentrantLock mainLock = this .mainLock;
1385 mainLock.lock();
1386 try {
1387 checkShutdownAccess();
1388 advanceRunState(STOP);
1389 interruptWorkers();
1390 tasks = drainQueue();
1391 } finally {
1392 mainLock.unlock();
1393 }
1394 tryTerminate();
1395 return tasks;
1396 }
1397
1398 public boolean isShutdown() {
1399 return !isRunning(ctl.get());
1400 }
1401
1402 /**
1403 * Returns true if this executor is in the process of terminating
1404 * after {@link #shutdown} or {@link #shutdownNow} but has not
1405 * completely terminated. This method may be useful for
1406 * debugging. A return of {@code true} reported a sufficient
1407 * period after shutdown may indicate that submitted tasks have
1408 * ignored or suppressed interruption, causing this executor not
1409 * to properly terminate.
1410 *
1411 * @return true if terminating but not yet terminated
1412 */
1413 public boolean isTerminating() {
1414 int c = ctl.get();
1415 return !isRunning(c) && runStateLessThan(c, TERMINATED);
1416 }
1417
1418 public boolean isTerminated() {
1419 return runStateAtLeast(ctl.get(), TERMINATED);
1420 }
1421
1422 public boolean awaitTermination(long timeout, TimeUnit unit)
1423 throws InterruptedException {
1424 long nanos = unit.toNanos(timeout);
1425 final ReentrantLock mainLock = this .mainLock;
1426 mainLock.lock();
1427 try {
1428 for (;;) {
1429 if (runStateAtLeast(ctl.get(), TERMINATED))
1430 return true;
1431 if (nanos <= 0)
1432 return false;
1433 nanos = termination.awaitNanos(nanos);
1434 }
1435 } finally {
1436 mainLock.unlock();
1437 }
1438 }
1439
1440 /**
1441 * Invokes {@code shutdown} when this executor is no longer
1442 * referenced and it has no threads.
1443 */
1444 protected void finalize() {
1445 shutdown();
1446 }
1447
1448 /**
1449 * Sets the thread factory used to create new threads.
1450 *
1451 * @param threadFactory the new thread factory
1452 * @throws NullPointerException if threadFactory is null
1453 * @see #getThreadFactory
1454 */
1455 public void setThreadFactory(ThreadFactory threadFactory) {
1456 if (threadFactory == null)
1457 throw new NullPointerException();
1458 this .threadFactory = threadFactory;
1459 }
1460
1461 /**
1462 * Returns the thread factory used to create new threads.
1463 *
1464 * @return the current thread factory
1465 * @see #setThreadFactory
1466 */
1467 public ThreadFactory getThreadFactory() {
1468 return threadFactory;
1469 }
1470
1471 /**
1472 * Sets a new handler for unexecutable tasks.
1473 *
1474 * @param handler the new handler
1475 * @throws NullPointerException if handler is null
1476 * @see #getRejectedExecutionHandler
1477 */
1478 public void setRejectedExecutionHandler(
1479 RejectedExecutionHandler handler) {
1480 if (handler == null)
1481 throw new NullPointerException();
1482 this .handler = handler;
1483 }
1484
1485 /**
1486 * Returns the current handler for unexecutable tasks.
1487 *
1488 * @return the current handler
1489 * @see #setRejectedExecutionHandler
1490 */
1491 public RejectedExecutionHandler getRejectedExecutionHandler() {
1492 return handler;
1493 }
1494
1495 /**
1496 * Sets the core number of threads. This overrides any value set
1497 * in the constructor. If the new value is smaller than the
1498 * current value, excess existing threads will be terminated when
1499 * they next become idle. If larger, new threads will, if needed,
1500 * be started to execute any queued tasks.
1501 *
1502 * @param corePoolSize the new core size
1503 * @throws IllegalArgumentException if {@code corePoolSize < 0}
1504 * @see #getCorePoolSize
1505 */
1506 public void setCorePoolSize(int corePoolSize) {
1507 if (corePoolSize < 0)
1508 throw new IllegalArgumentException();
1509 int delta = corePoolSize - this .corePoolSize;
1510 this .corePoolSize = corePoolSize;
1511 if (workerCountOf(ctl.get()) > corePoolSize)
1512 interruptIdleWorkers();
1513 else if (delta > 0) {
1514 // We don't really know how many new threads are "needed".
1515 // As a heuristic, prestart enough new workers (up to new
1516 // core size) to handle the current number of tasks in
1517 // queue, but stop if queue becomes empty while doing so.
1518 int k = Math.min(delta, workQueue.size());
1519 while (k-- > 0 && addWorker(null, true)) {
1520 if (workQueue.isEmpty())
1521 break;
1522 }
1523 }
1524 }
1525
1526 /**
1527 * Returns the core number of threads.
1528 *
1529 * @return the core number of threads
1530 * @see #setCorePoolSize
1531 */
1532 public int getCorePoolSize() {
1533 return corePoolSize;
1534 }
1535
1536 /**
1537 * Starts a core thread, causing it to idly wait for work. This
1538 * overrides the default policy of starting core threads only when
1539 * new tasks are executed. This method will return {@code false}
1540 * if all core threads have already been started.
1541 *
1542 * @return {@code true} if a thread was started
1543 */
1544 public boolean prestartCoreThread() {
1545 return workerCountOf(ctl.get()) < corePoolSize
1546 && addWorker(null, true);
1547 }
1548
1549 /**
1550 * Starts all core threads, causing them to idly wait for work. This
1551 * overrides the default policy of starting core threads only when
1552 * new tasks are executed.
1553 *
1554 * @return the number of threads started
1555 */
1556 public int prestartAllCoreThreads() {
1557 int n = 0;
1558 while (addWorker(null, true))
1559 ++n;
1560 return n;
1561 }
1562
1563 /**
1564 * Returns true if this pool allows core threads to time out and
1565 * terminate if no tasks arrive within the keepAlive time, being
1566 * replaced if needed when new tasks arrive. When true, the same
1567 * keep-alive policy applying to non-core threads applies also to
1568 * core threads. When false (the default), core threads are never
1569 * terminated due to lack of incoming tasks.
1570 *
1571 * @return {@code true} if core threads are allowed to time out,
1572 * else {@code false}
1573 *
1574 * @since 1.6
1575 */
1576 public boolean allowsCoreThreadTimeOut() {
1577 return allowCoreThreadTimeOut;
1578 }
1579
1580 /**
1581 * Sets the policy governing whether core threads may time out and
1582 * terminate if no tasks arrive within the keep-alive time, being
1583 * replaced if needed when new tasks arrive. When false, core
1584 * threads are never terminated due to lack of incoming
1585 * tasks. When true, the same keep-alive policy applying to
1586 * non-core threads applies also to core threads. To avoid
1587 * continual thread replacement, the keep-alive time must be
1588 * greater than zero when setting {@code true}. This method
1589 * should in general be called before the pool is actively used.
1590 *
1591 * @param value {@code true} if should time out, else {@code false}
1592 * @throws IllegalArgumentException if value is {@code true}
1593 * and the current keep-alive time is not greater than zero
1594 *
1595 * @since 1.6
1596 */
1597 public void allowCoreThreadTimeOut(boolean value) {
1598 if (value && keepAliveTime <= 0)
1599 throw new IllegalArgumentException(
1600 "Core threads must have nonzero keep alive times");
1601 if (value != allowCoreThreadTimeOut) {
1602 allowCoreThreadTimeOut = value;
1603 if (value)
1604 interruptIdleWorkers();
1605 }
1606 }
1607
1608 /**
1609 * Sets the maximum allowed number of threads. This overrides any
1610 * value set in the constructor. If the new value is smaller than
1611 * the current value, excess existing threads will be
1612 * terminated when they next become idle.
1613 *
1614 * @param maximumPoolSize the new maximum
1615 * @throws IllegalArgumentException if the new maximum is
1616 * less than or equal to zero, or
1617 * less than the {@linkplain #getCorePoolSize core pool size}
1618 * @see #getMaximumPoolSize
1619 */
1620 public void setMaximumPoolSize(int maximumPoolSize) {
1621 if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)
1622 throw new IllegalArgumentException();
1623 this .maximumPoolSize = maximumPoolSize;
1624 if (workerCountOf(ctl.get()) > maximumPoolSize)
1625 interruptIdleWorkers();
1626 }
1627
1628 /**
1629 * Returns the maximum allowed number of threads.
1630 *
1631 * @return the maximum allowed number of threads
1632 * @see #setMaximumPoolSize
1633 */
1634 public int getMaximumPoolSize() {
1635 return maximumPoolSize;
1636 }
1637
1638 /**
1639 * Sets the time limit for which threads may remain idle before
1640 * being terminated. If there are more than the core number of
1641 * threads currently in the pool, after waiting this amount of
1642 * time without processing a task, excess threads will be
1643 * terminated. This overrides any value set in the constructor.
1644 *
1645 * @param time the time to wait. A time value of zero will cause
1646 * excess threads to terminate immediately after executing tasks.
1647 * @param unit the time unit of the {@code time} argument
1648 * @throws IllegalArgumentException if {@code time} less than zero or
1649 * if {@code time} is zero and {@code allowsCoreThreadTimeOut}
1650 * @see #getKeepAliveTime
1651 */
1652 public void setKeepAliveTime(long time, TimeUnit unit) {
1653 if (time < 0)
1654 throw new IllegalArgumentException();
1655 if (time == 0 && allowsCoreThreadTimeOut())
1656 throw new IllegalArgumentException(
1657 "Core threads must have nonzero keep alive times");
1658 long keepAliveTime = unit.toNanos(time);
1659 long delta = keepAliveTime - this .keepAliveTime;
1660 this .keepAliveTime = keepAliveTime;
1661 if (delta < 0)
1662 interruptIdleWorkers();
1663 }
1664
1665 /**
1666 * Returns the thread keep-alive time, which is the amount of time
1667 * that threads in excess of the core pool size may remain
1668 * idle before being terminated.
1669 *
1670 * @param unit the desired time unit of the result
1671 * @return the time limit
1672 * @see #setKeepAliveTime
1673 */
1674 public long getKeepAliveTime(TimeUnit unit) {
1675 return unit.convert(keepAliveTime, TimeUnit.NANOSECONDS);
1676 }
1677
1678 /* User-level queue utilities */
1679
1680 /**
1681 * Returns the task queue used by this executor. Access to the
1682 * task queue is intended primarily for debugging and monitoring.
1683 * This queue may be in active use. Retrieving the task queue
1684 * does not prevent queued tasks from executing.
1685 *
1686 * @return the task queue
1687 */
1688 public BlockingQueue<Runnable> getQueue() {
1689 return workQueue;
1690 }
1691
1692 /**
1693 * Removes this task from the executor's internal queue if it is
1694 * present, thus causing it not to be run if it has not already
1695 * started.
1696 *
1697 * <p> This method may be useful as one part of a cancellation
1698 * scheme. It may fail to remove tasks that have been converted
1699 * into other forms before being placed on the internal queue. For
1700 * example, a task entered using {@code submit} might be
1701 * converted into a form that maintains {@code Future} status.
1702 * However, in such cases, method {@link #purge} may be used to
1703 * remove those Futures that have been cancelled.
1704 *
1705 * @param task the task to remove
1706 * @return true if the task was removed
1707 */
1708 public boolean remove(Runnable task) {
1709 boolean removed = workQueue.remove(task);
1710 tryTerminate(); // In case SHUTDOWN and now empty
1711 return removed;
1712 }
1713
1714 /**
1715 * Tries to remove from the work queue all {@link Future}
1716 * tasks that have been cancelled. This method can be useful as a
1717 * storage reclamation operation, that has no other impact on
1718 * functionality. Cancelled tasks are never executed, but may
1719 * accumulate in work queues until worker threads can actively
1720 * remove them. Invoking this method instead tries to remove them now.
1721 * However, this method may fail to remove tasks in
1722 * the presence of interference by other threads.
1723 */
1724 public void purge() {
1725 final BlockingQueue<Runnable> q = workQueue;
1726 try {
1727 Iterator<Runnable> it = q.iterator();
1728 while (it.hasNext()) {
1729 Runnable r = it.next();
1730 if (r instanceof Future<?>
1731 && ((Future<?>) r).isCancelled())
1732 it.remove();
1733 }
1734 } catch (ConcurrentModificationException fallThrough) {
1735 // Take slow path if we encounter interference during traversal.
1736 // Make copy for traversal and call remove for cancelled entries.
1737 // The slow path is more likely to be O(N*N).
1738 for (Object r : q.toArray())
1739 if (r instanceof Future<?>
1740 && ((Future<?>) r).isCancelled())
1741 q.remove(r);
1742 }
1743
1744 tryTerminate(); // In case SHUTDOWN and now empty
1745 }
1746
1747 /* Statistics */
1748
1749 /**
1750 * Returns the current number of threads in the pool.
1751 *
1752 * @return the number of threads
1753 */
1754 public int getPoolSize() {
1755 final ReentrantLock mainLock = this .mainLock;
1756 mainLock.lock();
1757 try {
1758 // Remove rare and surprising possibility of
1759 // isTerminated() && getPoolSize() > 0
1760 return runStateAtLeast(ctl.get(), TIDYING) ? 0 : workers
1761 .size();
1762 } finally {
1763 mainLock.unlock();
1764 }
1765 }
1766
1767 /**
1768 * Returns the approximate number of threads that are actively
1769 * executing tasks.
1770 *
1771 * @return the number of threads
1772 */
1773 public int getActiveCount() {
1774 final ReentrantLock mainLock = this .mainLock;
1775 mainLock.lock();
1776 try {
1777 int n = 0;
1778 for (Worker w : workers)
1779 if (w.isLocked())
1780 ++n;
1781 return n;
1782 } finally {
1783 mainLock.unlock();
1784 }
1785 }
1786
1787 /**
1788 * Returns the largest number of threads that have ever
1789 * simultaneously been in the pool.
1790 *
1791 * @return the number of threads
1792 */
1793 public int getLargestPoolSize() {
1794 final ReentrantLock mainLock = this .mainLock;
1795 mainLock.lock();
1796 try {
1797 return largestPoolSize;
1798 } finally {
1799 mainLock.unlock();
1800 }
1801 }
1802
1803 /**
1804 * Returns the approximate total number of tasks that have ever been
1805 * scheduled for execution. Because the states of tasks and
1806 * threads may change dynamically during computation, the returned
1807 * value is only an approximation.
1808 *
1809 * @return the number of tasks
1810 */
1811 public long getTaskCount() {
1812 final ReentrantLock mainLock = this .mainLock;
1813 mainLock.lock();
1814 try {
1815 long n = completedTaskCount;
1816 for (Worker w : workers) {
1817 n += w.completedTasks;
1818 if (w.isLocked())
1819 ++n;
1820 }
1821 return n + workQueue.size();
1822 } finally {
1823 mainLock.unlock();
1824 }
1825 }
1826
1827 /**
1828 * Returns the approximate total number of tasks that have
1829 * completed execution. Because the states of tasks and threads
1830 * may change dynamically during computation, the returned value
1831 * is only an approximation, but one that does not ever decrease
1832 * across successive calls.
1833 *
1834 * @return the number of tasks
1835 */
1836 public long getCompletedTaskCount() {
1837 final ReentrantLock mainLock = this .mainLock;
1838 mainLock.lock();
1839 try {
1840 long n = completedTaskCount;
1841 for (Worker w : workers)
1842 n += w.completedTasks;
1843 return n;
1844 } finally {
1845 mainLock.unlock();
1846 }
1847 }
1848
1849 /* Extension hooks */
1850
1851 /**
1852 * Method invoked prior to executing the given Runnable in the
1853 * given thread. This method is invoked by thread {@code t} that
1854 * will execute task {@code r}, and may be used to re-initialize
1855 * ThreadLocals, or to perform logging.
1856 *
1857 * <p>This implementation does nothing, but may be customized in
1858 * subclasses. Note: To properly nest multiple overridings, subclasses
1859 * should generally invoke {@code super.beforeExecute} at the end of
1860 * this method.
1861 *
1862 * @param t the thread that will run task {@code r}
1863 * @param r the task that will be executed
1864 */
1865 protected void beforeExecute(Thread t, Runnable r) {
1866 }
1867
1868 /**
1869 * Method invoked upon completion of execution of the given Runnable.
1870 * This method is invoked by the thread that executed the task. If
1871 * non-null, the Throwable is the uncaught {@code RuntimeException}
1872 * or {@code Error} that caused execution to terminate abruptly.
1873 *
1874 * <p>This implementation does nothing, but may be customized in
1875 * subclasses. Note: To properly nest multiple overridings, subclasses
1876 * should generally invoke {@code super.afterExecute} at the
1877 * beginning of this method.
1878 *
1879 * <p><b>Note:</b> When actions are enclosed in tasks (such as
1880 * {@link FutureTask}) either explicitly or via methods such as
1881 * {@code submit}, these task objects catch and maintain
1882 * computational exceptions, and so they do not cause abrupt
1883 * termination, and the internal exceptions are <em>not</em>
1884 * passed to this method. If you would like to trap both kinds of
1885 * failures in this method, you can further probe for such cases,
1886 * as in this sample subclass that prints either the direct cause
1887 * or the underlying exception if a task has been aborted:
1888 *
1889 * <pre> {@code
1890 * class ExtendedExecutor extends ThreadPoolExecutor {
1891 * // ...
1892 * protected void afterExecute(Runnable r, Throwable t) {
1893 * super.afterExecute(r, t);
1894 * if (t == null && r instanceof Future<?>) {
1895 * try {
1896 * Object result = ((Future<?>) r).get();
1897 * } catch (CancellationException ce) {
1898 * t = ce;
1899 * } catch (ExecutionException ee) {
1900 * t = ee.getCause();
1901 * } catch (InterruptedException ie) {
1902 * Thread.currentThread().interrupt(); // ignore/reset
1903 * }
1904 * }
1905 * if (t != null)
1906 * System.out.println(t);
1907 * }
1908 * }}</pre>
1909 *
1910 * @param r the runnable that has completed
1911 * @param t the exception that caused termination, or null if
1912 * execution completed normally
1913 */
1914 protected void afterExecute(Runnable r, Throwable t) {
1915 }
1916
1917 /**
1918 * Method invoked when the Executor has terminated. Default
1919 * implementation does nothing. Note: To properly nest multiple
1920 * overridings, subclasses should generally invoke
1921 * {@code super.terminated} within this method.
1922 */
1923 protected void terminated() {
1924 }
1925
1926 /* Predefined RejectedExecutionHandlers */
1927
1928 /**
1929 * A handler for rejected tasks that runs the rejected task
1930 * directly in the calling thread of the {@code execute} method,
1931 * unless the executor has been shut down, in which case the task
1932 * is discarded.
1933 */
1934 public static class CallerRunsPolicy implements
1935 RejectedExecutionHandler {
1936 /**
1937 * Creates a {@code CallerRunsPolicy}.
1938 */
1939 public CallerRunsPolicy() {
1940 }
1941
1942 /**
1943 * Executes task r in the caller's thread, unless the executor
1944 * has been shut down, in which case the task is discarded.
1945 *
1946 * @param r the runnable task requested to be executed
1947 * @param e the executor attempting to execute this task
1948 */
1949 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1950 if (!e.isShutdown()) {
1951 r.run();
1952 }
1953 }
1954 }
1955
1956 /**
1957 * A handler for rejected tasks that throws a
1958 * {@code RejectedExecutionException}.
1959 */
1960 public static class AbortPolicy implements RejectedExecutionHandler {
1961 /**
1962 * Creates an {@code AbortPolicy}.
1963 */
1964 public AbortPolicy() {
1965 }
1966
1967 /**
1968 * Always throws RejectedExecutionException.
1969 *
1970 * @param r the runnable task requested to be executed
1971 * @param e the executor attempting to execute this task
1972 * @throws RejectedExecutionException always.
1973 */
1974 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1975 throw new RejectedExecutionException();
1976 }
1977 }
1978
1979 /**
1980 * A handler for rejected tasks that silently discards the
1981 * rejected task.
1982 */
1983 public static class DiscardPolicy implements
1984 RejectedExecutionHandler {
1985 /**
1986 * Creates a {@code DiscardPolicy}.
1987 */
1988 public DiscardPolicy() {
1989 }
1990
1991 /**
1992 * Does nothing, which has the effect of discarding task r.
1993 *
1994 * @param r the runnable task requested to be executed
1995 * @param e the executor attempting to execute this task
1996 */
1997 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1998 }
1999 }
2000
2001 /**
2002 * A handler for rejected tasks that discards the oldest unhandled
2003 * request and then retries {@code execute}, unless the executor
2004 * is shut down, in which case the task is discarded.
2005 */
2006 public static class DiscardOldestPolicy implements
2007 RejectedExecutionHandler {
2008 /**
2009 * Creates a {@code DiscardOldestPolicy} for the given executor.
2010 */
2011 public DiscardOldestPolicy() {
2012 }
2013
2014 /**
2015 * Obtains and ignores the next task that the executor
2016 * would otherwise execute, if one is immediately available,
2017 * and then retries execution of task r, unless the executor
2018 * is shut down, in which case task r is instead discarded.
2019 *
2020 * @param r the runnable task requested to be executed
2021 * @param e the executor attempting to execute this task
2022 */
2023 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
2024 if (!e.isShutdown()) {
2025 e.getQueue().poll();
2026 e.execute(r);
2027 }
2028 }
2029 }
2030 }
|