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