1.1 --- /dev/null Thu Jan 01 00:00:00 1970 +0000
1.2 +++ b/rt/emul/compact/src/main/java/java/util/concurrent/ThreadPoolExecutor.java Sat Mar 19 10:46:31 2016 +0100
1.3 @@ -0,0 +1,2054 @@
1.4 +/*
1.5 + * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
1.6 + *
1.7 + * This code is free software; you can redistribute it and/or modify it
1.8 + * under the terms of the GNU General Public License version 2 only, as
1.9 + * published by the Free Software Foundation. Oracle designates this
1.10 + * particular file as subject to the "Classpath" exception as provided
1.11 + * by Oracle in the LICENSE file that accompanied this code.
1.12 + *
1.13 + * This code is distributed in the hope that it will be useful, but WITHOUT
1.14 + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
1.15 + * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
1.16 + * version 2 for more details (a copy is included in the LICENSE file that
1.17 + * accompanied this code).
1.18 + *
1.19 + * You should have received a copy of the GNU General Public License version
1.20 + * 2 along with this work; if not, write to the Free Software Foundation,
1.21 + * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
1.22 + *
1.23 + * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
1.24 + * or visit www.oracle.com if you need additional information or have any
1.25 + * questions.
1.26 + */
1.27 +
1.28 +/*
1.29 + * This file is available under and governed by the GNU General Public
1.30 + * License version 2 only, as published by the Free Software Foundation.
1.31 + * However, the following notice accompanied the original version of this
1.32 + * file:
1.33 + *
1.34 + * Written by Doug Lea with assistance from members of JCP JSR-166
1.35 + * Expert Group and released to the public domain, as explained at
1.36 + * http://creativecommons.org/publicdomain/zero/1.0/
1.37 + */
1.38 +
1.39 +package java.util.concurrent;
1.40 +import java.util.concurrent.locks.*;
1.41 +import java.util.concurrent.atomic.*;
1.42 +import java.util.*;
1.43 +
1.44 +/**
1.45 + * An {@link ExecutorService} that executes each submitted task using
1.46 + * one of possibly several pooled threads, normally configured
1.47 + * using {@link Executors} factory methods.
1.48 + *
1.49 + * <p>Thread pools address two different problems: they usually
1.50 + * provide improved performance when executing large numbers of
1.51 + * asynchronous tasks, due to reduced per-task invocation overhead,
1.52 + * and they provide a means of bounding and managing the resources,
1.53 + * including threads, consumed when executing a collection of tasks.
1.54 + * Each {@code ThreadPoolExecutor} also maintains some basic
1.55 + * statistics, such as the number of completed tasks.
1.56 + *
1.57 + * <p>To be useful across a wide range of contexts, this class
1.58 + * provides many adjustable parameters and extensibility
1.59 + * hooks. However, programmers are urged to use the more convenient
1.60 + * {@link Executors} factory methods {@link
1.61 + * Executors#newCachedThreadPool} (unbounded thread pool, with
1.62 + * automatic thread reclamation), {@link Executors#newFixedThreadPool}
1.63 + * (fixed size thread pool) and {@link
1.64 + * Executors#newSingleThreadExecutor} (single background thread), that
1.65 + * preconfigure settings for the most common usage
1.66 + * scenarios. Otherwise, use the following guide when manually
1.67 + * configuring and tuning this class:
1.68 + *
1.69 + * <dl>
1.70 + *
1.71 + * <dt>Core and maximum pool sizes</dt>
1.72 + *
1.73 + * <dd>A {@code ThreadPoolExecutor} will automatically adjust the
1.74 + * pool size (see {@link #getPoolSize})
1.75 + * according to the bounds set by
1.76 + * corePoolSize (see {@link #getCorePoolSize}) and
1.77 + * maximumPoolSize (see {@link #getMaximumPoolSize}).
1.78 + *
1.79 + * When a new task is submitted in method {@link #execute}, and fewer
1.80 + * than corePoolSize threads are running, a new thread is created to
1.81 + * handle the request, even if other worker threads are idle. If
1.82 + * there are more than corePoolSize but less than maximumPoolSize
1.83 + * threads running, a new thread will be created only if the queue is
1.84 + * full. By setting corePoolSize and maximumPoolSize the same, you
1.85 + * create a fixed-size thread pool. By setting maximumPoolSize to an
1.86 + * essentially unbounded value such as {@code Integer.MAX_VALUE}, you
1.87 + * allow the pool to accommodate an arbitrary number of concurrent
1.88 + * tasks. Most typically, core and maximum pool sizes are set only
1.89 + * upon construction, but they may also be changed dynamically using
1.90 + * {@link #setCorePoolSize} and {@link #setMaximumPoolSize}. </dd>
1.91 + *
1.92 + * <dt>On-demand construction</dt>
1.93 + *
1.94 + * <dd> By default, even core threads are initially created and
1.95 + * started only when new tasks arrive, but this can be overridden
1.96 + * dynamically using method {@link #prestartCoreThread} or {@link
1.97 + * #prestartAllCoreThreads}. You probably want to prestart threads if
1.98 + * you construct the pool with a non-empty queue. </dd>
1.99 + *
1.100 + * <dt>Creating new threads</dt>
1.101 + *
1.102 + * <dd>New threads are created using a {@link ThreadFactory}. If not
1.103 + * otherwise specified, a {@link Executors#defaultThreadFactory} is
1.104 + * used, that creates threads to all be in the same {@link
1.105 + * ThreadGroup} and with the same {@code NORM_PRIORITY} priority and
1.106 + * non-daemon status. By supplying a different ThreadFactory, you can
1.107 + * alter the thread's name, thread group, priority, daemon status,
1.108 + * etc. If a {@code ThreadFactory} fails to create a thread when asked
1.109 + * by returning null from {@code newThread}, the executor will
1.110 + * continue, but might not be able to execute any tasks. Threads
1.111 + * should possess the "modifyThread" {@code RuntimePermission}. If
1.112 + * worker threads or other threads using the pool do not possess this
1.113 + * permission, service may be degraded: configuration changes may not
1.114 + * take effect in a timely manner, and a shutdown pool may remain in a
1.115 + * state in which termination is possible but not completed.</dd>
1.116 + *
1.117 + * <dt>Keep-alive times</dt>
1.118 + *
1.119 + * <dd>If the pool currently has more than corePoolSize threads,
1.120 + * excess threads will be terminated if they have been idle for more
1.121 + * than the keepAliveTime (see {@link #getKeepAliveTime}). This
1.122 + * provides a means of reducing resource consumption when the pool is
1.123 + * not being actively used. If the pool becomes more active later, new
1.124 + * threads will be constructed. This parameter can also be changed
1.125 + * dynamically using method {@link #setKeepAliveTime}. Using a value
1.126 + * of {@code Long.MAX_VALUE} {@link TimeUnit#NANOSECONDS} effectively
1.127 + * disables idle threads from ever terminating prior to shut down. By
1.128 + * default, the keep-alive policy applies only when there are more
1.129 + * than corePoolSizeThreads. But method {@link
1.130 + * #allowCoreThreadTimeOut(boolean)} can be used to apply this
1.131 + * time-out policy to core threads as well, so long as the
1.132 + * keepAliveTime value is non-zero. </dd>
1.133 + *
1.134 + * <dt>Queuing</dt>
1.135 + *
1.136 + * <dd>Any {@link BlockingQueue} may be used to transfer and hold
1.137 + * submitted tasks. The use of this queue interacts with pool sizing:
1.138 + *
1.139 + * <ul>
1.140 + *
1.141 + * <li> If fewer than corePoolSize threads are running, the Executor
1.142 + * always prefers adding a new thread
1.143 + * rather than queuing.</li>
1.144 + *
1.145 + * <li> If corePoolSize or more threads are running, the Executor
1.146 + * always prefers queuing a request rather than adding a new
1.147 + * thread.</li>
1.148 + *
1.149 + * <li> If a request cannot be queued, a new thread is created unless
1.150 + * this would exceed maximumPoolSize, in which case, the task will be
1.151 + * rejected.</li>
1.152 + *
1.153 + * </ul>
1.154 + *
1.155 + * There are three general strategies for queuing:
1.156 + * <ol>
1.157 + *
1.158 + * <li> <em> Direct handoffs.</em> A good default choice for a work
1.159 + * queue is a {@link SynchronousQueue} that hands off tasks to threads
1.160 + * without otherwise holding them. Here, an attempt to queue a task
1.161 + * will fail if no threads are immediately available to run it, so a
1.162 + * new thread will be constructed. This policy avoids lockups when
1.163 + * handling sets of requests that might have internal dependencies.
1.164 + * Direct handoffs generally require unbounded maximumPoolSizes to
1.165 + * avoid rejection of new submitted tasks. This in turn admits the
1.166 + * possibility of unbounded thread growth when commands continue to
1.167 + * arrive on average faster than they can be processed. </li>
1.168 + *
1.169 + * <li><em> Unbounded queues.</em> Using an unbounded queue (for
1.170 + * example a {@link LinkedBlockingQueue} without a predefined
1.171 + * capacity) will cause new tasks to wait in the queue when all
1.172 + * corePoolSize threads are busy. Thus, no more than corePoolSize
1.173 + * threads will ever be created. (And the value of the maximumPoolSize
1.174 + * therefore doesn't have any effect.) This may be appropriate when
1.175 + * each task is completely independent of others, so tasks cannot
1.176 + * affect each others execution; for example, in a web page server.
1.177 + * While this style of queuing can be useful in smoothing out
1.178 + * transient bursts of requests, it admits the possibility of
1.179 + * unbounded work queue growth when commands continue to arrive on
1.180 + * average faster than they can be processed. </li>
1.181 + *
1.182 + * <li><em>Bounded queues.</em> A bounded queue (for example, an
1.183 + * {@link ArrayBlockingQueue}) helps prevent resource exhaustion when
1.184 + * used with finite maximumPoolSizes, but can be more difficult to
1.185 + * tune and control. Queue sizes and maximum pool sizes may be traded
1.186 + * off for each other: Using large queues and small pools minimizes
1.187 + * CPU usage, OS resources, and context-switching overhead, but can
1.188 + * lead to artificially low throughput. If tasks frequently block (for
1.189 + * example if they are I/O bound), a system may be able to schedule
1.190 + * time for more threads than you otherwise allow. Use of small queues
1.191 + * generally requires larger pool sizes, which keeps CPUs busier but
1.192 + * may encounter unacceptable scheduling overhead, which also
1.193 + * decreases throughput. </li>
1.194 + *
1.195 + * </ol>
1.196 + *
1.197 + * </dd>
1.198 + *
1.199 + * <dt>Rejected tasks</dt>
1.200 + *
1.201 + * <dd> New tasks submitted in method {@link #execute} will be
1.202 + * <em>rejected</em> when the Executor has been shut down, and also
1.203 + * when the Executor uses finite bounds for both maximum threads and
1.204 + * work queue capacity, and is saturated. In either case, the {@code
1.205 + * execute} method invokes the {@link
1.206 + * RejectedExecutionHandler#rejectedExecution} method of its {@link
1.207 + * RejectedExecutionHandler}. Four predefined handler policies are
1.208 + * provided:
1.209 + *
1.210 + * <ol>
1.211 + *
1.212 + * <li> In the default {@link ThreadPoolExecutor.AbortPolicy}, the
1.213 + * handler throws a runtime {@link RejectedExecutionException} upon
1.214 + * rejection. </li>
1.215 + *
1.216 + * <li> In {@link ThreadPoolExecutor.CallerRunsPolicy}, the thread
1.217 + * that invokes {@code execute} itself runs the task. This provides a
1.218 + * simple feedback control mechanism that will slow down the rate that
1.219 + * new tasks are submitted. </li>
1.220 + *
1.221 + * <li> In {@link ThreadPoolExecutor.DiscardPolicy}, a task that
1.222 + * cannot be executed is simply dropped. </li>
1.223 + *
1.224 + * <li>In {@link ThreadPoolExecutor.DiscardOldestPolicy}, if the
1.225 + * executor is not shut down, the task at the head of the work queue
1.226 + * is dropped, and then execution is retried (which can fail again,
1.227 + * causing this to be repeated.) </li>
1.228 + *
1.229 + * </ol>
1.230 + *
1.231 + * It is possible to define and use other kinds of {@link
1.232 + * RejectedExecutionHandler} classes. Doing so requires some care
1.233 + * especially when policies are designed to work only under particular
1.234 + * capacity or queuing policies. </dd>
1.235 + *
1.236 + * <dt>Hook methods</dt>
1.237 + *
1.238 + * <dd>This class provides {@code protected} overridable {@link
1.239 + * #beforeExecute} and {@link #afterExecute} methods that are called
1.240 + * before and after execution of each task. These can be used to
1.241 + * manipulate the execution environment; for example, reinitializing
1.242 + * ThreadLocals, gathering statistics, or adding log
1.243 + * entries. Additionally, method {@link #terminated} can be overridden
1.244 + * to perform any special processing that needs to be done once the
1.245 + * Executor has fully terminated.
1.246 + *
1.247 + * <p>If hook or callback methods throw exceptions, internal worker
1.248 + * threads may in turn fail and abruptly terminate.</dd>
1.249 + *
1.250 + * <dt>Queue maintenance</dt>
1.251 + *
1.252 + * <dd> Method {@link #getQueue} allows access to the work queue for
1.253 + * purposes of monitoring and debugging. Use of this method for any
1.254 + * other purpose is strongly discouraged. Two supplied methods,
1.255 + * {@link #remove} and {@link #purge} are available to assist in
1.256 + * storage reclamation when large numbers of queued tasks become
1.257 + * cancelled.</dd>
1.258 + *
1.259 + * <dt>Finalization</dt>
1.260 + *
1.261 + * <dd> A pool that is no longer referenced in a program <em>AND</em>
1.262 + * has no remaining threads will be {@code shutdown} automatically. If
1.263 + * you would like to ensure that unreferenced pools are reclaimed even
1.264 + * if users forget to call {@link #shutdown}, then you must arrange
1.265 + * that unused threads eventually die, by setting appropriate
1.266 + * keep-alive times, using a lower bound of zero core threads and/or
1.267 + * setting {@link #allowCoreThreadTimeOut(boolean)}. </dd>
1.268 + *
1.269 + * </dl>
1.270 + *
1.271 + * <p> <b>Extension example</b>. Most extensions of this class
1.272 + * override one or more of the protected hook methods. For example,
1.273 + * here is a subclass that adds a simple pause/resume feature:
1.274 + *
1.275 + * <pre> {@code
1.276 + * class PausableThreadPoolExecutor extends ThreadPoolExecutor {
1.277 + * private boolean isPaused;
1.278 + * private ReentrantLock pauseLock = new ReentrantLock();
1.279 + * private Condition unpaused = pauseLock.newCondition();
1.280 + *
1.281 + * public PausableThreadPoolExecutor(...) { super(...); }
1.282 + *
1.283 + * protected void beforeExecute(Thread t, Runnable r) {
1.284 + * super.beforeExecute(t, r);
1.285 + * pauseLock.lock();
1.286 + * try {
1.287 + * while (isPaused) unpaused.await();
1.288 + * } catch (InterruptedException ie) {
1.289 + * t.interrupt();
1.290 + * } finally {
1.291 + * pauseLock.unlock();
1.292 + * }
1.293 + * }
1.294 + *
1.295 + * public void pause() {
1.296 + * pauseLock.lock();
1.297 + * try {
1.298 + * isPaused = true;
1.299 + * } finally {
1.300 + * pauseLock.unlock();
1.301 + * }
1.302 + * }
1.303 + *
1.304 + * public void resume() {
1.305 + * pauseLock.lock();
1.306 + * try {
1.307 + * isPaused = false;
1.308 + * unpaused.signalAll();
1.309 + * } finally {
1.310 + * pauseLock.unlock();
1.311 + * }
1.312 + * }
1.313 + * }}</pre>
1.314 + *
1.315 + * @since 1.5
1.316 + * @author Doug Lea
1.317 + */
1.318 +public class ThreadPoolExecutor extends AbstractExecutorService {
1.319 + /**
1.320 + * The main pool control state, ctl, is an atomic integer packing
1.321 + * two conceptual fields
1.322 + * workerCount, indicating the effective number of threads
1.323 + * runState, indicating whether running, shutting down etc
1.324 + *
1.325 + * In order to pack them into one int, we limit workerCount to
1.326 + * (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
1.327 + * billion) otherwise representable. If this is ever an issue in
1.328 + * the future, the variable can be changed to be an AtomicLong,
1.329 + * and the shift/mask constants below adjusted. But until the need
1.330 + * arises, this code is a bit faster and simpler using an int.
1.331 + *
1.332 + * The workerCount is the number of workers that have been
1.333 + * permitted to start and not permitted to stop. The value may be
1.334 + * transiently different from the actual number of live threads,
1.335 + * for example when a ThreadFactory fails to create a thread when
1.336 + * asked, and when exiting threads are still performing
1.337 + * bookkeeping before terminating. The user-visible pool size is
1.338 + * reported as the current size of the workers set.
1.339 + *
1.340 + * The runState provides the main lifecyle control, taking on values:
1.341 + *
1.342 + * RUNNING: Accept new tasks and process queued tasks
1.343 + * SHUTDOWN: Don't accept new tasks, but process queued tasks
1.344 + * STOP: Don't accept new tasks, don't process queued tasks,
1.345 + * and interrupt in-progress tasks
1.346 + * TIDYING: All tasks have terminated, workerCount is zero,
1.347 + * the thread transitioning to state TIDYING
1.348 + * will run the terminated() hook method
1.349 + * TERMINATED: terminated() has completed
1.350 + *
1.351 + * The numerical order among these values matters, to allow
1.352 + * ordered comparisons. The runState monotonically increases over
1.353 + * time, but need not hit each state. The transitions are:
1.354 + *
1.355 + * RUNNING -> SHUTDOWN
1.356 + * On invocation of shutdown(), perhaps implicitly in finalize()
1.357 + * (RUNNING or SHUTDOWN) -> STOP
1.358 + * On invocation of shutdownNow()
1.359 + * SHUTDOWN -> TIDYING
1.360 + * When both queue and pool are empty
1.361 + * STOP -> TIDYING
1.362 + * When pool is empty
1.363 + * TIDYING -> TERMINATED
1.364 + * When the terminated() hook method has completed
1.365 + *
1.366 + * Threads waiting in awaitTermination() will return when the
1.367 + * state reaches TERMINATED.
1.368 + *
1.369 + * Detecting the transition from SHUTDOWN to TIDYING is less
1.370 + * straightforward than you'd like because the queue may become
1.371 + * empty after non-empty and vice versa during SHUTDOWN state, but
1.372 + * we can only terminate if, after seeing that it is empty, we see
1.373 + * that workerCount is 0 (which sometimes entails a recheck -- see
1.374 + * below).
1.375 + */
1.376 + private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
1.377 + private static final int COUNT_BITS = Integer.SIZE - 3;
1.378 + private static final int CAPACITY = (1 << COUNT_BITS) - 1;
1.379 +
1.380 + // runState is stored in the high-order bits
1.381 + private static final int RUNNING = -1 << COUNT_BITS;
1.382 + private static final int SHUTDOWN = 0 << COUNT_BITS;
1.383 + private static final int STOP = 1 << COUNT_BITS;
1.384 + private static final int TIDYING = 2 << COUNT_BITS;
1.385 + private static final int TERMINATED = 3 << COUNT_BITS;
1.386 +
1.387 + // Packing and unpacking ctl
1.388 + private static int runStateOf(int c) { return c & ~CAPACITY; }
1.389 + private static int workerCountOf(int c) { return c & CAPACITY; }
1.390 + private static int ctlOf(int rs, int wc) { return rs | wc; }
1.391 +
1.392 + /*
1.393 + * Bit field accessors that don't require unpacking ctl.
1.394 + * These depend on the bit layout and on workerCount being never negative.
1.395 + */
1.396 +
1.397 + private static boolean runStateLessThan(int c, int s) {
1.398 + return c < s;
1.399 + }
1.400 +
1.401 + private static boolean runStateAtLeast(int c, int s) {
1.402 + return c >= s;
1.403 + }
1.404 +
1.405 + private static boolean isRunning(int c) {
1.406 + return c < SHUTDOWN;
1.407 + }
1.408 +
1.409 + /**
1.410 + * Attempt to CAS-increment the workerCount field of ctl.
1.411 + */
1.412 + private boolean compareAndIncrementWorkerCount(int expect) {
1.413 + return ctl.compareAndSet(expect, expect + 1);
1.414 + }
1.415 +
1.416 + /**
1.417 + * Attempt to CAS-decrement the workerCount field of ctl.
1.418 + */
1.419 + private boolean compareAndDecrementWorkerCount(int expect) {
1.420 + return ctl.compareAndSet(expect, expect - 1);
1.421 + }
1.422 +
1.423 + /**
1.424 + * Decrements the workerCount field of ctl. This is called only on
1.425 + * abrupt termination of a thread (see processWorkerExit). Other
1.426 + * decrements are performed within getTask.
1.427 + */
1.428 + private void decrementWorkerCount() {
1.429 + do {} while (! compareAndDecrementWorkerCount(ctl.get()));
1.430 + }
1.431 +
1.432 + /**
1.433 + * The queue used for holding tasks and handing off to worker
1.434 + * threads. We do not require that workQueue.poll() returning
1.435 + * null necessarily means that workQueue.isEmpty(), so rely
1.436 + * solely on isEmpty to see if the queue is empty (which we must
1.437 + * do for example when deciding whether to transition from
1.438 + * SHUTDOWN to TIDYING). This accommodates special-purpose
1.439 + * queues such as DelayQueues for which poll() is allowed to
1.440 + * return null even if it may later return non-null when delays
1.441 + * expire.
1.442 + */
1.443 + private final BlockingQueue<Runnable> workQueue;
1.444 +
1.445 + /**
1.446 + * Lock held on access to workers set and related bookkeeping.
1.447 + * While we could use a concurrent set of some sort, it turns out
1.448 + * to be generally preferable to use a lock. Among the reasons is
1.449 + * that this serializes interruptIdleWorkers, which avoids
1.450 + * unnecessary interrupt storms, especially during shutdown.
1.451 + * Otherwise exiting threads would concurrently interrupt those
1.452 + * that have not yet interrupted. It also simplifies some of the
1.453 + * associated statistics bookkeeping of largestPoolSize etc. We
1.454 + * also hold mainLock on shutdown and shutdownNow, for the sake of
1.455 + * ensuring workers set is stable while separately checking
1.456 + * permission to interrupt and actually interrupting.
1.457 + */
1.458 + private final ReentrantLock mainLock = new ReentrantLock();
1.459 +
1.460 + /**
1.461 + * Set containing all worker threads in pool. Accessed only when
1.462 + * holding mainLock.
1.463 + */
1.464 + private final HashSet<Worker> workers = new HashSet<Worker>();
1.465 +
1.466 + /**
1.467 + * Wait condition to support awaitTermination
1.468 + */
1.469 + private final Condition termination = mainLock.newCondition();
1.470 +
1.471 + /**
1.472 + * Tracks largest attained pool size. Accessed only under
1.473 + * mainLock.
1.474 + */
1.475 + private int largestPoolSize;
1.476 +
1.477 + /**
1.478 + * Counter for completed tasks. Updated only on termination of
1.479 + * worker threads. Accessed only under mainLock.
1.480 + */
1.481 + private long completedTaskCount;
1.482 +
1.483 + /*
1.484 + * All user control parameters are declared as volatiles so that
1.485 + * ongoing actions are based on freshest values, but without need
1.486 + * for locking, since no internal invariants depend on them
1.487 + * changing synchronously with respect to other actions.
1.488 + */
1.489 +
1.490 + /**
1.491 + * Factory for new threads. All threads are created using this
1.492 + * factory (via method addWorker). All callers must be prepared
1.493 + * for addWorker to fail, which may reflect a system or user's
1.494 + * policy limiting the number of threads. Even though it is not
1.495 + * treated as an error, failure to create threads may result in
1.496 + * new tasks being rejected or existing ones remaining stuck in
1.497 + * the queue. On the other hand, no special precautions exist to
1.498 + * handle OutOfMemoryErrors that might be thrown while trying to
1.499 + * create threads, since there is generally no recourse from
1.500 + * within this class.
1.501 + */
1.502 + private volatile ThreadFactory threadFactory;
1.503 +
1.504 + /**
1.505 + * Handler called when saturated or shutdown in execute.
1.506 + */
1.507 + private volatile RejectedExecutionHandler handler;
1.508 +
1.509 + /**
1.510 + * Timeout in nanoseconds for idle threads waiting for work.
1.511 + * Threads use this timeout when there are more than corePoolSize
1.512 + * present or if allowCoreThreadTimeOut. Otherwise they wait
1.513 + * forever for new work.
1.514 + */
1.515 + private volatile long keepAliveTime;
1.516 +
1.517 + /**
1.518 + * If false (default), core threads stay alive even when idle.
1.519 + * If true, core threads use keepAliveTime to time out waiting
1.520 + * for work.
1.521 + */
1.522 + private volatile boolean allowCoreThreadTimeOut;
1.523 +
1.524 + /**
1.525 + * Core pool size is the minimum number of workers to keep alive
1.526 + * (and not allow to time out etc) unless allowCoreThreadTimeOut
1.527 + * is set, in which case the minimum is zero.
1.528 + */
1.529 + private volatile int corePoolSize;
1.530 +
1.531 + /**
1.532 + * Maximum pool size. Note that the actual maximum is internally
1.533 + * bounded by CAPACITY.
1.534 + */
1.535 + private volatile int maximumPoolSize;
1.536 +
1.537 + /**
1.538 + * The default rejected execution handler
1.539 + */
1.540 + private static final RejectedExecutionHandler defaultHandler =
1.541 + new AbortPolicy();
1.542 +
1.543 + /**
1.544 + * Permission required for callers of shutdown and shutdownNow.
1.545 + * We additionally require (see checkShutdownAccess) that callers
1.546 + * have permission to actually interrupt threads in the worker set
1.547 + * (as governed by Thread.interrupt, which relies on
1.548 + * ThreadGroup.checkAccess, which in turn relies on
1.549 + * SecurityManager.checkAccess). Shutdowns are attempted only if
1.550 + * these checks pass.
1.551 + *
1.552 + * All actual invocations of Thread.interrupt (see
1.553 + * interruptIdleWorkers and interruptWorkers) ignore
1.554 + * SecurityExceptions, meaning that the attempted interrupts
1.555 + * silently fail. In the case of shutdown, they should not fail
1.556 + * unless the SecurityManager has inconsistent policies, sometimes
1.557 + * allowing access to a thread and sometimes not. In such cases,
1.558 + * failure to actually interrupt threads may disable or delay full
1.559 + * termination. Other uses of interruptIdleWorkers are advisory,
1.560 + * and failure to actually interrupt will merely delay response to
1.561 + * configuration changes so is not handled exceptionally.
1.562 + */
1.563 + private static final RuntimePermission shutdownPerm =
1.564 + new RuntimePermission("modifyThread");
1.565 +
1.566 + /**
1.567 + * Class Worker mainly maintains interrupt control state for
1.568 + * threads running tasks, along with other minor bookkeeping.
1.569 + * This class opportunistically extends AbstractQueuedSynchronizer
1.570 + * to simplify acquiring and releasing a lock surrounding each
1.571 + * task execution. This protects against interrupts that are
1.572 + * intended to wake up a worker thread waiting for a task from
1.573 + * instead interrupting a task being run. We implement a simple
1.574 + * non-reentrant mutual exclusion lock rather than use ReentrantLock
1.575 + * because we do not want worker tasks to be able to reacquire the
1.576 + * lock when they invoke pool control methods like setCorePoolSize.
1.577 + */
1.578 + private final class Worker
1.579 + extends AbstractQueuedSynchronizer
1.580 + implements Runnable
1.581 + {
1.582 + /**
1.583 + * This class will never be serialized, but we provide a
1.584 + * serialVersionUID to suppress a javac warning.
1.585 + */
1.586 + private static final long serialVersionUID = 6138294804551838833L;
1.587 +
1.588 + /** Thread this worker is running in. Null if factory fails. */
1.589 + final Thread thread;
1.590 + /** Initial task to run. Possibly null. */
1.591 + Runnable firstTask;
1.592 + /** Per-thread task counter */
1.593 + volatile long completedTasks;
1.594 +
1.595 + /**
1.596 + * Creates with given first task and thread from ThreadFactory.
1.597 + * @param firstTask the first task (null if none)
1.598 + */
1.599 + Worker(Runnable firstTask) {
1.600 + this.firstTask = firstTask;
1.601 + this.thread = getThreadFactory().newThread(this);
1.602 + }
1.603 +
1.604 + /** Delegates main run loop to outer runWorker */
1.605 + public void run() {
1.606 + runWorker(this);
1.607 + }
1.608 +
1.609 + // Lock methods
1.610 + //
1.611 + // The value 0 represents the unlocked state.
1.612 + // The value 1 represents the locked state.
1.613 +
1.614 + protected boolean isHeldExclusively() {
1.615 + return getState() == 1;
1.616 + }
1.617 +
1.618 + protected boolean tryAcquire(int unused) {
1.619 + if (compareAndSetState(0, 1)) {
1.620 + setExclusiveOwnerThread(Thread.currentThread());
1.621 + return true;
1.622 + }
1.623 + return false;
1.624 + }
1.625 +
1.626 + protected boolean tryRelease(int unused) {
1.627 + setExclusiveOwnerThread(null);
1.628 + setState(0);
1.629 + return true;
1.630 + }
1.631 +
1.632 + public void lock() { acquire(1); }
1.633 + public boolean tryLock() { return tryAcquire(1); }
1.634 + public void unlock() { release(1); }
1.635 + public boolean isLocked() { return isHeldExclusively(); }
1.636 + }
1.637 +
1.638 + /*
1.639 + * Methods for setting control state
1.640 + */
1.641 +
1.642 + /**
1.643 + * Transitions runState to given target, or leaves it alone if
1.644 + * already at least the given target.
1.645 + *
1.646 + * @param targetState the desired state, either SHUTDOWN or STOP
1.647 + * (but not TIDYING or TERMINATED -- use tryTerminate for that)
1.648 + */
1.649 + private void advanceRunState(int targetState) {
1.650 + for (;;) {
1.651 + int c = ctl.get();
1.652 + if (runStateAtLeast(c, targetState) ||
1.653 + ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
1.654 + break;
1.655 + }
1.656 + }
1.657 +
1.658 + /**
1.659 + * Transitions to TERMINATED state if either (SHUTDOWN and pool
1.660 + * and queue empty) or (STOP and pool empty). If otherwise
1.661 + * eligible to terminate but workerCount is nonzero, interrupts an
1.662 + * idle worker to ensure that shutdown signals propagate. This
1.663 + * method must be called following any action that might make
1.664 + * termination possible -- reducing worker count or removing tasks
1.665 + * from the queue during shutdown. The method is non-private to
1.666 + * allow access from ScheduledThreadPoolExecutor.
1.667 + */
1.668 + final void tryTerminate() {
1.669 + for (;;) {
1.670 + int c = ctl.get();
1.671 + if (isRunning(c) ||
1.672 + runStateAtLeast(c, TIDYING) ||
1.673 + (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
1.674 + return;
1.675 + if (workerCountOf(c) != 0) { // Eligible to terminate
1.676 + interruptIdleWorkers(ONLY_ONE);
1.677 + return;
1.678 + }
1.679 +
1.680 + final ReentrantLock mainLock = this.mainLock;
1.681 + mainLock.lock();
1.682 + try {
1.683 + if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
1.684 + try {
1.685 + terminated();
1.686 + } finally {
1.687 + ctl.set(ctlOf(TERMINATED, 0));
1.688 + termination.signalAll();
1.689 + }
1.690 + return;
1.691 + }
1.692 + } finally {
1.693 + mainLock.unlock();
1.694 + }
1.695 + // else retry on failed CAS
1.696 + }
1.697 + }
1.698 +
1.699 + /*
1.700 + * Methods for controlling interrupts to worker threads.
1.701 + */
1.702 +
1.703 + /**
1.704 + * If there is a security manager, makes sure caller has
1.705 + * permission to shut down threads in general (see shutdownPerm).
1.706 + * If this passes, additionally makes sure the caller is allowed
1.707 + * to interrupt each worker thread. This might not be true even if
1.708 + * first check passed, if the SecurityManager treats some threads
1.709 + * specially.
1.710 + */
1.711 + private void checkShutdownAccess() {
1.712 + SecurityManager security = System.getSecurityManager();
1.713 + if (security != null) {
1.714 + security.checkPermission(shutdownPerm);
1.715 + final ReentrantLock mainLock = this.mainLock;
1.716 + mainLock.lock();
1.717 + try {
1.718 + for (Worker w : workers)
1.719 + security.checkAccess(w.thread);
1.720 + } finally {
1.721 + mainLock.unlock();
1.722 + }
1.723 + }
1.724 + }
1.725 +
1.726 + /**
1.727 + * Interrupts all threads, even if active. Ignores SecurityExceptions
1.728 + * (in which case some threads may remain uninterrupted).
1.729 + */
1.730 + private void interruptWorkers() {
1.731 + final ReentrantLock mainLock = this.mainLock;
1.732 + mainLock.lock();
1.733 + try {
1.734 + for (Worker w : workers) {
1.735 + try {
1.736 + w.thread.interrupt();
1.737 + } catch (SecurityException ignore) {
1.738 + }
1.739 + }
1.740 + } finally {
1.741 + mainLock.unlock();
1.742 + }
1.743 + }
1.744 +
1.745 + /**
1.746 + * Interrupts threads that might be waiting for tasks (as
1.747 + * indicated by not being locked) so they can check for
1.748 + * termination or configuration changes. Ignores
1.749 + * SecurityExceptions (in which case some threads may remain
1.750 + * uninterrupted).
1.751 + *
1.752 + * @param onlyOne If true, interrupt at most one worker. This is
1.753 + * called only from tryTerminate when termination is otherwise
1.754 + * enabled but there are still other workers. In this case, at
1.755 + * most one waiting worker is interrupted to propagate shutdown
1.756 + * signals in case all threads are currently waiting.
1.757 + * Interrupting any arbitrary thread ensures that newly arriving
1.758 + * workers since shutdown began will also eventually exit.
1.759 + * To guarantee eventual termination, it suffices to always
1.760 + * interrupt only one idle worker, but shutdown() interrupts all
1.761 + * idle workers so that redundant workers exit promptly, not
1.762 + * waiting for a straggler task to finish.
1.763 + */
1.764 + private void interruptIdleWorkers(boolean onlyOne) {
1.765 + final ReentrantLock mainLock = this.mainLock;
1.766 + mainLock.lock();
1.767 + try {
1.768 + for (Worker w : workers) {
1.769 + Thread t = w.thread;
1.770 + if (!t.isInterrupted() && w.tryLock()) {
1.771 + try {
1.772 + t.interrupt();
1.773 + } catch (SecurityException ignore) {
1.774 + } finally {
1.775 + w.unlock();
1.776 + }
1.777 + }
1.778 + if (onlyOne)
1.779 + break;
1.780 + }
1.781 + } finally {
1.782 + mainLock.unlock();
1.783 + }
1.784 + }
1.785 +
1.786 + /**
1.787 + * Common form of interruptIdleWorkers, to avoid having to
1.788 + * remember what the boolean argument means.
1.789 + */
1.790 + private void interruptIdleWorkers() {
1.791 + interruptIdleWorkers(false);
1.792 + }
1.793 +
1.794 + private static final boolean ONLY_ONE = true;
1.795 +
1.796 + /**
1.797 + * Ensures that unless the pool is stopping, the current thread
1.798 + * does not have its interrupt set. This requires a double-check
1.799 + * of state in case the interrupt was cleared concurrently with a
1.800 + * shutdownNow -- if so, the interrupt is re-enabled.
1.801 + */
1.802 + private void clearInterruptsForTaskRun() {
1.803 + if (runStateLessThan(ctl.get(), STOP) &&
1.804 + Thread.interrupted() &&
1.805 + runStateAtLeast(ctl.get(), STOP))
1.806 + Thread.currentThread().interrupt();
1.807 + }
1.808 +
1.809 + /*
1.810 + * Misc utilities, most of which are also exported to
1.811 + * ScheduledThreadPoolExecutor
1.812 + */
1.813 +
1.814 + /**
1.815 + * Invokes the rejected execution handler for the given command.
1.816 + * Package-protected for use by ScheduledThreadPoolExecutor.
1.817 + */
1.818 + final void reject(Runnable command) {
1.819 + handler.rejectedExecution(command, this);
1.820 + }
1.821 +
1.822 + /**
1.823 + * Performs any further cleanup following run state transition on
1.824 + * invocation of shutdown. A no-op here, but used by
1.825 + * ScheduledThreadPoolExecutor to cancel delayed tasks.
1.826 + */
1.827 + void onShutdown() {
1.828 + }
1.829 +
1.830 + /**
1.831 + * State check needed by ScheduledThreadPoolExecutor to
1.832 + * enable running tasks during shutdown.
1.833 + *
1.834 + * @param shutdownOK true if should return true if SHUTDOWN
1.835 + */
1.836 + final boolean isRunningOrShutdown(boolean shutdownOK) {
1.837 + int rs = runStateOf(ctl.get());
1.838 + return rs == RUNNING || (rs == SHUTDOWN && shutdownOK);
1.839 + }
1.840 +
1.841 + /**
1.842 + * Drains the task queue into a new list, normally using
1.843 + * drainTo. But if the queue is a DelayQueue or any other kind of
1.844 + * queue for which poll or drainTo may fail to remove some
1.845 + * elements, it deletes them one by one.
1.846 + */
1.847 + private List<Runnable> drainQueue() {
1.848 + BlockingQueue<Runnable> q = workQueue;
1.849 + List<Runnable> taskList = new ArrayList<Runnable>();
1.850 + q.drainTo(taskList);
1.851 + if (!q.isEmpty()) {
1.852 + for (Runnable r : q.toArray(new Runnable[0])) {
1.853 + if (q.remove(r))
1.854 + taskList.add(r);
1.855 + }
1.856 + }
1.857 + return taskList;
1.858 + }
1.859 +
1.860 + /*
1.861 + * Methods for creating, running and cleaning up after workers
1.862 + */
1.863 +
1.864 + /**
1.865 + * Checks if a new worker can be added with respect to current
1.866 + * pool state and the given bound (either core or maximum). If so,
1.867 + * the worker count is adjusted accordingly, and, if possible, a
1.868 + * new worker is created and started running firstTask as its
1.869 + * first task. This method returns false if the pool is stopped or
1.870 + * eligible to shut down. It also returns false if the thread
1.871 + * factory fails to create a thread when asked, which requires a
1.872 + * backout of workerCount, and a recheck for termination, in case
1.873 + * the existence of this worker was holding up termination.
1.874 + *
1.875 + * @param firstTask the task the new thread should run first (or
1.876 + * null if none). Workers are created with an initial first task
1.877 + * (in method execute()) to bypass queuing when there are fewer
1.878 + * than corePoolSize threads (in which case we always start one),
1.879 + * or when the queue is full (in which case we must bypass queue).
1.880 + * Initially idle threads are usually created via
1.881 + * prestartCoreThread or to replace other dying workers.
1.882 + *
1.883 + * @param core if true use corePoolSize as bound, else
1.884 + * maximumPoolSize. (A boolean indicator is used here rather than a
1.885 + * value to ensure reads of fresh values after checking other pool
1.886 + * state).
1.887 + * @return true if successful
1.888 + */
1.889 + private boolean addWorker(Runnable firstTask, boolean core) {
1.890 + retry:
1.891 + for (;;) {
1.892 + int c = ctl.get();
1.893 + int rs = runStateOf(c);
1.894 +
1.895 + // Check if queue empty only if necessary.
1.896 + if (rs >= SHUTDOWN &&
1.897 + ! (rs == SHUTDOWN &&
1.898 + firstTask == null &&
1.899 + ! workQueue.isEmpty()))
1.900 + return false;
1.901 +
1.902 + for (;;) {
1.903 + int wc = workerCountOf(c);
1.904 + if (wc >= CAPACITY ||
1.905 + wc >= (core ? corePoolSize : maximumPoolSize))
1.906 + return false;
1.907 + if (compareAndIncrementWorkerCount(c))
1.908 + break retry;
1.909 + c = ctl.get(); // Re-read ctl
1.910 + if (runStateOf(c) != rs)
1.911 + continue retry;
1.912 + // else CAS failed due to workerCount change; retry inner loop
1.913 + }
1.914 + }
1.915 +
1.916 + Worker w = new Worker(firstTask);
1.917 + Thread t = w.thread;
1.918 +
1.919 + final ReentrantLock mainLock = this.mainLock;
1.920 + mainLock.lock();
1.921 + try {
1.922 + // Recheck while holding lock.
1.923 + // Back out on ThreadFactory failure or if
1.924 + // shut down before lock acquired.
1.925 + int c = ctl.get();
1.926 + int rs = runStateOf(c);
1.927 +
1.928 + if (t == null ||
1.929 + (rs >= SHUTDOWN &&
1.930 + ! (rs == SHUTDOWN &&
1.931 + firstTask == null))) {
1.932 + decrementWorkerCount();
1.933 + tryTerminate();
1.934 + return false;
1.935 + }
1.936 +
1.937 + workers.add(w);
1.938 +
1.939 + int s = workers.size();
1.940 + if (s > largestPoolSize)
1.941 + largestPoolSize = s;
1.942 + } finally {
1.943 + mainLock.unlock();
1.944 + }
1.945 +
1.946 + t.start();
1.947 + // It is possible (but unlikely) for a thread to have been
1.948 + // added to workers, but not yet started, during transition to
1.949 + // STOP, which could result in a rare missed interrupt,
1.950 + // because Thread.interrupt is not guaranteed to have any effect
1.951 + // on a non-yet-started Thread (see Thread#interrupt).
1.952 + if (runStateOf(ctl.get()) == STOP && ! t.isInterrupted())
1.953 + t.interrupt();
1.954 +
1.955 + return true;
1.956 + }
1.957 +
1.958 + /**
1.959 + * Performs cleanup and bookkeeping for a dying worker. Called
1.960 + * only from worker threads. Unless completedAbruptly is set,
1.961 + * assumes that workerCount has already been adjusted to account
1.962 + * for exit. This method removes thread from worker set, and
1.963 + * possibly terminates the pool or replaces the worker if either
1.964 + * it exited due to user task exception or if fewer than
1.965 + * corePoolSize workers are running or queue is non-empty but
1.966 + * there are no workers.
1.967 + *
1.968 + * @param w the worker
1.969 + * @param completedAbruptly if the worker died due to user exception
1.970 + */
1.971 + private void processWorkerExit(Worker w, boolean completedAbruptly) {
1.972 + if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
1.973 + decrementWorkerCount();
1.974 +
1.975 + final ReentrantLock mainLock = this.mainLock;
1.976 + mainLock.lock();
1.977 + try {
1.978 + completedTaskCount += w.completedTasks;
1.979 + workers.remove(w);
1.980 + } finally {
1.981 + mainLock.unlock();
1.982 + }
1.983 +
1.984 + tryTerminate();
1.985 +
1.986 + int c = ctl.get();
1.987 + if (runStateLessThan(c, STOP)) {
1.988 + if (!completedAbruptly) {
1.989 + int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
1.990 + if (min == 0 && ! workQueue.isEmpty())
1.991 + min = 1;
1.992 + if (workerCountOf(c) >= min)
1.993 + return; // replacement not needed
1.994 + }
1.995 + addWorker(null, false);
1.996 + }
1.997 + }
1.998 +
1.999 + /**
1.1000 + * Performs blocking or timed wait for a task, depending on
1.1001 + * current configuration settings, or returns null if this worker
1.1002 + * must exit because of any of:
1.1003 + * 1. There are more than maximumPoolSize workers (due to
1.1004 + * a call to setMaximumPoolSize).
1.1005 + * 2. The pool is stopped.
1.1006 + * 3. The pool is shutdown and the queue is empty.
1.1007 + * 4. This worker timed out waiting for a task, and timed-out
1.1008 + * workers are subject to termination (that is,
1.1009 + * {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
1.1010 + * both before and after the timed wait.
1.1011 + *
1.1012 + * @return task, or null if the worker must exit, in which case
1.1013 + * workerCount is decremented
1.1014 + */
1.1015 + private Runnable getTask() {
1.1016 + boolean timedOut = false; // Did the last poll() time out?
1.1017 +
1.1018 + retry:
1.1019 + for (;;) {
1.1020 + int c = ctl.get();
1.1021 + int rs = runStateOf(c);
1.1022 +
1.1023 + // Check if queue empty only if necessary.
1.1024 + if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
1.1025 + decrementWorkerCount();
1.1026 + return null;
1.1027 + }
1.1028 +
1.1029 + boolean timed; // Are workers subject to culling?
1.1030 +
1.1031 + for (;;) {
1.1032 + int wc = workerCountOf(c);
1.1033 + timed = allowCoreThreadTimeOut || wc > corePoolSize;
1.1034 +
1.1035 + if (wc <= maximumPoolSize && ! (timedOut && timed))
1.1036 + break;
1.1037 + if (compareAndDecrementWorkerCount(c))
1.1038 + return null;
1.1039 + c = ctl.get(); // Re-read ctl
1.1040 + if (runStateOf(c) != rs)
1.1041 + continue retry;
1.1042 + // else CAS failed due to workerCount change; retry inner loop
1.1043 + }
1.1044 +
1.1045 + try {
1.1046 + Runnable r = timed ?
1.1047 + workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
1.1048 + workQueue.take();
1.1049 + if (r != null)
1.1050 + return r;
1.1051 + timedOut = true;
1.1052 + } catch (InterruptedException retry) {
1.1053 + timedOut = false;
1.1054 + }
1.1055 + }
1.1056 + }
1.1057 +
1.1058 + /**
1.1059 + * Main worker run loop. Repeatedly gets tasks from queue and
1.1060 + * executes them, while coping with a number of issues:
1.1061 + *
1.1062 + * 1. We may start out with an initial task, in which case we
1.1063 + * don't need to get the first one. Otherwise, as long as pool is
1.1064 + * running, we get tasks from getTask. If it returns null then the
1.1065 + * worker exits due to changed pool state or configuration
1.1066 + * parameters. Other exits result from exception throws in
1.1067 + * external code, in which case completedAbruptly holds, which
1.1068 + * usually leads processWorkerExit to replace this thread.
1.1069 + *
1.1070 + * 2. Before running any task, the lock is acquired to prevent
1.1071 + * other pool interrupts while the task is executing, and
1.1072 + * clearInterruptsForTaskRun called to ensure that unless pool is
1.1073 + * stopping, this thread does not have its interrupt set.
1.1074 + *
1.1075 + * 3. Each task run is preceded by a call to beforeExecute, which
1.1076 + * might throw an exception, in which case we cause thread to die
1.1077 + * (breaking loop with completedAbruptly true) without processing
1.1078 + * the task.
1.1079 + *
1.1080 + * 4. Assuming beforeExecute completes normally, we run the task,
1.1081 + * gathering any of its thrown exceptions to send to
1.1082 + * afterExecute. We separately handle RuntimeException, Error
1.1083 + * (both of which the specs guarantee that we trap) and arbitrary
1.1084 + * Throwables. Because we cannot rethrow Throwables within
1.1085 + * Runnable.run, we wrap them within Errors on the way out (to the
1.1086 + * thread's UncaughtExceptionHandler). Any thrown exception also
1.1087 + * conservatively causes thread to die.
1.1088 + *
1.1089 + * 5. After task.run completes, we call afterExecute, which may
1.1090 + * also throw an exception, which will also cause thread to
1.1091 + * die. According to JLS Sec 14.20, this exception is the one that
1.1092 + * will be in effect even if task.run throws.
1.1093 + *
1.1094 + * The net effect of the exception mechanics is that afterExecute
1.1095 + * and the thread's UncaughtExceptionHandler have as accurate
1.1096 + * information as we can provide about any problems encountered by
1.1097 + * user code.
1.1098 + *
1.1099 + * @param w the worker
1.1100 + */
1.1101 + final void runWorker(Worker w) {
1.1102 + Runnable task = w.firstTask;
1.1103 + w.firstTask = null;
1.1104 + boolean completedAbruptly = true;
1.1105 + try {
1.1106 + while (task != null || (task = getTask()) != null) {
1.1107 + w.lock();
1.1108 + clearInterruptsForTaskRun();
1.1109 + try {
1.1110 + beforeExecute(w.thread, task);
1.1111 + Throwable thrown = null;
1.1112 + try {
1.1113 + task.run();
1.1114 + } catch (RuntimeException x) {
1.1115 + thrown = x; throw x;
1.1116 + } catch (Error x) {
1.1117 + thrown = x; throw x;
1.1118 + } catch (Throwable x) {
1.1119 + thrown = x; throw new Error(x);
1.1120 + } finally {
1.1121 + afterExecute(task, thrown);
1.1122 + }
1.1123 + } finally {
1.1124 + task = null;
1.1125 + w.completedTasks++;
1.1126 + w.unlock();
1.1127 + }
1.1128 + }
1.1129 + completedAbruptly = false;
1.1130 + } finally {
1.1131 + processWorkerExit(w, completedAbruptly);
1.1132 + }
1.1133 + }
1.1134 +
1.1135 + // Public constructors and methods
1.1136 +
1.1137 + /**
1.1138 + * Creates a new {@code ThreadPoolExecutor} with the given initial
1.1139 + * parameters and default thread factory and rejected execution handler.
1.1140 + * It may be more convenient to use one of the {@link Executors} factory
1.1141 + * methods instead of this general purpose constructor.
1.1142 + *
1.1143 + * @param corePoolSize the number of threads to keep in the pool, even
1.1144 + * if they are idle, unless {@code allowCoreThreadTimeOut} is set
1.1145 + * @param maximumPoolSize the maximum number of threads to allow in the
1.1146 + * pool
1.1147 + * @param keepAliveTime when the number of threads is greater than
1.1148 + * the core, this is the maximum time that excess idle threads
1.1149 + * will wait for new tasks before terminating.
1.1150 + * @param unit the time unit for the {@code keepAliveTime} argument
1.1151 + * @param workQueue the queue to use for holding tasks before they are
1.1152 + * executed. This queue will hold only the {@code Runnable}
1.1153 + * tasks submitted by the {@code execute} method.
1.1154 + * @throws IllegalArgumentException if one of the following holds:<br>
1.1155 + * {@code corePoolSize < 0}<br>
1.1156 + * {@code keepAliveTime < 0}<br>
1.1157 + * {@code maximumPoolSize <= 0}<br>
1.1158 + * {@code maximumPoolSize < corePoolSize}
1.1159 + * @throws NullPointerException if {@code workQueue} is null
1.1160 + */
1.1161 + public ThreadPoolExecutor(int corePoolSize,
1.1162 + int maximumPoolSize,
1.1163 + long keepAliveTime,
1.1164 + TimeUnit unit,
1.1165 + BlockingQueue<Runnable> workQueue) {
1.1166 + this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
1.1167 + Executors.defaultThreadFactory(), defaultHandler);
1.1168 + }
1.1169 +
1.1170 + /**
1.1171 + * Creates a new {@code ThreadPoolExecutor} with the given initial
1.1172 + * parameters and default rejected execution handler.
1.1173 + *
1.1174 + * @param corePoolSize the number of threads to keep in the pool, even
1.1175 + * if they are idle, unless {@code allowCoreThreadTimeOut} is set
1.1176 + * @param maximumPoolSize the maximum number of threads to allow in the
1.1177 + * pool
1.1178 + * @param keepAliveTime when the number of threads is greater than
1.1179 + * the core, this is the maximum time that excess idle threads
1.1180 + * will wait for new tasks before terminating.
1.1181 + * @param unit the time unit for the {@code keepAliveTime} argument
1.1182 + * @param workQueue the queue to use for holding tasks before they are
1.1183 + * executed. This queue will hold only the {@code Runnable}
1.1184 + * tasks submitted by the {@code execute} method.
1.1185 + * @param threadFactory the factory to use when the executor
1.1186 + * creates a new thread
1.1187 + * @throws IllegalArgumentException if one of the following holds:<br>
1.1188 + * {@code corePoolSize < 0}<br>
1.1189 + * {@code keepAliveTime < 0}<br>
1.1190 + * {@code maximumPoolSize <= 0}<br>
1.1191 + * {@code maximumPoolSize < corePoolSize}
1.1192 + * @throws NullPointerException if {@code workQueue}
1.1193 + * or {@code threadFactory} is null
1.1194 + */
1.1195 + public ThreadPoolExecutor(int corePoolSize,
1.1196 + int maximumPoolSize,
1.1197 + long keepAliveTime,
1.1198 + TimeUnit unit,
1.1199 + BlockingQueue<Runnable> workQueue,
1.1200 + ThreadFactory threadFactory) {
1.1201 + this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
1.1202 + threadFactory, defaultHandler);
1.1203 + }
1.1204 +
1.1205 + /**
1.1206 + * Creates a new {@code ThreadPoolExecutor} with the given initial
1.1207 + * parameters and default thread factory.
1.1208 + *
1.1209 + * @param corePoolSize the number of threads to keep in the pool, even
1.1210 + * if they are idle, unless {@code allowCoreThreadTimeOut} is set
1.1211 + * @param maximumPoolSize the maximum number of threads to allow in the
1.1212 + * pool
1.1213 + * @param keepAliveTime when the number of threads is greater than
1.1214 + * the core, this is the maximum time that excess idle threads
1.1215 + * will wait for new tasks before terminating.
1.1216 + * @param unit the time unit for the {@code keepAliveTime} argument
1.1217 + * @param workQueue the queue to use for holding tasks before they are
1.1218 + * executed. This queue will hold only the {@code Runnable}
1.1219 + * tasks submitted by the {@code execute} method.
1.1220 + * @param handler the handler to use when execution is blocked
1.1221 + * because the thread bounds and queue capacities are reached
1.1222 + * @throws IllegalArgumentException if one of the following holds:<br>
1.1223 + * {@code corePoolSize < 0}<br>
1.1224 + * {@code keepAliveTime < 0}<br>
1.1225 + * {@code maximumPoolSize <= 0}<br>
1.1226 + * {@code maximumPoolSize < corePoolSize}
1.1227 + * @throws NullPointerException if {@code workQueue}
1.1228 + * or {@code handler} is null
1.1229 + */
1.1230 + public ThreadPoolExecutor(int corePoolSize,
1.1231 + int maximumPoolSize,
1.1232 + long keepAliveTime,
1.1233 + TimeUnit unit,
1.1234 + BlockingQueue<Runnable> workQueue,
1.1235 + RejectedExecutionHandler handler) {
1.1236 + this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
1.1237 + Executors.defaultThreadFactory(), handler);
1.1238 + }
1.1239 +
1.1240 + /**
1.1241 + * Creates a new {@code ThreadPoolExecutor} with the given initial
1.1242 + * parameters.
1.1243 + *
1.1244 + * @param corePoolSize the number of threads to keep in the pool, even
1.1245 + * if they are idle, unless {@code allowCoreThreadTimeOut} is set
1.1246 + * @param maximumPoolSize the maximum number of threads to allow in the
1.1247 + * pool
1.1248 + * @param keepAliveTime when the number of threads is greater than
1.1249 + * the core, this is the maximum time that excess idle threads
1.1250 + * will wait for new tasks before terminating.
1.1251 + * @param unit the time unit for the {@code keepAliveTime} argument
1.1252 + * @param workQueue the queue to use for holding tasks before they are
1.1253 + * executed. This queue will hold only the {@code Runnable}
1.1254 + * tasks submitted by the {@code execute} method.
1.1255 + * @param threadFactory the factory to use when the executor
1.1256 + * creates a new thread
1.1257 + * @param handler the handler to use when execution is blocked
1.1258 + * because the thread bounds and queue capacities are reached
1.1259 + * @throws IllegalArgumentException if one of the following holds:<br>
1.1260 + * {@code corePoolSize < 0}<br>
1.1261 + * {@code keepAliveTime < 0}<br>
1.1262 + * {@code maximumPoolSize <= 0}<br>
1.1263 + * {@code maximumPoolSize < corePoolSize}
1.1264 + * @throws NullPointerException if {@code workQueue}
1.1265 + * or {@code threadFactory} or {@code handler} is null
1.1266 + */
1.1267 + public ThreadPoolExecutor(int corePoolSize,
1.1268 + int maximumPoolSize,
1.1269 + long keepAliveTime,
1.1270 + TimeUnit unit,
1.1271 + BlockingQueue<Runnable> workQueue,
1.1272 + ThreadFactory threadFactory,
1.1273 + RejectedExecutionHandler handler) {
1.1274 + if (corePoolSize < 0 ||
1.1275 + maximumPoolSize <= 0 ||
1.1276 + maximumPoolSize < corePoolSize ||
1.1277 + keepAliveTime < 0)
1.1278 + throw new IllegalArgumentException();
1.1279 + if (workQueue == null || threadFactory == null || handler == null)
1.1280 + throw new NullPointerException();
1.1281 + this.corePoolSize = corePoolSize;
1.1282 + this.maximumPoolSize = maximumPoolSize;
1.1283 + this.workQueue = workQueue;
1.1284 + this.keepAliveTime = unit.toNanos(keepAliveTime);
1.1285 + this.threadFactory = threadFactory;
1.1286 + this.handler = handler;
1.1287 + }
1.1288 +
1.1289 + /**
1.1290 + * Executes the given task sometime in the future. The task
1.1291 + * may execute in a new thread or in an existing pooled thread.
1.1292 + *
1.1293 + * If the task cannot be submitted for execution, either because this
1.1294 + * executor has been shutdown or because its capacity has been reached,
1.1295 + * the task is handled by the current {@code RejectedExecutionHandler}.
1.1296 + *
1.1297 + * @param command the task to execute
1.1298 + * @throws RejectedExecutionException at discretion of
1.1299 + * {@code RejectedExecutionHandler}, if the task
1.1300 + * cannot be accepted for execution
1.1301 + * @throws NullPointerException if {@code command} is null
1.1302 + */
1.1303 + public void execute(Runnable command) {
1.1304 + if (command == null)
1.1305 + throw new NullPointerException();
1.1306 + /*
1.1307 + * Proceed in 3 steps:
1.1308 + *
1.1309 + * 1. If fewer than corePoolSize threads are running, try to
1.1310 + * start a new thread with the given command as its first
1.1311 + * task. The call to addWorker atomically checks runState and
1.1312 + * workerCount, and so prevents false alarms that would add
1.1313 + * threads when it shouldn't, by returning false.
1.1314 + *
1.1315 + * 2. If a task can be successfully queued, then we still need
1.1316 + * to double-check whether we should have added a thread
1.1317 + * (because existing ones died since last checking) or that
1.1318 + * the pool shut down since entry into this method. So we
1.1319 + * recheck state and if necessary roll back the enqueuing if
1.1320 + * stopped, or start a new thread if there are none.
1.1321 + *
1.1322 + * 3. If we cannot queue task, then we try to add a new
1.1323 + * thread. If it fails, we know we are shut down or saturated
1.1324 + * and so reject the task.
1.1325 + */
1.1326 + int c = ctl.get();
1.1327 + if (workerCountOf(c) < corePoolSize) {
1.1328 + if (addWorker(command, true))
1.1329 + return;
1.1330 + c = ctl.get();
1.1331 + }
1.1332 + if (isRunning(c) && workQueue.offer(command)) {
1.1333 + int recheck = ctl.get();
1.1334 + if (! isRunning(recheck) && remove(command))
1.1335 + reject(command);
1.1336 + else if (workerCountOf(recheck) == 0)
1.1337 + addWorker(null, false);
1.1338 + }
1.1339 + else if (!addWorker(command, false))
1.1340 + reject(command);
1.1341 + }
1.1342 +
1.1343 + /**
1.1344 + * Initiates an orderly shutdown in which previously submitted
1.1345 + * tasks are executed, but no new tasks will be accepted.
1.1346 + * Invocation has no additional effect if already shut down.
1.1347 + *
1.1348 + * <p>This method does not wait for previously submitted tasks to
1.1349 + * complete execution. Use {@link #awaitTermination awaitTermination}
1.1350 + * to do that.
1.1351 + *
1.1352 + * @throws SecurityException {@inheritDoc}
1.1353 + */
1.1354 + public void shutdown() {
1.1355 + final ReentrantLock mainLock = this.mainLock;
1.1356 + mainLock.lock();
1.1357 + try {
1.1358 + checkShutdownAccess();
1.1359 + advanceRunState(SHUTDOWN);
1.1360 + interruptIdleWorkers();
1.1361 + onShutdown(); // hook for ScheduledThreadPoolExecutor
1.1362 + } finally {
1.1363 + mainLock.unlock();
1.1364 + }
1.1365 + tryTerminate();
1.1366 + }
1.1367 +
1.1368 + /**
1.1369 + * Attempts to stop all actively executing tasks, halts the
1.1370 + * processing of waiting tasks, and returns a list of the tasks
1.1371 + * that were awaiting execution. These tasks are drained (removed)
1.1372 + * from the task queue upon return from this method.
1.1373 + *
1.1374 + * <p>This method does not wait for actively executing tasks to
1.1375 + * terminate. Use {@link #awaitTermination awaitTermination} to
1.1376 + * do that.
1.1377 + *
1.1378 + * <p>There are no guarantees beyond best-effort attempts to stop
1.1379 + * processing actively executing tasks. This implementation
1.1380 + * cancels tasks via {@link Thread#interrupt}, so any task that
1.1381 + * fails to respond to interrupts may never terminate.
1.1382 + *
1.1383 + * @throws SecurityException {@inheritDoc}
1.1384 + */
1.1385 + public List<Runnable> shutdownNow() {
1.1386 + List<Runnable> tasks;
1.1387 + final ReentrantLock mainLock = this.mainLock;
1.1388 + mainLock.lock();
1.1389 + try {
1.1390 + checkShutdownAccess();
1.1391 + advanceRunState(STOP);
1.1392 + interruptWorkers();
1.1393 + tasks = drainQueue();
1.1394 + } finally {
1.1395 + mainLock.unlock();
1.1396 + }
1.1397 + tryTerminate();
1.1398 + return tasks;
1.1399 + }
1.1400 +
1.1401 + public boolean isShutdown() {
1.1402 + return ! isRunning(ctl.get());
1.1403 + }
1.1404 +
1.1405 + /**
1.1406 + * Returns true if this executor is in the process of terminating
1.1407 + * after {@link #shutdown} or {@link #shutdownNow} but has not
1.1408 + * completely terminated. This method may be useful for
1.1409 + * debugging. A return of {@code true} reported a sufficient
1.1410 + * period after shutdown may indicate that submitted tasks have
1.1411 + * ignored or suppressed interruption, causing this executor not
1.1412 + * to properly terminate.
1.1413 + *
1.1414 + * @return true if terminating but not yet terminated
1.1415 + */
1.1416 + public boolean isTerminating() {
1.1417 + int c = ctl.get();
1.1418 + return ! isRunning(c) && runStateLessThan(c, TERMINATED);
1.1419 + }
1.1420 +
1.1421 + public boolean isTerminated() {
1.1422 + return runStateAtLeast(ctl.get(), TERMINATED);
1.1423 + }
1.1424 +
1.1425 + public boolean awaitTermination(long timeout, TimeUnit unit)
1.1426 + throws InterruptedException {
1.1427 + long nanos = unit.toNanos(timeout);
1.1428 + final ReentrantLock mainLock = this.mainLock;
1.1429 + mainLock.lock();
1.1430 + try {
1.1431 + for (;;) {
1.1432 + if (runStateAtLeast(ctl.get(), TERMINATED))
1.1433 + return true;
1.1434 + if (nanos <= 0)
1.1435 + return false;
1.1436 + nanos = termination.awaitNanos(nanos);
1.1437 + }
1.1438 + } finally {
1.1439 + mainLock.unlock();
1.1440 + }
1.1441 + }
1.1442 +
1.1443 + /**
1.1444 + * Invokes {@code shutdown} when this executor is no longer
1.1445 + * referenced and it has no threads.
1.1446 + */
1.1447 + protected void finalize() {
1.1448 + shutdown();
1.1449 + }
1.1450 +
1.1451 + /**
1.1452 + * Sets the thread factory used to create new threads.
1.1453 + *
1.1454 + * @param threadFactory the new thread factory
1.1455 + * @throws NullPointerException if threadFactory is null
1.1456 + * @see #getThreadFactory
1.1457 + */
1.1458 + public void setThreadFactory(ThreadFactory threadFactory) {
1.1459 + if (threadFactory == null)
1.1460 + throw new NullPointerException();
1.1461 + this.threadFactory = threadFactory;
1.1462 + }
1.1463 +
1.1464 + /**
1.1465 + * Returns the thread factory used to create new threads.
1.1466 + *
1.1467 + * @return the current thread factory
1.1468 + * @see #setThreadFactory
1.1469 + */
1.1470 + public ThreadFactory getThreadFactory() {
1.1471 + return threadFactory;
1.1472 + }
1.1473 +
1.1474 + /**
1.1475 + * Sets a new handler for unexecutable tasks.
1.1476 + *
1.1477 + * @param handler the new handler
1.1478 + * @throws NullPointerException if handler is null
1.1479 + * @see #getRejectedExecutionHandler
1.1480 + */
1.1481 + public void setRejectedExecutionHandler(RejectedExecutionHandler handler) {
1.1482 + if (handler == null)
1.1483 + throw new NullPointerException();
1.1484 + this.handler = handler;
1.1485 + }
1.1486 +
1.1487 + /**
1.1488 + * Returns the current handler for unexecutable tasks.
1.1489 + *
1.1490 + * @return the current handler
1.1491 + * @see #setRejectedExecutionHandler
1.1492 + */
1.1493 + public RejectedExecutionHandler getRejectedExecutionHandler() {
1.1494 + return handler;
1.1495 + }
1.1496 +
1.1497 + /**
1.1498 + * Sets the core number of threads. This overrides any value set
1.1499 + * in the constructor. If the new value is smaller than the
1.1500 + * current value, excess existing threads will be terminated when
1.1501 + * they next become idle. If larger, new threads will, if needed,
1.1502 + * be started to execute any queued tasks.
1.1503 + *
1.1504 + * @param corePoolSize the new core size
1.1505 + * @throws IllegalArgumentException if {@code corePoolSize < 0}
1.1506 + * @see #getCorePoolSize
1.1507 + */
1.1508 + public void setCorePoolSize(int corePoolSize) {
1.1509 + if (corePoolSize < 0)
1.1510 + throw new IllegalArgumentException();
1.1511 + int delta = corePoolSize - this.corePoolSize;
1.1512 + this.corePoolSize = corePoolSize;
1.1513 + if (workerCountOf(ctl.get()) > corePoolSize)
1.1514 + interruptIdleWorkers();
1.1515 + else if (delta > 0) {
1.1516 + // We don't really know how many new threads are "needed".
1.1517 + // As a heuristic, prestart enough new workers (up to new
1.1518 + // core size) to handle the current number of tasks in
1.1519 + // queue, but stop if queue becomes empty while doing so.
1.1520 + int k = Math.min(delta, workQueue.size());
1.1521 + while (k-- > 0 && addWorker(null, true)) {
1.1522 + if (workQueue.isEmpty())
1.1523 + break;
1.1524 + }
1.1525 + }
1.1526 + }
1.1527 +
1.1528 + /**
1.1529 + * Returns the core number of threads.
1.1530 + *
1.1531 + * @return the core number of threads
1.1532 + * @see #setCorePoolSize
1.1533 + */
1.1534 + public int getCorePoolSize() {
1.1535 + return corePoolSize;
1.1536 + }
1.1537 +
1.1538 + /**
1.1539 + * Starts a core thread, causing it to idly wait for work. This
1.1540 + * overrides the default policy of starting core threads only when
1.1541 + * new tasks are executed. This method will return {@code false}
1.1542 + * if all core threads have already been started.
1.1543 + *
1.1544 + * @return {@code true} if a thread was started
1.1545 + */
1.1546 + public boolean prestartCoreThread() {
1.1547 + return workerCountOf(ctl.get()) < corePoolSize &&
1.1548 + addWorker(null, true);
1.1549 + }
1.1550 +
1.1551 + /**
1.1552 + * Starts all core threads, causing them to idly wait for work. This
1.1553 + * overrides the default policy of starting core threads only when
1.1554 + * new tasks are executed.
1.1555 + *
1.1556 + * @return the number of threads started
1.1557 + */
1.1558 + public int prestartAllCoreThreads() {
1.1559 + int n = 0;
1.1560 + while (addWorker(null, true))
1.1561 + ++n;
1.1562 + return n;
1.1563 + }
1.1564 +
1.1565 + /**
1.1566 + * Returns true if this pool allows core threads to time out and
1.1567 + * terminate if no tasks arrive within the keepAlive time, being
1.1568 + * replaced if needed when new tasks arrive. When true, the same
1.1569 + * keep-alive policy applying to non-core threads applies also to
1.1570 + * core threads. When false (the default), core threads are never
1.1571 + * terminated due to lack of incoming tasks.
1.1572 + *
1.1573 + * @return {@code true} if core threads are allowed to time out,
1.1574 + * else {@code false}
1.1575 + *
1.1576 + * @since 1.6
1.1577 + */
1.1578 + public boolean allowsCoreThreadTimeOut() {
1.1579 + return allowCoreThreadTimeOut;
1.1580 + }
1.1581 +
1.1582 + /**
1.1583 + * Sets the policy governing whether core threads may time out and
1.1584 + * terminate if no tasks arrive within the keep-alive time, being
1.1585 + * replaced if needed when new tasks arrive. When false, core
1.1586 + * threads are never terminated due to lack of incoming
1.1587 + * tasks. When true, the same keep-alive policy applying to
1.1588 + * non-core threads applies also to core threads. To avoid
1.1589 + * continual thread replacement, the keep-alive time must be
1.1590 + * greater than zero when setting {@code true}. This method
1.1591 + * should in general be called before the pool is actively used.
1.1592 + *
1.1593 + * @param value {@code true} if should time out, else {@code false}
1.1594 + * @throws IllegalArgumentException if value is {@code true}
1.1595 + * and the current keep-alive time is not greater than zero
1.1596 + *
1.1597 + * @since 1.6
1.1598 + */
1.1599 + public void allowCoreThreadTimeOut(boolean value) {
1.1600 + if (value && keepAliveTime <= 0)
1.1601 + throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
1.1602 + if (value != allowCoreThreadTimeOut) {
1.1603 + allowCoreThreadTimeOut = value;
1.1604 + if (value)
1.1605 + interruptIdleWorkers();
1.1606 + }
1.1607 + }
1.1608 +
1.1609 + /**
1.1610 + * Sets the maximum allowed number of threads. This overrides any
1.1611 + * value set in the constructor. If the new value is smaller than
1.1612 + * the current value, excess existing threads will be
1.1613 + * terminated when they next become idle.
1.1614 + *
1.1615 + * @param maximumPoolSize the new maximum
1.1616 + * @throws IllegalArgumentException if the new maximum is
1.1617 + * less than or equal to zero, or
1.1618 + * less than the {@linkplain #getCorePoolSize core pool size}
1.1619 + * @see #getMaximumPoolSize
1.1620 + */
1.1621 + public void setMaximumPoolSize(int maximumPoolSize) {
1.1622 + if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)
1.1623 + throw new IllegalArgumentException();
1.1624 + this.maximumPoolSize = maximumPoolSize;
1.1625 + if (workerCountOf(ctl.get()) > maximumPoolSize)
1.1626 + interruptIdleWorkers();
1.1627 + }
1.1628 +
1.1629 + /**
1.1630 + * Returns the maximum allowed number of threads.
1.1631 + *
1.1632 + * @return the maximum allowed number of threads
1.1633 + * @see #setMaximumPoolSize
1.1634 + */
1.1635 + public int getMaximumPoolSize() {
1.1636 + return maximumPoolSize;
1.1637 + }
1.1638 +
1.1639 + /**
1.1640 + * Sets the time limit for which threads may remain idle before
1.1641 + * being terminated. If there are more than the core number of
1.1642 + * threads currently in the pool, after waiting this amount of
1.1643 + * time without processing a task, excess threads will be
1.1644 + * terminated. This overrides any value set in the constructor.
1.1645 + *
1.1646 + * @param time the time to wait. A time value of zero will cause
1.1647 + * excess threads to terminate immediately after executing tasks.
1.1648 + * @param unit the time unit of the {@code time} argument
1.1649 + * @throws IllegalArgumentException if {@code time} less than zero or
1.1650 + * if {@code time} is zero and {@code allowsCoreThreadTimeOut}
1.1651 + * @see #getKeepAliveTime
1.1652 + */
1.1653 + public void setKeepAliveTime(long time, TimeUnit unit) {
1.1654 + if (time < 0)
1.1655 + throw new IllegalArgumentException();
1.1656 + if (time == 0 && allowsCoreThreadTimeOut())
1.1657 + throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
1.1658 + long keepAliveTime = unit.toNanos(time);
1.1659 + long delta = keepAliveTime - this.keepAliveTime;
1.1660 + this.keepAliveTime = keepAliveTime;
1.1661 + if (delta < 0)
1.1662 + interruptIdleWorkers();
1.1663 + }
1.1664 +
1.1665 + /**
1.1666 + * Returns the thread keep-alive time, which is the amount of time
1.1667 + * that threads in excess of the core pool size may remain
1.1668 + * idle before being terminated.
1.1669 + *
1.1670 + * @param unit the desired time unit of the result
1.1671 + * @return the time limit
1.1672 + * @see #setKeepAliveTime
1.1673 + */
1.1674 + public long getKeepAliveTime(TimeUnit unit) {
1.1675 + return unit.convert(keepAliveTime, TimeUnit.NANOSECONDS);
1.1676 + }
1.1677 +
1.1678 + /* User-level queue utilities */
1.1679 +
1.1680 + /**
1.1681 + * Returns the task queue used by this executor. Access to the
1.1682 + * task queue is intended primarily for debugging and monitoring.
1.1683 + * This queue may be in active use. Retrieving the task queue
1.1684 + * does not prevent queued tasks from executing.
1.1685 + *
1.1686 + * @return the task queue
1.1687 + */
1.1688 + public BlockingQueue<Runnable> getQueue() {
1.1689 + return workQueue;
1.1690 + }
1.1691 +
1.1692 + /**
1.1693 + * Removes this task from the executor's internal queue if it is
1.1694 + * present, thus causing it not to be run if it has not already
1.1695 + * started.
1.1696 + *
1.1697 + * <p> This method may be useful as one part of a cancellation
1.1698 + * scheme. It may fail to remove tasks that have been converted
1.1699 + * into other forms before being placed on the internal queue. For
1.1700 + * example, a task entered using {@code submit} might be
1.1701 + * converted into a form that maintains {@code Future} status.
1.1702 + * However, in such cases, method {@link #purge} may be used to
1.1703 + * remove those Futures that have been cancelled.
1.1704 + *
1.1705 + * @param task the task to remove
1.1706 + * @return true if the task was removed
1.1707 + */
1.1708 + public boolean remove(Runnable task) {
1.1709 + boolean removed = workQueue.remove(task);
1.1710 + tryTerminate(); // In case SHUTDOWN and now empty
1.1711 + return removed;
1.1712 + }
1.1713 +
1.1714 + /**
1.1715 + * Tries to remove from the work queue all {@link Future}
1.1716 + * tasks that have been cancelled. This method can be useful as a
1.1717 + * storage reclamation operation, that has no other impact on
1.1718 + * functionality. Cancelled tasks are never executed, but may
1.1719 + * accumulate in work queues until worker threads can actively
1.1720 + * remove them. Invoking this method instead tries to remove them now.
1.1721 + * However, this method may fail to remove tasks in
1.1722 + * the presence of interference by other threads.
1.1723 + */
1.1724 + public void purge() {
1.1725 + final BlockingQueue<Runnable> q = workQueue;
1.1726 + try {
1.1727 + Iterator<Runnable> it = q.iterator();
1.1728 + while (it.hasNext()) {
1.1729 + Runnable r = it.next();
1.1730 + if (r instanceof Future<?> && ((Future<?>)r).isCancelled())
1.1731 + it.remove();
1.1732 + }
1.1733 + } catch (ConcurrentModificationException fallThrough) {
1.1734 + // Take slow path if we encounter interference during traversal.
1.1735 + // Make copy for traversal and call remove for cancelled entries.
1.1736 + // The slow path is more likely to be O(N*N).
1.1737 + for (Object r : q.toArray())
1.1738 + if (r instanceof Future<?> && ((Future<?>)r).isCancelled())
1.1739 + q.remove(r);
1.1740 + }
1.1741 +
1.1742 + tryTerminate(); // In case SHUTDOWN and now empty
1.1743 + }
1.1744 +
1.1745 + /* Statistics */
1.1746 +
1.1747 + /**
1.1748 + * Returns the current number of threads in the pool.
1.1749 + *
1.1750 + * @return the number of threads
1.1751 + */
1.1752 + public int getPoolSize() {
1.1753 + final ReentrantLock mainLock = this.mainLock;
1.1754 + mainLock.lock();
1.1755 + try {
1.1756 + // Remove rare and surprising possibility of
1.1757 + // isTerminated() && getPoolSize() > 0
1.1758 + return runStateAtLeast(ctl.get(), TIDYING) ? 0
1.1759 + : workers.size();
1.1760 + } finally {
1.1761 + mainLock.unlock();
1.1762 + }
1.1763 + }
1.1764 +
1.1765 + /**
1.1766 + * Returns the approximate number of threads that are actively
1.1767 + * executing tasks.
1.1768 + *
1.1769 + * @return the number of threads
1.1770 + */
1.1771 + public int getActiveCount() {
1.1772 + final ReentrantLock mainLock = this.mainLock;
1.1773 + mainLock.lock();
1.1774 + try {
1.1775 + int n = 0;
1.1776 + for (Worker w : workers)
1.1777 + if (w.isLocked())
1.1778 + ++n;
1.1779 + return n;
1.1780 + } finally {
1.1781 + mainLock.unlock();
1.1782 + }
1.1783 + }
1.1784 +
1.1785 + /**
1.1786 + * Returns the largest number of threads that have ever
1.1787 + * simultaneously been in the pool.
1.1788 + *
1.1789 + * @return the number of threads
1.1790 + */
1.1791 + public int getLargestPoolSize() {
1.1792 + final ReentrantLock mainLock = this.mainLock;
1.1793 + mainLock.lock();
1.1794 + try {
1.1795 + return largestPoolSize;
1.1796 + } finally {
1.1797 + mainLock.unlock();
1.1798 + }
1.1799 + }
1.1800 +
1.1801 + /**
1.1802 + * Returns the approximate total number of tasks that have ever been
1.1803 + * scheduled for execution. Because the states of tasks and
1.1804 + * threads may change dynamically during computation, the returned
1.1805 + * value is only an approximation.
1.1806 + *
1.1807 + * @return the number of tasks
1.1808 + */
1.1809 + public long getTaskCount() {
1.1810 + final ReentrantLock mainLock = this.mainLock;
1.1811 + mainLock.lock();
1.1812 + try {
1.1813 + long n = completedTaskCount;
1.1814 + for (Worker w : workers) {
1.1815 + n += w.completedTasks;
1.1816 + if (w.isLocked())
1.1817 + ++n;
1.1818 + }
1.1819 + return n + workQueue.size();
1.1820 + } finally {
1.1821 + mainLock.unlock();
1.1822 + }
1.1823 + }
1.1824 +
1.1825 + /**
1.1826 + * Returns the approximate total number of tasks that have
1.1827 + * completed execution. Because the states of tasks and threads
1.1828 + * may change dynamically during computation, the returned value
1.1829 + * is only an approximation, but one that does not ever decrease
1.1830 + * across successive calls.
1.1831 + *
1.1832 + * @return the number of tasks
1.1833 + */
1.1834 + public long getCompletedTaskCount() {
1.1835 + final ReentrantLock mainLock = this.mainLock;
1.1836 + mainLock.lock();
1.1837 + try {
1.1838 + long n = completedTaskCount;
1.1839 + for (Worker w : workers)
1.1840 + n += w.completedTasks;
1.1841 + return n;
1.1842 + } finally {
1.1843 + mainLock.unlock();
1.1844 + }
1.1845 + }
1.1846 +
1.1847 + /**
1.1848 + * Returns a string identifying this pool, as well as its state,
1.1849 + * including indications of run state and estimated worker and
1.1850 + * task counts.
1.1851 + *
1.1852 + * @return a string identifying this pool, as well as its state
1.1853 + */
1.1854 + public String toString() {
1.1855 + long ncompleted;
1.1856 + int nworkers, nactive;
1.1857 + final ReentrantLock mainLock = this.mainLock;
1.1858 + mainLock.lock();
1.1859 + try {
1.1860 + ncompleted = completedTaskCount;
1.1861 + nactive = 0;
1.1862 + nworkers = workers.size();
1.1863 + for (Worker w : workers) {
1.1864 + ncompleted += w.completedTasks;
1.1865 + if (w.isLocked())
1.1866 + ++nactive;
1.1867 + }
1.1868 + } finally {
1.1869 + mainLock.unlock();
1.1870 + }
1.1871 + int c = ctl.get();
1.1872 + String rs = (runStateLessThan(c, SHUTDOWN) ? "Running" :
1.1873 + (runStateAtLeast(c, TERMINATED) ? "Terminated" :
1.1874 + "Shutting down"));
1.1875 + return super.toString() +
1.1876 + "[" + rs +
1.1877 + ", pool size = " + nworkers +
1.1878 + ", active threads = " + nactive +
1.1879 + ", queued tasks = " + workQueue.size() +
1.1880 + ", completed tasks = " + ncompleted +
1.1881 + "]";
1.1882 + }
1.1883 +
1.1884 + /* Extension hooks */
1.1885 +
1.1886 + /**
1.1887 + * Method invoked prior to executing the given Runnable in the
1.1888 + * given thread. This method is invoked by thread {@code t} that
1.1889 + * will execute task {@code r}, and may be used to re-initialize
1.1890 + * ThreadLocals, or to perform logging.
1.1891 + *
1.1892 + * <p>This implementation does nothing, but may be customized in
1.1893 + * subclasses. Note: To properly nest multiple overridings, subclasses
1.1894 + * should generally invoke {@code super.beforeExecute} at the end of
1.1895 + * this method.
1.1896 + *
1.1897 + * @param t the thread that will run task {@code r}
1.1898 + * @param r the task that will be executed
1.1899 + */
1.1900 + protected void beforeExecute(Thread t, Runnable r) { }
1.1901 +
1.1902 + /**
1.1903 + * Method invoked upon completion of execution of the given Runnable.
1.1904 + * This method is invoked by the thread that executed the task. If
1.1905 + * non-null, the Throwable is the uncaught {@code RuntimeException}
1.1906 + * or {@code Error} that caused execution to terminate abruptly.
1.1907 + *
1.1908 + * <p>This implementation does nothing, but may be customized in
1.1909 + * subclasses. Note: To properly nest multiple overridings, subclasses
1.1910 + * should generally invoke {@code super.afterExecute} at the
1.1911 + * beginning of this method.
1.1912 + *
1.1913 + * <p><b>Note:</b> When actions are enclosed in tasks (such as
1.1914 + * {@link FutureTask}) either explicitly or via methods such as
1.1915 + * {@code submit}, these task objects catch and maintain
1.1916 + * computational exceptions, and so they do not cause abrupt
1.1917 + * termination, and the internal exceptions are <em>not</em>
1.1918 + * passed to this method. If you would like to trap both kinds of
1.1919 + * failures in this method, you can further probe for such cases,
1.1920 + * as in this sample subclass that prints either the direct cause
1.1921 + * or the underlying exception if a task has been aborted:
1.1922 + *
1.1923 + * <pre> {@code
1.1924 + * class ExtendedExecutor extends ThreadPoolExecutor {
1.1925 + * // ...
1.1926 + * protected void afterExecute(Runnable r, Throwable t) {
1.1927 + * super.afterExecute(r, t);
1.1928 + * if (t == null && r instanceof Future<?>) {
1.1929 + * try {
1.1930 + * Object result = ((Future<?>) r).get();
1.1931 + * } catch (CancellationException ce) {
1.1932 + * t = ce;
1.1933 + * } catch (ExecutionException ee) {
1.1934 + * t = ee.getCause();
1.1935 + * } catch (InterruptedException ie) {
1.1936 + * Thread.currentThread().interrupt(); // ignore/reset
1.1937 + * }
1.1938 + * }
1.1939 + * if (t != null)
1.1940 + * System.out.println(t);
1.1941 + * }
1.1942 + * }}</pre>
1.1943 + *
1.1944 + * @param r the runnable that has completed
1.1945 + * @param t the exception that caused termination, or null if
1.1946 + * execution completed normally
1.1947 + */
1.1948 + protected void afterExecute(Runnable r, Throwable t) { }
1.1949 +
1.1950 + /**
1.1951 + * Method invoked when the Executor has terminated. Default
1.1952 + * implementation does nothing. Note: To properly nest multiple
1.1953 + * overridings, subclasses should generally invoke
1.1954 + * {@code super.terminated} within this method.
1.1955 + */
1.1956 + protected void terminated() { }
1.1957 +
1.1958 + /* Predefined RejectedExecutionHandlers */
1.1959 +
1.1960 + /**
1.1961 + * A handler for rejected tasks that runs the rejected task
1.1962 + * directly in the calling thread of the {@code execute} method,
1.1963 + * unless the executor has been shut down, in which case the task
1.1964 + * is discarded.
1.1965 + */
1.1966 + public static class CallerRunsPolicy implements RejectedExecutionHandler {
1.1967 + /**
1.1968 + * Creates a {@code CallerRunsPolicy}.
1.1969 + */
1.1970 + public CallerRunsPolicy() { }
1.1971 +
1.1972 + /**
1.1973 + * Executes task r in the caller's thread, unless the executor
1.1974 + * has been shut down, in which case the task is discarded.
1.1975 + *
1.1976 + * @param r the runnable task requested to be executed
1.1977 + * @param e the executor attempting to execute this task
1.1978 + */
1.1979 + public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1.1980 + if (!e.isShutdown()) {
1.1981 + r.run();
1.1982 + }
1.1983 + }
1.1984 + }
1.1985 +
1.1986 + /**
1.1987 + * A handler for rejected tasks that throws a
1.1988 + * {@code RejectedExecutionException}.
1.1989 + */
1.1990 + public static class AbortPolicy implements RejectedExecutionHandler {
1.1991 + /**
1.1992 + * Creates an {@code AbortPolicy}.
1.1993 + */
1.1994 + public AbortPolicy() { }
1.1995 +
1.1996 + /**
1.1997 + * Always throws RejectedExecutionException.
1.1998 + *
1.1999 + * @param r the runnable task requested to be executed
1.2000 + * @param e the executor attempting to execute this task
1.2001 + * @throws RejectedExecutionException always.
1.2002 + */
1.2003 + public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1.2004 + throw new RejectedExecutionException("Task " + r.toString() +
1.2005 + " rejected from " +
1.2006 + e.toString());
1.2007 + }
1.2008 + }
1.2009 +
1.2010 + /**
1.2011 + * A handler for rejected tasks that silently discards the
1.2012 + * rejected task.
1.2013 + */
1.2014 + public static class DiscardPolicy implements RejectedExecutionHandler {
1.2015 + /**
1.2016 + * Creates a {@code DiscardPolicy}.
1.2017 + */
1.2018 + public DiscardPolicy() { }
1.2019 +
1.2020 + /**
1.2021 + * Does nothing, which has the effect of discarding task r.
1.2022 + *
1.2023 + * @param r the runnable task requested to be executed
1.2024 + * @param e the executor attempting to execute this task
1.2025 + */
1.2026 + public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1.2027 + }
1.2028 + }
1.2029 +
1.2030 + /**
1.2031 + * A handler for rejected tasks that discards the oldest unhandled
1.2032 + * request and then retries {@code execute}, unless the executor
1.2033 + * is shut down, in which case the task is discarded.
1.2034 + */
1.2035 + public static class DiscardOldestPolicy implements RejectedExecutionHandler {
1.2036 + /**
1.2037 + * Creates a {@code DiscardOldestPolicy} for the given executor.
1.2038 + */
1.2039 + public DiscardOldestPolicy() { }
1.2040 +
1.2041 + /**
1.2042 + * Obtains and ignores the next task that the executor
1.2043 + * would otherwise execute, if one is immediately available,
1.2044 + * and then retries execution of task r, unless the executor
1.2045 + * is shut down, in which case task r is instead discarded.
1.2046 + *
1.2047 + * @param r the runnable task requested to be executed
1.2048 + * @param e the executor attempting to execute this task
1.2049 + */
1.2050 + public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1.2051 + if (!e.isShutdown()) {
1.2052 + e.getQueue().poll();
1.2053 + e.execute(r);
1.2054 + }
1.2055 + }
1.2056 + }
1.2057 +}