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