1 /*
2 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
3 *
4 * This code is free software; you can redistribute it and/or modify it
5 * under the terms of the GNU General Public License version 2 only, as
6 * published by the Free Software Foundation. Oracle designates this
7 * particular file as subject to the "Classpath" exception as provided
8 * by Oracle in the LICENSE file that accompanied this code.
9 *
10 * This code is distributed in the hope that it will be useful, but WITHOUT
11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
13 * version 2 for more details (a copy is included in the LICENSE file that
14 * accompanied this code).
15 *
16 * You should have received a copy of the GNU General Public License version
17 * 2 along with this work; if not, write to the Free Software Foundation,
18 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
19 *
20 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
21 * or visit www.oracle.com if you need additional information or have any
22 * questions.
23 */
24
25 /*
26 * This file is available under and governed by the GNU General Public
27 * License version 2 only, as published by the Free Software Foundation.
28 * However, the following notice accompanied the original version of this
29 * file:
30 *
31 * Written by Doug Lea with assistance from members of JCP JSR-166
32 * Expert Group and released to the public domain, as explained at
33 * http://creativecommons.org/publicdomain/zero/1.0/
34 */
35
36 package java.util.concurrent;
37
38 import java.util.ArrayList;
39 import java.util.ConcurrentModificationException;
40 import java.util.HashSet;
41 import java.util.Iterator;
42 import java.util.List;
43 import java.util.concurrent.atomic.AtomicInteger;
44 import java.util.concurrent.locks.AbstractQueuedSynchronizer;
45 import java.util.concurrent.locks.Condition;
46 import java.util.concurrent.locks.ReentrantLock;
47
48 /**
49 * An {@link ExecutorService} that executes each submitted task using
50 * one of possibly several pooled threads, normally configured
51 * using {@link Executors} factory methods.
52 *
53 * <p>Thread pools address two different problems: they usually
54 * provide improved performance when executing large numbers of
55 * asynchronous tasks, due to reduced per-task invocation overhead,
56 * and they provide a means of bounding and managing the resources,
57 * including threads, consumed when executing a collection of tasks.
58 * Each {@code ThreadPoolExecutor} also maintains some basic
59 * statistics, such as the number of completed tasks.
60 *
61 * <p>To be useful across a wide range of contexts, this class
62 * provides many adjustable parameters and extensibility
63 * hooks. However, programmers are urged to use the more convenient
64 * {@link Executors} factory methods {@link
65 * Executors#newCachedThreadPool} (unbounded thread pool, with
66 * automatic thread reclamation), {@link Executors#newFixedThreadPool}
67 * (fixed size thread pool) and {@link
68 * Executors#newSingleThreadExecutor} (single background thread), that
69 * preconfigure settings for the most common usage
70 * scenarios. Otherwise, use the following guide when manually
71 * configuring and tuning this class:
72 *
73 * <dl>
74 *
75 * <dt>Core and maximum pool sizes</dt>
76 *
77 * <dd>A {@code ThreadPoolExecutor} will automatically adjust the
78 * pool size (see {@link #getPoolSize})
79 * according to the bounds set by
80 * corePoolSize (see {@link #getCorePoolSize}) and
81 * maximumPoolSize (see {@link #getMaximumPoolSize}).
82 *
83 * When a new task is submitted in method {@link #execute(Runnable)},
84 * if fewer than corePoolSize threads are running, a new thread is
85 * created to handle the request, even if other worker threads are
86 * idle. Else if fewer than maximumPoolSize threads are running, a
87 * new thread will be created to handle the request only if the queue
88 * is full. By setting corePoolSize and maximumPoolSize the same, you
89 * create a fixed-size thread pool. By setting maximumPoolSize to an
90 * essentially unbounded value such as {@code Integer.MAX_VALUE}, you
91 * allow the pool to accommodate an arbitrary number of concurrent
92 * tasks. Most typically, core and maximum pool sizes are set only
93 * upon construction, but they may also be changed dynamically using
94 * {@link #setCorePoolSize} and {@link #setMaximumPoolSize}. </dd>
95 *
96 * <dt>On-demand construction</dt>
97 *
98 * <dd>By default, even core threads are initially created and
99 * started only when new tasks arrive, but this can be overridden
100 * dynamically using method {@link #prestartCoreThread} or {@link
101 * #prestartAllCoreThreads}. You probably want to prestart threads if
102 * you construct the pool with a non-empty queue. </dd>
103 *
104 * <dt>Creating new threads</dt>
105 *
106 * <dd>New threads are created using a {@link ThreadFactory}. If not
107 * otherwise specified, a {@link Executors#defaultThreadFactory} is
108 * used, that creates threads to all be in the same {@link
109 * ThreadGroup} and with the same {@code NORM_PRIORITY} priority and
110 * non-daemon status. By supplying a different ThreadFactory, you can
111 * alter the thread's name, thread group, priority, daemon status,
112 * etc. If a {@code ThreadFactory} fails to create a thread when asked
113 * by returning null from {@code newThread}, the executor will
114 * continue, but might not be able to execute any tasks. Threads
115 * should possess the "modifyThread" {@code RuntimePermission}. If
116 * worker threads or other threads using the pool do not possess this
117 * permission, service may be degraded: configuration changes may not
118 * take effect in a timely manner, and a shutdown pool may remain in a
119 * state in which termination is possible but not completed.</dd>
120 *
121 * <dt>Keep-alive times</dt>
122 *
123 * <dd>If the pool currently has more than corePoolSize threads,
124 * excess threads will be terminated if they have been idle for more
125 * than the keepAliveTime (see {@link #getKeepAliveTime(TimeUnit)}).
126 * This provides a means of reducing resource consumption when the
127 * pool is not being actively used. If the pool becomes more active
128 * later, new threads will be constructed. This parameter can also be
129 * changed dynamically using method {@link #setKeepAliveTime(long,
130 * TimeUnit)}. Using a value of {@code Long.MAX_VALUE} {@link
131 * TimeUnit#NANOSECONDS} effectively disables idle threads from ever
132 * terminating prior to shut down. By default, the keep-alive policy
133 * applies only when there are more than corePoolSize threads, but
134 * method {@link #allowCoreThreadTimeOut(boolean)} can be used to
135 * apply this time-out policy to core threads as well, so long as the
136 * keepAliveTime value is non-zero. </dd>
137 *
138 * <dt>Queuing</dt>
139 *
140 * <dd>Any {@link BlockingQueue} may be used to transfer and hold
141 * submitted tasks. The use of this queue interacts with pool sizing:
142 *
143 * <ul>
144 *
145 * <li>If fewer than corePoolSize threads are running, the Executor
146 * always prefers adding a new thread
147 * rather than queuing.
148 *
149 * <li>If corePoolSize or more threads are running, the Executor
150 * always prefers queuing a request rather than adding a new
151 * thread.
152 *
153 * <li>If a request cannot be queued, a new thread is created unless
154 * this would exceed maximumPoolSize, in which case, the task will be
155 * rejected.
156 *
157 * </ul>
158 *
159 * There are three general strategies for queuing:
160 * <ol>
161 *
162 * <li><em> Direct handoffs.</em> A good default choice for a work
163 * queue is a {@link SynchronousQueue} that hands off tasks to threads
164 * without otherwise holding them. Here, an attempt to queue a task
165 * will fail if no threads are immediately available to run it, so a
166 * new thread will be constructed. This policy avoids lockups when
167 * handling sets of requests that might have internal dependencies.
168 * Direct handoffs generally require unbounded maximumPoolSizes to
169 * avoid rejection of new submitted tasks. This in turn admits the
170 * possibility of unbounded thread growth when commands continue to
171 * arrive on average faster than they can be processed.
172 *
173 * <li><em> Unbounded queues.</em> Using an unbounded queue (for
174 * example a {@link LinkedBlockingQueue} without a predefined
175 * capacity) will cause new tasks to wait in the queue when all
176 * corePoolSize threads are busy. Thus, no more than corePoolSize
177 * threads will ever be created. (And the value of the maximumPoolSize
178 * therefore doesn't have any effect.) This may be appropriate when
179 * each task is completely independent of others, so tasks cannot
180 * affect each others execution; for example, in a web page server.
181 * While this style of queuing can be useful in smoothing out
182 * transient bursts of requests, it admits the possibility of
183 * unbounded work queue growth when commands continue to arrive on
184 * average faster than they can be processed.
185 *
186 * <li><em>Bounded queues.</em> A bounded queue (for example, an
187 * {@link ArrayBlockingQueue}) helps prevent resource exhaustion when
188 * used with finite maximumPoolSizes, but can be more difficult to
189 * tune and control. Queue sizes and maximum pool sizes may be traded
190 * off for each other: Using large queues and small pools minimizes
191 * CPU usage, OS resources, and context-switching overhead, but can
192 * lead to artificially low throughput. If tasks frequently block (for
193 * example if they are I/O bound), a system may be able to schedule
194 * time for more threads than you otherwise allow. Use of small queues
195 * generally requires larger pool sizes, which keeps CPUs busier but
196 * may encounter unacceptable scheduling overhead, which also
197 * decreases throughput.
198 *
199 * </ol>
200 *
201 * </dd>
202 *
203 * <dt>Rejected tasks</dt>
204 *
205 * <dd>New tasks submitted in method {@link #execute(Runnable)} will be
206 * <em>rejected</em> when the Executor has been shut down, and also when
207 * the Executor uses finite bounds for both maximum threads and work queue
208 * capacity, and is saturated. In either case, the {@code execute} method
209 * invokes the {@link
210 * RejectedExecutionHandler#rejectedExecution(Runnable, ThreadPoolExecutor)}
211 * method of its {@link RejectedExecutionHandler}. Four predefined handler
212 * policies are provided:
213 *
214 * <ol>
215 *
216 * <li>In the default {@link ThreadPoolExecutor.AbortPolicy}, the handler
217 * throws a runtime {@link RejectedExecutionException} upon rejection.
218 *
219 * <li>In {@link ThreadPoolExecutor.CallerRunsPolicy}, the thread
220 * that invokes {@code execute} itself runs the task. This provides a
221 * simple feedback control mechanism that will slow down the rate that
222 * new tasks are submitted.
223 *
224 * <li>In {@link ThreadPoolExecutor.DiscardPolicy}, a task that cannot
225 * be executed is simply dropped. This policy is designed only for
226 * those rare cases in which task completion is never relied upon.
227 *
228 * <li>In {@link ThreadPoolExecutor.DiscardOldestPolicy}, if the
229 * executor is not shut down, the task at the head of the work queue
230 * is dropped, and then execution is retried (which can fail again,
231 * causing this to be repeated.) This policy is rarely acceptable. In
232 * nearly all cases, you should also cancel the task to cause an
233 * exception in any component waiting for its completion, and/or log
234 * the failure, as illustrated in {@link
235 * ThreadPoolExecutor.DiscardOldestPolicy} documentation.
236 *
237 * </ol>
238 *
239 * It is possible to define and use other kinds of {@link
240 * RejectedExecutionHandler} classes. Doing so requires some care
241 * especially when policies are designed to work only under particular
242 * capacity or queuing policies. </dd>
243 *
244 * <dt>Hook methods</dt>
245 *
246 * <dd>This class provides {@code protected} overridable
247 * {@link #beforeExecute(Thread, Runnable)} and
248 * {@link #afterExecute(Runnable, Throwable)} methods that are called
249 * before and after execution of each task. These can be used to
250 * manipulate the execution environment; for example, reinitializing
251 * ThreadLocals, gathering statistics, or adding log entries.
252 * Additionally, method {@link #terminated} can be overridden to perform
253 * any special processing that needs to be done once the Executor has
254 * fully terminated.
255 *
256 * <p>If hook, callback, or BlockingQueue methods throw exceptions,
257 * internal worker threads may in turn fail, abruptly terminate, and
258 * possibly be replaced.</dd>
259 *
260 * <dt>Queue maintenance</dt>
261 *
262 * <dd>Method {@link #getQueue()} allows access to the work queue
263 * for purposes of monitoring and debugging. Use of this method for
264 * any other purpose is strongly discouraged. Two supplied methods,
265 * {@link #remove(Runnable)} and {@link #purge} are available to
266 * assist in storage reclamation when large numbers of queued tasks
267 * become cancelled.</dd>
268 *
269 * <dt>Reclamation</dt>
270 *
271 * <dd>A pool that is no longer referenced in a program <em>AND</em>
272 * has no remaining threads may be reclaimed (garbage collected)
273 * without being explicitly shutdown. You can configure a pool to
274 * allow all unused threads to eventually die by setting appropriate
275 * keep-alive times, using a lower bound of zero core threads and/or
276 * setting {@link #allowCoreThreadTimeOut(boolean)}. </dd>
277 *
278 * </dl>
279 *
280 * <p><b>Extension example.</b> Most extensions of this class
281 * override one or more of the protected hook methods. For example,
282 * here is a subclass that adds a simple pause/resume feature:
283 *
284 * <pre> {@code
285 * class PausableThreadPoolExecutor extends ThreadPoolExecutor {
286 * private boolean isPaused;
287 * private ReentrantLock pauseLock = new ReentrantLock();
288 * private Condition unpaused = pauseLock.newCondition();
289 *
290 * public PausableThreadPoolExecutor(...) { super(...); }
291 *
292 * protected void beforeExecute(Thread t, Runnable r) {
293 * super.beforeExecute(t, r);
294 * pauseLock.lock();
295 * try {
296 * while (isPaused) unpaused.await();
297 * } catch (InterruptedException ie) {
298 * t.interrupt();
299 * } finally {
300 * pauseLock.unlock();
301 * }
302 * }
303 *
304 * public void pause() {
305 * pauseLock.lock();
306 * try {
307 * isPaused = true;
308 * } finally {
309 * pauseLock.unlock();
310 * }
311 * }
312 *
313 * public void resume() {
314 * pauseLock.lock();
315 * try {
316 * isPaused = false;
317 * unpaused.signalAll();
318 * } finally {
319 * pauseLock.unlock();
320 * }
321 * }
322 * }}</pre>
323 *
324 * @since 1.5
325 * @author Doug Lea
326 */
327 public class ThreadPoolExecutor extends AbstractExecutorService {
328 /**
329 * The main pool control state, ctl, is an atomic integer packing
330 * two conceptual fields
331 * workerCount, indicating the effective number of threads
332 * runState, indicating whether running, shutting down etc
333 *
334 * In order to pack them into one int, we limit workerCount to
335 * (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
336 * billion) otherwise representable. If this is ever an issue in
337 * the future, the variable can be changed to be an AtomicLong,
338 * and the shift/mask constants below adjusted. But until the need
339 * arises, this code is a bit faster and simpler using an int.
340 *
341 * The workerCount is the number of workers that have been
342 * permitted to start and not permitted to stop. The value may be
343 * transiently different from the actual number of live threads,
344 * for example when a ThreadFactory fails to create a thread when
345 * asked, and when exiting threads are still performing
346 * bookkeeping before terminating. The user-visible pool size is
347 * reported as the current size of the workers set.
348 *
349 * The runState provides the main lifecycle control, taking on values:
350 *
351 * RUNNING: Accept new tasks and process queued tasks
352 * SHUTDOWN: Don't accept new tasks, but process queued tasks
353 * STOP: Don't accept new tasks, don't process queued tasks,
354 * and interrupt in-progress tasks
355 * TIDYING: All tasks have terminated, workerCount is zero,
356 * the thread transitioning to state TIDYING
357 * will run the terminated() hook method
358 * TERMINATED: terminated() has completed
359 *
360 * The numerical order among these values matters, to allow
361 * ordered comparisons. The runState monotonically increases over
362 * time, but need not hit each state. The transitions are:
363 *
364 * RUNNING -> SHUTDOWN
365 * On invocation of shutdown()
366 * (RUNNING or SHUTDOWN) -> STOP
367 * On invocation of shutdownNow()
368 * SHUTDOWN -> TIDYING
369 * When both queue and pool are empty
370 * STOP -> TIDYING
371 * When pool is empty
372 * TIDYING -> TERMINATED
373 * When the terminated() hook method has completed
374 *
375 * Threads waiting in awaitTermination() will return when the
376 * state reaches TERMINATED.
377 *
378 * Detecting the transition from SHUTDOWN to TIDYING is less
379 * straightforward than you'd like because the queue may become
380 * empty after non-empty and vice versa during SHUTDOWN state, but
381 * we can only terminate if, after seeing that it is empty, we see
382 * that workerCount is 0 (which sometimes entails a recheck -- see
383 * below).
384 */
385 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
386 private static final int COUNT_BITS = Integer.SIZE - 3;
387 private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
388
389 // runState is stored in the high-order bits
390 private static final int RUNNING = -1 << COUNT_BITS;
391 private static final int SHUTDOWN = 0 << COUNT_BITS;
392 private static final int STOP = 1 << COUNT_BITS;
393 private static final int TIDYING = 2 << COUNT_BITS;
394 private static final int TERMINATED = 3 << COUNT_BITS;
395
396 // Packing and unpacking ctl
397 private static int runStateOf(int c) { return c & ~COUNT_MASK; }
398 private static int workerCountOf(int c) { return c & COUNT_MASK; }
399 private static int ctlOf(int rs, int wc) { return rs | wc; }
400
401 /*
402 * Bit field accessors that don't require unpacking ctl.
403 * These depend on the bit layout and on workerCount being never negative.
404 */
405
406 private static boolean runStateLessThan(int c, int s) {
407 return c < s;
408 }
409
410 private static boolean runStateAtLeast(int c, int s) {
411 return c >= s;
412 }
413
414 private static boolean isRunning(int c) {
415 return c < SHUTDOWN;
416 }
417
418 /**
419 * Attempts to CAS-increment the workerCount field of ctl.
420 */
421 private boolean compareAndIncrementWorkerCount(int expect) {
422 return ctl.compareAndSet(expect, expect + 1);
423 }
424
425 /**
426 * Attempts to CAS-decrement the workerCount field of ctl.
427 */
428 private boolean compareAndDecrementWorkerCount(int expect) {
429 return ctl.compareAndSet(expect, expect - 1);
430 }
431
432 /**
433 * Decrements the workerCount field of ctl. This is called only on
434 * abrupt termination of a thread (see processWorkerExit). Other
435 * decrements are performed within getTask.
436 */
437 private void decrementWorkerCount() {
438 ctl.addAndGet(-1);
439 }
440
441 /**
442 * The queue used for holding tasks and handing off to worker
443 * threads. We do not require that workQueue.poll() returning
444 * null necessarily means that workQueue.isEmpty(), so rely
445 * solely on isEmpty to see if the queue is empty (which we must
446 * do for example when deciding whether to transition from
447 * SHUTDOWN to TIDYING). This accommodates special-purpose
448 * queues such as DelayQueues for which poll() is allowed to
449 * return null even if it may later return non-null when delays
450 * expire.
451 */
452 private final BlockingQueue<Runnable> workQueue;
453
454 /**
455 * Lock held on access to workers set and related bookkeeping.
456 * While we could use a concurrent set of some sort, it turns out
457 * to be generally preferable to use a lock. Among the reasons is
458 * that this serializes interruptIdleWorkers, which avoids
459 * unnecessary interrupt storms, especially during shutdown.
460 * Otherwise exiting threads would concurrently interrupt those
461 * that have not yet interrupted. It also simplifies some of the
462 * associated statistics bookkeeping of largestPoolSize etc. We
463 * also hold mainLock on shutdown and shutdownNow, for the sake of
464 * ensuring workers set is stable while separately checking
465 * permission to interrupt and actually interrupting.
466 */
467 private final ReentrantLock mainLock = new ReentrantLock();
468
469 /**
470 * Set containing all worker threads in pool. Accessed only when
471 * holding mainLock.
472 */
473 private final HashSet<Worker> workers = new HashSet<>();
474
475 /**
476 * Wait condition to support awaitTermination.
477 */
478 private final Condition termination = mainLock.newCondition();
479
480 /**
481 * Tracks largest attained pool size. Accessed only under
482 * mainLock.
483 */
484 private int largestPoolSize;
485
486 /**
487 * Counter for completed tasks. Updated only on termination of
488 * worker threads. Accessed only under mainLock.
489 */
490 private long completedTaskCount;
491
492 /*
493 * All user control parameters are declared as volatiles so that
494 * ongoing actions are based on freshest values, but without need
495 * for locking, since no internal invariants depend on them
496 * changing synchronously with respect to other actions.
497 */
498
499 /**
500 * Factory for new threads. All threads are created using this
501 * factory (via method addWorker). All callers must be prepared
502 * for addWorker to fail, which may reflect a system or user's
503 * policy limiting the number of threads. Even though it is not
504 * treated as an error, failure to create threads may result in
505 * new tasks being rejected or existing ones remaining stuck in
506 * the queue.
507 *
508 * We go further and preserve pool invariants even in the face of
509 * errors such as OutOfMemoryError, that might be thrown while
510 * trying to create threads. Such errors are rather common due to
511 * the need to allocate a native stack in Thread.start, and users
512 * will want to perform clean pool shutdown to clean up. There
513 * will likely be enough memory available for the cleanup code to
514 * complete without encountering yet another OutOfMemoryError.
515 */
516 private volatile ThreadFactory threadFactory;
517
518 /**
519 * Handler called when saturated or shutdown in execute.
520 */
521 private volatile RejectedExecutionHandler handler;
522
523 /**
524 * Timeout in nanoseconds for idle threads waiting for work.
525 * Threads use this timeout when there are more than corePoolSize
526 * present or if allowCoreThreadTimeOut. Otherwise they wait
527 * forever for new work.
528 */
529 private volatile long keepAliveTime;
530
531 /**
532 * If false (default), core threads stay alive even when idle.
533 * If true, core threads use keepAliveTime to time out waiting
534 * for work.
535 */
536 private volatile boolean allowCoreThreadTimeOut;
537
538 /**
539 * Core pool size is the minimum number of workers to keep alive
540 * (and not allow to time out etc) unless allowCoreThreadTimeOut
541 * is set, in which case the minimum is zero.
542 *
543 * Since the worker count is actually stored in COUNT_BITS bits,
544 * the effective limit is {@code corePoolSize & COUNT_MASK}.
545 */
546 private volatile int corePoolSize;
547
548 /**
549 * Maximum pool size.
550 *
551 * Since the worker count is actually stored in COUNT_BITS bits,
552 * the effective limit is {@code maximumPoolSize & COUNT_MASK}.
553 */
554 private volatile int maximumPoolSize;
555
556 /**
557 * The default rejected execution handler.
558 */
559 private static final RejectedExecutionHandler defaultHandler =
560 new AbortPolicy();
561
562 /**
563 * Permission required for callers of shutdown and shutdownNow.
564 * We additionally require (see checkShutdownAccess) that callers
565 * have permission to actually interrupt threads in the worker set
566 * (as governed by Thread.interrupt, which relies on
567 * ThreadGroup.checkAccess, which in turn relies on
568 * SecurityManager.checkAccess). Shutdowns are attempted only if
569 * these checks pass.
570 *
571 * All actual invocations of Thread.interrupt (see
572 * interruptIdleWorkers and interruptWorkers) ignore
573 * SecurityExceptions, meaning that the attempted interrupts
574 * silently fail. In the case of shutdown, they should not fail
575 * unless the SecurityManager has inconsistent policies, sometimes
576 * allowing access to a thread and sometimes not. In such cases,
577 * failure to actually interrupt threads may disable or delay full
578 * termination. Other uses of interruptIdleWorkers are advisory,
579 * and failure to actually interrupt will merely delay response to
580 * configuration changes so is not handled exceptionally.
581 */
582 private static final RuntimePermission shutdownPerm =
583 new RuntimePermission("modifyThread");
584
585 /**
586 * Class Worker mainly maintains interrupt control state for
587 * threads running tasks, along with other minor bookkeeping.
588 * This class opportunistically extends AbstractQueuedSynchronizer
589 * to simplify acquiring and releasing a lock surrounding each
590 * task execution. This protects against interrupts that are
591 * intended to wake up a worker thread waiting for a task from
592 * instead interrupting a task being run. We implement a simple
593 * non-reentrant mutual exclusion lock rather than use
594 * ReentrantLock because we do not want worker tasks to be able to
595 * reacquire the lock when they invoke pool control methods like
596 * setCorePoolSize. Additionally, to suppress interrupts until
597 * the thread actually starts running tasks, we initialize lock
598 * state to a negative value, and clear it upon start (in
599 * runWorker).
600 */
601 private final class Worker
602 extends AbstractQueuedSynchronizer
603 implements Runnable
604 {
605 /**
606 * This class will never be serialized, but we provide a
607 * serialVersionUID to suppress a javac warning.
608 */
609 private static final long serialVersionUID = 6138294804551838833L;
610
611 /** Thread this worker is running in. Null if factory fails. */
612 @SuppressWarnings("serial") // Unlikely to be serializable
613 final Thread thread;
614 /** Initial task to run. Possibly null. */
615 @SuppressWarnings("serial") // Not statically typed as Serializable
616 Runnable firstTask;
617 /** Per-thread task counter */
618 volatile long completedTasks;
619
620 // TODO: switch to AbstractQueuedLongSynchronizer and move
621 // completedTasks into the lock word.
622
623 /**
624 * Creates with given first task and thread from ThreadFactory.
625 * @param firstTask the first task (null if none)
626 */
627 Worker(Runnable firstTask) {
628 setState(-1); // inhibit interrupts until runWorker
629 this.firstTask = firstTask;
630 this.thread = getThreadFactory().newThread(this);
631 }
632
633 /** Delegates main run loop to outer runWorker. */
634 public void run() {
635 runWorker(this);
636 }
637
638 // Lock methods
639 //
640 // The value 0 represents the unlocked state.
641 // The value 1 represents the locked state.
642
643 protected boolean isHeldExclusively() {
644 return getState() != 0;
645 }
646
647 protected boolean tryAcquire(int unused) {
648 if (compareAndSetState(0, 1)) {
649 setExclusiveOwnerThread(Thread.currentThread());
650 return true;
651 }
652 return false;
653 }
654
655 protected boolean tryRelease(int unused) {
656 setExclusiveOwnerThread(null);
657 setState(0);
658 return true;
659 }
660
661 public void lock() { acquire(1); }
662 public boolean tryLock() { return tryAcquire(1); }
663 public void unlock() { release(1); }
664 public boolean isLocked() { return isHeldExclusively(); }
665
666 void interruptIfStarted() {
667 Thread t;
668 if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
669 try {
670 t.interrupt();
671 } catch (SecurityException ignore) {
672 }
673 }
674 }
675 }
676
677 /*
678 * Methods for setting control state
679 */
680
681 /**
682 * Transitions runState to given target, or leaves it alone if
683 * already at least the given target.
684 *
685 * @param targetState the desired state, either SHUTDOWN or STOP
686 * (but not TIDYING or TERMINATED -- use tryTerminate for that)
687 */
688 private void advanceRunState(int targetState) {
689 // assert targetState == SHUTDOWN || targetState == STOP;
690 for (;;) {
691 int c = ctl.get();
692 if (runStateAtLeast(c, targetState) ||
693 ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))))
694 break;
695 }
696 }
697
698 /**
699 * Transitions to TERMINATED state if either (SHUTDOWN and pool
700 * and queue empty) or (STOP and pool empty). If otherwise
701 * eligible to terminate but workerCount is nonzero, interrupts an
702 * idle worker to ensure that shutdown signals propagate. This
703 * method must be called following any action that might make
704 * termination possible -- reducing worker count or removing tasks
705 * from the queue during shutdown. The method is non-private to
706 * allow access from ScheduledThreadPoolExecutor.
707 */
708 final void tryTerminate() {
709 for (;;) {
710 int c = ctl.get();
711 if (isRunning(c) ||
712 runStateAtLeast(c, TIDYING) ||
713 (runStateLessThan(c, STOP) && ! workQueue.isEmpty()))
714 return;
715 if (workerCountOf(c) != 0) { // Eligible to terminate
716 interruptIdleWorkers(ONLY_ONE);
717 return;
718 }
719
720 final ReentrantLock mainLock = this.mainLock;
721 mainLock.lock();
722 try {
723 if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
724 try {
725 terminated();
726 } finally {
727 ctl.set(ctlOf(TERMINATED, 0));
728 termination.signalAll();
729 }
730 return;
731 }
732 } finally {
733 mainLock.unlock();
734 }
735 // else retry on failed CAS
736 }
737 }
738
739 /*
740 * Methods for controlling interrupts to worker threads.
741 */
742
743 /**
744 * If there is a security manager, makes sure caller has
745 * permission to shut down threads in general (see shutdownPerm).
746 * If this passes, additionally makes sure the caller is allowed
747 * to interrupt each worker thread. This might not be true even if
748 * first check passed, if the SecurityManager treats some threads
749 * specially.
750 */
751 private void checkShutdownAccess() {
752 // assert mainLock.isHeldByCurrentThread();
753 @SuppressWarnings("removal")
754 SecurityManager security = System.getSecurityManager();
755 if (security != null) {
756 security.checkPermission(shutdownPerm);
757 for (Worker w : workers)
758 security.checkAccess(w.thread);
759 }
760 }
761
762 /**
763 * Interrupts all threads, even if active. Ignores SecurityExceptions
764 * (in which case some threads may remain uninterrupted).
765 */
766 private void interruptWorkers() {
767 // assert mainLock.isHeldByCurrentThread();
768 for (Worker w : workers)
769 w.interruptIfStarted();
770 }
771
772 /**
773 * Interrupts threads that might be waiting for tasks (as
774 * indicated by not being locked) so they can check for
775 * termination or configuration changes. Ignores
776 * SecurityExceptions (in which case some threads may remain
777 * uninterrupted).
778 *
779 * @param onlyOne If true, interrupt at most one worker. This is
780 * called only from tryTerminate when termination is otherwise
781 * enabled but there are still other workers. In this case, at
782 * most one waiting worker is interrupted to propagate shutdown
783 * signals in case all threads are currently waiting.
784 * Interrupting any arbitrary thread ensures that newly arriving
785 * workers since shutdown began will also eventually exit.
786 * To guarantee eventual termination, it suffices to always
787 * interrupt only one idle worker, but shutdown() interrupts all
788 * idle workers so that redundant workers exit promptly, not
789 * waiting for a straggler task to finish.
790 */
791 private void interruptIdleWorkers(boolean onlyOne) {
792 final ReentrantLock mainLock = this.mainLock;
793 mainLock.lock();
794 try {
795 for (Worker w : workers) {
796 Thread t = w.thread;
797 if (!t.isInterrupted() && w.tryLock()) {
798 try {
799 t.interrupt();
800 } catch (SecurityException ignore) {
801 } finally {
802 w.unlock();
803 }
804 }
805 if (onlyOne)
806 break;
807 }
808 } finally {
809 mainLock.unlock();
810 }
811 }
812
813 /**
814 * Common form of interruptIdleWorkers, to avoid having to
815 * remember what the boolean argument means.
816 */
817 private void interruptIdleWorkers() {
818 interruptIdleWorkers(false);
819 }
820
821 private static final boolean ONLY_ONE = true;
822
823 /*
824 * Misc utilities, most of which are also exported to
825 * ScheduledThreadPoolExecutor
826 */
827
828 /**
829 * Invokes the rejected execution handler for the given command.
830 * Package-protected for use by ScheduledThreadPoolExecutor.
831 */
832 final void reject(Runnable command) {
833 handler.rejectedExecution(command, this);
834 }
835
836 /**
837 * Performs any further cleanup following run state transition on
838 * invocation of shutdown. A no-op here, but used by
839 * ScheduledThreadPoolExecutor to cancel delayed tasks.
840 */
841 void onShutdown() {
842 }
843
844 /**
845 * Drains the task queue into a new list, normally using
846 * drainTo. But if the queue is a DelayQueue or any other kind of
847 * queue for which poll or drainTo may fail to remove some
848 * elements, it deletes them one by one.
849 */
850 private List<Runnable> drainQueue() {
851 BlockingQueue<Runnable> q = workQueue;
852 ArrayList<Runnable> taskList = new ArrayList<>();
853 q.drainTo(taskList);
854 if (!q.isEmpty()) {
855 for (Runnable r : q.toArray(new Runnable[0])) {
856 if (q.remove(r))
857 taskList.add(r);
858 }
859 }
860 return taskList;
861 }
862
863 /*
864 * Methods for creating, running and cleaning up after workers
865 */
866
867 /**
868 * Checks if a new worker can be added with respect to current
869 * pool state and the given bound (either core or maximum). If so,
870 * the worker count is adjusted accordingly, and, if possible, a
871 * new worker is created and started, running firstTask as its
872 * first task. This method returns false if the pool is stopped or
873 * eligible to shut down. It also returns false if the thread
874 * factory fails to create a thread when asked. If the thread
875 * creation fails, either due to the thread factory returning
876 * null, or due to an exception (typically OutOfMemoryError in
877 * Thread.start()), we roll back cleanly.
878 *
879 * @param firstTask the task the new thread should run first (or
880 * null if none). Workers are created with an initial first task
881 * (in method execute()) to bypass queuing when there are fewer
882 * than corePoolSize threads (in which case we always start one),
883 * or when the queue is full (in which case we must bypass queue).
884 * Initially idle threads are usually created via
885 * prestartCoreThread or to replace other dying workers.
886 *
887 * @param core if true use corePoolSize as bound, else
888 * maximumPoolSize. (A boolean indicator is used here rather than a
889 * value to ensure reads of fresh values after checking other pool
890 * state).
891 * @return true if successful
892 */
893 private boolean addWorker(Runnable firstTask, boolean core) {
894 retry:
895 for (int c = ctl.get();;) {
896 // Check if queue empty only if necessary.
897 if (runStateAtLeast(c, SHUTDOWN)
898 && (runStateAtLeast(c, STOP)
899 || firstTask != null
900 || workQueue.isEmpty()))
901 return false;
902
903 for (;;) {
904 if (workerCountOf(c)
905 >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
906 return false;
907 if (compareAndIncrementWorkerCount(c))
908 break retry;
909 c = ctl.get(); // Re-read ctl
910 if (runStateAtLeast(c, SHUTDOWN))
911 continue retry;
912 // else CAS failed due to workerCount change; retry inner loop
913 }
914 }
915
916 boolean workerStarted = false;
917 boolean workerAdded = false;
918 Worker w = null;
919 try {
920 w = new Worker(firstTask);
921 final Thread t = w.thread;
922 if (t != null) {
923 final ReentrantLock mainLock = this.mainLock;
924 mainLock.lock();
925 try {
926 // Recheck while holding lock.
927 // Back out on ThreadFactory failure or if
928 // shut down before lock acquired.
929 int c = ctl.get();
930
931 if (isRunning(c) ||
932 (runStateLessThan(c, STOP) && firstTask == null)) {
933 if (t.getState() != Thread.State.NEW)
934 throw new IllegalThreadStateException();
935 workers.add(w);
936 workerAdded = true;
937 int s = workers.size();
938 if (s > largestPoolSize)
939 largestPoolSize = s;
940 }
941 } finally {
942 mainLock.unlock();
943 }
944 if (workerAdded) {
945 t.start();
946 workerStarted = true;
947 }
948 }
949 } finally {
950 if (! workerStarted)
951 addWorkerFailed(w);
952 }
953 return workerStarted;
954 }
955
956 /**
957 * Rolls back the worker thread creation.
958 * - removes worker from workers, if present
959 * - decrements worker count
960 * - rechecks for termination, in case the existence of this
961 * worker was holding up termination
962 */
963 private void addWorkerFailed(Worker w) {
964 final ReentrantLock mainLock = this.mainLock;
965 mainLock.lock();
966 try {
967 if (w != null)
968 workers.remove(w);
969 decrementWorkerCount();
970 tryTerminate();
971 } finally {
972 mainLock.unlock();
973 }
974 }
975
976 /**
977 * Performs cleanup and bookkeeping for a dying worker. Called
978 * only from worker threads. Unless completedAbruptly is set,
979 * assumes that workerCount has already been adjusted to account
980 * for exit. This method removes thread from worker set, and
981 * possibly terminates the pool or replaces the worker if either
982 * it exited due to user task exception or if fewer than
983 * corePoolSize workers are running or queue is non-empty but
984 * there are no workers.
985 *
986 * @param w the worker
987 * @param completedAbruptly if the worker died due to user exception
988 */
989 private void processWorkerExit(Worker w, boolean completedAbruptly) {
990 if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
991 decrementWorkerCount();
992
993 final ReentrantLock mainLock = this.mainLock;
994 mainLock.lock();
995 try {
996 completedTaskCount += w.completedTasks;
997 workers.remove(w);
998 } finally {
999 mainLock.unlock();
1000 }
1001
1002 tryTerminate();
1003
1004 int c = ctl.get();
1005 if (runStateLessThan(c, STOP)) {
1006 if (!completedAbruptly) {
1007 int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
1008 if (min == 0 && ! workQueue.isEmpty())
1009 min = 1;
1010 if (workerCountOf(c) >= min)
1011 return; // replacement not needed
1012 }
1013 addWorker(null, false);
1014 }
1015 }
1016
1017 /**
1018 * Performs blocking or timed wait for a task, depending on
1019 * current configuration settings, or returns null if this worker
1020 * must exit because of any of:
1021 * 1. There are more than maximumPoolSize workers (due to
1022 * a call to setMaximumPoolSize).
1023 * 2. The pool is stopped.
1024 * 3. The pool is shutdown and the queue is empty.
1025 * 4. This worker timed out waiting for a task, and timed-out
1026 * workers are subject to termination (that is,
1027 * {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
1028 * both before and after the timed wait, and if the queue is
1029 * non-empty, this worker is not the last thread in the pool.
1030 *
1031 * @return task, or null if the worker must exit, in which case
1032 * workerCount is decremented
1033 */
1034 private Runnable getTask() {
1035 boolean timedOut = false; // Did the last poll() time out?
1036
1037 for (;;) {
1038 int c = ctl.get();
1039
1040 // Check if queue empty only if necessary.
1041 if (runStateAtLeast(c, SHUTDOWN)
1042 && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
1043 decrementWorkerCount();
1044 return null;
1045 }
1046
1047 int wc = workerCountOf(c);
1048
1049 // Are workers subject to culling?
1050 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
1051
1052 if ((wc > maximumPoolSize || (timed && timedOut))
1053 && (wc > 1 || workQueue.isEmpty())) {
1054 if (compareAndDecrementWorkerCount(c))
1055 return null;
1056 continue;
1057 }
1058
1059 try {
1060 Runnable r = timed ?
1061 workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
1062 workQueue.take();
1063 if (r != null)
1064 return r;
1065 timedOut = true;
1066 } catch (InterruptedException retry) {
1067 timedOut = false;
1068 }
1069 }
1070 }
1071
1072 /**
1073 * Main worker run loop. Repeatedly gets tasks from queue and
1074 * executes them, while coping with a number of issues:
1075 *
1076 * 1. We may start out with an initial task, in which case we
1077 * don't need to get the first one. Otherwise, as long as pool is
1078 * running, we get tasks from getTask. If it returns null then the
1079 * worker exits due to changed pool state or configuration
1080 * parameters. Other exits result from exception throws in
1081 * external code, in which case completedAbruptly holds, which
1082 * usually leads processWorkerExit to replace this thread.
1083 *
1084 * 2. Before running any task, the lock is acquired to prevent
1085 * other pool interrupts while the task is executing, and then we
1086 * ensure that unless pool is stopping, this thread does not have
1087 * its interrupt set.
1088 *
1089 * 3. Each task run is preceded by a call to beforeExecute, which
1090 * might throw an exception, in which case we cause thread to die
1091 * (breaking loop with completedAbruptly true) without processing
1092 * the task.
1093 *
1094 * 4. Assuming beforeExecute completes normally, we run the task,
1095 * gathering any of its thrown exceptions to send to afterExecute.
1096 * We separately handle RuntimeException, Error (both of which the
1097 * specs guarantee that we trap) and arbitrary Throwables.
1098 * Because we cannot rethrow Throwables within Runnable.run, we
1099 * wrap them within Errors on the way out (to the thread's
1100 * UncaughtExceptionHandler). Any thrown exception also
1101 * conservatively causes thread to die.
1102 *
1103 * 5. After task.run completes, we call afterExecute, which may
1104 * also throw an exception, which will also cause thread to
1105 * die. According to JLS Sec 14.20, this exception is the one that
1106 * will be in effect even if task.run throws.
1107 *
1108 * The net effect of the exception mechanics is that afterExecute
1109 * and the thread's UncaughtExceptionHandler have as accurate
1110 * information as we can provide about any problems encountered by
1111 * user code.
1112 *
1113 * @param w the worker
1114 */
1115 final void runWorker(Worker w) {
1116 Thread wt = Thread.currentThread();
1117 Runnable task = w.firstTask;
1118 w.firstTask = null;
1119 w.unlock(); // allow interrupts
1120 boolean completedAbruptly = true;
1121 try {
1122 while (task != null || (task = getTask()) != null) {
1123 w.lock();
1124 // If pool is stopping, ensure thread is interrupted;
1125 // if not, ensure thread is not interrupted. This
1126 // requires a recheck in second case to deal with
1127 // shutdownNow race while clearing interrupt
1128 if ((runStateAtLeast(ctl.get(), STOP) ||
1129 (Thread.interrupted() &&
1130 runStateAtLeast(ctl.get(), STOP))) &&
1131 !wt.isInterrupted())
1132 wt.interrupt();
1133 try {
1134 beforeExecute(wt, task);
1135 try {
1136 task.run();
1137 afterExecute(task, null);
1138 } catch (Throwable ex) {
1139 afterExecute(task, ex);
1140 throw ex;
1141 }
1142 } finally {
1143 task = null;
1144 w.completedTasks++;
1145 w.unlock();
1146 }
1147 }
1148 completedAbruptly = false;
1149 } finally {
1150 processWorkerExit(w, completedAbruptly);
1151 }
1152 }
1153
1154 // Public constructors and methods
1155
1156 /**
1157 * Creates a new {@code ThreadPoolExecutor} with the given initial
1158 * parameters, the
1159 * {@linkplain Executors#defaultThreadFactory default thread factory}
1160 * and the {@linkplain ThreadPoolExecutor.AbortPolicy
1161 * default rejected execution handler}.
1162 *
1163 * <p>It may be more convenient to use one of the {@link Executors}
1164 * factory methods instead of this general purpose constructor.
1165 *
1166 * @param corePoolSize the number of threads to keep in the pool, even
1167 * if they are idle, unless {@code allowCoreThreadTimeOut} is set
1168 * @param maximumPoolSize the maximum number of threads to allow in the
1169 * pool
1170 * @param keepAliveTime when the number of threads is greater than
1171 * the core, this is the maximum time that excess idle threads
1172 * will wait for new tasks before terminating.
1173 * @param unit the time unit for the {@code keepAliveTime} argument
1174 * @param workQueue the queue to use for holding tasks before they are
1175 * executed. This queue will hold only the {@code Runnable}
1176 * tasks submitted by the {@code execute} method.
1177 * @throws IllegalArgumentException if one of the following holds:<br>
1178 * {@code corePoolSize < 0}<br>
1179 * {@code keepAliveTime < 0}<br>
1180 * {@code maximumPoolSize <= 0}<br>
1181 * {@code maximumPoolSize < corePoolSize}
1182 * @throws NullPointerException if {@code workQueue} is null
1183 */
1184 public ThreadPoolExecutor(int corePoolSize,
1185 int maximumPoolSize,
1186 long keepAliveTime,
1187 TimeUnit unit,
1188 BlockingQueue<Runnable> workQueue) {
1189 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
1190 Executors.defaultThreadFactory(), defaultHandler);
1191 }
1192
1193 /**
1194 * Creates a new {@code ThreadPoolExecutor} with the given initial
1195 * parameters and the {@linkplain ThreadPoolExecutor.AbortPolicy
1196 * default rejected execution handler}.
1197 *
1198 * @param corePoolSize the number of threads to keep in the pool, even
1199 * if they are idle, unless {@code allowCoreThreadTimeOut} is set
1200 * @param maximumPoolSize the maximum number of threads to allow in the
1201 * pool
1202 * @param keepAliveTime when the number of threads is greater than
1203 * the core, this is the maximum time that excess idle threads
1204 * will wait for new tasks before terminating.
1205 * @param unit the time unit for the {@code keepAliveTime} argument
1206 * @param workQueue the queue to use for holding tasks before they are
1207 * executed. This queue will hold only the {@code Runnable}
1208 * tasks submitted by the {@code execute} method.
1209 * @param threadFactory the factory to use when the executor
1210 * creates a new thread
1211 * @throws IllegalArgumentException if one of the following holds:<br>
1212 * {@code corePoolSize < 0}<br>
1213 * {@code keepAliveTime < 0}<br>
1214 * {@code maximumPoolSize <= 0}<br>
1215 * {@code maximumPoolSize < corePoolSize}
1216 * @throws NullPointerException if {@code workQueue}
1217 * or {@code threadFactory} is null
1218 */
1219 public ThreadPoolExecutor(int corePoolSize,
1220 int maximumPoolSize,
1221 long keepAliveTime,
1222 TimeUnit unit,
1223 BlockingQueue<Runnable> workQueue,
1224 ThreadFactory threadFactory) {
1225 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
1226 threadFactory, defaultHandler);
1227 }
1228
1229 /**
1230 * Creates a new {@code ThreadPoolExecutor} with the given initial
1231 * parameters and the
1232 * {@linkplain Executors#defaultThreadFactory default thread factory}.
1233 *
1234 * @param corePoolSize the number of threads to keep in the pool, even
1235 * if they are idle, unless {@code allowCoreThreadTimeOut} is set
1236 * @param maximumPoolSize the maximum number of threads to allow in the
1237 * pool
1238 * @param keepAliveTime when the number of threads is greater than
1239 * the core, this is the maximum time that excess idle threads
1240 * will wait for new tasks before terminating.
1241 * @param unit the time unit for the {@code keepAliveTime} argument
1242 * @param workQueue the queue to use for holding tasks before they are
1243 * executed. This queue will hold only the {@code Runnable}
1244 * tasks submitted by the {@code execute} method.
1245 * @param handler the handler to use when execution is blocked
1246 * because the thread bounds and queue capacities are reached
1247 * @throws IllegalArgumentException if one of the following holds:<br>
1248 * {@code corePoolSize < 0}<br>
1249 * {@code keepAliveTime < 0}<br>
1250 * {@code maximumPoolSize <= 0}<br>
1251 * {@code maximumPoolSize < corePoolSize}
1252 * @throws NullPointerException if {@code workQueue}
1253 * or {@code handler} is null
1254 */
1255 public ThreadPoolExecutor(int corePoolSize,
1256 int maximumPoolSize,
1257 long keepAliveTime,
1258 TimeUnit unit,
1259 BlockingQueue<Runnable> workQueue,
1260 RejectedExecutionHandler handler) {
1261 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
1262 Executors.defaultThreadFactory(), handler);
1263 }
1264
1265 /**
1266 * Creates a new {@code ThreadPoolExecutor} with the given initial
1267 * parameters.
1268 *
1269 * @param corePoolSize the number of threads to keep in the pool, even
1270 * if they are idle, unless {@code allowCoreThreadTimeOut} is set
1271 * @param maximumPoolSize the maximum number of threads to allow in the
1272 * pool
1273 * @param keepAliveTime when the number of threads is greater than
1274 * the core, this is the maximum time that excess idle threads
1275 * will wait for new tasks before terminating.
1276 * @param unit the time unit for the {@code keepAliveTime} argument
1277 * @param workQueue the queue to use for holding tasks before they are
1278 * executed. This queue will hold only the {@code Runnable}
1279 * tasks submitted by the {@code execute} method.
1280 * @param threadFactory the factory to use when the executor
1281 * creates a new thread
1282 * @param handler the handler to use when execution is blocked
1283 * because the thread bounds and queue capacities are reached
1284 * @throws IllegalArgumentException if one of the following holds:<br>
1285 * {@code corePoolSize < 0}<br>
1286 * {@code keepAliveTime < 0}<br>
1287 * {@code maximumPoolSize <= 0}<br>
1288 * {@code maximumPoolSize < corePoolSize}
1289 * @throws NullPointerException if {@code workQueue}
1290 * or {@code threadFactory} or {@code handler} is null
1291 */
1292 public ThreadPoolExecutor(int corePoolSize,
1293 int maximumPoolSize,
1294 long keepAliveTime,
1295 TimeUnit unit,
1296 BlockingQueue<Runnable> workQueue,
1297 ThreadFactory threadFactory,
1298 RejectedExecutionHandler handler) {
1299 if (corePoolSize < 0 ||
1300 maximumPoolSize <= 0 ||
1301 maximumPoolSize < corePoolSize ||
1302 keepAliveTime < 0)
1303 throw new IllegalArgumentException();
1304 if (workQueue == null || threadFactory == null || handler == null)
1305 throw new NullPointerException();
1306 this.corePoolSize = corePoolSize;
1307 this.maximumPoolSize = maximumPoolSize;
1308 this.workQueue = workQueue;
1309 this.keepAliveTime = unit.toNanos(keepAliveTime);
1310 this.threadFactory = threadFactory;
1311 this.handler = handler;
1312 }
1313
1314 /**
1315 * Executes the given task sometime in the future. The task
1316 * may execute in a new thread or in an existing pooled thread.
1317 *
1318 * If the task cannot be submitted for execution, either because this
1319 * executor has been shutdown or because its capacity has been reached,
1320 * the task is handled by the current {@link RejectedExecutionHandler}.
1321 *
1322 * @param command the task to execute
1323 * @throws RejectedExecutionException at discretion of
1324 * {@code RejectedExecutionHandler}, if the task
1325 * cannot be accepted for execution
1326 * @throws NullPointerException if {@code command} is null
1327 */
1328 public void execute(Runnable command) {
1329 if (command == null)
1330 throw new NullPointerException();
1331 /*
1332 * Proceed in 3 steps:
1333 *
1334 * 1. If fewer than corePoolSize threads are running, try to
1335 * start a new thread with the given command as its first
1336 * task. The call to addWorker atomically checks runState and
1337 * workerCount, and so prevents false alarms that would add
1338 * threads when it shouldn't, by returning false.
1339 *
1340 * 2. If a task can be successfully queued, then we still need
1341 * to double-check whether we should have added a thread
1342 * (because existing ones died since last checking) or that
1343 * the pool shut down since entry into this method. So we
1344 * recheck state and if necessary roll back the enqueuing if
1345 * stopped, or start a new thread if there are none.
1346 *
1347 * 3. If we cannot queue task, then we try to add a new
1348 * thread. If it fails, we know we are shut down or saturated
1349 * and so reject the task.
1350 */
1351 int c = ctl.get();
1352 if (workerCountOf(c) < corePoolSize) {
1353 if (addWorker(command, true))
1354 return;
1355 c = ctl.get();
1356 }
1357 if (isRunning(c) && workQueue.offer(command)) {
1358 int recheck = ctl.get();
1359 if (! isRunning(recheck) && remove(command))
1360 reject(command);
1361 else if (workerCountOf(recheck) == 0)
1362 addWorker(null, false);
1363 }
1364 else if (!addWorker(command, false))
1365 reject(command);
1366 }
1367
1368 /**
1369 * Initiates an orderly shutdown in which previously submitted
1370 * tasks are executed, but no new tasks will be accepted.
1371 * Invocation has no additional effect if already shut down.
1372 *
1373 * <p>This method does not wait for previously submitted tasks to
1374 * complete execution. Use {@link #awaitTermination awaitTermination}
1375 * to do that.
1376 *
1377 * @throws SecurityException {@inheritDoc}
1378 */
1379 public void shutdown() {
1380 final ReentrantLock mainLock = this.mainLock;
1381 mainLock.lock();
1382 try {
1383 checkShutdownAccess();
1384 advanceRunState(SHUTDOWN);
1385 interruptIdleWorkers();
1386 onShutdown(); // hook for ScheduledThreadPoolExecutor
1387 } finally {
1388 mainLock.unlock();
1389 }
1390 tryTerminate();
1391 }
1392
1393 /**
1394 * Attempts to stop all actively executing tasks, halts the
1395 * processing of waiting tasks, and returns a list of the tasks
1396 * that were awaiting execution. These tasks are drained (removed)
1397 * from the task queue upon return from this method.
1398 *
1399 * <p>This method does not wait for actively executing tasks to
1400 * terminate. Use {@link #awaitTermination awaitTermination} to
1401 * do that.
1402 *
1403 * <p>There are no guarantees beyond best-effort attempts to stop
1404 * processing actively executing tasks. This implementation
1405 * interrupts tasks via {@link Thread#interrupt}; any task that
1406 * fails to respond to interrupts may never terminate.
1407 *
1408 * @throws SecurityException {@inheritDoc}
1409 */
1410 public List<Runnable> shutdownNow() {
1411 List<Runnable> tasks;
1412 final ReentrantLock mainLock = this.mainLock;
1413 mainLock.lock();
1414 try {
1415 checkShutdownAccess();
1416 advanceRunState(STOP);
1417 interruptWorkers();
1418 tasks = drainQueue();
1419 } finally {
1420 mainLock.unlock();
1421 }
1422 tryTerminate();
1423 return tasks;
1424 }
1425
1426 public boolean isShutdown() {
1427 return runStateAtLeast(ctl.get(), SHUTDOWN);
1428 }
1429
1430 /** Used by ScheduledThreadPoolExecutor. */
1431 boolean isStopped() {
1432 return runStateAtLeast(ctl.get(), STOP);
1433 }
1434
1435 /**
1436 * Returns true if this executor is in the process of terminating
1437 * after {@link #shutdown} or {@link #shutdownNow} but has not
1438 * completely terminated. This method may be useful for
1439 * debugging. A return of {@code true} reported a sufficient
1440 * period after shutdown may indicate that submitted tasks have
1441 * ignored or suppressed interruption, causing this executor not
1442 * to properly terminate.
1443 *
1444 * @return {@code true} if terminating but not yet terminated
1445 */
1446 public boolean isTerminating() {
1447 int c = ctl.get();
1448 return runStateAtLeast(c, SHUTDOWN) && runStateLessThan(c, TERMINATED);
1449 }
1450
1451 public boolean isTerminated() {
1452 return runStateAtLeast(ctl.get(), TERMINATED);
1453 }
1454
1455 public boolean awaitTermination(long timeout, TimeUnit unit)
1456 throws InterruptedException {
1457 long nanos = unit.toNanos(timeout);
1458 final ReentrantLock mainLock = this.mainLock;
1459 mainLock.lock();
1460 try {
1461 while (runStateLessThan(ctl.get(), TERMINATED)) {
1462 if (nanos <= 0L)
1463 return false;
1464 nanos = termination.awaitNanos(nanos);
1465 }
1466 return true;
1467 } finally {
1468 mainLock.unlock();
1469 }
1470 }
1471
1472 // Override without "throws Throwable" for compatibility with subclasses
1473 // whose finalize method invokes super.finalize() (as is recommended).
1474 // Before JDK 11, finalize() had a non-empty method body.
1475
1476 /**
1477 * @implNote Previous versions of this class had a finalize method
1478 * that shut down this executor, but in this version, finalize
1479 * does nothing.
1480 */
1481 @Deprecated(since="9")
1482 protected void finalize() {}
1483
1484 /**
1485 * Sets the thread factory used to create new threads.
1486 *
1487 * @param threadFactory the new thread factory
1488 * @throws NullPointerException if threadFactory is null
1489 * @see #getThreadFactory
1490 */
1491 public void setThreadFactory(ThreadFactory threadFactory) {
1492 if (threadFactory == null)
1493 throw new NullPointerException();
1494 this.threadFactory = threadFactory;
1495 }
1496
1497 /**
1498 * Returns the thread factory used to create new threads.
1499 *
1500 * @return the current thread factory
1501 * @see #setThreadFactory(ThreadFactory)
1502 */
1503 public ThreadFactory getThreadFactory() {
1504 return threadFactory;
1505 }
1506
1507 /**
1508 * Sets a new handler for unexecutable tasks.
1509 *
1510 * @param handler the new handler
1511 * @throws NullPointerException if handler is null
1512 * @see #getRejectedExecutionHandler
1513 */
1514 public void setRejectedExecutionHandler(RejectedExecutionHandler handler) {
1515 if (handler == null)
1516 throw new NullPointerException();
1517 this.handler = handler;
1518 }
1519
1520 /**
1521 * Returns the current handler for unexecutable tasks.
1522 *
1523 * @return the current handler
1524 * @see #setRejectedExecutionHandler(RejectedExecutionHandler)
1525 */
1526 public RejectedExecutionHandler getRejectedExecutionHandler() {
1527 return handler;
1528 }
1529
1530 /**
1531 * Sets the core number of threads. This overrides any value set
1532 * in the constructor. If the new value is smaller than the
1533 * current value, excess existing threads will be terminated when
1534 * they next become idle. If larger, new threads will, if needed,
1535 * be started to execute any queued tasks.
1536 *
1537 * @param corePoolSize the new core size
1538 * @throws IllegalArgumentException if {@code corePoolSize < 0}
1539 * or {@code corePoolSize} is greater than the {@linkplain
1540 * #getMaximumPoolSize() maximum pool size}
1541 * @see #getCorePoolSize
1542 */
1543 public void setCorePoolSize(int corePoolSize) {
1544 if (corePoolSize < 0 || maximumPoolSize < corePoolSize)
1545 throw new IllegalArgumentException();
1546 int delta = corePoolSize - this.corePoolSize;
1547 this.corePoolSize = corePoolSize;
1548 if (workerCountOf(ctl.get()) > corePoolSize)
1549 interruptIdleWorkers();
1550 else if (delta > 0) {
1551 // We don't really know how many new threads are "needed".
1552 // As a heuristic, prestart enough new workers (up to new
1553 // core size) to handle the current number of tasks in
1554 // queue, but stop if queue becomes empty while doing so.
1555 int k = Math.min(delta, workQueue.size());
1556 while (k-- > 0 && addWorker(null, true)) {
1557 if (workQueue.isEmpty())
1558 break;
1559 }
1560 }
1561 }
1562
1563 /**
1564 * Returns the core number of threads.
1565 *
1566 * @return the core number of threads
1567 * @see #setCorePoolSize
1568 */
1569 public int getCorePoolSize() {
1570 return corePoolSize;
1571 }
1572
1573 /**
1574 * Starts a core thread, causing it to idly wait for work. This
1575 * overrides the default policy of starting core threads only when
1576 * new tasks are executed. This method will return {@code false}
1577 * if all core threads have already been started.
1578 *
1579 * @return {@code true} if a thread was started
1580 */
1581 public boolean prestartCoreThread() {
1582 return workerCountOf(ctl.get()) < corePoolSize &&
1583 addWorker(null, true);
1584 }
1585
1586 /**
1587 * Same as prestartCoreThread except arranges that at least one
1588 * thread is started even if corePoolSize is 0.
1589 */
1590 void ensurePrestart() {
1591 int wc = workerCountOf(ctl.get());
1592 if (wc < corePoolSize)
1593 addWorker(null, true);
1594 else if (wc == 0)
1595 addWorker(null, false);
1596 }
1597
1598 /**
1599 * Starts all core threads, causing them to idly wait for work. This
1600 * overrides the default policy of starting core threads only when
1601 * new tasks are executed.
1602 *
1603 * @return the number of threads started
1604 */
1605 public int prestartAllCoreThreads() {
1606 int n = 0;
1607 while (addWorker(null, true))
1608 ++n;
1609 return n;
1610 }
1611
1612 /**
1613 * Returns true if this pool allows core threads to time out and
1614 * terminate if no tasks arrive within the keepAlive time, being
1615 * replaced if needed when new tasks arrive. When true, the same
1616 * keep-alive policy applying to non-core threads applies also to
1617 * core threads. When false (the default), core threads are never
1618 * terminated due to lack of incoming tasks.
1619 *
1620 * @return {@code true} if core threads are allowed to time out,
1621 * else {@code false}
1622 *
1623 * @since 1.6
1624 */
1625 public boolean allowsCoreThreadTimeOut() {
1626 return allowCoreThreadTimeOut;
1627 }
1628
1629 /**
1630 * Sets the policy governing whether core threads may time out and
1631 * terminate if no tasks arrive within the keep-alive time, being
1632 * replaced if needed when new tasks arrive. When false, core
1633 * threads are never terminated due to lack of incoming
1634 * tasks. When true, the same keep-alive policy applying to
1635 * non-core threads applies also to core threads. To avoid
1636 * continual thread replacement, the keep-alive time must be
1637 * greater than zero when setting {@code true}. This method
1638 * should in general be called before the pool is actively used.
1639 *
1640 * @param value {@code true} if should time out, else {@code false}
1641 * @throws IllegalArgumentException if value is {@code true}
1642 * and the current keep-alive time is not greater than zero
1643 *
1644 * @since 1.6
1645 */
1646 public void allowCoreThreadTimeOut(boolean value) {
1647 if (value && keepAliveTime <= 0)
1648 throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
1649 if (value != allowCoreThreadTimeOut) {
1650 allowCoreThreadTimeOut = value;
1651 if (value)
1652 interruptIdleWorkers();
1653 }
1654 }
1655
1656 /**
1657 * Sets the maximum allowed number of threads. This overrides any
1658 * value set in the constructor. If the new value is smaller than
1659 * the current value, excess existing threads will be
1660 * terminated when they next become idle.
1661 *
1662 * @param maximumPoolSize the new maximum
1663 * @throws IllegalArgumentException if the new maximum is
1664 * less than or equal to zero, or
1665 * less than the {@linkplain #getCorePoolSize core pool size}
1666 * @see #getMaximumPoolSize
1667 */
1668 public void setMaximumPoolSize(int maximumPoolSize) {
1669 if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)
1670 throw new IllegalArgumentException();
1671 this.maximumPoolSize = maximumPoolSize;
1672 if (workerCountOf(ctl.get()) > maximumPoolSize)
1673 interruptIdleWorkers();
1674 }
1675
1676 /**
1677 * Returns the maximum allowed number of threads.
1678 *
1679 * @return the maximum allowed number of threads
1680 * @see #setMaximumPoolSize
1681 */
1682 public int getMaximumPoolSize() {
1683 return maximumPoolSize;
1684 }
1685
1686 /**
1687 * Sets the thread keep-alive time, which is the amount of time
1688 * that threads may remain idle before being terminated.
1689 * Threads that wait this amount of time without processing a
1690 * task will be terminated if there are more than the core
1691 * number of threads currently in the pool, or if this pool
1692 * {@linkplain #allowsCoreThreadTimeOut() allows core thread timeout}.
1693 * This overrides any value set in the constructor.
1694 *
1695 * @param time the time to wait. A time value of zero will cause
1696 * excess threads to terminate immediately after executing tasks.
1697 * @param unit the time unit of the {@code time} argument
1698 * @throws IllegalArgumentException if {@code time} less than zero or
1699 * if {@code time} is zero and {@code allowsCoreThreadTimeOut}
1700 * @see #getKeepAliveTime(TimeUnit)
1701 */
1702 public void setKeepAliveTime(long time, TimeUnit unit) {
1703 if (time < 0)
1704 throw new IllegalArgumentException();
1705 if (time == 0 && allowsCoreThreadTimeOut())
1706 throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
1707 long keepAliveTime = unit.toNanos(time);
1708 long delta = keepAliveTime - this.keepAliveTime;
1709 this.keepAliveTime = keepAliveTime;
1710 if (delta < 0)
1711 interruptIdleWorkers();
1712 }
1713
1714 /**
1715 * Returns the thread keep-alive time, which is the amount of time
1716 * that threads may remain idle before being terminated.
1717 * Threads that wait this amount of time without processing a
1718 * task will be terminated if there are more than the core
1719 * number of threads currently in the pool, or if this pool
1720 * {@linkplain #allowsCoreThreadTimeOut() allows core thread timeout}.
1721 *
1722 * @param unit the desired time unit of the result
1723 * @return the time limit
1724 * @see #setKeepAliveTime(long, TimeUnit)
1725 */
1726 public long getKeepAliveTime(TimeUnit unit) {
1727 return unit.convert(keepAliveTime, TimeUnit.NANOSECONDS);
1728 }
1729
1730 /* User-level queue utilities */
1731
1732 /**
1733 * Returns the task queue used by this executor. Access to the
1734 * task queue is intended primarily for debugging and monitoring.
1735 * This queue may be in active use. Retrieving the task queue
1736 * does not prevent queued tasks from executing.
1737 *
1738 * @return the task queue
1739 */
1740 public BlockingQueue<Runnable> getQueue() {
1741 return workQueue;
1742 }
1743
1744 /**
1745 * Removes this task from the executor's internal queue if it is
1746 * present, thus causing it not to be run if it has not already
1747 * started.
1748 *
1749 * <p>This method may be useful as one part of a cancellation
1750 * scheme. It may fail to remove tasks that have been converted
1751 * into other forms before being placed on the internal queue.
1752 * For example, a task entered using {@code submit} might be
1753 * converted into a form that maintains {@code Future} status.
1754 * However, in such cases, method {@link #purge} may be used to
1755 * remove those Futures that have been cancelled.
1756 *
1757 * @param task the task to remove
1758 * @return {@code true} if the task was removed
1759 */
1760 public boolean remove(Runnable task) {
1761 boolean removed = workQueue.remove(task);
1762 tryTerminate(); // In case SHUTDOWN and now empty
1763 return removed;
1764 }
1765
1766 /**
1767 * Tries to remove from the work queue all {@link Future}
1768 * tasks that have been cancelled. This method can be useful as a
1769 * storage reclamation operation, that has no other impact on
1770 * functionality. Cancelled tasks are never executed, but may
1771 * accumulate in work queues until worker threads can actively
1772 * remove them. Invoking this method instead tries to remove them now.
1773 * However, this method may fail to remove tasks in
1774 * the presence of interference by other threads.
1775 */
1776 public void purge() {
1777 final BlockingQueue<Runnable> q = workQueue;
1778 try {
1779 Iterator<Runnable> it = q.iterator();
1780 while (it.hasNext()) {
1781 Runnable r = it.next();
1782 if (r instanceof Future<?> && ((Future<?>)r).isCancelled())
1783 it.remove();
1784 }
1785 } catch (ConcurrentModificationException fallThrough) {
1786 // Take slow path if we encounter interference during traversal.
1787 // Make copy for traversal and call remove for cancelled entries.
1788 // The slow path is more likely to be O(N*N).
1789 for (Object r : q.toArray())
1790 if (r instanceof Future<?> && ((Future<?>)r).isCancelled())
1791 q.remove(r);
1792 }
1793
1794 tryTerminate(); // In case SHUTDOWN and now empty
1795 }
1796
1797 /* Statistics */
1798
1799 /**
1800 * Returns the current number of threads in the pool.
1801 *
1802 * @return the number of threads
1803 */
1804 public int getPoolSize() {
1805 final ReentrantLock mainLock = this.mainLock;
1806 mainLock.lock();
1807 try {
1808 // Remove rare and surprising possibility of
1809 // isTerminated() && getPoolSize() > 0
1810 return runStateAtLeast(ctl.get(), TIDYING) ? 0
1811 : workers.size();
1812 } finally {
1813 mainLock.unlock();
1814 }
1815 }
1816
1817 /**
1818 * Returns the approximate number of threads that are actively
1819 * executing tasks.
1820 *
1821 * @return the number of threads
1822 */
1823 public int getActiveCount() {
1824 final ReentrantLock mainLock = this.mainLock;
1825 mainLock.lock();
1826 try {
1827 int n = 0;
1828 for (Worker w : workers)
1829 if (w.isLocked())
1830 ++n;
1831 return n;
1832 } finally {
1833 mainLock.unlock();
1834 }
1835 }
1836
1837 /**
1838 * Returns the largest number of threads that have ever
1839 * simultaneously been in the pool.
1840 *
1841 * @return the number of threads
1842 */
1843 public int getLargestPoolSize() {
1844 final ReentrantLock mainLock = this.mainLock;
1845 mainLock.lock();
1846 try {
1847 return largestPoolSize;
1848 } finally {
1849 mainLock.unlock();
1850 }
1851 }
1852
1853 /**
1854 * Returns the approximate total number of tasks that have ever been
1855 * scheduled for execution. Because the states of tasks and
1856 * threads may change dynamically during computation, the returned
1857 * value is only an approximation.
1858 *
1859 * @return the number of tasks
1860 */
1861 public long getTaskCount() {
1862 final ReentrantLock mainLock = this.mainLock;
1863 mainLock.lock();
1864 try {
1865 long n = completedTaskCount;
1866 for (Worker w : workers) {
1867 n += w.completedTasks;
1868 if (w.isLocked())
1869 ++n;
1870 }
1871 return n + workQueue.size();
1872 } finally {
1873 mainLock.unlock();
1874 }
1875 }
1876
1877 /**
1878 * Returns the approximate total number of tasks that have
1879 * completed execution. Because the states of tasks and threads
1880 * may change dynamically during computation, the returned value
1881 * is only an approximation, but one that does not ever decrease
1882 * across successive calls.
1883 *
1884 * @return the number of tasks
1885 */
1886 public long getCompletedTaskCount() {
1887 final ReentrantLock mainLock = this.mainLock;
1888 mainLock.lock();
1889 try {
1890 long n = completedTaskCount;
1891 for (Worker w : workers)
1892 n += w.completedTasks;
1893 return n;
1894 } finally {
1895 mainLock.unlock();
1896 }
1897 }
1898
1899 /**
1900 * Returns a string identifying this pool, as well as its state,
1901 * including indications of run state and estimated worker and
1902 * task counts.
1903 *
1904 * @return a string identifying this pool, as well as its state
1905 */
1906 public String toString() {
1907 long ncompleted;
1908 int nworkers, nactive;
1909 final ReentrantLock mainLock = this.mainLock;
1910 mainLock.lock();
1911 try {
1912 ncompleted = completedTaskCount;
1913 nactive = 0;
1914 nworkers = workers.size();
1915 for (Worker w : workers) {
1916 ncompleted += w.completedTasks;
1917 if (w.isLocked())
1918 ++nactive;
1919 }
1920 } finally {
1921 mainLock.unlock();
1922 }
1923 int c = ctl.get();
1924 String runState =
1925 isRunning(c) ? "Running" :
1926 runStateAtLeast(c, TERMINATED) ? "Terminated" :
1927 "Shutting down";
1928 return super.toString() +
1929 "[" + runState +
1930 ", pool size = " + nworkers +
1931 ", active threads = " + nactive +
1932 ", queued tasks = " + workQueue.size() +
1933 ", completed tasks = " + ncompleted +
1934 "]";
1935 }
1936
1937 /* Extension hooks */
1938
1939 /**
1940 * Method invoked prior to executing the given Runnable in the
1941 * given thread. This method is invoked by thread {@code t} that
1942 * will execute task {@code r}, and may be used to re-initialize
1943 * ThreadLocals, or to perform logging.
1944 *
1945 * <p>This implementation does nothing, but may be customized in
1946 * subclasses. Note: To properly nest multiple overridings, subclasses
1947 * should generally invoke {@code super.beforeExecute} at the end of
1948 * this method.
1949 *
1950 * @param t the thread that will run task {@code r}
1951 * @param r the task that will be executed
1952 */
1953 protected void beforeExecute(Thread t, Runnable r) { }
1954
1955 /**
1956 * Method invoked upon completion of execution of the given Runnable.
1957 * This method is invoked by the thread that executed the task. If
1958 * non-null, the Throwable is the uncaught {@code RuntimeException}
1959 * or {@code Error} that caused execution to terminate abruptly.
1960 *
1961 * <p>This implementation does nothing, but may be customized in
1962 * subclasses. Note: To properly nest multiple overridings, subclasses
1963 * should generally invoke {@code super.afterExecute} at the
1964 * beginning of this method.
1965 *
1966 * <p><b>Note:</b> When actions are enclosed in tasks (such as
1967 * {@link FutureTask}) either explicitly or via methods such as
1968 * {@code submit}, these task objects catch and maintain
1969 * computational exceptions, and so they do not cause abrupt
1970 * termination, and the internal exceptions are <em>not</em>
1971 * passed to this method. If you would like to trap both kinds of
1972 * failures in this method, you can further probe for such cases,
1973 * as in this sample subclass that prints either the direct cause
1974 * or the underlying exception if a task has been aborted:
1975 *
1976 * <pre> {@code
1977 * class ExtendedExecutor extends ThreadPoolExecutor {
1978 * // ...
1979 * protected void afterExecute(Runnable r, Throwable t) {
1980 * super.afterExecute(r, t);
1981 * if (t == null
1982 * && r instanceof Future<?>
1983 * && ((Future<?>)r).isDone()) {
1984 * try {
1985 * Object result = ((Future<?>) r).get();
1986 * } catch (CancellationException ce) {
1987 * t = ce;
1988 * } catch (ExecutionException ee) {
1989 * t = ee.getCause();
1990 * } catch (InterruptedException ie) {
1991 * // ignore/reset
1992 * Thread.currentThread().interrupt();
1993 * }
1994 * }
1995 * if (t != null)
1996 * System.out.println(t);
1997 * }
1998 * }}</pre>
1999 *
2000 * @param r the runnable that has completed
2001 * @param t the exception that caused termination, or null if
2002 * execution completed normally
2003 */
2004 protected void afterExecute(Runnable r, Throwable t) { }
2005
2006 /**
2007 * Method invoked when the Executor has terminated. Default
2008 * implementation does nothing. Note: To properly nest multiple
2009 * overridings, subclasses should generally invoke
2010 * {@code super.terminated} within this method.
2011 */
2012 protected void terminated() { }
2013
2014 /* Predefined RejectedExecutionHandlers */
2015
2016 /**
2017 * A handler for rejected tasks that runs the rejected task
2018 * directly in the calling thread of the {@code execute} method,
2019 * unless the executor has been shut down, in which case the task
2020 * is discarded.
2021 */
2022 public static class CallerRunsPolicy implements RejectedExecutionHandler {
2023 /**
2024 * Creates a {@code CallerRunsPolicy}.
2025 */
2026 public CallerRunsPolicy() { }
2027
2028 /**
2029 * Executes task r in the caller's thread, unless the executor
2030 * has been shut down, in which case the task is discarded.
2031 *
2032 * @param r the runnable task requested to be executed
2033 * @param e the executor attempting to execute this task
2034 */
2035 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
2036 if (!e.isShutdown()) {
2037 r.run();
2038 }
2039 }
2040 }
2041
2042 /**
2043 * A handler for rejected tasks that throws a
2044 * {@link RejectedExecutionException}.
2045 *
2046 * This is the default handler for {@link ThreadPoolExecutor} and
2047 * {@link ScheduledThreadPoolExecutor}.
2048 */
2049 public static class AbortPolicy implements RejectedExecutionHandler {
2050 /**
2051 * Creates an {@code AbortPolicy}.
2052 */
2053 public AbortPolicy() { }
2054
2055 /**
2056 * Always throws RejectedExecutionException.
2057 *
2058 * @param r the runnable task requested to be executed
2059 * @param e the executor attempting to execute this task
2060 * @throws RejectedExecutionException always
2061 */
2062 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
2063 throw new RejectedExecutionException("Task " + r.toString() +
2064 " rejected from " +
2065 e.toString());
2066 }
2067 }
2068
2069 /**
2070 * A handler for rejected tasks that silently discards the
2071 * rejected task.
2072 */
2073 public static class DiscardPolicy implements RejectedExecutionHandler {
2074 /**
2075 * Creates a {@code DiscardPolicy}.
2076 */
2077 public DiscardPolicy() { }
2078
2079 /**
2080 * Does nothing, which has the effect of discarding task r.
2081 *
2082 * @param r the runnable task requested to be executed
2083 * @param e the executor attempting to execute this task
2084 */
2085 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
2086 }
2087 }
2088
2089 /**
2090 * A handler for rejected tasks that discards the oldest unhandled
2091 * request and then retries {@code execute}, unless the executor
2092 * is shut down, in which case the task is discarded. This policy is
2093 * rarely useful in cases where other threads may be waiting for
2094 * tasks to terminate, or failures must be recorded. Instead consider
2095 * using a handler of the form:
2096 * <pre> {@code
2097 * new RejectedExecutionHandler() {
2098 * public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
2099 * Runnable dropped = e.getQueue().poll();
2100 * if (dropped instanceof Future<?>) {
2101 * ((Future<?>)dropped).cancel(false);
2102 * // also consider logging the failure
2103 * }
2104 * e.execute(r); // retry
2105 * }}}</pre>
2106 */
2107 public static class DiscardOldestPolicy implements RejectedExecutionHandler {
2108 /**
2109 * Creates a {@code DiscardOldestPolicy} for the given executor.
2110 */
2111 public DiscardOldestPolicy() { }
2112
2113 /**
2114 * Obtains and ignores the next task that the executor
2115 * would otherwise execute, if one is immediately available,
2116 * and then retries execution of task r, unless the executor
2117 * is shut down, in which case task r is instead discarded.
2118 *
2119 * @param r the runnable task requested to be executed
2120 * @param e the executor attempting to execute this task
2121 */
2122 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
2123 if (!e.isShutdown()) {
2124 e.getQueue().poll();
2125 e.execute(r);
2126 }
2127 }
2128 }
2129 }