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