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