ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/ThreadPoolExecutor.java
Revision: 1.77
Committed: Tue Feb 7 20:54:24 2006 UTC (18 years, 3 months ago) by jsr166
Branch: MAIN
Changes since 1.76: +0 -1 lines
Log Message:
6378729: Remove workaround for 6280605

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> </dl>
233 *
234 * <p> <b>Extension example</b>. Most extensions of this class
235 * override one or more of the protected hook methods. For example,
236 * here is a subclass that adds a simple pause/resume feature:
237 *
238 * <pre>
239 * class PausableThreadPoolExecutor extends ThreadPoolExecutor {
240 * private boolean isPaused;
241 * private ReentrantLock pauseLock = new ReentrantLock();
242 * private Condition unpaused = pauseLock.newCondition();
243 *
244 * public PausableThreadPoolExecutor(...) { super(...); }
245 *
246 * protected void beforeExecute(Thread t, Runnable r) {
247 * super.beforeExecute(t, r);
248 * pauseLock.lock();
249 * try {
250 * while (isPaused) unpaused.await();
251 * } catch (InterruptedException ie) {
252 * t.interrupt();
253 * } finally {
254 * pauseLock.unlock();
255 * }
256 * }
257 *
258 * public void pause() {
259 * pauseLock.lock();
260 * try {
261 * isPaused = true;
262 * } finally {
263 * pauseLock.unlock();
264 * }
265 * }
266 *
267 * public void resume() {
268 * pauseLock.lock();
269 * try {
270 * isPaused = false;
271 * unpaused.signalAll();
272 * } finally {
273 * pauseLock.unlock();
274 * }
275 * }
276 * }
277 * </pre>
278 * @since 1.5
279 * @author Doug Lea
280 */
281 public class ThreadPoolExecutor extends AbstractExecutorService {
282 /**
283 * Only used to force toArray() to produce a Runnable[].
284 */
285 private static final Runnable[] EMPTY_RUNNABLE_ARRAY = new Runnable[0];
286
287 /**
288 * Permission for checking shutdown
289 */
290 private static final RuntimePermission shutdownPerm =
291 new RuntimePermission("modifyThread");
292
293 /**
294 * Queue used for holding tasks and handing off to worker threads.
295 */
296 private final BlockingQueue<Runnable> workQueue;
297
298 /**
299 * Lock held on updates to poolSize, corePoolSize, maximumPoolSize, and
300 * workers set.
301 */
302 private final ReentrantLock mainLock = new ReentrantLock();
303
304 /**
305 * Wait condition to support awaitTermination
306 */
307 private final Condition termination = mainLock.newCondition();
308
309 /**
310 * Set containing all worker threads in pool.
311 */
312 private final HashSet<Worker> workers = new HashSet<Worker>();
313
314 /**
315 * Timeout in nanoseconds for idle threads waiting for work.
316 * Threads use this timeout only when there are more than
317 * corePoolSize present. Otherwise they wait forever for new work.
318 */
319 private volatile long keepAliveTime;
320
321 /**
322 * If false (default) core threads stay alive even when idle.
323 * If true, core threads use keepAliveTime to time out waiting for work.
324 */
325 private boolean allowCoreThreadTimeOut;
326
327 /**
328 * Core pool size, updated only while holding mainLock,
329 * but volatile to allow concurrent readability even
330 * during updates.
331 */
332 private volatile int corePoolSize;
333
334 /**
335 * Maximum pool size, updated only while holding mainLock
336 * but volatile to allow concurrent readability even
337 * during updates.
338 */
339 private volatile int maximumPoolSize;
340
341 /**
342 * Current pool size, updated only while holding mainLock
343 * but volatile to allow concurrent readability even
344 * during updates.
345 */
346 private volatile int poolSize;
347
348 /**
349 * Lifecycle state
350 */
351 volatile int runState;
352
353 // Special values for runState
354 /** Normal, not-shutdown mode */
355 static final int RUNNING = 0;
356 /** Controlled shutdown mode */
357 static final int SHUTDOWN = 1;
358 /** Immediate shutdown mode */
359 static final int STOP = 2;
360 /** Final state */
361 static final int TERMINATED = 3;
362
363 /**
364 * Handler called when saturated or shutdown in execute.
365 */
366 private volatile RejectedExecutionHandler handler;
367
368 /**
369 * Factory for new threads.
370 */
371 private volatile ThreadFactory threadFactory;
372
373 /**
374 * Tracks largest attained pool size.
375 */
376 private int largestPoolSize;
377
378 /**
379 * Counter for completed tasks. Updated only on termination of
380 * worker threads.
381 */
382 private long completedTaskCount;
383
384 /**
385 * The default rejected execution handler
386 */
387 private static final RejectedExecutionHandler defaultHandler =
388 new AbortPolicy();
389
390 /**
391 * Invokes the rejected execution handler for the given command.
392 */
393 void reject(Runnable command) {
394 handler.rejectedExecution(command, this);
395 }
396
397 /**
398 * Creates and returns a new thread running firstTask as its first
399 * task. Call only while holding mainLock.
400 * @param firstTask the task the new thread should run first (or
401 * null if none)
402 * @return the new thread, or null if threadFactory fails to create thread
403 */
404 private Thread addThread(Runnable firstTask) {
405 Worker w = new Worker(firstTask);
406 Thread t = threadFactory.newThread(w);
407 if (t != null) {
408 w.thread = t;
409 workers.add(w);
410 int nt = ++poolSize;
411 if (nt > largestPoolSize)
412 largestPoolSize = nt;
413 }
414 return t;
415 }
416
417 /**
418 * Creates and starts a new thread running firstTask as its first
419 * task, only if fewer than corePoolSize threads are running.
420 * @param firstTask the task the new thread should run first (or
421 * null if none)
422 * @return true if successful.
423 */
424 private boolean addIfUnderCorePoolSize(Runnable firstTask) {
425 Thread t = null;
426 final ReentrantLock mainLock = this.mainLock;
427 mainLock.lock();
428 try {
429 if (poolSize < corePoolSize)
430 t = addThread(firstTask);
431 } finally {
432 mainLock.unlock();
433 }
434 if (t == null)
435 return false;
436 t.start();
437 return true;
438 }
439
440 /**
441 * Creates and starts a new thread only if fewer than maximumPoolSize
442 * threads are running. The new thread runs as its first task the
443 * next task in queue, or if there is none, the given task.
444 * @param firstTask the task the new thread should run first (or
445 * null if none)
446 * @return 0 if a new thread cannot be created, a positive number
447 * if firstTask will be run in a new thread, or a negative number
448 * if a new thread was created but is running some other task, in
449 * which case the caller must try some other way to run firstTask
450 * (perhaps by calling this method again).
451 */
452 private int addIfUnderMaximumPoolSize(Runnable firstTask) {
453 Thread t = null;
454 int status = 0;
455 final ReentrantLock mainLock = this.mainLock;
456 mainLock.lock();
457 try {
458 if (poolSize < maximumPoolSize) {
459 Runnable next = workQueue.poll();
460 if (next == null) {
461 next = firstTask;
462 status = 1;
463 } else
464 status = -1;
465 t = addThread(next);
466 }
467 } finally {
468 mainLock.unlock();
469 }
470 if (t == null)
471 return 0;
472 t.start();
473 return status;
474 }
475
476
477 /**
478 * Gets the next task for a worker thread to run.
479 * @return the task
480 */
481 Runnable getTask() {
482 for (;;) {
483 try {
484 switch (runState) {
485 case RUNNING: {
486 // untimed wait if core and not allowing core timeout
487 if (poolSize <= corePoolSize && !allowCoreThreadTimeOut)
488 return workQueue.take();
489
490 long timeout = keepAliveTime;
491 if (timeout <= 0) // die immediately for 0 timeout
492 return null;
493 Runnable r = workQueue.poll(timeout, TimeUnit.NANOSECONDS);
494 if (r != null)
495 return r;
496 if (poolSize > corePoolSize || allowCoreThreadTimeOut)
497 return null; // timed out
498 // Else, after timeout, the pool shrank. Retry
499 break;
500 }
501
502 case SHUTDOWN: {
503 // Help drain queue
504 Runnable r = workQueue.poll();
505 if (r != null)
506 return r;
507
508 // Check if can terminate
509 if (workQueue.isEmpty()) {
510 interruptIdleWorkers();
511 return null;
512 }
513
514 // Else there could still be delayed tasks in queue.
515 return workQueue.take();
516 }
517
518 case STOP:
519 return null;
520 default:
521 assert false;
522 }
523 } catch (InterruptedException ie) {
524 // On interruption, re-check runstate
525 }
526 }
527 }
528
529 /**
530 * Wakes up all threads that might be waiting for tasks.
531 */
532 void interruptIdleWorkers() {
533 final ReentrantLock mainLock = this.mainLock;
534 mainLock.lock();
535 try {
536 for (Worker w : workers)
537 w.interruptIfIdle();
538 } finally {
539 mainLock.unlock();
540 }
541 }
542
543 /**
544 * Performs bookkeeping for a terminated worker thread.
545 * @param w the worker
546 */
547 void workerDone(Worker w) {
548 final ReentrantLock mainLock = this.mainLock;
549 mainLock.lock();
550 try {
551 completedTaskCount += w.completedTasks;
552 workers.remove(w);
553 if (--poolSize > 0)
554 return;
555
556 // Else, this is the last thread. Deal with potential shutdown.
557
558 int state = runState;
559 assert state != TERMINATED;
560
561 if (state != STOP) {
562 // If there are queued tasks but no threads, create
563 // replacement thread. We must create it initially
564 // idle to avoid orphaned tasks in case addThread
565 // fails. This also handles case of delayed tasks
566 // that will sometime later become runnable.
567 if (!workQueue.isEmpty()) {
568 Thread t = addThread(null);
569 if (t != null)
570 t.start();
571 return;
572 }
573
574 // Otherwise, we can exit without replacement
575 if (state == RUNNING)
576 return;
577 }
578
579 // Either state is STOP, or state is SHUTDOWN and there is
580 // no work to do. So we can terminate.
581 termination.signalAll();
582 runState = TERMINATED;
583 // fall through to call terminate() outside of lock.
584 } finally {
585 mainLock.unlock();
586 }
587
588 assert runState == TERMINATED;
589 terminated();
590 }
591
592 /**
593 * Worker threads
594 */
595 private class Worker implements Runnable {
596
597 /**
598 * The runLock is acquired and released surrounding each task
599 * execution. It mainly protects against interrupts that are
600 * intended to cancel the worker thread from instead
601 * interrupting the task being run.
602 */
603 private final ReentrantLock runLock = new ReentrantLock();
604
605 /**
606 * Initial task to run before entering run loop
607 */
608 private Runnable firstTask;
609
610 /**
611 * Per thread completed task counter; accumulated
612 * into completedTaskCount upon termination.
613 */
614 volatile long completedTasks;
615
616 /**
617 * Thread this worker is running in. Acts as a final field,
618 * but cannot be set until thread is created.
619 */
620 Thread thread;
621
622 Worker(Runnable firstTask) {
623 this.firstTask = firstTask;
624 }
625
626 boolean isActive() {
627 return runLock.isLocked();
628 }
629
630 /**
631 * Interrupts thread if not running a task.
632 */
633 void interruptIfIdle() {
634 final ReentrantLock runLock = this.runLock;
635 if (runLock.tryLock()) {
636 try {
637 thread.interrupt();
638 } finally {
639 runLock.unlock();
640 }
641 }
642 }
643
644 /**
645 * Interrupts thread even if running a task.
646 */
647 void interruptNow() {
648 thread.interrupt();
649 }
650
651 /**
652 * Runs a single task between before/after methods.
653 */
654 private void runTask(Runnable task) {
655 final ReentrantLock runLock = this.runLock;
656 runLock.lock();
657 try {
658 Thread.interrupted(); // clear interrupt status on entry
659 // Abort now if immediate cancel. Otherwise, we have
660 // committed to run this task.
661 if (runState == STOP)
662 return;
663
664 boolean ran = false;
665 beforeExecute(thread, task);
666 try {
667 task.run();
668 ran = true;
669 afterExecute(task, null);
670 ++completedTasks;
671 } catch (RuntimeException ex) {
672 if (!ran)
673 afterExecute(task, ex);
674 // Else the exception occurred within
675 // afterExecute itself in which case we don't
676 // want to call it again.
677 throw ex;
678 }
679 } finally {
680 runLock.unlock();
681 }
682 }
683
684 /**
685 * Main run loop
686 */
687 public void run() {
688 try {
689 Runnable task = firstTask;
690 firstTask = null;
691 while (task != null || (task = getTask()) != null) {
692 runTask(task);
693 task = null; // unnecessary but can help GC
694 }
695 } finally {
696 workerDone(this);
697 }
698 }
699 }
700
701 // Public methods
702
703 /**
704 * Creates a new <tt>ThreadPoolExecutor</tt> with the given initial
705 * parameters and default thread factory and rejected execution handler.
706 * It may be more convenient to use one of the {@link Executors} factory
707 * methods instead of this general purpose constructor.
708 *
709 * @param corePoolSize the number of threads to keep in the
710 * pool, even if they are idle.
711 * @param maximumPoolSize the maximum number of threads to allow in the
712 * pool.
713 * @param keepAliveTime when the number of threads is greater than
714 * the core, this is the maximum time that excess idle threads
715 * will wait for new tasks before terminating.
716 * @param unit the time unit for the keepAliveTime
717 * argument.
718 * @param workQueue the queue to use for holding tasks before they
719 * are executed. This queue will hold only the <tt>Runnable</tt>
720 * tasks submitted by the <tt>execute</tt> method.
721 * @throws IllegalArgumentException if corePoolSize, or
722 * keepAliveTime less than zero, or if maximumPoolSize less than or
723 * equal to zero, or if corePoolSize greater than maximumPoolSize.
724 * @throws NullPointerException if <tt>workQueue</tt> is null
725 */
726 public ThreadPoolExecutor(int corePoolSize,
727 int maximumPoolSize,
728 long keepAliveTime,
729 TimeUnit unit,
730 BlockingQueue<Runnable> workQueue) {
731 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
732 Executors.defaultThreadFactory(), defaultHandler);
733 }
734
735 /**
736 * Creates a new <tt>ThreadPoolExecutor</tt> with the given initial
737 * parameters and default rejected execution handler.
738 *
739 * @param corePoolSize the number of threads to keep in the
740 * pool, even if they are idle.
741 * @param maximumPoolSize the maximum number of threads to allow in the
742 * pool.
743 * @param keepAliveTime when the number of threads is greater than
744 * the core, this is the maximum time that excess idle threads
745 * will wait for new tasks before terminating.
746 * @param unit the time unit for the keepAliveTime
747 * argument.
748 * @param workQueue the queue to use for holding tasks before they
749 * are executed. This queue will hold only the <tt>Runnable</tt>
750 * tasks submitted by the <tt>execute</tt> method.
751 * @param threadFactory the factory to use when the executor
752 * creates a new thread.
753 * @throws IllegalArgumentException if corePoolSize, or
754 * keepAliveTime less than zero, or if maximumPoolSize less than or
755 * equal to zero, or if corePoolSize greater than maximumPoolSize.
756 * @throws NullPointerException if <tt>workQueue</tt>
757 * or <tt>threadFactory</tt> are null.
758 */
759 public ThreadPoolExecutor(int corePoolSize,
760 int maximumPoolSize,
761 long keepAliveTime,
762 TimeUnit unit,
763 BlockingQueue<Runnable> workQueue,
764 ThreadFactory threadFactory) {
765 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
766 threadFactory, defaultHandler);
767 }
768
769 /**
770 * Creates a new <tt>ThreadPoolExecutor</tt> with the given initial
771 * parameters and default thread factory.
772 *
773 * @param corePoolSize the number of threads to keep in the
774 * pool, even if they are idle.
775 * @param maximumPoolSize the maximum number of threads to allow in the
776 * pool.
777 * @param keepAliveTime when the number of threads is greater than
778 * the core, this is the maximum time that excess idle threads
779 * will wait for new tasks before terminating.
780 * @param unit the time unit for the keepAliveTime
781 * argument.
782 * @param workQueue the queue to use for holding tasks before they
783 * are executed. This queue will hold only the <tt>Runnable</tt>
784 * tasks submitted by the <tt>execute</tt> method.
785 * @param handler the handler to use when execution is blocked
786 * because the thread bounds and queue capacities are reached.
787 * @throws IllegalArgumentException if corePoolSize, or
788 * keepAliveTime less than zero, or if maximumPoolSize less than or
789 * equal to zero, or if corePoolSize greater than maximumPoolSize.
790 * @throws NullPointerException if <tt>workQueue</tt>
791 * or <tt>handler</tt> are null.
792 */
793 public ThreadPoolExecutor(int corePoolSize,
794 int maximumPoolSize,
795 long keepAliveTime,
796 TimeUnit unit,
797 BlockingQueue<Runnable> workQueue,
798 RejectedExecutionHandler handler) {
799 this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
800 Executors.defaultThreadFactory(), handler);
801 }
802
803 /**
804 * Creates a new <tt>ThreadPoolExecutor</tt> with the given initial
805 * parameters.
806 *
807 * @param corePoolSize the number of threads to keep in the
808 * pool, even if they are idle.
809 * @param maximumPoolSize the maximum number of threads to allow in the
810 * pool.
811 * @param keepAliveTime when the number of threads is greater than
812 * the core, this is the maximum time that excess idle threads
813 * will wait for new tasks before terminating.
814 * @param unit the time unit for the keepAliveTime
815 * argument.
816 * @param workQueue the queue to use for holding tasks before they
817 * are executed. This queue will hold only the <tt>Runnable</tt>
818 * tasks submitted by the <tt>execute</tt> method.
819 * @param threadFactory the factory to use when the executor
820 * creates a new thread.
821 * @param handler the handler to use when execution is blocked
822 * because the thread bounds and queue capacities are reached.
823 * @throws IllegalArgumentException if corePoolSize, or
824 * keepAliveTime less than zero, or if maximumPoolSize less than or
825 * equal to zero, or if corePoolSize greater than maximumPoolSize.
826 * @throws NullPointerException if <tt>workQueue</tt>
827 * or <tt>threadFactory</tt> or <tt>handler</tt> are null.
828 */
829 public ThreadPoolExecutor(int corePoolSize,
830 int maximumPoolSize,
831 long keepAliveTime,
832 TimeUnit unit,
833 BlockingQueue<Runnable> workQueue,
834 ThreadFactory threadFactory,
835 RejectedExecutionHandler handler) {
836 if (corePoolSize < 0 ||
837 maximumPoolSize <= 0 ||
838 maximumPoolSize < corePoolSize ||
839 keepAliveTime < 0)
840 throw new IllegalArgumentException();
841 if (workQueue == null || threadFactory == null || handler == null)
842 throw new NullPointerException();
843 this.corePoolSize = corePoolSize;
844 this.maximumPoolSize = maximumPoolSize;
845 this.workQueue = workQueue;
846 this.keepAliveTime = unit.toNanos(keepAliveTime);
847 this.threadFactory = threadFactory;
848 this.handler = handler;
849 }
850
851
852 /**
853 * Executes the given task sometime in the future. The task
854 * may execute in a new thread or in an existing pooled thread.
855 *
856 * If the task cannot be submitted for execution, either because this
857 * executor has been shutdown or because its capacity has been reached,
858 * the task is handled by the current <tt>RejectedExecutionHandler</tt>.
859 *
860 * @param command the task to execute
861 * @throws RejectedExecutionException at discretion of
862 * <tt>RejectedExecutionHandler</tt>, if task cannot be accepted
863 * for execution
864 * @throws NullPointerException if command is null
865 */
866 public void execute(Runnable command) {
867 if (command == null)
868 throw new NullPointerException();
869 for (;;) {
870 if (runState != RUNNING) {
871 reject(command);
872 return;
873 }
874 if (poolSize < corePoolSize && addIfUnderCorePoolSize(command))
875 return;
876 if (workQueue.offer(command))
877 return;
878 int status = addIfUnderMaximumPoolSize(command);
879 if (status > 0) // created new thread
880 return;
881 if (status == 0) { // failed to create thread
882 reject(command);
883 return;
884 }
885 // Retry if created a new thread but it is busy with another task
886 }
887 }
888
889 /**
890 * Initiates an orderly shutdown in which previously submitted
891 * tasks are executed, but no new tasks will be
892 * accepted. Invocation has no additional effect if already shut
893 * down.
894 * @throws SecurityException if a security manager exists and
895 * shutting down this ExecutorService may manipulate threads that
896 * the caller is not permitted to modify because it does not hold
897 * {@link java.lang.RuntimePermission}<tt>("modifyThread")</tt>,
898 * or the security manager's <tt>checkAccess</tt> method denies access.
899 */
900 public void shutdown() {
901 // Fail if caller doesn't have modifyThread permission. We
902 // explicitly check permissions directly because we can't trust
903 // implementations of SecurityManager to correctly override
904 // the "check access" methods such that our documented
905 // security policy is implemented.
906 SecurityManager security = System.getSecurityManager();
907 if (security != null)
908 java.security.AccessController.checkPermission(shutdownPerm);
909
910 boolean fullyTerminated = false;
911 final ReentrantLock mainLock = this.mainLock;
912 mainLock.lock();
913 try {
914 if (workers.size() > 0) {
915 // Check if caller can modify worker threads. This
916 // might not be true even if passed above check, if
917 // the SecurityManager treats some threads specially.
918 if (security != null) {
919 for (Worker w: workers)
920 security.checkAccess(w.thread);
921 }
922
923 int state = runState;
924 if (state == RUNNING) // don't override shutdownNow
925 runState = SHUTDOWN;
926
927 try {
928 for (Worker w: workers)
929 w.interruptIfIdle();
930 } catch (SecurityException se) {
931 // If SecurityManager allows above checks, but
932 // then unexpectedly throws exception when
933 // interrupting threads (which it ought not do),
934 // back out as cleanly as we can. Some threads may
935 // have been killed but we remain in non-shutdown
936 // state.
937 runState = state;
938 throw se;
939 }
940 }
941 else { // If no workers, trigger full termination now
942 fullyTerminated = true;
943 runState = TERMINATED;
944 termination.signalAll();
945 }
946 } finally {
947 mainLock.unlock();
948 }
949 if (fullyTerminated)
950 terminated();
951 }
952
953
954 /**
955 * Attempts to stop all actively executing tasks, halts the
956 * processing of waiting tasks, and returns a list of the tasks
957 * that were awaiting execution.
958 *
959 * <p>There are no guarantees beyond best-effort attempts to stop
960 * processing actively executing tasks. This implementation
961 * cancels tasks via {@link Thread#interrupt}, so any task that
962 * fails to respond to interrupts may never terminate.
963 *
964 * @return list of tasks that never commenced execution
965 * @throws SecurityException if a security manager exists and
966 * shutting down this ExecutorService may manipulate threads that
967 * the caller is not permitted to modify because it does not hold
968 * {@link java.lang.RuntimePermission}<tt>("modifyThread")</tt>,
969 * or the security manager's <tt>checkAccess</tt> method denies access.
970 */
971 public List<Runnable> shutdownNow() {
972 // Almost the same code as shutdown()
973 SecurityManager security = System.getSecurityManager();
974 if (security != null)
975 java.security.AccessController.checkPermission(shutdownPerm);
976
977 boolean fullyTerminated = false;
978 final ReentrantLock mainLock = this.mainLock;
979 mainLock.lock();
980 try {
981 if (workers.size() > 0) {
982 if (security != null) {
983 for (Worker w: workers)
984 security.checkAccess(w.thread);
985 }
986
987 int state = runState;
988 if (state != TERMINATED)
989 runState = STOP;
990 try {
991 for (Worker w : workers)
992 w.interruptNow();
993 } catch (SecurityException se) {
994 runState = state; // back out;
995 throw se;
996 }
997 }
998 else { // If no workers, trigger full termination now
999 fullyTerminated = true;
1000 runState = TERMINATED;
1001 termination.signalAll();
1002 }
1003 } finally {
1004 mainLock.unlock();
1005 }
1006 if (fullyTerminated)
1007 terminated();
1008 return Arrays.asList(workQueue.toArray(EMPTY_RUNNABLE_ARRAY));
1009 }
1010
1011 public boolean isShutdown() {
1012 return runState != RUNNING;
1013 }
1014
1015 /**
1016 * Returns true if this executor is in the process of terminating
1017 * after <tt>shutdown</tt> or <tt>shutdownNow</tt> but has not
1018 * completely terminated. This method may be useful for
1019 * debugging. A return of <tt>true</tt> reported a sufficient
1020 * period after shutdown may indicate that submitted tasks have
1021 * ignored or suppressed interruption, causing this executor not
1022 * to properly terminate.
1023 * @return true if terminating but not yet terminated.
1024 */
1025 public boolean isTerminating() {
1026 return runState == STOP;
1027 }
1028
1029 public boolean isTerminated() {
1030 return runState == TERMINATED;
1031 }
1032
1033 public boolean awaitTermination(long timeout, TimeUnit unit)
1034 throws InterruptedException {
1035 long nanos = unit.toNanos(timeout);
1036 final ReentrantLock mainLock = this.mainLock;
1037 mainLock.lock();
1038 try {
1039 for (;;) {
1040 if (runState == TERMINATED)
1041 return true;
1042 if (nanos <= 0)
1043 return false;
1044 nanos = termination.awaitNanos(nanos);
1045 }
1046 } finally {
1047 mainLock.unlock();
1048 }
1049 }
1050
1051 /**
1052 * Invokes <tt>shutdown</tt> when this executor is no longer
1053 * referenced.
1054 */
1055 protected void finalize() {
1056 shutdown();
1057 }
1058
1059 /**
1060 * Sets the thread factory used to create new threads.
1061 *
1062 * @param threadFactory the new thread factory
1063 * @throws NullPointerException if threadFactory is null
1064 * @see #getThreadFactory
1065 */
1066 public void setThreadFactory(ThreadFactory threadFactory) {
1067 if (threadFactory == null)
1068 throw new NullPointerException();
1069 this.threadFactory = threadFactory;
1070 }
1071
1072 /**
1073 * Returns the thread factory used to create new threads.
1074 *
1075 * @return the current thread factory
1076 * @see #setThreadFactory
1077 */
1078 public ThreadFactory getThreadFactory() {
1079 return threadFactory;
1080 }
1081
1082 /**
1083 * Sets a new handler for unexecutable tasks.
1084 *
1085 * @param handler the new handler
1086 * @throws NullPointerException if handler is null
1087 * @see #getRejectedExecutionHandler
1088 */
1089 public void setRejectedExecutionHandler(RejectedExecutionHandler handler) {
1090 if (handler == null)
1091 throw new NullPointerException();
1092 this.handler = handler;
1093 }
1094
1095 /**
1096 * Returns the current handler for unexecutable tasks.
1097 *
1098 * @return the current handler
1099 * @see #setRejectedExecutionHandler
1100 */
1101 public RejectedExecutionHandler getRejectedExecutionHandler() {
1102 return handler;
1103 }
1104
1105 /**
1106 * Returns the task queue used by this executor. Access to the
1107 * task queue is intended primarily for debugging and monitoring.
1108 * This queue may be in active use. Retrieving the task queue
1109 * does not prevent queued tasks from executing.
1110 *
1111 * @return the task queue
1112 */
1113 public BlockingQueue<Runnable> getQueue() {
1114 return workQueue;
1115 }
1116
1117 /**
1118 * Removes this task from the executor's internal queue if it is
1119 * present, thus causing it not to be run if it has not already
1120 * started.
1121 *
1122 * <p> This method may be useful as one part of a cancellation
1123 * scheme. It may fail to remove tasks that have been converted
1124 * into other forms before being placed on the internal queue. For
1125 * example, a task entered using <tt>submit</tt> might be
1126 * converted into a form that maintains <tt>Future</tt> status.
1127 * However, in such cases, method {@link ThreadPoolExecutor#purge}
1128 * may be used to remove those Futures that have been cancelled.
1129 *
1130 * @param task the task to remove
1131 * @return true if the task was removed
1132 */
1133 public boolean remove(Runnable task) {
1134 return getQueue().remove(task);
1135 }
1136
1137
1138 /**
1139 * Tries to remove from the work queue all {@link Future}
1140 * tasks that have been cancelled. This method can be useful as a
1141 * storage reclamation operation, that has no other impact on
1142 * functionality. Cancelled tasks are never executed, but may
1143 * accumulate in work queues until worker threads can actively
1144 * remove them. Invoking this method instead tries to remove them now.
1145 * However, this method may fail to remove tasks in
1146 * the presence of interference by other threads.
1147 */
1148 public void purge() {
1149 // Fail if we encounter interference during traversal
1150 try {
1151 Iterator<Runnable> it = getQueue().iterator();
1152 while (it.hasNext()) {
1153 Runnable r = it.next();
1154 if (r instanceof Future<?>) {
1155 Future<?> c = (Future<?>)r;
1156 if (c.isCancelled())
1157 it.remove();
1158 }
1159 }
1160 }
1161 catch (ConcurrentModificationException ex) {
1162 return;
1163 }
1164 }
1165
1166 /**
1167 * Sets the core number of threads. This overrides any value set
1168 * in the constructor. If the new value is smaller than the
1169 * current value, excess existing threads will be terminated when
1170 * they next become idle. If larger, new threads will, if needed,
1171 * be started to execute any queued tasks.
1172 *
1173 * @param corePoolSize the new core size
1174 * @throws IllegalArgumentException if <tt>corePoolSize</tt>
1175 * less than zero
1176 * @see #getCorePoolSize
1177 */
1178 public void setCorePoolSize(int corePoolSize) {
1179 if (corePoolSize < 0)
1180 throw new IllegalArgumentException();
1181 final ReentrantLock mainLock = this.mainLock;
1182 mainLock.lock();
1183 try {
1184 int extra = this.corePoolSize - corePoolSize;
1185 this.corePoolSize = corePoolSize;
1186 if (extra < 0) {
1187 int n = workQueue.size();
1188 // We have to create initially-idle threads here
1189 // because we otherwise have no recourse about
1190 // what to do with a dequeued task if addThread fails.
1191 while (extra++ < 0 && n-- > 0 && poolSize < corePoolSize ) {
1192 Thread t = addThread(null);
1193 if (t != null)
1194 t.start();
1195 else
1196 break;
1197 }
1198 }
1199 else if (extra > 0 && poolSize > corePoolSize) {
1200 Iterator<Worker> it = workers.iterator();
1201 while (it.hasNext() &&
1202 extra-- > 0 &&
1203 poolSize > corePoolSize &&
1204 workQueue.remainingCapacity() == 0)
1205 it.next().interruptIfIdle();
1206 }
1207 } finally {
1208 mainLock.unlock();
1209 }
1210 }
1211
1212 /**
1213 * Returns the core number of threads.
1214 *
1215 * @return the core number of threads
1216 * @see #setCorePoolSize
1217 */
1218 public int getCorePoolSize() {
1219 return corePoolSize;
1220 }
1221
1222 /**
1223 * Starts a core thread, causing it to idly wait for work. This
1224 * overrides the default policy of starting core threads only when
1225 * new tasks are executed. This method will return <tt>false</tt>
1226 * if all core threads have already been started.
1227 * @return true if a thread was started
1228 */
1229 public boolean prestartCoreThread() {
1230 return addIfUnderCorePoolSize(null);
1231 }
1232
1233 /**
1234 * Starts all core threads, causing them to idly wait for work. This
1235 * overrides the default policy of starting core threads only when
1236 * new tasks are executed.
1237 * @return the number of threads started.
1238 */
1239 public int prestartAllCoreThreads() {
1240 int n = 0;
1241 while (addIfUnderCorePoolSize(null))
1242 ++n;
1243 return n;
1244 }
1245
1246 /**
1247 * Returns true if this pool allows core threads to time out and
1248 * terminate if no tasks arrive within the keepAlive time, being
1249 * replaced if needed when new tasks arrive. When true, the same
1250 * keep-alive policy applying to non-core threads applies also to
1251 * core threads. When false (the default), core threads are never
1252 * terminated due to lack of incoming tasks.
1253 * @return <tt>true</tt> if core threads are allowed to time out,
1254 * else <tt>false</tt>
1255 *
1256 * @since 1.6
1257 */
1258 public boolean allowsCoreThreadTimeOut() {
1259 return allowCoreThreadTimeOut;
1260 }
1261
1262 /**
1263 * Sets the policy governing whether core threads may time out and
1264 * terminate if no tasks arrive within the keep-alive time, being
1265 * replaced if needed when new tasks arrive. When false, core
1266 * threads are never terminated due to lack of incoming
1267 * tasks. When true, the same keep-alive policy applying to
1268 * non-core threads applies also to core threads. To avoid
1269 * continual thread replacement, the keep-alive time must be
1270 * greater than zero when setting <tt>true</tt>. This method
1271 * should in general be called before the pool is actively used.
1272 * @param value <tt>true</tt> if should time out, else <tt>false</tt>
1273 * @throws IllegalArgumentException if value is <tt>true</tt>
1274 * and the current keep-alive time is not greater than zero.
1275 *
1276 * @since 1.6
1277 */
1278 public void allowCoreThreadTimeOut(boolean value) {
1279 if (value && keepAliveTime <= 0)
1280 throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
1281
1282 allowCoreThreadTimeOut = value;
1283 }
1284
1285 /**
1286 * Sets the maximum allowed number of threads. This overrides any
1287 * value set in the constructor. If the new value is smaller than
1288 * the current value, excess existing threads will be
1289 * terminated when they next become idle.
1290 *
1291 * @param maximumPoolSize the new maximum
1292 * @throws IllegalArgumentException if maximumPoolSize less than zero or
1293 * the {@link #getCorePoolSize core pool size}
1294 * @see #getMaximumPoolSize
1295 */
1296 public void setMaximumPoolSize(int maximumPoolSize) {
1297 if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)
1298 throw new IllegalArgumentException();
1299 final ReentrantLock mainLock = this.mainLock;
1300 mainLock.lock();
1301 try {
1302 int extra = this.maximumPoolSize - maximumPoolSize;
1303 this.maximumPoolSize = maximumPoolSize;
1304 if (extra > 0 && poolSize > maximumPoolSize) {
1305 Iterator<Worker> it = workers.iterator();
1306 while (it.hasNext() &&
1307 extra > 0 &&
1308 poolSize > maximumPoolSize) {
1309 it.next().interruptIfIdle();
1310 --extra;
1311 }
1312 }
1313 } finally {
1314 mainLock.unlock();
1315 }
1316 }
1317
1318 /**
1319 * Returns the maximum allowed number of threads.
1320 *
1321 * @return the maximum allowed number of threads
1322 * @see #setMaximumPoolSize
1323 */
1324 public int getMaximumPoolSize() {
1325 return maximumPoolSize;
1326 }
1327
1328 /**
1329 * Sets the time limit for which threads may remain idle before
1330 * being terminated. If there are more than the core number of
1331 * threads currently in the pool, after waiting this amount of
1332 * time without processing a task, excess threads will be
1333 * terminated. This overrides any value set in the constructor.
1334 * @param time the time to wait. A time value of zero will cause
1335 * excess threads to terminate immediately after executing tasks.
1336 * @param unit the time unit of the time argument
1337 * @throws IllegalArgumentException if time less than zero or
1338 * if time is zero and allowsCoreThreadTimeOut
1339 * @see #getKeepAliveTime
1340 */
1341 public void setKeepAliveTime(long time, TimeUnit unit) {
1342 if (time < 0)
1343 throw new IllegalArgumentException();
1344 if (time == 0 && allowsCoreThreadTimeOut())
1345 throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
1346 this.keepAliveTime = unit.toNanos(time);
1347 }
1348
1349 /**
1350 * Returns the thread keep-alive time, which is the amount of time
1351 * which threads in excess of the core pool size may remain
1352 * idle before being terminated.
1353 *
1354 * @param unit the desired time unit of the result
1355 * @return the time limit
1356 * @see #setKeepAliveTime
1357 */
1358 public long getKeepAliveTime(TimeUnit unit) {
1359 return unit.convert(keepAliveTime, TimeUnit.NANOSECONDS);
1360 }
1361
1362 /* Statistics */
1363
1364 /**
1365 * Returns the current number of threads in the pool.
1366 *
1367 * @return the number of threads
1368 */
1369 public int getPoolSize() {
1370 return poolSize;
1371 }
1372
1373 /**
1374 * Returns the approximate number of threads that are actively
1375 * executing tasks.
1376 *
1377 * @return the number of threads
1378 */
1379 public int getActiveCount() {
1380 final ReentrantLock mainLock = this.mainLock;
1381 mainLock.lock();
1382 try {
1383 int n = 0;
1384 for (Worker w : workers) {
1385 if (w.isActive())
1386 ++n;
1387 }
1388 return n;
1389 } finally {
1390 mainLock.unlock();
1391 }
1392 }
1393
1394 /**
1395 * Returns the largest number of threads that have ever
1396 * simultaneously been in the pool.
1397 *
1398 * @return the number of threads
1399 */
1400 public int getLargestPoolSize() {
1401 final ReentrantLock mainLock = this.mainLock;
1402 mainLock.lock();
1403 try {
1404 return largestPoolSize;
1405 } finally {
1406 mainLock.unlock();
1407 }
1408 }
1409
1410 /**
1411 * Returns the approximate total number of tasks that have been
1412 * scheduled for execution. Because the states of tasks and
1413 * threads may change dynamically during computation, the returned
1414 * value is only an approximation, but one that does not ever
1415 * decrease across successive calls.
1416 *
1417 * @return the number of tasks
1418 */
1419 public long getTaskCount() {
1420 final ReentrantLock mainLock = this.mainLock;
1421 mainLock.lock();
1422 try {
1423 long n = completedTaskCount;
1424 for (Worker w : workers) {
1425 n += w.completedTasks;
1426 if (w.isActive())
1427 ++n;
1428 }
1429 return n + workQueue.size();
1430 } finally {
1431 mainLock.unlock();
1432 }
1433 }
1434
1435 /**
1436 * Returns the approximate total number of tasks that have
1437 * completed execution. Because the states of tasks and threads
1438 * may change dynamically during computation, the returned value
1439 * is only an approximation, but one that does not ever decrease
1440 * across successive calls.
1441 *
1442 * @return the number of tasks
1443 */
1444 public long getCompletedTaskCount() {
1445 final ReentrantLock mainLock = this.mainLock;
1446 mainLock.lock();
1447 try {
1448 long n = completedTaskCount;
1449 for (Worker w : workers)
1450 n += w.completedTasks;
1451 return n;
1452 } finally {
1453 mainLock.unlock();
1454 }
1455 }
1456
1457 /**
1458 * Method invoked prior to executing the given Runnable in the
1459 * given thread. This method is invoked by thread <tt>t</tt> that
1460 * will execute task <tt>r</tt>, and may be used to re-initialize
1461 * ThreadLocals, or to perform logging.
1462 *
1463 * <p>This implementation does nothing, but may be customized in
1464 * subclasses. Note: To properly nest multiple overridings, subclasses
1465 * should generally invoke <tt>super.beforeExecute</tt> at the end of
1466 * this method.
1467 *
1468 * @param t the thread that will run task r.
1469 * @param r the task that will be executed.
1470 */
1471 protected void beforeExecute(Thread t, Runnable r) { }
1472
1473 /**
1474 * Method invoked upon completion of execution of the given Runnable.
1475 * This method is invoked by the thread that executed the task. If
1476 * non-null, the Throwable is the uncaught <tt>RuntimeException</tt>
1477 * or <tt>Error</tt> that caused execution to terminate abruptly.
1478 *
1479 * <p><b>Note:</b> When actions are enclosed in tasks (such as
1480 * {@link FutureTask}) either explicitly or via methods such as
1481 * <tt>submit</tt>, these task objects catch and maintain
1482 * computational exceptions, and so they do not cause abrupt
1483 * termination, and the internal exceptions are <em>not</em>
1484 * passed to this method.
1485 *
1486 * <p>This implementation does nothing, but may be customized in
1487 * subclasses. Note: To properly nest multiple overridings, subclasses
1488 * should generally invoke <tt>super.afterExecute</tt> at the
1489 * beginning of this method.
1490 *
1491 * @param r the runnable that has completed.
1492 * @param t the exception that caused termination, or null if
1493 * execution completed normally.
1494 */
1495 protected void afterExecute(Runnable r, Throwable t) { }
1496
1497 /**
1498 * Method invoked when the Executor has terminated. Default
1499 * implementation does nothing. Note: To properly nest multiple
1500 * overridings, subclasses should generally invoke
1501 * <tt>super.terminated</tt> within this method.
1502 */
1503 protected void terminated() { }
1504
1505 /**
1506 * A handler for rejected tasks that runs the rejected task
1507 * directly in the calling thread of the <tt>execute</tt> method,
1508 * unless the executor has been shut down, in which case the task
1509 * is discarded.
1510 */
1511 public static class CallerRunsPolicy implements RejectedExecutionHandler {
1512 /**
1513 * Creates a <tt>CallerRunsPolicy</tt>.
1514 */
1515 public CallerRunsPolicy() { }
1516
1517 /**
1518 * Executes task r in the caller's thread, unless the executor
1519 * has been shut down, in which case the task is discarded.
1520 * @param r the runnable task requested to be executed
1521 * @param e the executor attempting to execute this task
1522 */
1523 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1524 if (!e.isShutdown()) {
1525 r.run();
1526 }
1527 }
1528 }
1529
1530 /**
1531 * A handler for rejected tasks that throws a
1532 * <tt>RejectedExecutionException</tt>.
1533 */
1534 public static class AbortPolicy implements RejectedExecutionHandler {
1535 /**
1536 * Creates an <tt>AbortPolicy</tt>.
1537 */
1538 public AbortPolicy() { }
1539
1540 /**
1541 * Always throws RejectedExecutionException.
1542 * @param r the runnable task requested to be executed
1543 * @param e the executor attempting to execute this task
1544 * @throws RejectedExecutionException always.
1545 */
1546 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1547 throw new RejectedExecutionException();
1548 }
1549 }
1550
1551 /**
1552 * A handler for rejected tasks that silently discards the
1553 * rejected task.
1554 */
1555 public static class DiscardPolicy implements RejectedExecutionHandler {
1556 /**
1557 * Creates a <tt>DiscardPolicy</tt>.
1558 */
1559 public DiscardPolicy() { }
1560
1561 /**
1562 * Does nothing, which has the effect of discarding task r.
1563 * @param r the runnable task requested to be executed
1564 * @param e the executor attempting to execute this task
1565 */
1566 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1567 }
1568 }
1569
1570 /**
1571 * A handler for rejected tasks that discards the oldest unhandled
1572 * request and then retries <tt>execute</tt>, unless the executor
1573 * is shut down, in which case the task is discarded.
1574 */
1575 public static class DiscardOldestPolicy implements RejectedExecutionHandler {
1576 /**
1577 * Creates a <tt>DiscardOldestPolicy</tt> for the given executor.
1578 */
1579 public DiscardOldestPolicy() { }
1580
1581 /**
1582 * Obtains and ignores the next task that the executor
1583 * would otherwise execute, if one is immediately available,
1584 * and then retries execution of task r, unless the executor
1585 * is shut down, in which case task r is instead discarded.
1586 * @param r the runnable task requested to be executed
1587 * @param e the executor attempting to execute this task
1588 */
1589 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
1590 if (!e.isShutdown()) {
1591 e.getQueue().poll();
1592 e.execute(r);
1593 }
1594 }
1595 }
1596 }