ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/ThreadPoolExecutor.java
Revision: 1.80
Committed: Thu Apr 20 07:20:42 2006 UTC (18 years, 1 month ago) by jsr166
Branch: MAIN
Changes since 1.79: +2 -2 lines
Log Message:
whitespace

File Contents

# Content
1 /*
2 * Written by Doug Lea with assistance from members of JCP JSR-166
3 * Expert Group and released to the public domain, as explained at
4 * http://creativecommons.org/licenses/publicdomain
5 */
6
7 package java.util.concurrent;
8 import java.util.concurrent.locks.*;
9 import java.util.*;
10
11 /**
12 * An {@link ExecutorService} that executes each submitted task using
13 * one of possibly several pooled threads, normally configured
14 * using {@link Executors} factory methods.
15 *
16 * <p>Thread pools address two different problems: they usually
17 * provide improved performance when executing large numbers of
18 * asynchronous tasks, due to reduced per-task invocation overhead,
19 * and they provide a means of bounding and managing the resources,
20 * including threads, consumed when executing a collection of tasks.
21 * Each <tt>ThreadPoolExecutor</tt> also maintains some basic
22 * statistics, such as the number of completed tasks.
23 *
24 * <p>To be useful across a wide range of contexts, this class
25 * provides many adjustable parameters and extensibility
26 * hooks. However, programmers are urged to use the more convenient
27 * {@link Executors} factory methods {@link
28 * Executors#newCachedThreadPool} (unbounded thread pool, with
29 * automatic thread reclamation), {@link Executors#newFixedThreadPool}
30 * (fixed size thread pool) and {@link
31 * Executors#newSingleThreadExecutor} (single background thread), that
32 * preconfigure settings for the most common usage
33 * scenarios. Otherwise, use the following guide when manually
34 * configuring and tuning this class:
35 *
36 * <dl>
37 *
38 * <dt>Core and maximum pool sizes</dt>
39 *
40 * <dd>A <tt>ThreadPoolExecutor</tt> will automatically adjust the
41 * pool size
42 * (see {@link ThreadPoolExecutor#getPoolSize})
43 * according to the bounds set by corePoolSize
44 * (see {@link ThreadPoolExecutor#getCorePoolSize})
45 * and
46 * maximumPoolSize
47 * (see {@link ThreadPoolExecutor#getMaximumPoolSize}).
48 * When a new task is submitted in method {@link
49 * ThreadPoolExecutor#execute}, and fewer than corePoolSize threads
50 * are running, a new thread is created to handle the request, even if
51 * other worker threads are idle. If there are more than
52 * corePoolSize but less than maximumPoolSize threads running, a new
53 * thread will be created only if the queue is full. By setting
54 * corePoolSize and maximumPoolSize the same, you create a fixed-size
55 * thread pool. By setting maximumPoolSize to an essentially unbounded
56 * value such as <tt>Integer.MAX_VALUE</tt>, you allow the pool to
57 * accommodate an arbitrary number of concurrent tasks. Most typically,
58 * core and maximum pool sizes are set only upon construction, but they
59 * may also be changed dynamically using {@link
60 * ThreadPoolExecutor#setCorePoolSize} and {@link
61 * ThreadPoolExecutor#setMaximumPoolSize}. <dd>
62 *
63 * <dt> On-demand construction
64 *
65 * <dd> By default, even core threads are initially created and
66 * started only when new tasks arrive, but this can be overridden
67 * dynamically using method {@link
68 * ThreadPoolExecutor#prestartCoreThread} or
69 * {@link ThreadPoolExecutor#prestartAllCoreThreads}.
70 * You probably want to prestart threads if you construct the
71 * pool with a non-empty queue. </dd>
72 *
73 * <dt>Creating new threads</dt>
74 *
75 * <dd>New threads are created using a {@link
76 * java.util.concurrent.ThreadFactory}. If not otherwise specified, a
77 * {@link Executors#defaultThreadFactory} is used, that creates threads to all
78 * be in the same {@link ThreadGroup} and with the same
79 * <tt>NORM_PRIORITY</tt> priority and non-daemon status. By supplying
80 * a different ThreadFactory, you can alter the thread's name, thread
81 * group, priority, daemon status, etc. If a <tt>ThreadFactory</tt> fails to create
82 * a thread when asked by returning null from <tt>newThread</tt>,
83 * the executor will continue, but might
84 * not be able to execute any tasks. </dd>
85 *
86 * <dt>Keep-alive times</dt>
87 *
88 * <dd>If the pool currently has more than corePoolSize threads,
89 * excess threads will be terminated if they have been idle for more
90 * than the keepAliveTime (see {@link
91 * ThreadPoolExecutor#getKeepAliveTime}). This provides a means of
92 * reducing resource consumption when the pool is not being actively
93 * used. If the pool becomes more active later, new threads will be
94 * constructed. This parameter can also be changed dynamically using
95 * method {@link ThreadPoolExecutor#setKeepAliveTime}. Using a value
96 * of <tt>Long.MAX_VALUE</tt> {@link TimeUnit#NANOSECONDS} effectively
97 * disables idle threads from ever terminating prior to shut down. By
98 * default, the keep-alive policy applies only when there are more
99 * than corePoolSizeThreads. But method {@link
100 * ThreadPoolExecutor#allowCoreThreadTimeOut} can be used to apply
101 * this time-out policy to core threads as well, so long as
102 * the keepAliveTime value is non-zero. </dd>
103 *
104 * <dt>Queuing</dt>
105 *
106 * <dd>Any {@link BlockingQueue} may be used to transfer and hold
107 * submitted tasks. The use of this queue interacts with pool sizing:
108 *
109 * <ul>
110 *
111 * <li> If fewer than corePoolSize threads are running, the Executor
112 * always prefers adding a new thread
113 * rather than queuing.</li>
114 *
115 * <li> If corePoolSize or more threads are running, the Executor
116 * always prefers queuing a request rather than adding a new
117 * thread.</li>
118 *
119 * <li> If a request cannot be queued, a new thread is created unless
120 * this would exceed maximumPoolSize, in which case, the task will be
121 * rejected.</li>
122 *
123 * </ul>
124 *
125 * There are three general strategies for queuing:
126 * <ol>
127 *
128 * <li> <em> Direct handoffs.</em> A good default choice for a work
129 * queue is a {@link SynchronousQueue} that hands off tasks to threads
130 * without otherwise holding them. Here, an attempt to queue a task
131 * will fail if no threads are immediately available to run it, so a
132 * new thread will be constructed. This policy avoids lockups when
133 * handling sets of requests that might have internal dependencies.
134 * Direct handoffs generally require unbounded maximumPoolSizes to
135 * avoid rejection of new submitted tasks. This in turn admits the
136 * possibility of unbounded thread growth when commands continue to
137 * arrive on average faster than they can be processed. </li>
138 *
139 * <li><em> Unbounded queues.</em> Using an unbounded queue (for
140 * example a {@link LinkedBlockingQueue} without a predefined
141 * capacity) will cause new tasks to wait in the queue when all
142 * corePoolSize threads are busy. Thus, no more than corePoolSize
143 * threads will ever be created. (And the value of the maximumPoolSize
144 * therefore doesn't have any effect.) This may be appropriate when
145 * each task is completely independent of others, so tasks cannot
146 * affect each others execution; for example, in a web page server.
147 * While this style of queuing can be useful in smoothing out
148 * transient bursts of requests, it admits the possibility of
149 * unbounded work queue growth when commands continue to arrive on
150 * average faster than they can be processed. </li>
151 *
152 * <li><em>Bounded queues.</em> A bounded queue (for example, an
153 * {@link ArrayBlockingQueue}) helps prevent resource exhaustion when
154 * used with finite maximumPoolSizes, but can be more difficult to
155 * tune and control. Queue sizes and maximum pool sizes may be traded
156 * off for each other: Using large queues and small pools minimizes
157 * CPU usage, OS resources, and context-switching overhead, but can
158 * lead to artificially low throughput. If tasks frequently block (for
159 * example if they are I/O bound), a system may be able to schedule
160 * time for more threads than you otherwise allow. Use of small queues
161 * generally requires larger pool sizes, which keeps CPUs busier but
162 * may encounter unacceptable scheduling overhead, which also
163 * decreases throughput. </li>
164 *
165 * </ol>
166 *
167 * </dd>
168 *
169 * <dt>Rejected tasks</dt>
170 *
171 * <dd> New tasks submitted in method {@link
172 * ThreadPoolExecutor#execute} will be <em>rejected</em> when the
173 * Executor has been shut down, and also when the Executor uses finite
174 * bounds for both maximum threads and work queue capacity, and is
175 * saturated. In either case, the <tt>execute</tt> method invokes the
176 * {@link RejectedExecutionHandler#rejectedExecution} method of its
177 * {@link RejectedExecutionHandler}. Four predefined handler policies
178 * are provided:
179 *
180 * <ol>
181 *
182 * <li> In the
183 * default {@link ThreadPoolExecutor.AbortPolicy}, the handler throws a
184 * runtime {@link RejectedExecutionException} upon rejection. </li>
185 *
186 * <li> In {@link
187 * ThreadPoolExecutor.CallerRunsPolicy}, the thread that invokes
188 * <tt>execute</tt> itself runs the task. This provides a simple
189 * feedback control mechanism that will slow down the rate that new
190 * tasks are submitted. </li>
191 *
192 * <li> In {@link ThreadPoolExecutor.DiscardPolicy},
193 * a task that cannot be executed is simply dropped. </li>
194 *
195 * <li>In {@link
196 * ThreadPoolExecutor.DiscardOldestPolicy}, if the executor is not
197 * shut down, the task at the head of the work queue is dropped, and
198 * then execution is retried (which can fail again, causing this to be
199 * repeated.) </li>
200 *
201 * </ol>
202 *
203 * It is possible to define and use other kinds of {@link
204 * RejectedExecutionHandler} classes. Doing so requires some care
205 * especially when policies are designed to work only under particular
206 * capacity or queuing policies. </dd>
207 *
208 * <dt>Hook methods</dt>
209 *
210 * <dd>This class provides <tt>protected</tt> overridable {@link
211 * ThreadPoolExecutor#beforeExecute} and {@link
212 * ThreadPoolExecutor#afterExecute} methods that are called before and
213 * after execution of each task. These can be used to manipulate the
214 * execution environment; for example, reinitializing ThreadLocals,
215 * gathering statistics, or adding log entries. Additionally, method
216 * {@link ThreadPoolExecutor#terminated} can be overridden to perform
217 * any special processing that needs to be done once the Executor has
218 * fully terminated.
219 *
220 * <p>If hook or callback methods throw
221 * exceptions, internal worker threads may in turn fail and
222 * abruptly terminate.</dd>
223 *
224 * <dt>Queue maintenance</dt>
225 *
226 * <dd> Method {@link ThreadPoolExecutor#getQueue} allows access to
227 * the work queue for purposes of monitoring and debugging. Use of
228 * this method for any other purpose is strongly discouraged. Two
229 * supplied methods, {@link ThreadPoolExecutor#remove} and {@link
230 * ThreadPoolExecutor#purge} are available to assist in storage
231 * reclamation when large numbers of queued tasks become
232 * cancelled.</dd>
233 *
234 * <dt>Finalization</dt>
235 *
236 * <dd> A pool that is no longer referenced in a program <em>AND</em>
237 * has no remaining threads will be <tt>shutdown</tt>
238 * automatically. If you would like to ensure that unreferenced pools
239 * are reclaimed even if users forget to call {@link
240 * ThreadPoolExecutor#shutdown}, then you must arrange that unused
241 * threads eventually die, by setting appropriate keep-alive times,
242 * using a lower bound of zero core threads and/or setting {@link
243 * ThreadPoolExecutor#allowCoreThreadTimeOut}. </dd> </dl>
244 *
245 * <p> <b>Extension example</b>. Most extensions of this class
246 * override one or more of the protected hook methods. For example,
247 * here is a subclass that adds a simple pause/resume feature:
248 *
249 * <pre>
250 * class PausableThreadPoolExecutor extends ThreadPoolExecutor {
251 * private boolean isPaused;
252 * private ReentrantLock pauseLock = new ReentrantLock();
253 * private Condition unpaused = pauseLock.newCondition();
254 *
255 * public PausableThreadPoolExecutor(...) { super(...); }
256 *
257 * protected void beforeExecute(Thread t, Runnable r) {
258 * super.beforeExecute(t, r);
259 * pauseLock.lock();
260 * try {
261 * while (isPaused) unpaused.await();
262 * } catch (InterruptedException ie) {
263 * t.interrupt();
264 * } finally {
265 * pauseLock.unlock();
266 * }
267 * }
268 *
269 * public void pause() {
270 * pauseLock.lock();
271 * try {
272 * isPaused = true;
273 * } finally {
274 * pauseLock.unlock();
275 * }
276 * }
277 *
278 * public void resume() {
279 * pauseLock.lock();
280 * try {
281 * isPaused = false;
282 * unpaused.signalAll();
283 * } finally {
284 * pauseLock.unlock();
285 * }
286 * }
287 * }
288 * </pre>
289 * @since 1.5
290 * @author Doug Lea
291 */
292 public class ThreadPoolExecutor extends AbstractExecutorService {
293 /**
294 * Only used to force toArray() to produce a Runnable[].
295 */
296 private static final Runnable[] EMPTY_RUNNABLE_ARRAY = new Runnable[0];
297
298 /**
299 * Permission for checking shutdown
300 */
301 private static final RuntimePermission shutdownPerm =
302 new RuntimePermission("modifyThread");
303
304 /**
305 * Queue used for holding tasks and handing off to worker threads.
306 */
307 private final BlockingQueue<Runnable> workQueue;
308
309 /**
310 * Lock held on updates to poolSize, corePoolSize, maximumPoolSize, and
311 * workers set.
312 */
313 private final ReentrantLock mainLock = new ReentrantLock();
314
315 /**
316 * Wait condition to support awaitTermination
317 */
318 private final Condition termination = mainLock.newCondition();
319
320 /**
321 * Set containing all worker threads in pool.
322 */
323 private final HashSet<Worker> workers = new HashSet<Worker>();
324
325 /**
326 * Timeout in nanoseconds for idle threads waiting for work.
327 * Threads use this timeout only when there are more than
328 * corePoolSize present. Otherwise they wait forever for new work.
329 */
330 private volatile long keepAliveTime;
331
332 /**
333 * If false (default) core threads stay alive even when idle.
334 * If true, core threads use keepAliveTime to time out waiting for work.
335 */
336 private boolean allowCoreThreadTimeOut;
337
338 /**
339 * Core pool size, updated only while holding mainLock,
340 * but volatile to allow concurrent readability even
341 * during updates.
342 */
343 private volatile int corePoolSize;
344
345 /**
346 * Maximum pool size, updated only while holding mainLock
347 * but volatile to allow concurrent readability even
348 * during updates.
349 */
350 private volatile int maximumPoolSize;
351
352 /**
353 * Current pool size, updated only while holding mainLock
354 * but volatile to allow concurrent readability even
355 * during updates.
356 */
357 private volatile int poolSize;
358
359 /**
360 * Lifecycle state
361 */
362 volatile int runState;
363
364 // Special values for runState
365 /** Normal, not-shutdown mode */
366 static final int RUNNING = 0;
367 /** Controlled shutdown mode */
368 static final int SHUTDOWN = 1;
369 /** Immediate shutdown mode */
370 static final int STOP = 2;
371 /** Final state */
372 static final int TERMINATED = 3;
373
374 /**
375 * Handler called when saturated or shutdown in execute.
376 */
377 private volatile RejectedExecutionHandler handler;
378
379 /**
380 * Factory for new threads.
381 */
382 private volatile ThreadFactory threadFactory;
383
384 /**
385 * Tracks largest attained pool size.
386 */
387 private int largestPoolSize;
388
389 /**
390 * Counter for completed tasks. Updated only on termination of
391 * worker threads.
392 */
393 private long completedTaskCount;
394
395 /**
396 * The default rejected execution handler
397 */
398 private static final RejectedExecutionHandler defaultHandler =
399 new AbortPolicy();
400
401 /**
402 * Invokes the rejected execution handler for the given command.
403 */
404 void reject(Runnable command) {
405 handler.rejectedExecution(command, this);
406 }
407
408 /**
409 * Creates and returns a new thread running firstTask as its first
410 * task. Call only while holding mainLock.
411 * @param firstTask the task the new thread should run first (or
412 * null if none)
413 * @return the new thread, or null if threadFactory fails to create thread
414 */
415 private Thread addThread(Runnable firstTask) {
416 Worker w = new Worker(firstTask);
417 Thread t = threadFactory.newThread(w);
418 if (t != null) {
419 w.thread = t;
420 workers.add(w);
421 int nt = ++poolSize;
422 if (nt > largestPoolSize)
423 largestPoolSize = nt;
424 }
425 return t;
426 }
427
428 /**
429 * Creates and starts a new thread running firstTask as its first
430 * task, only if fewer than corePoolSize threads are running.
431 * @param firstTask the task the new thread should run first (or
432 * null if none)
433 * @return true if successful.
434 */
435 private boolean addIfUnderCorePoolSize(Runnable firstTask) {
436 Thread t = null;
437 final ReentrantLock mainLock = this.mainLock;
438 mainLock.lock();
439 try {
440 if (poolSize < corePoolSize)
441 t = addThread(firstTask);
442 } finally {
443 mainLock.unlock();
444 }
445 if (t == null)
446 return false;
447 t.start();
448 return true;
449 }
450
451 /**
452 * Creates and starts a new thread only if fewer than maximumPoolSize
453 * threads are running. The new thread runs as its first task the
454 * next task in queue, or if there is none, the given task.
455 * @param firstTask the task the new thread should run first (or
456 * null if none)
457 * @return 0 if a new thread cannot be created, a positive number
458 * if firstTask will be run in a new thread, or a negative number
459 * if a new thread was created but is running some other task, in
460 * which case the caller must try some other way to run firstTask
461 * (perhaps by calling this method again).
462 */
463 private int addIfUnderMaximumPoolSize(Runnable firstTask) {
464 Thread t = null;
465 int status = 0;
466 final ReentrantLock mainLock = this.mainLock;
467 mainLock.lock();
468 try {
469 if (poolSize < maximumPoolSize) {
470 Runnable next = workQueue.poll();
471 if (next == null) {
472 next = firstTask;
473 status = 1;
474 } else
475 status = -1;
476 t = addThread(next);
477 }
478 } finally {
479 mainLock.unlock();
480 }
481 if (t == null)
482 return 0;
483 t.start();
484 return status;
485 }
486
487
488 /**
489 * Gets the next task for a worker thread to run.
490 * @return the task
491 */
492 Runnable getTask() {
493 for (;;) {
494 try {
495 switch (runState) {
496 case RUNNING: {
497 // untimed wait if core and not allowing core timeout
498 if (poolSize <= corePoolSize && !allowCoreThreadTimeOut)
499 return workQueue.take();
500
501 long timeout = keepAliveTime;
502 if (timeout <= 0) // die immediately for 0 timeout
503 return null;
504 Runnable r = workQueue.poll(timeout, TimeUnit.NANOSECONDS);
505 if (r != null)
506 return r;
507 if (poolSize > corePoolSize || allowCoreThreadTimeOut)
508 return null; // timed out
509 // Else, after timeout, the pool shrank. Retry
510 break;
511 }
512
513 case SHUTDOWN: {
514 // Help drain queue
515 Runnable r = workQueue.poll();
516 if (r != null)
517 return r;
518
519 // Check if can terminate
520 if (workQueue.isEmpty()) {
521 interruptIdleWorkers();
522 return null;
523 }
524
525 // Else there could still be delayed tasks in queue.
526 return workQueue.take();
527 }
528
529 case STOP:
530 return null;
531 default:
532 assert false;
533 }
534 } catch (InterruptedException ie) {
535 // On interruption, re-check runstate
536 }
537 }
538 }
539
540 /**
541 * Wakes up all threads that might be waiting for tasks.
542 */
543 void interruptIdleWorkers() {
544 final ReentrantLock mainLock = this.mainLock;
545 mainLock.lock();
546 try {
547 for (Worker w : workers)
548 w.interruptIfIdle();
549 } finally {
550 mainLock.unlock();
551 }
552 }
553
554 /**
555 * Performs bookkeeping for a terminated worker thread.
556 * @param w the worker
557 */
558 void workerDone(Worker w) {
559 final ReentrantLock mainLock = this.mainLock;
560 mainLock.lock();
561 try {
562 completedTaskCount += w.completedTasks;
563 workers.remove(w);
564 if (--poolSize > 0)
565 return;
566
567 // Else, this is the last thread. Deal with potential shutdown.
568
569 int state = runState;
570 assert state != TERMINATED;
571
572 if (state != STOP) {
573 // If there are queued tasks but no threads, create
574 // replacement thread. We must create it initially
575 // idle to avoid orphaned tasks in case addThread
576 // fails. This also handles case of delayed tasks
577 // that will sometime later become runnable.
578 if (!workQueue.isEmpty()) {
579 Thread t = addThread(null);
580 if (t != null)
581 t.start();
582 return;
583 }
584
585 // Otherwise, we can exit without replacement
586 if (state == RUNNING)
587 return;
588 }
589
590 // Either state is STOP, or state is SHUTDOWN and there is
591 // no work to do. So we can terminate.
592 termination.signalAll();
593 runState = TERMINATED;
594 // fall through to call terminate() outside of lock.
595 } finally {
596 mainLock.unlock();
597 }
598
599 assert runState == TERMINATED;
600 terminated();
601 }
602
603 /**
604 * Worker threads
605 */
606 private class Worker implements Runnable {
607
608 /**
609 * The runLock is acquired and released surrounding each task
610 * execution. It mainly protects against interrupts that are
611 * intended to cancel the worker thread from instead
612 * interrupting the task being run.
613 */
614 private final ReentrantLock runLock = new ReentrantLock();
615
616 /**
617 * Initial task to run before entering run loop
618 */
619 private Runnable firstTask;
620
621 /**
622 * Per thread completed task counter; accumulated
623 * into completedTaskCount upon termination.
624 */
625 volatile long completedTasks;
626
627 /**
628 * Thread this worker is running in. Acts as a final field,
629 * but cannot be set until thread is created.
630 */
631 Thread thread;
632
633 Worker(Runnable firstTask) {
634 this.firstTask = firstTask;
635 }
636
637 boolean isActive() {
638 return runLock.isLocked();
639 }
640
641 /**
642 * Interrupts thread if not running a task.
643 */
644 void interruptIfIdle() {
645 final ReentrantLock runLock = this.runLock;
646 if (runLock.tryLock()) {
647 try {
648 thread.interrupt();
649 } finally {
650 runLock.unlock();
651 }
652 }
653 }
654
655 /**
656 * Interrupts thread even if running a task.
657 */
658 void interruptNow() {
659 thread.interrupt();
660 }
661
662 /**
663 * Runs a single task between before/after methods.
664 */
665 private void runTask(Runnable task) {
666 final ReentrantLock runLock = this.runLock;
667 runLock.lock();
668 try {
669 Thread.interrupted(); // clear interrupt status on entry
670 // Abort now if immediate cancel. Otherwise, we have
671 // committed to run this task.
672 if (runState == STOP)
673 return;
674
675 boolean ran = false;
676 beforeExecute(thread, task);
677 try {
678 task.run();
679 ran = true;
680 afterExecute(task, null);
681 ++completedTasks;
682 } catch (RuntimeException ex) {
683 if (!ran)
684 afterExecute(task, ex);
685 // Else the exception occurred within
686 // afterExecute itself in which case we don't
687 // want to call it again.
688 throw ex;
689 }
690 } finally {
691 runLock.unlock();
692 }
693 }
694
695 /**
696 * Main run loop
697 */
698 public void run() {
699 try {
700 Runnable task = firstTask;
701 firstTask = null;
702 while (task != null || (task = getTask()) != null) {
703 runTask(task);
704 task = null; // unnecessary but can help GC
705 }
706 } finally {
707 workerDone(this);
708 }
709 }
710 }
711
712 // Public methods
713
714 /**
715 * Creates a new <tt>ThreadPoolExecutor</tt> with the given initial
716 * parameters and default thread factory and rejected execution handler.
717 * It may be more convenient to use one of the {@link Executors} factory
718 * methods instead of this general purpose constructor.
719 *
720 * @param corePoolSize the number of threads to keep in the
721 * pool, even if they are idle.
722 * @param maximumPoolSize the maximum number of threads to allow in the
723 * pool.
724 * @param keepAliveTime when the number of threads is greater than
725 * the core, this is the maximum time that excess idle threads
726 * will wait for new tasks before terminating.
727 * @param unit the time unit for the keepAliveTime
728 * argument.
729 * @param workQueue the queue to use for holding tasks before they
730 * are executed. This queue will hold only the <tt>Runnable</tt>
731 * tasks submitted by the <tt>execute</tt> method.
732 * @throws IllegalArgumentException if corePoolSize, or
733 * keepAliveTime less than zero, or if maximumPoolSize less than or
734 * equal to zero, or if corePoolSize greater than maximumPoolSize.
735 * @throws NullPointerException if <tt>workQueue</tt> is null
736 */
737 public ThreadPoolExecutor(int corePoolSize,
738 int maximumPoolSize,
739 long keepAliveTime,
740 TimeUnit unit,
741 BlockingQueue<Runnable> workQueue) {
742 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
743 Executors.defaultThreadFactory(), defaultHandler);
744 }
745
746 /**
747 * Creates a new <tt>ThreadPoolExecutor</tt> with the given initial
748 * parameters and default rejected execution handler.
749 *
750 * @param corePoolSize the number of threads to keep in the
751 * pool, even if they are idle.
752 * @param maximumPoolSize the maximum number of threads to allow in the
753 * pool.
754 * @param keepAliveTime when the number of threads is greater than
755 * the core, this is the maximum time that excess idle threads
756 * will wait for new tasks before terminating.
757 * @param unit the time unit for the keepAliveTime
758 * argument.
759 * @param workQueue the queue to use for holding tasks before they
760 * are executed. This queue will hold only the <tt>Runnable</tt>
761 * tasks submitted by the <tt>execute</tt> method.
762 * @param threadFactory the factory to use when the executor
763 * creates a new thread.
764 * @throws IllegalArgumentException if corePoolSize, or
765 * keepAliveTime less than zero, or if maximumPoolSize less than or
766 * equal to zero, or if corePoolSize greater than maximumPoolSize.
767 * @throws NullPointerException if <tt>workQueue</tt>
768 * or <tt>threadFactory</tt> are null.
769 */
770 public ThreadPoolExecutor(int corePoolSize,
771 int maximumPoolSize,
772 long keepAliveTime,
773 TimeUnit unit,
774 BlockingQueue<Runnable> workQueue,
775 ThreadFactory threadFactory) {
776 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
777 threadFactory, defaultHandler);
778 }
779
780 /**
781 * Creates a new <tt>ThreadPoolExecutor</tt> with the given initial
782 * parameters and default thread factory.
783 *
784 * @param corePoolSize the number of threads to keep in the
785 * pool, even if they are idle.
786 * @param maximumPoolSize the maximum number of threads to allow in the
787 * pool.
788 * @param keepAliveTime when the number of threads is greater than
789 * the core, this is the maximum time that excess idle threads
790 * will wait for new tasks before terminating.
791 * @param unit the time unit for the keepAliveTime
792 * argument.
793 * @param workQueue the queue to use for holding tasks before they
794 * are executed. This queue will hold only the <tt>Runnable</tt>
795 * tasks submitted by the <tt>execute</tt> method.
796 * @param handler the handler to use when execution is blocked
797 * because the thread bounds and queue capacities are reached.
798 * @throws IllegalArgumentException if corePoolSize, or
799 * keepAliveTime less than zero, or if maximumPoolSize less than or
800 * equal to zero, or if corePoolSize greater than maximumPoolSize.
801 * @throws NullPointerException if <tt>workQueue</tt>
802 * or <tt>handler</tt> are null.
803 */
804 public ThreadPoolExecutor(int corePoolSize,
805 int maximumPoolSize,
806 long keepAliveTime,
807 TimeUnit unit,
808 BlockingQueue<Runnable> workQueue,
809 RejectedExecutionHandler handler) {
810 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
811 Executors.defaultThreadFactory(), handler);
812 }
813
814 /**
815 * Creates a new <tt>ThreadPoolExecutor</tt> with the given initial
816 * parameters.
817 *
818 * @param corePoolSize the number of threads to keep in the
819 * pool, even if they are idle.
820 * @param maximumPoolSize the maximum number of threads to allow in the
821 * pool.
822 * @param keepAliveTime when the number of threads is greater than
823 * the core, this is the maximum time that excess idle threads
824 * will wait for new tasks before terminating.
825 * @param unit the time unit for the keepAliveTime
826 * argument.
827 * @param workQueue the queue to use for holding tasks before they
828 * are executed. This queue will hold only the <tt>Runnable</tt>
829 * tasks submitted by the <tt>execute</tt> method.
830 * @param threadFactory the factory to use when the executor
831 * creates a new thread.
832 * @param handler the handler to use when execution is blocked
833 * because the thread bounds and queue capacities are reached.
834 * @throws IllegalArgumentException if corePoolSize, or
835 * keepAliveTime less than zero, or if maximumPoolSize less than or
836 * equal to zero, or if corePoolSize greater than maximumPoolSize.
837 * @throws NullPointerException if <tt>workQueue</tt>
838 * or <tt>threadFactory</tt> or <tt>handler</tt> are null.
839 */
840 public ThreadPoolExecutor(int corePoolSize,
841 int maximumPoolSize,
842 long keepAliveTime,
843 TimeUnit unit,
844 BlockingQueue<Runnable> workQueue,
845 ThreadFactory threadFactory,
846 RejectedExecutionHandler handler) {
847 if (corePoolSize < 0 ||
848 maximumPoolSize <= 0 ||
849 maximumPoolSize < corePoolSize ||
850 keepAliveTime < 0)
851 throw new IllegalArgumentException();
852 if (workQueue == null || threadFactory == null || handler == null)
853 throw new NullPointerException();
854 this.corePoolSize = corePoolSize;
855 this.maximumPoolSize = maximumPoolSize;
856 this.workQueue = workQueue;
857 this.keepAliveTime = unit.toNanos(keepAliveTime);
858 this.threadFactory = threadFactory;
859 this.handler = handler;
860 }
861
862
863 /**
864 * Executes the given task sometime in the future. The task
865 * may execute in a new thread or in an existing pooled thread.
866 *
867 * If the task cannot be submitted for execution, either because this
868 * executor has been shutdown or because its capacity has been reached,
869 * the task is handled by the current <tt>RejectedExecutionHandler</tt>.
870 *
871 * @param command the task to execute
872 * @throws RejectedExecutionException at discretion of
873 * <tt>RejectedExecutionHandler</tt>, if task cannot be accepted
874 * for execution
875 * @throws NullPointerException if command is null
876 */
877 public void execute(Runnable command) {
878 if (command == null)
879 throw new NullPointerException();
880 for (;;) {
881 if (runState != RUNNING) {
882 reject(command);
883 return;
884 }
885 if (poolSize < corePoolSize && addIfUnderCorePoolSize(command))
886 return;
887 if (workQueue.offer(command))
888 return;
889 int status = addIfUnderMaximumPoolSize(command);
890 if (status > 0) // created new thread
891 return;
892 if (status == 0) { // failed to create thread
893 reject(command);
894 return;
895 }
896 // Retry if created a new thread but it is busy with another task
897 }
898 }
899
900 /**
901 * Initiates an orderly shutdown in which previously submitted
902 * tasks are executed, but no new tasks will be
903 * accepted. Invocation has no additional effect if already shut
904 * down.
905 * @throws SecurityException if a security manager exists and
906 * shutting down this ExecutorService may manipulate threads that
907 * the caller is not permitted to modify because it does not hold
908 * {@link java.lang.RuntimePermission}<tt>("modifyThread")</tt>,
909 * or the security manager's <tt>checkAccess</tt> method denies access.
910 */
911 public void shutdown() {
912 // Fail if caller doesn't have modifyThread permission.
913 SecurityManager security = System.getSecurityManager();
914 if (security != null)
915 security.checkPermission(shutdownPerm);
916
917 boolean fullyTerminated = false;
918 final ReentrantLock mainLock = this.mainLock;
919 mainLock.lock();
920 try {
921 if (workers.size() > 0) {
922 // Check if caller can modify worker threads. This
923 // might not be true even if passed above check, if
924 // the SecurityManager treats some threads specially.
925 if (security != null) {
926 for (Worker w: workers)
927 security.checkAccess(w.thread);
928 }
929
930 int state = runState;
931 if (state == RUNNING) // don't override shutdownNow
932 runState = SHUTDOWN;
933
934 try {
935 for (Worker w: workers)
936 w.interruptIfIdle();
937 } catch (SecurityException se) {
938 // If SecurityManager allows above checks, but
939 // then unexpectedly throws exception when
940 // interrupting threads (which it ought not do),
941 // back out as cleanly as we can. Some threads may
942 // have been killed but we remain in non-shutdown
943 // state.
944 runState = state;
945 throw se;
946 }
947 }
948 else { // If no workers, trigger full termination now
949 fullyTerminated = true;
950 runState = TERMINATED;
951 termination.signalAll();
952 }
953 } finally {
954 mainLock.unlock();
955 }
956 if (fullyTerminated)
957 terminated();
958 }
959
960
961 /**
962 * Attempts to stop all actively executing tasks, halts the
963 * processing of waiting tasks, and returns a list of the tasks
964 * that were awaiting execution.
965 *
966 * <p>There are no guarantees beyond best-effort attempts to stop
967 * processing actively executing tasks. This implementation
968 * cancels tasks via {@link Thread#interrupt}, so any task that
969 * fails to respond to interrupts may never terminate.
970 *
971 * @return list of tasks that never commenced execution
972 * @throws SecurityException if a security manager exists and
973 * shutting down this ExecutorService may manipulate threads that
974 * the caller is not permitted to modify because it does not hold
975 * {@link java.lang.RuntimePermission}<tt>("modifyThread")</tt>,
976 * or the security manager's <tt>checkAccess</tt> method denies access.
977 */
978 public List<Runnable> shutdownNow() {
979 // Almost the same code as shutdown()
980 SecurityManager security = System.getSecurityManager();
981 if (security != null)
982 security.checkPermission(shutdownPerm);
983
984 boolean fullyTerminated = false;
985 final ReentrantLock mainLock = this.mainLock;
986 mainLock.lock();
987 try {
988 if (workers.size() > 0) {
989 if (security != null) {
990 for (Worker w: workers)
991 security.checkAccess(w.thread);
992 }
993
994 int state = runState;
995 if (state != TERMINATED)
996 runState = STOP;
997 try {
998 for (Worker w : workers)
999 w.interruptNow();
1000 } catch (SecurityException se) {
1001 runState = state; // back out;
1002 throw se;
1003 }
1004 }
1005 else { // If no workers, trigger full termination now
1006 fullyTerminated = true;
1007 runState = TERMINATED;
1008 termination.signalAll();
1009 }
1010 } finally {
1011 mainLock.unlock();
1012 }
1013 if (fullyTerminated)
1014 terminated();
1015 return Arrays.asList(workQueue.toArray(EMPTY_RUNNABLE_ARRAY));
1016 }
1017
1018 public boolean isShutdown() {
1019 return runState != RUNNING;
1020 }
1021
1022 /**
1023 * Returns true if this executor is in the process of terminating
1024 * after <tt>shutdown</tt> or <tt>shutdownNow</tt> but has not
1025 * completely terminated. This method may be useful for
1026 * debugging. A return of <tt>true</tt> reported a sufficient
1027 * period after shutdown may indicate that submitted tasks have
1028 * ignored or suppressed interruption, causing this executor not
1029 * to properly terminate.
1030 * @return true if terminating but not yet terminated.
1031 */
1032 public boolean isTerminating() {
1033 return runState == STOP;
1034 }
1035
1036 public boolean isTerminated() {
1037 return runState == TERMINATED;
1038 }
1039
1040 public boolean awaitTermination(long timeout, TimeUnit unit)
1041 throws InterruptedException {
1042 long nanos = unit.toNanos(timeout);
1043 final ReentrantLock mainLock = this.mainLock;
1044 mainLock.lock();
1045 try {
1046 for (;;) {
1047 if (runState == TERMINATED)
1048 return true;
1049 if (nanos <= 0)
1050 return false;
1051 nanos = termination.awaitNanos(nanos);
1052 }
1053 } finally {
1054 mainLock.unlock();
1055 }
1056 }
1057
1058 /**
1059 * Invokes <tt>shutdown</tt> when this executor is no longer
1060 * referenced.
1061 */
1062 protected void finalize() {
1063 shutdown();
1064 }
1065
1066 /**
1067 * Sets the thread factory used to create new threads.
1068 *
1069 * @param threadFactory the new thread factory
1070 * @throws NullPointerException if threadFactory is null
1071 * @see #getThreadFactory
1072 */
1073 public void setThreadFactory(ThreadFactory threadFactory) {
1074 if (threadFactory == null)
1075 throw new NullPointerException();
1076 this.threadFactory = threadFactory;
1077 }
1078
1079 /**
1080 * Returns the thread factory used to create new threads.
1081 *
1082 * @return the current thread factory
1083 * @see #setThreadFactory
1084 */
1085 public ThreadFactory getThreadFactory() {
1086 return threadFactory;
1087 }
1088
1089 /**
1090 * Sets a new handler for unexecutable tasks.
1091 *
1092 * @param handler the new handler
1093 * @throws NullPointerException if handler is null
1094 * @see #getRejectedExecutionHandler
1095 */
1096 public void setRejectedExecutionHandler(RejectedExecutionHandler handler) {
1097 if (handler == null)
1098 throw new NullPointerException();
1099 this.handler = handler;
1100 }
1101
1102 /**
1103 * Returns the current handler for unexecutable tasks.
1104 *
1105 * @return the current handler
1106 * @see #setRejectedExecutionHandler
1107 */
1108 public RejectedExecutionHandler getRejectedExecutionHandler() {
1109 return handler;
1110 }
1111
1112 /**
1113 * Returns the task queue used by this executor. Access to the
1114 * task queue is intended primarily for debugging and monitoring.
1115 * This queue may be in active use. Retrieving the task queue
1116 * does not prevent queued tasks from executing.
1117 *
1118 * @return the task queue
1119 */
1120 public BlockingQueue<Runnable> getQueue() {
1121 return workQueue;
1122 }
1123
1124 /**
1125 * Removes this task from the executor's internal queue if it is
1126 * present, thus causing it not to be run if it has not already
1127 * started.
1128 *
1129 * <p> This method may be useful as one part of a cancellation
1130 * scheme. It may fail to remove tasks that have been converted
1131 * into other forms before being placed on the internal queue. For
1132 * example, a task entered using <tt>submit</tt> might be
1133 * converted into a form that maintains <tt>Future</tt> status.
1134 * However, in such cases, method {@link ThreadPoolExecutor#purge}
1135 * may be used to remove those Futures that have been cancelled.
1136 *
1137 * @param task the task to remove
1138 * @return true if the task was removed
1139 */
1140 public boolean remove(Runnable task) {
1141 return getQueue().remove(task);
1142 }
1143
1144
1145 /**
1146 * Tries to remove from the work queue all {@link Future}
1147 * tasks that have been cancelled. This method can be useful as a
1148 * storage reclamation operation, that has no other impact on
1149 * functionality. Cancelled tasks are never executed, but may
1150 * accumulate in work queues until worker threads can actively
1151 * remove them. Invoking this method instead tries to remove them now.
1152 * However, this method may fail to remove tasks in
1153 * the presence of interference by other threads.
1154 */
1155 public void purge() {
1156 // Fail if we encounter interference during traversal
1157 try {
1158 Iterator<Runnable> it = getQueue().iterator();
1159 while (it.hasNext()) {
1160 Runnable r = it.next();
1161 if (r instanceof Future<?>) {
1162 Future<?> c = (Future<?>)r;
1163 if (c.isCancelled())
1164 it.remove();
1165 }
1166 }
1167 }
1168 catch (ConcurrentModificationException ex) {
1169 return;
1170 }
1171 }
1172
1173 /**
1174 * Sets the core number of threads. This overrides any value set
1175 * in the constructor. If the new value is smaller than the
1176 * current value, excess existing threads will be terminated when
1177 * they next become idle. If larger, new threads will, if needed,
1178 * be started to execute any queued tasks.
1179 *
1180 * @param corePoolSize the new core size
1181 * @throws IllegalArgumentException if <tt>corePoolSize</tt>
1182 * less than zero
1183 * @see #getCorePoolSize
1184 */
1185 public void setCorePoolSize(int corePoolSize) {
1186 if (corePoolSize < 0)
1187 throw new IllegalArgumentException();
1188 final ReentrantLock mainLock = this.mainLock;
1189 mainLock.lock();
1190 try {
1191 int extra = this.corePoolSize - corePoolSize;
1192 this.corePoolSize = corePoolSize;
1193 if (extra < 0) {
1194 int n = workQueue.size();
1195 // We have to create initially-idle threads here
1196 // because we otherwise have no recourse about
1197 // what to do with a dequeued task if addThread fails.
1198 while (extra++ < 0 && n-- > 0 && poolSize < corePoolSize ) {
1199 Thread t = addThread(null);
1200 if (t != null)
1201 t.start();
1202 else
1203 break;
1204 }
1205 }
1206 else if (extra > 0 && poolSize > corePoolSize) {
1207 Iterator<Worker> it = workers.iterator();
1208 while (it.hasNext() &&
1209 extra-- > 0 &&
1210 poolSize > corePoolSize &&
1211 workQueue.remainingCapacity() == 0)
1212 it.next().interruptIfIdle();
1213 }
1214 } finally {
1215 mainLock.unlock();
1216 }
1217 }
1218
1219 /**
1220 * Returns the core number of threads.
1221 *
1222 * @return the core number of threads
1223 * @see #setCorePoolSize
1224 */
1225 public int getCorePoolSize() {
1226 return corePoolSize;
1227 }
1228
1229 /**
1230 * Starts a core thread, causing it to idly wait for work. This
1231 * overrides the default policy of starting core threads only when
1232 * new tasks are executed. This method will return <tt>false</tt>
1233 * if all core threads have already been started.
1234 * @return true if a thread was started
1235 */
1236 public boolean prestartCoreThread() {
1237 return addIfUnderCorePoolSize(null);
1238 }
1239
1240 /**
1241 * Starts all core threads, causing them to idly wait for work. This
1242 * overrides the default policy of starting core threads only when
1243 * new tasks are executed.
1244 * @return the number of threads started.
1245 */
1246 public int prestartAllCoreThreads() {
1247 int n = 0;
1248 while (addIfUnderCorePoolSize(null))
1249 ++n;
1250 return n;
1251 }
1252
1253 /**
1254 * Returns true if this pool allows core threads to time out and
1255 * terminate if no tasks arrive within the keepAlive time, being
1256 * replaced if needed when new tasks arrive. When true, the same
1257 * keep-alive policy applying to non-core threads applies also to
1258 * core threads. When false (the default), core threads are never
1259 * terminated due to lack of incoming tasks.
1260 * @return <tt>true</tt> if core threads are allowed to time out,
1261 * else <tt>false</tt>
1262 *
1263 * @since 1.6
1264 */
1265 public boolean allowsCoreThreadTimeOut() {
1266 return allowCoreThreadTimeOut;
1267 }
1268
1269 /**
1270 * Sets the policy governing whether core threads may time out and
1271 * terminate if no tasks arrive within the keep-alive time, being
1272 * replaced if needed when new tasks arrive. When false, core
1273 * threads are never terminated due to lack of incoming
1274 * tasks. When true, the same keep-alive policy applying to
1275 * non-core threads applies also to core threads. To avoid
1276 * continual thread replacement, the keep-alive time must be
1277 * greater than zero when setting <tt>true</tt>. This method
1278 * should in general be called before the pool is actively used.
1279 * @param value <tt>true</tt> if should time out, else <tt>false</tt>
1280 * @throws IllegalArgumentException if value is <tt>true</tt>
1281 * and the current keep-alive time is not greater than zero.
1282 *
1283 * @since 1.6
1284 */
1285 public void allowCoreThreadTimeOut(boolean value) {
1286 if (value && keepAliveTime <= 0)
1287 throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
1288
1289 allowCoreThreadTimeOut = value;
1290 }
1291
1292 /**
1293 * Sets the maximum allowed number of threads. This overrides any
1294 * value set in the constructor. If the new value is smaller than
1295 * the current value, excess existing threads will be
1296 * terminated when they next become idle.
1297 *
1298 * @param maximumPoolSize the new maximum
1299 * @throws IllegalArgumentException if maximumPoolSize less than zero or
1300 * the {@link #getCorePoolSize core pool size}
1301 * @see #getMaximumPoolSize
1302 */
1303 public void setMaximumPoolSize(int maximumPoolSize) {
1304 if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)
1305 throw new IllegalArgumentException();
1306 final ReentrantLock mainLock = this.mainLock;
1307 mainLock.lock();
1308 try {
1309 int extra = this.maximumPoolSize - maximumPoolSize;
1310 this.maximumPoolSize = maximumPoolSize;
1311 if (extra > 0 && poolSize > maximumPoolSize) {
1312 Iterator<Worker> it = workers.iterator();
1313 while (it.hasNext() &&
1314 extra > 0 &&
1315 poolSize > maximumPoolSize) {
1316 it.next().interruptIfIdle();
1317 --extra;
1318 }
1319 }
1320 } finally {
1321 mainLock.unlock();
1322 }
1323 }
1324
1325 /**
1326 * Returns the maximum allowed number of threads.
1327 *
1328 * @return the maximum allowed number of threads
1329 * @see #setMaximumPoolSize
1330 */
1331 public int getMaximumPoolSize() {
1332 return maximumPoolSize;
1333 }
1334
1335 /**
1336 * Sets the time limit for which threads may remain idle before
1337 * being terminated. If there are more than the core number of
1338 * threads currently in the pool, after waiting this amount of
1339 * time without processing a task, excess threads will be
1340 * terminated. This overrides any value set in the constructor.
1341 * @param time the time to wait. A time value of zero will cause
1342 * excess threads to terminate immediately after executing tasks.
1343 * @param unit the time unit of the time argument
1344 * @throws IllegalArgumentException if time less than zero or
1345 * if time is zero and allowsCoreThreadTimeOut
1346 * @see #getKeepAliveTime
1347 */
1348 public void setKeepAliveTime(long time, TimeUnit unit) {
1349 if (time < 0)
1350 throw new IllegalArgumentException();
1351 if (time == 0 && allowsCoreThreadTimeOut())
1352 throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
1353 this.keepAliveTime = unit.toNanos(time);
1354 }
1355
1356 /**
1357 * Returns the thread keep-alive time, which is the amount of time
1358 * which threads in excess of the core pool size may remain
1359 * idle before being terminated.
1360 *
1361 * @param unit the desired time unit of the result
1362 * @return the time limit
1363 * @see #setKeepAliveTime
1364 */
1365 public long getKeepAliveTime(TimeUnit unit) {
1366 return unit.convert(keepAliveTime, TimeUnit.NANOSECONDS);
1367 }
1368
1369 /* Statistics */
1370
1371 /**
1372 * Returns the current number of threads in the pool.
1373 *
1374 * @return the number of threads
1375 */
1376 public int getPoolSize() {
1377 return poolSize;
1378 }
1379
1380 /**
1381 * Returns the approximate number of threads that are actively
1382 * executing tasks.
1383 *
1384 * @return the number of threads
1385 */
1386 public int getActiveCount() {
1387 final ReentrantLock mainLock = this.mainLock;
1388 mainLock.lock();
1389 try {
1390 int n = 0;
1391 for (Worker w : workers) {
1392 if (w.isActive())
1393 ++n;
1394 }
1395 return n;
1396 } finally {
1397 mainLock.unlock();
1398 }
1399 }
1400
1401 /**
1402 * Returns the largest number of threads that have ever
1403 * simultaneously been in the pool.
1404 *
1405 * @return the number of threads
1406 */
1407 public int getLargestPoolSize() {
1408 final ReentrantLock mainLock = this.mainLock;
1409 mainLock.lock();
1410 try {
1411 return largestPoolSize;
1412 } finally {
1413 mainLock.unlock();
1414 }
1415 }
1416
1417 /**
1418 * Returns the approximate total number of tasks that have been
1419 * scheduled for execution. Because the states of tasks and
1420 * threads may change dynamically during computation, the returned
1421 * value is only an approximation, but one that does not ever
1422 * decrease across successive calls.
1423 *
1424 * @return the number of tasks
1425 */
1426 public long getTaskCount() {
1427 final ReentrantLock mainLock = this.mainLock;
1428 mainLock.lock();
1429 try {
1430 long n = completedTaskCount;
1431 for (Worker w : workers) {
1432 n += w.completedTasks;
1433 if (w.isActive())
1434 ++n;
1435 }
1436 return n + workQueue.size();
1437 } finally {
1438 mainLock.unlock();
1439 }
1440 }
1441
1442 /**
1443 * Returns the approximate total number of tasks that have
1444 * completed execution. Because the states of tasks and threads
1445 * may change dynamically during computation, the returned value
1446 * is only an approximation, but one that does not ever decrease
1447 * across successive calls.
1448 *
1449 * @return the number of tasks
1450 */
1451 public long getCompletedTaskCount() {
1452 final ReentrantLock mainLock = this.mainLock;
1453 mainLock.lock();
1454 try {
1455 long n = completedTaskCount;
1456 for (Worker w : workers)
1457 n += w.completedTasks;
1458 return n;
1459 } finally {
1460 mainLock.unlock();
1461 }
1462 }
1463
1464 /**
1465 * Method invoked prior to executing the given Runnable in the
1466 * given thread. This method is invoked by thread <tt>t</tt> that
1467 * will execute task <tt>r</tt>, and may be used to re-initialize
1468 * ThreadLocals, or to perform logging.
1469 *
1470 * <p>This implementation does nothing, but may be customized in
1471 * subclasses. Note: To properly nest multiple overridings, subclasses
1472 * should generally invoke <tt>super.beforeExecute</tt> at the end of
1473 * this method.
1474 *
1475 * @param t the thread that will run task r.
1476 * @param r the task that will be executed.
1477 */
1478 protected void beforeExecute(Thread t, Runnable r) { }
1479
1480 /**
1481 * Method invoked upon completion of execution of the given Runnable.
1482 * This method is invoked by the thread that executed the task. If
1483 * non-null, the Throwable is the uncaught <tt>RuntimeException</tt>
1484 * or <tt>Error</tt> that caused execution to terminate abruptly.
1485 *
1486 * <p><b>Note:</b> When actions are enclosed in tasks (such as
1487 * {@link FutureTask}) either explicitly or via methods such as
1488 * <tt>submit</tt>, these task objects catch and maintain
1489 * computational exceptions, and so they do not cause abrupt
1490 * termination, and the internal exceptions are <em>not</em>
1491 * passed to this method.
1492 *
1493 * <p>This implementation does nothing, but may be customized in
1494 * subclasses. Note: To properly nest multiple overridings, subclasses
1495 * should generally invoke <tt>super.afterExecute</tt> at the
1496 * beginning of this method.
1497 *
1498 * @param r the runnable that has completed.
1499 * @param t the exception that caused termination, or null if
1500 * execution completed normally.
1501 */
1502 protected void afterExecute(Runnable r, Throwable t) { }
1503
1504 /**
1505 * Method invoked when the Executor has terminated. Default
1506 * implementation does nothing. Note: To properly nest multiple
1507 * overridings, subclasses should generally invoke
1508 * <tt>super.terminated</tt> within this method.
1509 */
1510 protected void terminated() { }
1511
1512 /**
1513 * A handler for rejected tasks that runs the rejected task
1514 * directly in the calling thread of the <tt>execute</tt> method,
1515 * unless the executor has been shut down, in which case the task
1516 * is discarded.
1517 */
1518 public static class CallerRunsPolicy implements RejectedExecutionHandler {
1519 /**
1520 * Creates a <tt>CallerRunsPolicy</tt>.
1521 */
1522 public CallerRunsPolicy() { }
1523
1524 /**
1525 * Executes task r in the caller's thread, unless the executor
1526 * has been shut down, in which case the task is discarded.
1527 * @param r the runnable task requested to be executed
1528 * @param e the executor attempting to execute this task
1529 */
1530 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1531 if (!e.isShutdown()) {
1532 r.run();
1533 }
1534 }
1535 }
1536
1537 /**
1538 * A handler for rejected tasks that throws a
1539 * <tt>RejectedExecutionException</tt>.
1540 */
1541 public static class AbortPolicy implements RejectedExecutionHandler {
1542 /**
1543 * Creates an <tt>AbortPolicy</tt>.
1544 */
1545 public AbortPolicy() { }
1546
1547 /**
1548 * Always throws RejectedExecutionException.
1549 * @param r the runnable task requested to be executed
1550 * @param e the executor attempting to execute this task
1551 * @throws RejectedExecutionException always.
1552 */
1553 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1554 throw new RejectedExecutionException();
1555 }
1556 }
1557
1558 /**
1559 * A handler for rejected tasks that silently discards the
1560 * rejected task.
1561 */
1562 public static class DiscardPolicy implements RejectedExecutionHandler {
1563 /**
1564 * Creates a <tt>DiscardPolicy</tt>.
1565 */
1566 public DiscardPolicy() { }
1567
1568 /**
1569 * Does nothing, which has the effect of discarding task r.
1570 * @param r the runnable task requested to be executed
1571 * @param e the executor attempting to execute this task
1572 */
1573 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1574 }
1575 }
1576
1577 /**
1578 * A handler for rejected tasks that discards the oldest unhandled
1579 * request and then retries <tt>execute</tt>, unless the executor
1580 * is shut down, in which case the task is discarded.
1581 */
1582 public static class DiscardOldestPolicy implements RejectedExecutionHandler {
1583 /**
1584 * Creates a <tt>DiscardOldestPolicy</tt> for the given executor.
1585 */
1586 public DiscardOldestPolicy() { }
1587
1588 /**
1589 * Obtains and ignores the next task that the executor
1590 * would otherwise execute, if one is immediately available,
1591 * and then retries execution of task r, unless the executor
1592 * is shut down, in which case task r is instead discarded.
1593 * @param r the runnable task requested to be executed
1594 * @param e the executor attempting to execute this task
1595 */
1596 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1597 if (!e.isShutdown()) {
1598 e.getQueue().poll();
1599 e.execute(r);
1600 }
1601 }
1602 }
1603 }