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