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