1 /* 2 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. 3 * 4 * This code is free software; you can redistribute it and/or modify it 5 * under the terms of the GNU General Public License version 2 only, as 6 * published by the Free Software Foundation. Oracle designates this 7 * particular file as subject to the "Classpath" exception as provided 8 * by Oracle in the LICENSE file that accompanied this code. 9 * 10 * This code is distributed in the hope that it will be useful, but WITHOUT 11 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 12 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License 13 * version 2 for more details (a copy is included in the LICENSE file that 14 * accompanied this code). 15 * 16 * You should have received a copy of the GNU General Public License version 17 * 2 along with this work; if not, write to the Free Software Foundation, 18 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. 19 * 20 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA 21 * or visit www.oracle.com if you need additional information or have any 22 * questions. 23 */ 24 25 /* 26 * This file is available under and governed by the GNU General Public 27 * License version 2 only, as published by the Free Software Foundation. 28 * However, the following notice accompanied the original version of this 29 * file: 30 * 31 * Written by Doug Lea with assistance from members of JCP JSR-166 32 * Expert Group and released to the public domain, as explained at 33 * http://creativecommons.org/publicdomain/zero/1.0/ 34 */ 35 36 package java.util.concurrent; 37 38 import java.lang.Thread.UncaughtExceptionHandler; 39 import java.lang.invoke.MethodHandles; 40 import java.lang.invoke.VarHandle; 41 import java.security.AccessController; 42 import java.security.AccessControlContext; 43 import java.security.Permission; 44 import java.security.Permissions; 45 import java.security.PrivilegedAction; 46 import java.security.ProtectionDomain; 47 import java.util.ArrayList; 48 import java.util.Collection; 49 import java.util.Collections; 50 import java.util.List; 51 import java.util.function.Predicate; 52 import java.util.concurrent.atomic.AtomicInteger; 53 import java.util.concurrent.locks.LockSupport; 54 import java.util.concurrent.locks.ReentrantLock; 55 import java.util.concurrent.locks.Condition; 56 57 // Android-changed: Substituted @systemProperty tag with @code. 58 /** 59 * An {@link ExecutorService} for running {@link ForkJoinTask}s. 60 * A {@code ForkJoinPool} provides the entry point for submissions 61 * from non-{@code ForkJoinTask} clients, as well as management and 62 * monitoring operations. 63 * 64 * <p>A {@code ForkJoinPool} differs from other kinds of {@link 65 * ExecutorService} mainly by virtue of employing 66 * <em>work-stealing</em>: all threads in the pool attempt to find and 67 * execute tasks submitted to the pool and/or created by other active 68 * tasks (eventually blocking waiting for work if none exist). This 69 * enables efficient processing when most tasks spawn other subtasks 70 * (as do most {@code ForkJoinTask}s), as well as when many small 71 * tasks are submitted to the pool from external clients. Especially 72 * when setting <em>asyncMode</em> to true in constructors, {@code 73 * ForkJoinPool}s may also be appropriate for use with event-style 74 * tasks that are never joined. All worker threads are initialized 75 * with {@link Thread#isDaemon} set {@code true}. 76 * 77 * <p>A static {@link #commonPool()} is available and appropriate for 78 * most applications. The common pool is used by any ForkJoinTask that 79 * is not explicitly submitted to a specified pool. Using the common 80 * pool normally reduces resource usage (its threads are slowly 81 * reclaimed during periods of non-use, and reinstated upon subsequent 82 * use). 83 * 84 * <p>For applications that require separate or custom pools, a {@code 85 * ForkJoinPool} may be constructed with a given target parallelism 86 * level; by default, equal to the number of available processors. 87 * The pool attempts to maintain enough active (or available) threads 88 * by dynamically adding, suspending, or resuming internal worker 89 * threads, even if some tasks are stalled waiting to join others. 90 * However, no such adjustments are guaranteed in the face of blocked 91 * I/O or other unmanaged synchronization. The nested {@link 92 * ManagedBlocker} interface enables extension of the kinds of 93 * synchronization accommodated. The default policies may be 94 * overridden using a constructor with parameters corresponding to 95 * those documented in class {@link ThreadPoolExecutor}. 96 * 97 * <p>In addition to execution and lifecycle control methods, this 98 * class provides status check methods (for example 99 * {@link #getStealCount}) that are intended to aid in developing, 100 * tuning, and monitoring fork/join applications. Also, method 101 * {@link #toString} returns indications of pool state in a 102 * convenient form for informal monitoring. 103 * 104 * <p>As is the case with other ExecutorServices, there are three 105 * main task execution methods summarized in the following table. 106 * These are designed to be used primarily by clients not already 107 * engaged in fork/join computations in the current pool. The main 108 * forms of these methods accept instances of {@code ForkJoinTask}, 109 * but overloaded forms also allow mixed execution of plain {@code 110 * Runnable}- or {@code Callable}- based activities as well. However, 111 * tasks that are already executing in a pool should normally instead 112 * use the within-computation forms listed in the table unless using 113 * async event-style tasks that are not usually joined, in which case 114 * there is little difference among choice of methods. 115 * 116 * <table class="plain"> 117 * <caption>Summary of task execution methods</caption> 118 * <tr> 119 * <td></td> 120 * <th scope="col"> Call from non-fork/join clients</th> 121 * <th scope="col"> Call from within fork/join computations</th> 122 * </tr> 123 * <tr> 124 * <th scope="row" style="text-align:left"> Arrange async execution</th> 125 * <td> {@link #execute(ForkJoinTask)}</td> 126 * <td> {@link ForkJoinTask#fork}</td> 127 * </tr> 128 * <tr> 129 * <th scope="row" style="text-align:left"> Await and obtain result</th> 130 * <td> {@link #invoke(ForkJoinTask)}</td> 131 * <td> {@link ForkJoinTask#invoke}</td> 132 * </tr> 133 * <tr> 134 * <th scope="row" style="text-align:left"> Arrange exec and obtain Future</th> 135 * <td> {@link #submit(ForkJoinTask)}</td> 136 * <td> {@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td> 137 * </tr> 138 * </table> 139 * 140 * <p>The parameters used to construct the common pool may be controlled by 141 * setting the following {@linkplain System#getProperty system properties}: 142 * <ul> 143 * <li>{@code java.util.concurrent.ForkJoinPool.common.parallelism} 144 * - the parallelism level, a non-negative integer 145 * <li>{@code java.util.concurrent.ForkJoinPool.common.threadFactory} 146 * - the class name of a {@link ForkJoinWorkerThreadFactory}. 147 * The {@linkplain ClassLoader#getSystemClassLoader() system class loader} 148 * is used to load this class. 149 * <li>{@code java.util.concurrent.ForkJoinPool.common.exceptionHandler} 150 * - the class name of a {@link UncaughtExceptionHandler}. 151 * The {@linkplain ClassLoader#getSystemClassLoader() system class loader} 152 * is used to load this class. 153 * <li>{@code java.util.concurrent.ForkJoinPool.common.maximumSpares} 154 * - the maximum number of allowed extra threads to maintain target 155 * parallelism (default 256). 156 * </ul> 157 * If no thread factory is supplied via a system property, then the 158 * common pool uses a factory that uses the system class loader as the 159 * {@linkplain Thread#getContextClassLoader() thread context class loader}. 160 * In addition, if a {@link SecurityManager} is present, then 161 * the common pool uses a factory supplying threads that have no 162 * {@link Permissions} enabled. 163 * 164 * Upon any error in establishing these settings, default parameters 165 * are used. It is possible to disable or limit the use of threads in 166 * the common pool by setting the parallelism property to zero, and/or 167 * using a factory that may return {@code null}. However doing so may 168 * cause unjoined tasks to never be executed. 169 * 170 * <p><b>Implementation notes:</b> This implementation restricts the 171 * maximum number of running threads to 32767. Attempts to create 172 * pools with greater than the maximum number result in 173 * {@code IllegalArgumentException}. 174 * 175 * <p>This implementation rejects submitted tasks (that is, by throwing 176 * {@link RejectedExecutionException}) only when the pool is shut down 177 * or internal resources have been exhausted. 178 * 179 * @since 1.7 180 * @author Doug Lea 181 */ 182 @jdk.internal.vm.annotation.Contended 183 public class ForkJoinPool extends AbstractExecutorService { 184 185 // Android-changed: Substituted time reference with Android API version reference. 186 /* 187 * Implementation Overview 188 * 189 * This class and its nested classes provide the main 190 * functionality and control for a set of worker threads: 191 * Submissions from non-FJ threads enter into submission queues. 192 * Workers take these tasks and typically split them into subtasks 193 * that may be stolen by other workers. Work-stealing based on 194 * randomized scans generally leads to better throughput than 195 * "work dealing" in which producers assign tasks to idle threads, 196 * in part because threads that have finished other tasks before 197 * the signalled thread wakes up (which can be a long time) can 198 * take the task instead. Preference rules give first priority to 199 * processing tasks from their own queues (LIFO or FIFO, depending 200 * on mode), then to randomized FIFO steals of tasks in other 201 * queues. This framework began as vehicle for supporting 202 * tree-structured parallelism using work-stealing. Over time, 203 * its scalability advantages led to extensions and changes to 204 * better support more diverse usage contexts. Because most 205 * internal methods and nested classes are interrelated, their 206 * main rationale and descriptions are presented here; individual 207 * methods and nested classes contain only brief comments about 208 * details. 209 * 210 * WorkQueues 211 * ========== 212 * 213 * Most operations occur within work-stealing queues (in nested 214 * class WorkQueue). These are special forms of Deques that 215 * support only three of the four possible end-operations -- push, 216 * pop, and poll (aka steal), under the further constraints that 217 * push and pop are called only from the owning thread (or, as 218 * extended here, under a lock), while poll may be called from 219 * other threads. (If you are unfamiliar with them, you probably 220 * want to read Herlihy and Shavit's book "The Art of 221 * Multiprocessor programming", chapter 16 describing these in 222 * more detail before proceeding.) The main work-stealing queue 223 * design is roughly similar to those in the papers "Dynamic 224 * Circular Work-Stealing Deque" by Chase and Lev, SPAA 2005 225 * (http://research.sun.com/scalable/pubs/index.html) and 226 * "Idempotent work stealing" by Michael, Saraswat, and Vechev, 227 * PPoPP 2009 (http://portal.acm.org/citation.cfm?id=1504186). 228 * The main differences ultimately stem from GC requirements that 229 * we null out taken slots as soon as we can, to maintain as small 230 * a footprint as possible even in programs generating huge 231 * numbers of tasks. To accomplish this, we shift the CAS 232 * arbitrating pop vs poll (steal) from being on the indices 233 * ("base" and "top") to the slots themselves. 234 * 235 * Adding tasks then takes the form of a classic array push(task) 236 * in a circular buffer: 237 * q.array[q.top++ % length] = task; 238 * 239 * The actual code needs to null-check and size-check the array, 240 * uses masking, not mod, for indexing a power-of-two-sized array, 241 * enforces memory ordering, supports resizing, and possibly 242 * signals waiting workers to start scanning -- see below. 243 * 244 * The pop operation (always performed by owner) is of the form: 245 * if ((task = getAndSet(q.array, (q.top-1) % length, null)) != null) 246 * decrement top and return task; 247 * If this fails, the queue is empty. 248 * 249 * The poll operation by another stealer thread is, basically: 250 * if (CAS nonnull task at q.array[q.base % length] to null) 251 * increment base and return task; 252 * 253 * This may fail due to contention, and may be retried. 254 * Implementations must ensure a consistent snapshot of the base 255 * index and the task (by looping or trying elsewhere) before 256 * trying CAS. There isn't actually a method of this form, 257 * because failure due to inconsistency or contention is handled 258 * in different ways in different contexts, normally by first 259 * trying other queues. (For the most straightforward example, see 260 * method pollScan.) There are further variants for cases 261 * requiring inspection of elements before extracting them, so 262 * must interleave these with variants of this code. Also, a more 263 * efficient version (nextLocalTask) is used for polls by owners. 264 * It avoids some overhead because the queue cannot be growing 265 * during call. 266 * 267 * Memory ordering. See "Correct and Efficient Work-Stealing for 268 * Weak Memory Models" by Le, Pop, Cohen, and Nardelli, PPoPP 2013 269 * (http://www.di.ens.fr/~zappa/readings/ppopp13.pdf) for an 270 * analysis of memory ordering requirements in work-stealing 271 * algorithms similar to the one used here. Inserting and 272 * extracting tasks in array slots via volatile or atomic accesses 273 * or explicit fences provides primary synchronization. 274 * 275 * Operations on deque elements require reads and writes of both 276 * indices and slots. When possible, we allow these to occur in 277 * any order. Because the base and top indices (along with other 278 * pool or array fields accessed in many methods) only imprecisely 279 * guide where to extract from, we let accesses other than the 280 * element getAndSet/CAS/setVolatile appear in any order, using 281 * plain mode. But we must still preface some methods (mainly 282 * those that may be accessed externally) with an acquireFence to 283 * avoid unbounded staleness. This is equivalent to acting as if 284 * callers use an acquiring read of the reference to the pool or 285 * queue when invoking the method, even when they do not. We use 286 * explicit acquiring reads (getSlot) rather than plain array 287 * access when acquire mode is required but not otherwise ensured 288 * by context. To reduce stalls by other stealers, we encourage 289 * timely writes to the base index by immediately following 290 * updates with a write of a volatile field that must be updated 291 * anyway, or an Opaque-mode write if there is no such 292 * opportunity. 293 * 294 * Because indices and slot contents cannot always be consistent, 295 * the emptiness check base == top is only quiescently accurate 296 * (and so used where this suffices). Otherwise, it may err on the 297 * side of possibly making the queue appear nonempty when a push, 298 * pop, or poll have not fully committed, or making it appear 299 * empty when an update of top or base has not yet been seen. 300 * Similarly, the check in push for the queue array being full may 301 * trigger when not completely full, causing a resize earlier than 302 * required. 303 * 304 * Mainly because of these potential inconsistencies among slots 305 * vs indices, the poll operation, considered individually, is not 306 * wait-free. One thief cannot successfully continue until another 307 * in-progress one (or, if previously empty, a push) visibly 308 * completes. This can stall threads when required to consume 309 * from a given queue (which may spin). However, in the 310 * aggregate, we ensure probabilistic non-blockingness at least 311 * until checking quiescence (which is intrinsically blocking): 312 * If an attempted steal fails, a scanning thief chooses a 313 * different victim target to try next. So, in order for one thief 314 * to progress, it suffices for any in-progress poll or new push 315 * on any empty queue to complete. The worst cases occur when many 316 * threads are looking for tasks being produced by a stalled 317 * producer. 318 * 319 * This approach also enables support of a user mode in which 320 * local task processing is in FIFO, not LIFO order, simply by 321 * using poll rather than pop. This can be useful in 322 * message-passing frameworks in which tasks are never joined, 323 * although with increased contention among task producers and 324 * consumers. 325 * 326 * WorkQueues are also used in a similar way for tasks submitted 327 * to the pool. We cannot mix these tasks in the same queues used 328 * by workers. Instead, we randomly associate submission queues 329 * with submitting threads, using a form of hashing. The 330 * ThreadLocalRandom probe value serves as a hash code for 331 * choosing existing queues, and may be randomly repositioned upon 332 * contention with other submitters. In essence, submitters act 333 * like workers except that they are restricted to executing local 334 * tasks that they submitted (or when known, subtasks thereof). 335 * Insertion of tasks in shared mode requires a lock. We use only 336 * a simple spinlock (using field "source"), because submitters 337 * encountering a busy queue move to a different position to use 338 * or create other queues. They block only when registering new 339 * queues. 340 * 341 * Management 342 * ========== 343 * 344 * The main throughput advantages of work-stealing stem from 345 * decentralized control -- workers mostly take tasks from 346 * themselves or each other, at rates that can exceed a billion 347 * per second. Most non-atomic control is performed by some form 348 * of scanning across or within queues. The pool itself creates, 349 * activates (enables scanning for and running tasks), 350 * deactivates, blocks, and terminates threads, all with minimal 351 * central information. There are only a few properties that we 352 * can globally track or maintain, so we pack them into a small 353 * number of variables, often maintaining atomicity without 354 * blocking or locking. Nearly all essentially atomic control 355 * state is held in a few volatile variables that are by far most 356 * often read (not written) as status and consistency checks. We 357 * pack as much information into them as we can. 358 * 359 * Field "ctl" contains 64 bits holding information needed to 360 * atomically decide to add, enqueue (on an event queue), and 361 * dequeue and release workers. To enable this packing, we 362 * restrict maximum parallelism to (1<<15)-1 (which is far in 363 * excess of normal operating range) to allow ids, counts, and 364 * their negations (used for thresholding) to fit into 16bit 365 * subfields. 366 * 367 * Field "mode" holds configuration parameters as well as lifetime 368 * status, atomically and monotonically setting SHUTDOWN, STOP, 369 * and finally TERMINATED bits. It is updated only via bitwise 370 * atomics (getAndBitwiseOr). 371 * 372 * Array "queues" holds references to WorkQueues. It is updated 373 * (only during worker creation and termination) under the 374 * registrationLock, but is otherwise concurrently readable, and 375 * accessed directly (although always prefaced by acquireFences or 376 * other acquiring reads). To simplify index-based operations, the 377 * array size is always a power of two, and all readers must 378 * tolerate null slots. Worker queues are at odd indices. Worker 379 * ids masked with SMASK match their index. Shared (submission) 380 * queues are at even indices. Grouping them together in this way 381 * simplifies and speeds up task scanning. 382 * 383 * All worker thread creation is on-demand, triggered by task 384 * submissions, replacement of terminated workers, and/or 385 * compensation for blocked workers. However, all other support 386 * code is set up to work with other policies. To ensure that we 387 * do not hold on to worker or task references that would prevent 388 * GC, all accesses to workQueues are via indices into the 389 * queues array (which is one source of some of the messy code 390 * constructions here). In essence, the queues array serves as 391 * a weak reference mechanism. Thus for example the stack top 392 * subfield of ctl stores indices, not references. 393 * 394 * Queuing Idle Workers. Unlike HPC work-stealing frameworks, we 395 * cannot let workers spin indefinitely scanning for tasks when 396 * none can be found immediately, and we cannot start/resume 397 * workers unless there appear to be tasks available. On the 398 * other hand, we must quickly prod them into action when new 399 * tasks are submitted or generated. These latencies are mainly a 400 * function of JVM park/unpark (and underlying OS) performance, 401 * which can be slow and variable. In many usages, ramp-up time 402 * is the main limiting factor in overall performance, which is 403 * compounded at program start-up by JIT compilation and 404 * allocation. On the other hand, throughput degrades when too 405 * many threads poll for too few tasks. 406 * 407 * The "ctl" field atomically maintains total and "released" 408 * worker counts, plus the head of the available worker queue 409 * (actually stack, represented by the lower 32bit subfield of 410 * ctl). Released workers are those known to be scanning for 411 * and/or running tasks. Unreleased ("available") workers are 412 * recorded in the ctl stack. These workers are made available for 413 * signalling by enqueuing in ctl (see method awaitWork). The 414 * "queue" is a form of Treiber stack. This is ideal for 415 * activating threads in most-recently used order, and improves 416 * performance and locality, outweighing the disadvantages of 417 * being prone to contention and inability to release a worker 418 * unless it is topmost on stack. The top stack state holds the 419 * value of the "phase" field of the worker: its index and status, 420 * plus a version counter that, in addition to the count subfields 421 * (also serving as version stamps) provide protection against 422 * Treiber stack ABA effects. 423 * 424 * Creating workers. To create a worker, we pre-increment counts 425 * (serving as a reservation), and attempt to construct a 426 * ForkJoinWorkerThread via its factory. On starting, the new 427 * thread first invokes registerWorker, where it constructs a 428 * WorkQueue and is assigned an index in the queues array 429 * (expanding the array if necessary). Upon any exception across 430 * these steps, or null return from factory, deregisterWorker 431 * adjusts counts and records accordingly. If a null return, the 432 * pool continues running with fewer than the target number 433 * workers. If exceptional, the exception is propagated, generally 434 * to some external caller. 435 * 436 * WorkQueue field "phase" is used by both workers and the pool to 437 * manage and track whether a worker is UNSIGNALLED (possibly 438 * blocked waiting for a signal). When a worker is enqueued its 439 * phase field is set negative. Note that phase field updates lag 440 * queue CAS releases; seeing a negative phase does not guarantee 441 * that the worker is available. When queued, the lower 16 bits of 442 * its phase must hold its pool index. So we place the index there 443 * upon initialization and never modify these bits. 444 * 445 * The ctl field also serves as the basis for memory 446 * synchronization surrounding activation. This uses a more 447 * efficient version of a Dekker-like rule that task producers and 448 * consumers sync with each other by both writing/CASing ctl (even 449 * if to its current value). However, rather than CASing ctl to 450 * its current value in the common case where no action is 451 * required, we reduce write contention by ensuring that 452 * signalWork invocations are prefaced with a full-volatile memory 453 * access (which is usually needed anyway). 454 * 455 * Signalling. Signals (in signalWork) cause new or reactivated 456 * workers to scan for tasks. Method signalWork and its callers 457 * try to approximate the unattainable goal of having the right 458 * number of workers activated for the tasks at hand, but must err 459 * on the side of too many workers vs too few to avoid stalls. If 460 * computations are purely tree structured, it suffices for every 461 * worker to activate another when it pushes a task into an empty 462 * queue, resulting in O(log(#threads)) steps to full activation. 463 * If instead, tasks come in serially from only a single producer, 464 * each worker taking its first (since the last quiescence) task 465 * from a queue should signal another if there are more tasks in 466 * that queue. This is equivalent to, but generally faster than, 467 * arranging the stealer take two tasks, re-pushing one on its own 468 * queue, and signalling (because its queue is empty), also 469 * resulting in logarithmic full activation time. Because we don't 470 * know about usage patterns (or most commonly, mixtures), we use 471 * both approaches. We approximate the second rule by arranging 472 * that workers in scan() do not repeat signals when repeatedly 473 * taking tasks from any given queue, by remembering the previous 474 * one. There are narrow windows in which both rules may apply, 475 * leading to duplicate or unnecessary signals. Despite such 476 * limitations, these rules usually avoid slowdowns that otherwise 477 * occur when too many workers contend to take too few tasks, or 478 * when producers waste most of their time resignalling. However, 479 * contention and overhead effects may still occur during ramp-up, 480 * ramp-down, and small computations involving only a few workers. 481 * 482 * Scanning. Method scan performs top-level scanning for (and 483 * execution of) tasks. Scans by different workers and/or at 484 * different times are unlikely to poll queues in the same 485 * order. Each scan traverses and tries to poll from each queue in 486 * a pseudorandom permutation order by starting at a random index, 487 * and using a constant cyclically exhaustive stride; restarting 488 * upon contention. (Non-top-level scans; for example in 489 * helpJoin, use simpler linear probes because they do not 490 * systematically contend with top-level scans.) The pseudorandom 491 * generator need not have high-quality statistical properties in 492 * the long term. We use Marsaglia XorShifts, seeded with the Weyl 493 * sequence from ThreadLocalRandom probes, which are cheap and 494 * suffice. Scans do not otherwise explicitly take into account 495 * core affinities, loads, cache localities, etc, However, they do 496 * exploit temporal locality (which usually approximates these) by 497 * preferring to re-poll from the same queue after a successful 498 * poll before trying others (see method topLevelExec). This 499 * reduces fairness, which is partially counteracted by using a 500 * one-shot form of poll (tryPoll) that may lose to other workers. 501 * 502 * Deactivation. Method scan returns a sentinel when no tasks are 503 * found, leading to deactivation (see awaitWork). The count 504 * fields in ctl allow accurate discovery of quiescent states 505 * (i.e., when all workers are idle) after deactivation. However, 506 * this may also race with new (external) submissions, so a 507 * recheck is also needed to determine quiescence. Upon apparently 508 * triggering quiescence, awaitWork re-scans and self-signals if 509 * it may have missed a signal. In other cases, a missed signal 510 * may transiently lower parallelism because deactivation does not 511 * necessarily mean that there is no more work, only that that 512 * there were no tasks not taken by other workers. But more 513 * signals are generated (see above) to eventually reactivate if 514 * needed. 515 * 516 * Trimming workers. To release resources after periods of lack of 517 * use, a worker starting to wait when the pool is quiescent will 518 * time out and terminate if the pool has remained quiescent for 519 * period given by field keepAlive. 520 * 521 * Shutdown and Termination. A call to shutdownNow invokes 522 * tryTerminate to atomically set a mode bit. The calling thread, 523 * as well as every other worker thereafter terminating, helps 524 * terminate others by cancelling their unprocessed tasks, and 525 * waking them up. Calls to non-abrupt shutdown() preface this by 526 * checking isQuiescent before triggering the "STOP" phase of 527 * termination. To conform to ExecutorService invoke, invokeAll, 528 * and invokeAny specs, we must track pool status while waiting, 529 * and interrupt interruptible callers on termination (see 530 * ForkJoinTask.joinForPoolInvoke etc). 531 * 532 * Joining Tasks 533 * ============= 534 * 535 * Normally, the first option when joining a task that is not done 536 * is to try to unfork it from local queue and run it. Otherwise, 537 * any of several actions may be taken when one worker is waiting 538 * to join a task stolen (or always held) by another. Because we 539 * are multiplexing many tasks on to a pool of workers, we can't 540 * always just let them block (as in Thread.join). We also cannot 541 * just reassign the joiner's run-time stack with another and 542 * replace it later, which would be a form of "continuation", that 543 * even if possible is not necessarily a good idea since we may 544 * need both an unblocked task and its continuation to progress. 545 * Instead we combine two tactics: 546 * 547 * Helping: Arranging for the joiner to execute some task that it 548 * could be running if the steal had not occurred. 549 * 550 * Compensating: Unless there are already enough live threads, 551 * method tryCompensate() may create or re-activate a spare 552 * thread to compensate for blocked joiners until they unblock. 553 * 554 * A third form (implemented via tryRemove) amounts to helping a 555 * hypothetical compensator: If we can readily tell that a 556 * possible action of a compensator is to steal and execute the 557 * task being joined, the joining thread can do so directly, 558 * without the need for a compensation thread; although with a 559 * (rare) possibility of reduced parallelism because of a 560 * transient gap in the queue array. 561 * 562 * Other intermediate forms available for specific task types (for 563 * example helpAsyncBlocker) often avoid or postpone the need for 564 * blocking or compensation. 565 * 566 * The ManagedBlocker extension API can't use helping so relies 567 * only on compensation in method awaitBlocker. 568 * 569 * The algorithm in helpJoin entails a form of "linear helping". 570 * Each worker records (in field "source") the id of the queue 571 * from which it last stole a task. The scan in method helpJoin 572 * uses these markers to try to find a worker to help (i.e., steal 573 * back a task from and execute it) that could hasten completion 574 * of the actively joined task. Thus, the joiner executes a task 575 * that would be on its own local deque if the to-be-joined task 576 * had not been stolen. This is a conservative variant of the 577 * approach described in Wagner & Calder "Leapfrogging: a portable 578 * technique for implementing efficient futures" SIGPLAN Notices, 579 * 1993 (http://portal.acm.org/citation.cfm?id=155354). It differs 580 * mainly in that we only record queue ids, not full dependency 581 * links. This requires a linear scan of the queues array to 582 * locate stealers, but isolates cost to when it is needed, rather 583 * than adding to per-task overhead. Also, searches are limited to 584 * direct and at most two levels of indirect stealers, after which 585 * there are rapidly diminishing returns on increased overhead. 586 * Searches can fail to locate stealers when stalls delay 587 * recording sources. Further, even when accurately identified, 588 * stealers might not ever produce a task that the joiner can in 589 * turn help with. So, compensation is tried upon failure to find 590 * tasks to run. 591 * 592 * Joining CountedCompleters (see helpComplete) differs from (and 593 * is generally more efficient than) other cases because task 594 * eligibility is determined by checking completion chains rather 595 * than tracking stealers. 596 * 597 * Joining under timeouts (ForkJoinTask timed get) uses a 598 * constrained mixture of helping and compensating in part because 599 * pools (actually, only the common pool) may not have any 600 * available threads: If the pool is saturated (all available 601 * workers are busy), the caller tries to remove and otherwise 602 * help; else it blocks under compensation so that it may time out 603 * independently of any tasks. 604 * 605 * Compensation does not by default aim to keep exactly the target 606 * parallelism number of unblocked threads running at any given 607 * time. Some previous versions of this class employed immediate 608 * compensations for any blocked join. However, in practice, the 609 * vast majority of blockages are transient byproducts of GC and 610 * other JVM or OS activities that are made worse by replacement 611 * when they cause longer-term oversubscription. Rather than 612 * impose arbitrary policies, we allow users to override the 613 * default of only adding threads upon apparent starvation. The 614 * compensation mechanism may also be bounded. Bounds for the 615 * commonPool (see COMMON_MAX_SPARES) better enable JVMs to cope 616 * with programming errors and abuse before running out of 617 * resources to do so. 618 * 619 * Common Pool 620 * =========== 621 * 622 * The static common pool always exists after static 623 * initialization. Since it (or any other created pool) need 624 * never be used, we minimize initial construction overhead and 625 * footprint to the setup of about a dozen fields. 626 * 627 * When external threads submit to the common pool, they can 628 * perform subtask processing (see helpComplete and related 629 * methods) upon joins. This caller-helps policy makes it 630 * sensible to set common pool parallelism level to one (or more) 631 * less than the total number of available cores, or even zero for 632 * pure caller-runs. We do not need to record whether external 633 * submissions are to the common pool -- if not, external help 634 * methods return quickly. These submitters would otherwise be 635 * blocked waiting for completion, so the extra effort (with 636 * liberally sprinkled task status checks) in inapplicable cases 637 * amounts to an odd form of limited spin-wait before blocking in 638 * ForkJoinTask.join. 639 * 640 * Guarantees for common pool parallelism zero are limited to 641 * tasks that are joined by their callers in a tree-structured 642 * fashion or use CountedCompleters (as is true for jdk 643 * parallelStreams). Support infiltrates several methods, 644 * including those that retry helping steps until we are sure that 645 * none apply if there are no workers. 646 * 647 * As a more appropriate default in managed environments, unless 648 * overridden by system properties, we use workers of subclass 649 * InnocuousForkJoinWorkerThread when there is a SecurityManager 650 * present. These workers have no permissions set, do not belong 651 * to any user-defined ThreadGroup, and erase all ThreadLocals 652 * after executing any top-level task. The associated mechanics 653 * may be JVM-dependent and must access particular Thread class 654 * fields to achieve this effect. 655 * 656 * Interrupt handling 657 * ================== 658 * 659 * The framework is designed to manage task cancellation 660 * (ForkJoinTask.cancel) independently from the interrupt status 661 * of threads running tasks. (See the public ForkJoinTask 662 * documentation for rationale.) Interrupts are issued only in 663 * tryTerminate, when workers should be terminating and tasks 664 * should be cancelled anyway. Interrupts are cleared only when 665 * necessary to ensure that calls to LockSupport.park do not loop 666 * indefinitely (park returns immediately if the current thread is 667 * interrupted). If so, interruption is reinstated after blocking 668 * if status could be visible during the scope of any task. For 669 * cases in which task bodies are specified or desired to 670 * interrupt upon cancellation, ForkJoinTask.cancel can be 671 * overridden to do so (as is done for invoke{Any,All}). 672 * 673 * Memory placement 674 * ================ 675 * 676 * Performance can be very sensitive to placement of instances of 677 * ForkJoinPool and WorkQueues and their queue arrays. To reduce 678 * false-sharing impact, the @Contended annotation isolates the 679 * ForkJoinPool.ctl field as well as the most heavily written 680 * WorkQueue fields. These mainly reduce cache traffic by scanners. 681 * WorkQueue arrays are presized large enough to avoid resizing 682 * (which transiently reduces throughput) in most tree-like 683 * computations, although not in some streaming usages. Initial 684 * sizes are not large enough to avoid secondary contention 685 * effects (especially for GC cardmarks) when queues are placed 686 * near each other in memory. This is common, but has different 687 * impact in different collectors and remains incompletely 688 * addressed. 689 * 690 * Style notes 691 * =========== 692 * 693 * Memory ordering relies mainly on atomic operations (CAS, 694 * getAndSet, getAndAdd) along with explicit fences. This can be 695 * awkward and ugly, but also reflects the need to control 696 * outcomes across the unusual cases that arise in very racy code 697 * with very few invariants. All fields are read into locals 698 * before use, and null-checked if they are references, even if 699 * they can never be null under current usages. Array accesses 700 * using masked indices include checks (that are always true) that 701 * the array length is non-zero to avoid compilers inserting more 702 * expensive traps. This is usually done in a "C"-like style of 703 * listing declarations at the heads of methods or blocks, and 704 * using inline assignments on first encounter. Nearly all 705 * explicit checks lead to bypass/return, not exception throws, 706 * because they may legitimately arise during shutdown. 707 * 708 * There is a lot of representation-level coupling among classes 709 * ForkJoinPool, ForkJoinWorkerThread, and ForkJoinTask. The 710 * fields of WorkQueue maintain data structures managed by 711 * ForkJoinPool, so are directly accessed. There is little point 712 * trying to reduce this, since any associated future changes in 713 * representations will need to be accompanied by algorithmic 714 * changes anyway. Several methods intrinsically sprawl because 715 * they must accumulate sets of consistent reads of fields held in 716 * local variables. Some others are artificially broken up to 717 * reduce producer/consumer imbalances due to dynamic compilation. 718 * There are also other coding oddities (including several 719 * unnecessary-looking hoisted null checks) that help some methods 720 * perform reasonably even when interpreted (not compiled). 721 * 722 * The order of declarations in this file is (with a few exceptions): 723 * (1) Static utility functions 724 * (2) Nested (static) classes 725 * (3) Static fields 726 * (4) Fields, along with constants used when unpacking some of them 727 * (5) Internal control methods 728 * (6) Callbacks and other support for ForkJoinTask methods 729 * (7) Exported methods 730 * (8) Static block initializing statics in minimally dependent order 731 * 732 * Revision notes 733 * ============== 734 * 735 * The main sources of differences of ForkJoin classes from previous 736 * versions, up to Android API level 33, are: 737 * 738 * * ForkJoinTask now uses field "aux" to support blocking joins 739 * and/or record exceptions, replacing reliance on builtin 740 * monitors and side tables. 741 * * Scans probe slots (vs compare indices), along with related 742 * changes that reduce performance differences across most 743 * garbage collectors, and reduce contention. 744 * * Refactoring for better integration of special task types and 745 * other capabilities that had been incrementally tacked on. Plus 746 * many minor reworkings to improve consistency. 747 */ 748 749 // Static utilities 750 751 /** 752 * If there is a security manager, makes sure caller has 753 * permission to modify threads. 754 */ checkPermission()755 private static void checkPermission() { 756 @SuppressWarnings("removal") 757 SecurityManager security = System.getSecurityManager(); 758 if (security != null) 759 security.checkPermission(modifyThreadPermission); 760 } 761 762 @SuppressWarnings("removal") contextWithPermissions(Permission .... perms)763 static AccessControlContext contextWithPermissions(Permission ... perms) { 764 Permissions permissions = new Permissions(); 765 for (Permission perm : perms) 766 permissions.add(perm); 767 return new AccessControlContext( 768 new ProtectionDomain[] { new ProtectionDomain(null, permissions) }); 769 } 770 771 // Nested classes 772 773 /** 774 * Factory for creating new {@link ForkJoinWorkerThread}s. 775 * A {@code ForkJoinWorkerThreadFactory} must be defined and used 776 * for {@code ForkJoinWorkerThread} subclasses that extend base 777 * functionality or initialize threads with different contexts. 778 */ 779 public static interface ForkJoinWorkerThreadFactory { 780 /** 781 * Returns a new worker thread operating in the given pool. 782 * Returning null or throwing an exception may result in tasks 783 * never being executed. If this method throws an exception, 784 * it is relayed to the caller of the method (for example 785 * {@code execute}) causing attempted thread creation. If this 786 * method returns null or throws an exception, it is not 787 * retried until the next attempted creation (for example 788 * another call to {@code execute}). 789 * 790 * @param pool the pool this thread works in 791 * @return the new worker thread, or {@code null} if the request 792 * to create a thread is rejected 793 * @throws NullPointerException if the pool is null 794 */ newThread(ForkJoinPool pool)795 public ForkJoinWorkerThread newThread(ForkJoinPool pool); 796 } 797 798 /** 799 * Default ForkJoinWorkerThreadFactory implementation; creates a 800 * new ForkJoinWorkerThread using the system class loader as the 801 * thread context class loader. 802 */ 803 static final class DefaultForkJoinWorkerThreadFactory 804 implements ForkJoinWorkerThreadFactory { 805 // ACC for access to the factory 806 @SuppressWarnings("removal") 807 private static final AccessControlContext ACC = contextWithPermissions( 808 new RuntimePermission("getClassLoader"), 809 new RuntimePermission("setContextClassLoader")); 810 @SuppressWarnings("removal") newThread(ForkJoinPool pool)811 public final ForkJoinWorkerThread newThread(ForkJoinPool pool) { 812 return AccessController.doPrivileged( 813 new PrivilegedAction<>() { 814 public ForkJoinWorkerThread run() { 815 return new ForkJoinWorkerThread(null, pool, true, false); 816 }}, 817 ACC); 818 } 819 } 820 821 /** 822 * Factory for CommonPool unless overridden by System property. 823 * Creates InnocuousForkJoinWorkerThreads if a security manager is 824 * present at time of invocation. Support requires that we break 825 * quite a lot of encapsulation (some via helper methods in 826 * ThreadLocalRandom) to access and set Thread fields. 827 */ 828 static final class DefaultCommonPoolForkJoinWorkerThreadFactory 829 implements ForkJoinWorkerThreadFactory { 830 @SuppressWarnings("removal") 831 private static final AccessControlContext ACC = contextWithPermissions( 832 modifyThreadPermission, 833 new RuntimePermission("enableContextClassLoaderOverride"), 834 new RuntimePermission("modifyThreadGroup"), 835 new RuntimePermission("getClassLoader"), 836 new RuntimePermission("setContextClassLoader")); 837 838 @SuppressWarnings("removal") 839 public final ForkJoinWorkerThread newThread(ForkJoinPool pool) { 840 return AccessController.doPrivileged( 841 new PrivilegedAction<>() { 842 public ForkJoinWorkerThread run() { 843 return System.getSecurityManager() == null ? 844 new ForkJoinWorkerThread(null, pool, true, true): 845 new ForkJoinWorkerThread. 846 InnocuousForkJoinWorkerThread(pool); }}, 847 ACC); 848 } 849 } 850 851 // Constants shared across ForkJoinPool and WorkQueue 852 853 // Bounds 854 static final int SWIDTH = 16; // width of short 855 static final int SMASK = 0xffff; // short bits == max index 856 static final int MAX_CAP = 0x7fff; // max #workers - 1 857 858 // Masks and units for WorkQueue.phase and ctl sp subfield 859 static final int UNSIGNALLED = 1 << 31; // must be negative 860 static final int SS_SEQ = 1 << 16; // version count 861 862 // Mode bits and sentinels, some also used in WorkQueue fields 863 static final int FIFO = 1 << 16; // fifo queue or access mode 864 static final int SRC = 1 << 17; // set for valid queue ids 865 static final int INNOCUOUS = 1 << 18; // set for Innocuous workers 866 static final int QUIET = 1 << 19; // quiescing phase or source 867 static final int SHUTDOWN = 1 << 24; 868 static final int TERMINATED = 1 << 25; 869 static final int STOP = 1 << 31; // must be negative 870 static final int UNCOMPENSATE = 1 << 16; // tryCompensate return 871 872 /** 873 * Initial capacity of work-stealing queue array. Must be a power 874 * of two, at least 2. See above. 875 */ 876 static final int INITIAL_QUEUE_CAPACITY = 1 << 8; 877 878 /** 879 * Queues supporting work-stealing as well as external task 880 * submission. See above for descriptions and algorithms. 881 */ 882 static final class WorkQueue { 883 volatile int phase; // versioned, negative if inactive 884 int stackPred; // pool stack (ctl) predecessor link 885 int config; // index, mode, ORed with SRC after init 886 int base; // index of next slot for poll 887 ForkJoinTask<?>[] array; // the queued tasks; power of 2 size 888 final ForkJoinWorkerThread owner; // owning thread or null if shared 889 890 // segregate fields frequently updated but not read by scans or steals 891 @jdk.internal.vm.annotation.Contended("w") 892 int top; // index of next slot for push 893 @jdk.internal.vm.annotation.Contended("w") 894 volatile int source; // source queue id, lock, or sentinel 895 @jdk.internal.vm.annotation.Contended("w") 896 int nsteals; // number of steals from other queues 897 898 // Support for atomic operations 899 private static final VarHandle QA; // for array slots 900 private static final VarHandle SOURCE; 901 private static final VarHandle BASE; 902 static final ForkJoinTask<?> getSlot(ForkJoinTask<?>[] a, int i) { 903 return (ForkJoinTask<?>)QA.getAcquire(a, i); 904 } 905 static final ForkJoinTask<?> getAndClearSlot(ForkJoinTask<?>[] a, 906 int i) { 907 return (ForkJoinTask<?>)QA.getAndSet(a, i, null); 908 } 909 static final void setSlotVolatile(ForkJoinTask<?>[] a, int i, 910 ForkJoinTask<?> v) { 911 QA.setVolatile(a, i, v); 912 } 913 static final boolean casSlotToNull(ForkJoinTask<?>[] a, int i, 914 ForkJoinTask<?> c) { 915 return QA.compareAndSet(a, i, c, null); 916 } 917 final boolean tryLock() { 918 return SOURCE.compareAndSet(this, 0, 1); 919 } 920 final void setBaseOpaque(int b) { 921 BASE.setOpaque(this, b); 922 } 923 924 /** 925 * Constructor used by ForkJoinWorkerThreads. Most fields 926 * are initialized upon thread start, in pool.registerWorker. 927 */ 928 WorkQueue(ForkJoinWorkerThread owner, boolean isInnocuous) { 929 this.config = (isInnocuous) ? INNOCUOUS : 0; 930 this.owner = owner; 931 } 932 933 /** 934 * Constructor used for external queues. 935 */ 936 WorkQueue(int config) { 937 array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY]; 938 this.config = config; 939 owner = null; 940 phase = -1; 941 } 942 943 /** 944 * Returns an exportable index (used by ForkJoinWorkerThread). 945 */ 946 final int getPoolIndex() { 947 return (config & 0xffff) >>> 1; // ignore odd/even tag bit 948 } 949 950 /** 951 * Returns the approximate number of tasks in the queue. 952 */ 953 final int queueSize() { 954 VarHandle.acquireFence(); // ensure fresh reads by external callers 955 int n = top - base; 956 return (n < 0) ? 0 : n; // ignore transient negative 957 } 958 959 /** 960 * Provides a more conservative estimate of whether this queue 961 * has any tasks than does queueSize. 962 */ 963 final boolean isEmpty() { 964 return !((source != 0 && owner == null) || top - base > 0); 965 } 966 967 /** 968 * Pushes a task. Call only by owner in unshared queues. 969 * 970 * @param task the task. Caller must ensure non-null. 971 * @param pool (no-op if null) 972 * @throws RejectedExecutionException if array cannot be resized 973 */ 974 final void push(ForkJoinTask<?> task, ForkJoinPool pool) { 975 ForkJoinTask<?>[] a = array; 976 int s = top++, d = s - base, cap, m; // skip insert if disabled 977 if (a != null && pool != null && (cap = a.length) > 0) { 978 setSlotVolatile(a, (m = cap - 1) & s, task); 979 if (d == m) 980 growArray(); 981 if (d == m || a[m & (s - 1)] == null) 982 pool.signalWork(); // signal if was empty or resized 983 } 984 } 985 986 /** 987 * Pushes task to a shared queue with lock already held, and unlocks. 988 * 989 * @return true if caller should signal work 990 */ 991 final boolean lockedPush(ForkJoinTask<?> task) { 992 ForkJoinTask<?>[] a = array; 993 int s = top++, d = s - base, cap, m; 994 if (a != null && (cap = a.length) > 0) { 995 a[(m = cap - 1) & s] = task; 996 if (d == m) 997 growArray(); 998 source = 0; // unlock 999 if (d == m || a[m & (s - 1)] == null) 1000 return true; 1001 } 1002 return false; 1003 } 1004 1005 /** 1006 * Doubles the capacity of array. Called by owner or with lock 1007 * held after pre-incrementing top, which is reverted on 1008 * allocation failure. 1009 */ 1010 final void growArray() { 1011 ForkJoinTask<?>[] oldArray = array, newArray; 1012 int s = top - 1, oldCap, newCap; 1013 if (oldArray != null && (oldCap = oldArray.length) > 0 && 1014 (newCap = oldCap << 1) > 0) { // skip if disabled 1015 try { 1016 newArray = new ForkJoinTask<?>[newCap]; 1017 } catch (Throwable ex) { 1018 top = s; 1019 if (owner == null) 1020 source = 0; // unlock 1021 throw new RejectedExecutionException( 1022 "Queue capacity exceeded"); 1023 } 1024 int newMask = newCap - 1, oldMask = oldCap - 1; 1025 for (int k = oldCap; k > 0; --k, --s) { 1026 ForkJoinTask<?> x; // poll old, push to new 1027 if ((x = getAndClearSlot(oldArray, s & oldMask)) == null) 1028 break; // others already taken 1029 newArray[s & newMask] = x; 1030 } 1031 VarHandle.releaseFence(); // fill before publish 1032 array = newArray; 1033 } 1034 } 1035 1036 // Variants of pop 1037 1038 /** 1039 * Pops and returns task, or null if empty. Called only by owner. 1040 */ 1041 private ForkJoinTask<?> pop() { 1042 ForkJoinTask<?> t = null; 1043 int s = top, cap; ForkJoinTask<?>[] a; 1044 if ((a = array) != null && (cap = a.length) > 0 && base != s-- && 1045 (t = getAndClearSlot(a, (cap - 1) & s)) != null) 1046 top = s; 1047 return t; 1048 } 1049 1050 /** 1051 * Pops the given task for owner only if it is at the current top. 1052 */ 1053 final boolean tryUnpush(ForkJoinTask<?> task) { 1054 int s = top, cap; ForkJoinTask<?>[] a; 1055 if ((a = array) != null && (cap = a.length) > 0 && base != s-- && 1056 casSlotToNull(a, (cap - 1) & s, task)) { 1057 top = s; 1058 return true; 1059 } 1060 return false; 1061 } 1062 1063 /** 1064 * Locking version of tryUnpush. 1065 */ 1066 final boolean externalTryUnpush(ForkJoinTask<?> task) { 1067 boolean taken = false; 1068 for (;;) { 1069 int s = top, cap, k; ForkJoinTask<?>[] a; 1070 if ((a = array) == null || (cap = a.length) <= 0 || 1071 a[k = (cap - 1) & (s - 1)] != task) 1072 break; 1073 if (tryLock()) { 1074 if (top == s && array == a) { 1075 if (taken = casSlotToNull(a, k, task)) { 1076 top = s - 1; 1077 source = 0; 1078 break; 1079 } 1080 } 1081 source = 0; // release lock for retry 1082 } 1083 Thread.yield(); // trylock failure 1084 } 1085 return taken; 1086 } 1087 1088 /** 1089 * Deep form of tryUnpush: Traverses from top and removes task if 1090 * present, shifting others to fill gap. 1091 */ 1092 final boolean tryRemove(ForkJoinTask<?> task, boolean owned) { 1093 boolean taken = false; 1094 int p = top, cap; ForkJoinTask<?>[] a; ForkJoinTask<?> t; 1095 if ((a = array) != null && task != null && (cap = a.length) > 0) { 1096 int m = cap - 1, s = p - 1, d = p - base; 1097 for (int i = s, k; d > 0; --i, --d) { 1098 if ((t = a[k = i & m]) == task) { 1099 if (owned || tryLock()) { 1100 if ((owned || (array == a && top == p)) && 1101 (taken = casSlotToNull(a, k, t))) { 1102 for (int j = i; j != s; ) // shift down 1103 a[j & m] = getAndClearSlot(a, ++j & m); 1104 top = s; 1105 } 1106 if (!owned) 1107 source = 0; 1108 } 1109 break; 1110 } 1111 } 1112 } 1113 return taken; 1114 } 1115 1116 // variants of poll 1117 1118 /** 1119 * Tries once to poll next task in FIFO order, failing on 1120 * inconsistency or contention. 1121 */ 1122 final ForkJoinTask<?> tryPoll() { 1123 int cap, b, k; ForkJoinTask<?>[] a; 1124 if ((a = array) != null && (cap = a.length) > 0) { 1125 ForkJoinTask<?> t = getSlot(a, k = (cap - 1) & (b = base)); 1126 if (base == b++ && t != null && casSlotToNull(a, k, t)) { 1127 setBaseOpaque(b); 1128 return t; 1129 } 1130 } 1131 return null; 1132 } 1133 1134 /** 1135 * Takes next task, if one exists, in order specified by mode. 1136 */ 1137 final ForkJoinTask<?> nextLocalTask(int cfg) { 1138 ForkJoinTask<?> t = null; 1139 int s = top, cap; ForkJoinTask<?>[] a; 1140 if ((a = array) != null && (cap = a.length) > 0) { 1141 for (int b, d;;) { 1142 if ((d = s - (b = base)) <= 0) 1143 break; 1144 if (d == 1 || (cfg & FIFO) == 0) { 1145 if ((t = getAndClearSlot(a, --s & (cap - 1))) != null) 1146 top = s; 1147 break; 1148 } 1149 if ((t = getAndClearSlot(a, b++ & (cap - 1))) != null) { 1150 setBaseOpaque(b); 1151 break; 1152 } 1153 } 1154 } 1155 return t; 1156 } 1157 1158 /** 1159 * Takes next task, if one exists, using configured mode. 1160 */ 1161 final ForkJoinTask<?> nextLocalTask() { 1162 return nextLocalTask(config); 1163 } 1164 1165 /** 1166 * Returns next task, if one exists, in order specified by mode. 1167 */ 1168 final ForkJoinTask<?> peek() { 1169 VarHandle.acquireFence(); 1170 int cap; ForkJoinTask<?>[] a; 1171 return ((a = array) != null && (cap = a.length) > 0) ? 1172 a[(cap - 1) & ((config & FIFO) != 0 ? base : top - 1)] : null; 1173 } 1174 1175 // specialized execution methods 1176 1177 /** 1178 * Runs the given (stolen) task if nonnull, as well as 1179 * remaining local tasks and/or others available from the 1180 * given queue. 1181 */ 1182 final void topLevelExec(ForkJoinTask<?> task, WorkQueue q) { 1183 int cfg = config, nstolen = 1; 1184 while (task != null) { 1185 task.doExec(); 1186 if ((task = nextLocalTask(cfg)) == null && 1187 q != null && (task = q.tryPoll()) != null) 1188 ++nstolen; 1189 } 1190 nsteals += nstolen; 1191 source = 0; 1192 if ((cfg & INNOCUOUS) != 0) 1193 ThreadLocalRandom.eraseThreadLocals(Thread.currentThread()); 1194 } 1195 1196 /** 1197 * Tries to pop and run tasks within the target's computation 1198 * until done, not found, or limit exceeded. 1199 * 1200 * @param task root of CountedCompleter computation 1201 * @param owned true if owned by a ForkJoinWorkerThread 1202 * @param limit max runs, or zero for no limit 1203 * @return task status on exit 1204 */ 1205 final int helpComplete(ForkJoinTask<?> task, boolean owned, int limit) { 1206 int status = 0, cap, k, p, s; ForkJoinTask<?>[] a; ForkJoinTask<?> t; 1207 while (task != null && (status = task.status) >= 0 && 1208 (a = array) != null && (cap = a.length) > 0 && 1209 (t = a[k = (cap - 1) & (s = (p = top) - 1)]) 1210 instanceof CountedCompleter) { 1211 CountedCompleter<?> f = (CountedCompleter<?>)t; 1212 boolean taken = false; 1213 for (;;) { // exec if root task is a completer of t 1214 if (f == task) { 1215 if (owned) { 1216 if ((taken = casSlotToNull(a, k, t))) 1217 top = s; 1218 } 1219 else if (tryLock()) { 1220 if (top == p && array == a && 1221 (taken = casSlotToNull(a, k, t))) 1222 top = s; 1223 source = 0; 1224 } 1225 if (taken) 1226 t.doExec(); 1227 else if (!owned) 1228 Thread.yield(); // tryLock failure 1229 break; 1230 } 1231 else if ((f = f.completer) == null) 1232 break; 1233 } 1234 if (taken && limit != 0 && --limit == 0) 1235 break; 1236 } 1237 return status; 1238 } 1239 1240 /** 1241 * Tries to poll and run AsynchronousCompletionTasks until 1242 * none found or blocker is released. 1243 * 1244 * @param blocker the blocker 1245 */ 1246 final void helpAsyncBlocker(ManagedBlocker blocker) { 1247 int cap, b, d, k; ForkJoinTask<?>[] a; ForkJoinTask<?> t; 1248 while (blocker != null && (d = top - (b = base)) > 0 && 1249 (a = array) != null && (cap = a.length) > 0 && 1250 (((t = getSlot(a, k = (cap - 1) & b)) == null && d > 1) || 1251 t instanceof 1252 CompletableFuture.AsynchronousCompletionTask) && 1253 !blocker.isReleasable()) { 1254 if (t != null && base == b++ && casSlotToNull(a, k, t)) { 1255 setBaseOpaque(b); 1256 t.doExec(); 1257 } 1258 } 1259 } 1260 1261 // misc 1262 1263 /** AccessControlContext for innocuous workers, created on 1st use. */ 1264 @SuppressWarnings("removal") 1265 private static AccessControlContext INNOCUOUS_ACC; 1266 1267 /** 1268 * Initializes (upon registration) InnocuousForkJoinWorkerThreads. 1269 */ 1270 @SuppressWarnings("removal") 1271 final void initializeInnocuousWorker() { 1272 AccessControlContext acc; // racy construction OK 1273 if ((acc = INNOCUOUS_ACC) == null) 1274 INNOCUOUS_ACC = acc = new AccessControlContext( 1275 new ProtectionDomain[] { new ProtectionDomain(null, null) }); 1276 Thread t = Thread.currentThread(); 1277 ThreadLocalRandom.setInheritedAccessControlContext(t, acc); 1278 ThreadLocalRandom.eraseThreadLocals(t); 1279 } 1280 1281 /** 1282 * Returns true if owned by a worker thread and not known to be blocked. 1283 */ 1284 final boolean isApparentlyUnblocked() { 1285 Thread wt; Thread.State s; 1286 return ((wt = owner) != null && 1287 (s = wt.getState()) != Thread.State.BLOCKED && 1288 s != Thread.State.WAITING && 1289 s != Thread.State.TIMED_WAITING); 1290 } 1291 1292 static { 1293 try { 1294 QA = MethodHandles.arrayElementVarHandle(ForkJoinTask[].class); 1295 MethodHandles.Lookup l = MethodHandles.lookup(); 1296 SOURCE = l.findVarHandle(WorkQueue.class, "source", int.class); 1297 BASE = l.findVarHandle(WorkQueue.class, "base", int.class); 1298 } catch (ReflectiveOperationException e) { 1299 throw new ExceptionInInitializerError(e); 1300 } 1301 } 1302 } 1303 1304 // static fields (initialized in static initializer below) 1305 1306 /** 1307 * Creates a new ForkJoinWorkerThread. This factory is used unless 1308 * overridden in ForkJoinPool constructors. 1309 */ 1310 public static final ForkJoinWorkerThreadFactory 1311 defaultForkJoinWorkerThreadFactory; 1312 1313 /** 1314 * Permission required for callers of methods that may start or 1315 * kill threads. 1316 */ 1317 static final RuntimePermission modifyThreadPermission; 1318 1319 /** 1320 * Common (static) pool. Non-null for public use unless a static 1321 * construction exception, but internal usages null-check on use 1322 * to paranoically avoid potential initialization circularities 1323 * as well as to simplify generated code. 1324 */ 1325 static final ForkJoinPool common; 1326 1327 /** 1328 * Common pool parallelism. To allow simpler use and management 1329 * when common pool threads are disabled, we allow the underlying 1330 * common.parallelism field to be zero, but in that case still report 1331 * parallelism as 1 to reflect resulting caller-runs mechanics. 1332 */ 1333 static final int COMMON_PARALLELISM; 1334 1335 /** 1336 * Limit on spare thread construction in tryCompensate. 1337 */ 1338 private static final int COMMON_MAX_SPARES; 1339 1340 /** 1341 * Sequence number for creating worker names 1342 */ 1343 private static volatile int poolIds; 1344 1345 // static configuration constants 1346 1347 /** 1348 * Default idle timeout value (in milliseconds) for the thread 1349 * triggering quiescence to park waiting for new work 1350 */ 1351 private static final long DEFAULT_KEEPALIVE = 60_000L; 1352 1353 /** 1354 * Undershoot tolerance for idle timeouts 1355 */ 1356 private static final long TIMEOUT_SLOP = 20L; 1357 1358 /** 1359 * The default value for COMMON_MAX_SPARES. Overridable using the 1360 * "java.util.concurrent.ForkJoinPool.common.maximumSpares" system 1361 * property. The default value is far in excess of normal 1362 * requirements, but also far short of MAX_CAP and typical OS 1363 * thread limits, so allows JVMs to catch misuse/abuse before 1364 * running out of resources needed to do so. 1365 */ 1366 private static final int DEFAULT_COMMON_MAX_SPARES = 256; 1367 1368 /* 1369 * Bits and masks for field ctl, packed with 4 16 bit subfields: 1370 * RC: Number of released (unqueued) workers minus target parallelism 1371 * TC: Number of total workers minus target parallelism 1372 * SS: version count and status of top waiting thread 1373 * ID: poolIndex of top of Treiber stack of waiters 1374 * 1375 * When convenient, we can extract the lower 32 stack top bits 1376 * (including version bits) as sp=(int)ctl. The offsets of counts 1377 * by the target parallelism and the positionings of fields makes 1378 * it possible to perform the most common checks via sign tests of 1379 * fields: When ac is negative, there are not enough unqueued 1380 * workers, when tc is negative, there are not enough total 1381 * workers. When sp is non-zero, there are waiting workers. To 1382 * deal with possibly negative fields, we use casts in and out of 1383 * "short" and/or signed shifts to maintain signedness. 1384 * 1385 * Because it occupies uppermost bits, we can add one release 1386 * count using getAndAdd of RC_UNIT, rather than CAS, when 1387 * returning from a blocked join. Other updates entail multiple 1388 * subfields and masking, requiring CAS. 1389 * 1390 * The limits packed in field "bounds" are also offset by the 1391 * parallelism level to make them comparable to the ctl rc and tc 1392 * fields. 1393 */ 1394 1395 // Lower and upper word masks 1396 private static final long SP_MASK = 0xffffffffL; 1397 private static final long UC_MASK = ~SP_MASK; 1398 1399 // Release counts 1400 private static final int RC_SHIFT = 48; 1401 private static final long RC_UNIT = 0x0001L << RC_SHIFT; 1402 private static final long RC_MASK = 0xffffL << RC_SHIFT; 1403 1404 // Total counts 1405 private static final int TC_SHIFT = 32; 1406 private static final long TC_UNIT = 0x0001L << TC_SHIFT; 1407 private static final long TC_MASK = 0xffffL << TC_SHIFT; 1408 private static final long ADD_WORKER = 0x0001L << (TC_SHIFT + 15); // sign 1409 1410 // Instance fields 1411 1412 final long keepAlive; // milliseconds before dropping if idle 1413 volatile long stealCount; // collects worker nsteals 1414 int scanRover; // advances across pollScan calls 1415 volatile int threadIds; // for worker thread names 1416 final int bounds; // min, max threads packed as shorts 1417 volatile int mode; // parallelism, runstate, queue mode 1418 WorkQueue[] queues; // main registry 1419 final ReentrantLock registrationLock; 1420 Condition termination; // lazily constructed 1421 final String workerNamePrefix; // null for common pool 1422 final ForkJoinWorkerThreadFactory factory; 1423 final UncaughtExceptionHandler ueh; // per-worker UEH 1424 final Predicate<? super ForkJoinPool> saturate; 1425 1426 @jdk.internal.vm.annotation.Contended("fjpctl") // segregate 1427 volatile long ctl; // main pool control 1428 1429 // Support for atomic operations 1430 private static final VarHandle CTL; 1431 private static final VarHandle MODE; 1432 private static final VarHandle THREADIDS; 1433 private static final VarHandle POOLIDS; 1434 private boolean compareAndSetCtl(long c, long v) { 1435 return CTL.compareAndSet(this, c, v); 1436 } 1437 private long compareAndExchangeCtl(long c, long v) { 1438 return (long)CTL.compareAndExchange(this, c, v); 1439 } 1440 private long getAndAddCtl(long v) { 1441 return (long)CTL.getAndAdd(this, v); 1442 } 1443 private int getAndBitwiseOrMode(int v) { 1444 return (int)MODE.getAndBitwiseOr(this, v); 1445 } 1446 private int getAndAddThreadIds(int x) { 1447 return (int)THREADIDS.getAndAdd(this, x); 1448 } 1449 private static int getAndAddPoolIds(int x) { 1450 return (int)POOLIDS.getAndAdd(x); 1451 } 1452 1453 // Creating, registering and deregistering workers 1454 1455 /** 1456 * Tries to construct and start one worker. Assumes that total 1457 * count has already been incremented as a reservation. Invokes 1458 * deregisterWorker on any failure. 1459 * 1460 * @return true if successful 1461 */ 1462 private boolean createWorker() { 1463 ForkJoinWorkerThreadFactory fac = factory; 1464 Throwable ex = null; 1465 ForkJoinWorkerThread wt = null; 1466 try { 1467 if (fac != null && (wt = fac.newThread(this)) != null) { 1468 wt.start(); 1469 return true; 1470 } 1471 } catch (Throwable rex) { 1472 ex = rex; 1473 } 1474 deregisterWorker(wt, ex); 1475 return false; 1476 } 1477 1478 /** 1479 * Provides a name for ForkJoinWorkerThread constructor. 1480 */ 1481 final String nextWorkerThreadName() { 1482 String prefix = workerNamePrefix; 1483 int tid = getAndAddThreadIds(1) + 1; 1484 if (prefix == null) // commonPool has no prefix 1485 prefix = "ForkJoinPool.commonPool-worker-"; 1486 return prefix.concat(Integer.toString(tid)); 1487 } 1488 1489 /** 1490 * Finishes initializing and records owned queue. 1491 * 1492 * @param w caller's WorkQueue 1493 */ 1494 final void registerWorker(WorkQueue w) { 1495 ReentrantLock lock = registrationLock; 1496 ThreadLocalRandom.localInit(); 1497 int seed = ThreadLocalRandom.getProbe(); 1498 if (w != null && lock != null) { 1499 int modebits = (mode & FIFO) | w.config; 1500 w.array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY]; 1501 w.stackPred = seed; // stash for runWorker 1502 if ((modebits & INNOCUOUS) != 0) 1503 w.initializeInnocuousWorker(); 1504 int id = (seed << 1) | 1; // initial index guess 1505 lock.lock(); 1506 try { 1507 WorkQueue[] qs; int n; // find queue index 1508 if ((qs = queues) != null && (n = qs.length) > 0) { 1509 int k = n, m = n - 1; 1510 for (; qs[id &= m] != null && k > 0; id -= 2, k -= 2); 1511 if (k == 0) 1512 id = n | 1; // resize below 1513 w.phase = w.config = id | modebits; // now publishable 1514 1515 if (id < n) 1516 qs[id] = w; 1517 else { // expand array 1518 int an = n << 1, am = an - 1; 1519 WorkQueue[] as = new WorkQueue[an]; 1520 as[id & am] = w; 1521 for (int j = 1; j < n; j += 2) 1522 as[j] = qs[j]; 1523 for (int j = 0; j < n; j += 2) { 1524 WorkQueue q; 1525 if ((q = qs[j]) != null) // shared queues may move 1526 as[q.config & am] = q; 1527 } 1528 VarHandle.releaseFence(); // fill before publish 1529 queues = as; 1530 } 1531 } 1532 } finally { 1533 lock.unlock(); 1534 } 1535 } 1536 } 1537 1538 /** 1539 * Final callback from terminating worker, as well as upon failure 1540 * to construct or start a worker. Removes record of worker from 1541 * array, and adjusts counts. If pool is shutting down, tries to 1542 * complete termination. 1543 * 1544 * @param wt the worker thread, or null if construction failed 1545 * @param ex the exception causing failure, or null if none 1546 */ 1547 final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) { 1548 ReentrantLock lock = registrationLock; 1549 WorkQueue w = null; 1550 int cfg = 0; 1551 if (wt != null && (w = wt.workQueue) != null && lock != null) { 1552 WorkQueue[] qs; int n, i; 1553 cfg = w.config; 1554 long ns = w.nsteals & 0xffffffffL; 1555 lock.lock(); // remove index from array 1556 if ((qs = queues) != null && (n = qs.length) > 0 && 1557 qs[i = cfg & (n - 1)] == w) 1558 qs[i] = null; 1559 stealCount += ns; // accumulate steals 1560 lock.unlock(); 1561 long c = ctl; 1562 if ((cfg & QUIET) == 0) // unless self-signalled, decrement counts 1563 do {} while (c != (c = compareAndExchangeCtl( 1564 c, ((RC_MASK & (c - RC_UNIT)) | 1565 (TC_MASK & (c - TC_UNIT)) | 1566 (SP_MASK & c))))); 1567 else if ((int)c == 0) // was dropped on timeout 1568 cfg = 0; // suppress signal if last 1569 for (ForkJoinTask<?> t; (t = w.pop()) != null; ) 1570 ForkJoinTask.cancelIgnoringExceptions(t); // cancel tasks 1571 } 1572 1573 if (!tryTerminate(false, false) && w != null && (cfg & SRC) != 0) 1574 signalWork(); // possibly replace worker 1575 if (ex != null) 1576 ForkJoinTask.rethrow(ex); 1577 } 1578 1579 /* 1580 * Tries to create or release a worker if too few are running. 1581 */ 1582 final void signalWork() { 1583 for (long c = ctl; c < 0L;) { 1584 int sp, i; WorkQueue[] qs; WorkQueue v; 1585 if ((sp = (int)c & ~UNSIGNALLED) == 0) { // no idle workers 1586 if ((c & ADD_WORKER) == 0L) // enough total workers 1587 break; 1588 if (c == (c = compareAndExchangeCtl( 1589 c, ((RC_MASK & (c + RC_UNIT)) | 1590 (TC_MASK & (c + TC_UNIT)))))) { 1591 createWorker(); 1592 break; 1593 } 1594 } 1595 else if ((qs = queues) == null) 1596 break; // unstarted/terminated 1597 else if (qs.length <= (i = sp & SMASK)) 1598 break; // terminated 1599 else if ((v = qs[i]) == null) 1600 break; // terminating 1601 else { 1602 long nc = (v.stackPred & SP_MASK) | (UC_MASK & (c + RC_UNIT)); 1603 Thread vt = v.owner; 1604 if (c == (c = compareAndExchangeCtl(c, nc))) { 1605 v.phase = sp; 1606 LockSupport.unpark(vt); // release idle worker 1607 break; 1608 } 1609 } 1610 } 1611 } 1612 1613 /** 1614 * Top-level runloop for workers, called by ForkJoinWorkerThread.run. 1615 * See above for explanation. 1616 * 1617 * @param w caller's WorkQueue (may be null on failed initialization) 1618 */ 1619 final void runWorker(WorkQueue w) { 1620 if (mode >= 0 && w != null) { // skip on failed init 1621 w.config |= SRC; // mark as valid source 1622 int r = w.stackPred, src = 0; // use seed from registerWorker 1623 do { 1624 r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift 1625 } while ((src = scan(w, src, r)) >= 0 || 1626 (src = awaitWork(w)) == 0); 1627 } 1628 } 1629 1630 /** 1631 * Scans for and if found executes top-level tasks: Tries to poll 1632 * each queue starting at a random index with random stride, 1633 * returning source id or retry indicator if contended or 1634 * inconsistent. 1635 * 1636 * @param w caller's WorkQueue 1637 * @param prevSrc the previous queue stolen from in current phase, or 0 1638 * @param r random seed 1639 * @return id of queue if taken, negative if none found, prevSrc for retry 1640 */ 1641 private int scan(WorkQueue w, int prevSrc, int r) { 1642 WorkQueue[] qs = queues; 1643 int n = (w == null || qs == null) ? 0 : qs.length; 1644 for (int step = (r >>> 16) | 1, i = n; i > 0; --i, r += step) { 1645 int j, cap, b; WorkQueue q; ForkJoinTask<?>[] a; 1646 if ((q = qs[j = r & (n - 1)]) != null && // poll at qs[j].array[k] 1647 (a = q.array) != null && (cap = a.length) > 0) { 1648 int k = (cap - 1) & (b = q.base), nextBase = b + 1; 1649 int nextIndex = (cap - 1) & nextBase, src = j | SRC; 1650 ForkJoinTask<?> t = WorkQueue.getSlot(a, k); 1651 if (q.base != b) // inconsistent 1652 return prevSrc; 1653 else if (t != null && WorkQueue.casSlotToNull(a, k, t)) { 1654 q.base = nextBase; 1655 ForkJoinTask<?> next = a[nextIndex]; 1656 if ((w.source = src) != prevSrc && next != null) 1657 signalWork(); // propagate 1658 w.topLevelExec(t, q); 1659 return src; 1660 } 1661 else if (a[nextIndex] != null) // revisit 1662 return prevSrc; 1663 } 1664 } 1665 return (queues != qs) ? prevSrc: -1; // possibly resized 1666 } 1667 1668 /** 1669 * Advances worker phase, pushes onto ctl stack, and awaits signal 1670 * or reports termination. 1671 * 1672 * @return negative if terminated, else 0 1673 */ 1674 private int awaitWork(WorkQueue w) { 1675 if (w == null) 1676 return -1; // already terminated 1677 int phase = (w.phase + SS_SEQ) & ~UNSIGNALLED; 1678 w.phase = phase | UNSIGNALLED; // advance phase 1679 long prevCtl = ctl, c; // enqueue 1680 do { 1681 w.stackPred = (int)prevCtl; 1682 c = ((prevCtl - RC_UNIT) & UC_MASK) | (phase & SP_MASK); 1683 } while (prevCtl != (prevCtl = compareAndExchangeCtl(prevCtl, c))); 1684 1685 Thread.interrupted(); // clear status 1686 LockSupport.setCurrentBlocker(this); // prepare to block (exit also OK) 1687 long deadline = 0L; // nonzero if possibly quiescent 1688 int ac = (int)(c >> RC_SHIFT), md; 1689 if ((md = mode) < 0) // pool is terminating 1690 return -1; 1691 else if ((md & SMASK) + ac <= 0) { 1692 boolean checkTermination = (md & SHUTDOWN) != 0; 1693 if ((deadline = System.currentTimeMillis() + keepAlive) == 0L) 1694 deadline = 1L; // avoid zero 1695 WorkQueue[] qs = queues; // check for racing submission 1696 int n = (qs == null) ? 0 : qs.length; 1697 for (int i = 0; i < n; i += 2) { 1698 WorkQueue q; ForkJoinTask<?>[] a; int cap, b; 1699 if (ctl != c) { // already signalled 1700 checkTermination = false; 1701 break; 1702 } 1703 else if ((q = qs[i]) != null && 1704 (a = q.array) != null && (cap = a.length) > 0 && 1705 ((b = q.base) != q.top || a[(cap - 1) & b] != null || 1706 q.source != 0)) { 1707 if (compareAndSetCtl(c, prevCtl)) 1708 w.phase = phase; // self-signal 1709 checkTermination = false; 1710 break; 1711 } 1712 } 1713 if (checkTermination && tryTerminate(false, false)) 1714 return -1; // trigger quiescent termination 1715 } 1716 1717 for (boolean alt = false;;) { // await activation or termination 1718 if (w.phase >= 0) 1719 break; 1720 else if (mode < 0) 1721 return -1; 1722 else if ((c = ctl) == prevCtl) 1723 Thread.onSpinWait(); // signal in progress 1724 else if (!(alt = !alt)) // check between park calls 1725 Thread.interrupted(); 1726 else if (deadline == 0L) 1727 LockSupport.park(); 1728 else if (deadline - System.currentTimeMillis() > TIMEOUT_SLOP) 1729 LockSupport.parkUntil(deadline); 1730 else if (((int)c & SMASK) == (w.config & SMASK) && 1731 compareAndSetCtl(c, ((UC_MASK & (c - TC_UNIT)) | 1732 (prevCtl & SP_MASK)))) { 1733 w.config |= QUIET; // sentinel for deregisterWorker 1734 return -1; // drop on timeout 1735 } 1736 else if ((deadline += keepAlive) == 0L) 1737 deadline = 1L; // not at head; restart timer 1738 } 1739 return 0; 1740 } 1741 1742 // Utilities used by ForkJoinTask 1743 1744 /** 1745 * Returns true if can start terminating if enabled, or already terminated 1746 */ 1747 final boolean canStop() { 1748 outer: for (long oldSum = 0L;;) { // repeat until stable 1749 int md; WorkQueue[] qs; long c; 1750 if ((qs = queues) == null || ((md = mode) & STOP) != 0) 1751 return true; 1752 if ((md & SMASK) + (int)((c = ctl) >> RC_SHIFT) > 0) 1753 break; 1754 long checkSum = c; 1755 for (int i = 1; i < qs.length; i += 2) { // scan submitters 1756 WorkQueue q; ForkJoinTask<?>[] a; int s = 0, cap; 1757 if ((q = qs[i]) != null && (a = q.array) != null && 1758 (cap = a.length) > 0 && 1759 ((s = q.top) != q.base || a[(cap - 1) & s] != null || 1760 q.source != 0)) 1761 break outer; 1762 checkSum += (((long)i) << 32) ^ s; 1763 } 1764 if (oldSum == (oldSum = checkSum) && queues == qs) 1765 return true; 1766 } 1767 return (mode & STOP) != 0; // recheck mode on false return 1768 } 1769 1770 /** 1771 * Tries to decrement counts (sometimes implicitly) and possibly 1772 * arrange for a compensating worker in preparation for 1773 * blocking. May fail due to interference, in which case -1 is 1774 * returned so caller may retry. A zero return value indicates 1775 * that the caller doesn't need to re-adjust counts when later 1776 * unblocked. 1777 * 1778 * @param c incoming ctl value 1779 * @return UNCOMPENSATE: block then adjust, 0: block, -1 : retry 1780 */ 1781 private int tryCompensate(long c) { 1782 Predicate<? super ForkJoinPool> sat; 1783 int md = mode, b = bounds; 1784 // counts are signed; centered at parallelism level == 0 1785 int minActive = (short)(b & SMASK), 1786 maxTotal = b >>> SWIDTH, 1787 active = (int)(c >> RC_SHIFT), 1788 total = (short)(c >>> TC_SHIFT), 1789 sp = (int)c & ~UNSIGNALLED; 1790 if ((md & SMASK) == 0) 1791 return 0; // cannot compensate if parallelism zero 1792 else if (total >= 0) { 1793 if (sp != 0) { // activate idle worker 1794 WorkQueue[] qs; int n; WorkQueue v; 1795 if ((qs = queues) != null && (n = qs.length) > 0 && 1796 (v = qs[sp & (n - 1)]) != null) { 1797 Thread vt = v.owner; 1798 long nc = ((long)v.stackPred & SP_MASK) | (UC_MASK & c); 1799 if (compareAndSetCtl(c, nc)) { 1800 v.phase = sp; 1801 LockSupport.unpark(vt); 1802 return UNCOMPENSATE; 1803 } 1804 } 1805 return -1; // retry 1806 } 1807 else if (active > minActive) { // reduce parallelism 1808 long nc = ((RC_MASK & (c - RC_UNIT)) | (~RC_MASK & c)); 1809 return compareAndSetCtl(c, nc) ? UNCOMPENSATE : -1; 1810 } 1811 } 1812 if (total < maxTotal) { // expand pool 1813 long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK); 1814 return (!compareAndSetCtl(c, nc) ? -1 : 1815 !createWorker() ? 0 : UNCOMPENSATE); 1816 } 1817 else if (!compareAndSetCtl(c, c)) // validate 1818 return -1; 1819 else if ((sat = saturate) != null && sat.test(this)) 1820 return 0; 1821 else 1822 throw new RejectedExecutionException( 1823 "Thread limit exceeded replacing blocked worker"); 1824 } 1825 1826 /** 1827 * Readjusts RC count; called from ForkJoinTask after blocking. 1828 */ 1829 final void uncompensate() { 1830 getAndAddCtl(RC_UNIT); 1831 } 1832 1833 /** 1834 * Helps if possible until the given task is done. Scans other 1835 * queues for a task produced by one of w's stealers; returning 1836 * compensated blocking sentinel if none are found. 1837 * 1838 * @param task the task 1839 * @param w caller's WorkQueue 1840 * @param canHelp if false, compensate only 1841 * @return task status on exit, or UNCOMPENSATE for compensated blocking 1842 */ 1843 final int helpJoin(ForkJoinTask<?> task, WorkQueue w, boolean canHelp) { 1844 int s = 0; 1845 if (task != null && w != null) { 1846 int wsrc = w.source, wid = w.config & SMASK, r = wid + 2; 1847 boolean scan = true; 1848 long c = 0L; // track ctl stability 1849 outer: for (;;) { 1850 if ((s = task.status) < 0) 1851 break; 1852 else if (scan = !scan) { // previous scan was empty 1853 if (mode < 0) 1854 ForkJoinTask.cancelIgnoringExceptions(task); 1855 else if (c == (c = ctl) && (s = tryCompensate(c)) >= 0) 1856 break; // block 1857 } 1858 else if (canHelp) { // scan for subtasks 1859 WorkQueue[] qs = queues; 1860 int n = (qs == null) ? 0 : qs.length, m = n - 1; 1861 for (int i = n; i > 0; i -= 2, r += 2) { 1862 int j; WorkQueue q, x, y; ForkJoinTask<?>[] a; 1863 if ((q = qs[j = r & m]) != null) { 1864 int sq = q.source & SMASK, cap, b; 1865 if ((a = q.array) != null && (cap = a.length) > 0) { 1866 int k = (cap - 1) & (b = q.base); 1867 int nextBase = b + 1, src = j | SRC, sx; 1868 ForkJoinTask<?> t = WorkQueue.getSlot(a, k); 1869 boolean eligible = sq == wid || 1870 ((x = qs[sq & m]) != null && // indirect 1871 ((sx = (x.source & SMASK)) == wid || 1872 ((y = qs[sx & m]) != null && // 2-indirect 1873 (y.source & SMASK) == wid))); 1874 if ((s = task.status) < 0) 1875 break outer; 1876 else if ((q.source & SMASK) != sq || 1877 q.base != b) 1878 scan = true; // inconsistent 1879 else if (t == null) 1880 scan |= (a[nextBase & (cap - 1)] != null || 1881 q.top != b); // lagging 1882 else if (eligible) { 1883 if (WorkQueue.casSlotToNull(a, k, t)) { 1884 q.base = nextBase; 1885 w.source = src; 1886 t.doExec(); 1887 w.source = wsrc; 1888 } 1889 scan = true; 1890 break; 1891 } 1892 } 1893 } 1894 } 1895 } 1896 } 1897 } 1898 return s; 1899 } 1900 1901 /** 1902 * Extra helpJoin steps for CountedCompleters. Scans for and runs 1903 * subtasks of the given root task, returning if none are found. 1904 * 1905 * @param task root of CountedCompleter computation 1906 * @param w caller's WorkQueue 1907 * @param owned true if owned by a ForkJoinWorkerThread 1908 * @return task status on exit 1909 */ 1910 final int helpComplete(ForkJoinTask<?> task, WorkQueue w, boolean owned) { 1911 int s = 0; 1912 if (task != null && w != null) { 1913 int r = w.config; 1914 boolean scan = true, locals = true; 1915 long c = 0L; 1916 outer: for (;;) { 1917 if (locals) { // try locals before scanning 1918 if ((s = w.helpComplete(task, owned, 0)) < 0) 1919 break; 1920 locals = false; 1921 } 1922 else if ((s = task.status) < 0) 1923 break; 1924 else if (scan = !scan) { 1925 if (c == (c = ctl)) 1926 break; 1927 } 1928 else { // scan for subtasks 1929 WorkQueue[] qs = queues; 1930 int n = (qs == null) ? 0 : qs.length; 1931 for (int i = n; i > 0; --i, ++r) { 1932 int j, cap, b; WorkQueue q; ForkJoinTask<?>[] a; 1933 boolean eligible = false; 1934 if ((q = qs[j = r & (n - 1)]) != null && 1935 (a = q.array) != null && (cap = a.length) > 0) { 1936 int k = (cap - 1) & (b = q.base), nextBase = b + 1; 1937 ForkJoinTask<?> t = WorkQueue.getSlot(a, k); 1938 if (t instanceof CountedCompleter) { 1939 CountedCompleter<?> f = (CountedCompleter<?>)t; 1940 do {} while (!(eligible = (f == task)) && 1941 (f = f.completer) != null); 1942 } 1943 if ((s = task.status) < 0) 1944 break outer; 1945 else if (q.base != b) 1946 scan = true; // inconsistent 1947 else if (t == null) 1948 scan |= (a[nextBase & (cap - 1)] != null || 1949 q.top != b); 1950 else if (eligible) { 1951 if (WorkQueue.casSlotToNull(a, k, t)) { 1952 q.setBaseOpaque(nextBase); 1953 t.doExec(); 1954 locals = true; 1955 } 1956 scan = true; 1957 break; 1958 } 1959 } 1960 } 1961 } 1962 } 1963 } 1964 return s; 1965 } 1966 1967 /** 1968 * Scans for and returns a polled task, if available. Used only 1969 * for untracked polls. Begins scan at an index (scanRover) 1970 * advanced on each call, to avoid systematic unfairness. 1971 * 1972 * @param submissionsOnly if true, only scan submission queues 1973 */ 1974 private ForkJoinTask<?> pollScan(boolean submissionsOnly) { 1975 VarHandle.acquireFence(); 1976 int r = scanRover += 0x61c88647; // Weyl increment; raciness OK 1977 if (submissionsOnly) // even indices only 1978 r &= ~1; 1979 int step = (submissionsOnly) ? 2 : 1; 1980 WorkQueue[] qs; int n; 1981 while ((qs = queues) != null && (n = qs.length) > 0) { 1982 boolean scan = false; 1983 for (int i = 0; i < n; i += step) { 1984 int j, cap, b; WorkQueue q; ForkJoinTask<?>[] a; 1985 if ((q = qs[j = (n - 1) & (r + i)]) != null && 1986 (a = q.array) != null && (cap = a.length) > 0) { 1987 int k = (cap - 1) & (b = q.base), nextBase = b + 1; 1988 ForkJoinTask<?> t = WorkQueue.getSlot(a, k); 1989 if (q.base != b) 1990 scan = true; 1991 else if (t == null) 1992 scan |= (q.top != b || a[nextBase & (cap - 1)] != null); 1993 else if (!WorkQueue.casSlotToNull(a, k, t)) 1994 scan = true; 1995 else { 1996 q.setBaseOpaque(nextBase); 1997 return t; 1998 } 1999 } 2000 } 2001 if (!scan && queues == qs) 2002 break; 2003 } 2004 return null; 2005 } 2006 2007 /** 2008 * Runs tasks until {@code isQuiescent()}. Rather than blocking 2009 * when tasks cannot be found, rescans until all others cannot 2010 * find tasks either. 2011 * 2012 * @param nanos max wait time (Long.MAX_VALUE if effectively untimed) 2013 * @param interruptible true if return on interrupt 2014 * @return positive if quiescent, negative if interrupted, else 0 2015 */ 2016 final int helpQuiescePool(WorkQueue w, long nanos, boolean interruptible) { 2017 if (w == null) 2018 return 0; 2019 long startTime = System.nanoTime(), parkTime = 0L; 2020 int prevSrc = w.source, wsrc = prevSrc, cfg = w.config, r = cfg + 1; 2021 for (boolean active = true, locals = true;;) { 2022 boolean busy = false, scan = false; 2023 if (locals) { // run local tasks before (re)polling 2024 locals = false; 2025 for (ForkJoinTask<?> u; (u = w.nextLocalTask(cfg)) != null;) 2026 u.doExec(); 2027 } 2028 WorkQueue[] qs = queues; 2029 int n = (qs == null) ? 0 : qs.length; 2030 for (int i = n; i > 0; --i, ++r) { 2031 int j, b, cap; WorkQueue q; ForkJoinTask<?>[] a; 2032 if ((q = qs[j = (n - 1) & r]) != null && q != w && 2033 (a = q.array) != null && (cap = a.length) > 0) { 2034 int k = (cap - 1) & (b = q.base); 2035 int nextBase = b + 1, src = j | SRC; 2036 ForkJoinTask<?> t = WorkQueue.getSlot(a, k); 2037 if (q.base != b) 2038 busy = scan = true; 2039 else if (t != null) { 2040 busy = scan = true; 2041 if (!active) { // increment before taking 2042 active = true; 2043 getAndAddCtl(RC_UNIT); 2044 } 2045 if (WorkQueue.casSlotToNull(a, k, t)) { 2046 q.base = nextBase; 2047 w.source = src; 2048 t.doExec(); 2049 w.source = wsrc = prevSrc; 2050 locals = true; 2051 } 2052 break; 2053 } 2054 else if (!busy) { 2055 if (q.top != b || a[nextBase & (cap - 1)] != null) 2056 busy = scan = true; 2057 else if (q.source != QUIET && q.phase >= 0) 2058 busy = true; 2059 } 2060 } 2061 } 2062 VarHandle.acquireFence(); 2063 if (!scan && queues == qs) { 2064 boolean interrupted; 2065 if (!busy) { 2066 w.source = prevSrc; 2067 if (!active) 2068 getAndAddCtl(RC_UNIT); 2069 return 1; 2070 } 2071 if (wsrc != QUIET) 2072 w.source = wsrc = QUIET; 2073 if (active) { // decrement 2074 active = false; 2075 parkTime = 0L; 2076 getAndAddCtl(RC_MASK & -RC_UNIT); 2077 } 2078 else if (parkTime == 0L) { 2079 parkTime = 1L << 10; // initially about 1 usec 2080 Thread.yield(); 2081 } 2082 else if ((interrupted = interruptible && Thread.interrupted()) || 2083 System.nanoTime() - startTime > nanos) { 2084 getAndAddCtl(RC_UNIT); 2085 return interrupted ? -1 : 0; 2086 } 2087 else { 2088 LockSupport.parkNanos(this, parkTime); 2089 if (parkTime < nanos >>> 8 && parkTime < 1L << 20) 2090 parkTime <<= 1; // max sleep approx 1 sec or 1% nanos 2091 } 2092 } 2093 } 2094 } 2095 2096 /** 2097 * Helps quiesce from external caller until done, interrupted, or timeout 2098 * 2099 * @param nanos max wait time (Long.MAX_VALUE if effectively untimed) 2100 * @param interruptible true if return on interrupt 2101 * @return positive if quiescent, negative if interrupted, else 0 2102 */ 2103 final int externalHelpQuiescePool(long nanos, boolean interruptible) { 2104 for (long startTime = System.nanoTime(), parkTime = 0L;;) { 2105 ForkJoinTask<?> t; 2106 if ((t = pollScan(false)) != null) { 2107 t.doExec(); 2108 parkTime = 0L; 2109 } 2110 else if (canStop()) 2111 return 1; 2112 else if (parkTime == 0L) { 2113 parkTime = 1L << 10; 2114 Thread.yield(); 2115 } 2116 else if ((System.nanoTime() - startTime) > nanos) 2117 return 0; 2118 else if (interruptible && Thread.interrupted()) 2119 return -1; 2120 else { 2121 LockSupport.parkNanos(this, parkTime); 2122 if (parkTime < nanos >>> 8 && parkTime < 1L << 20) 2123 parkTime <<= 1; 2124 } 2125 } 2126 } 2127 2128 /** 2129 * Gets and removes a local or stolen task for the given worker. 2130 * 2131 * @return a task, if available 2132 */ 2133 final ForkJoinTask<?> nextTaskFor(WorkQueue w) { 2134 ForkJoinTask<?> t; 2135 if (w == null || (t = w.nextLocalTask(w.config)) == null) 2136 t = pollScan(false); 2137 return t; 2138 } 2139 2140 // External operations 2141 2142 /** 2143 * Finds and locks a WorkQueue for an external submitter, or 2144 * returns null if shutdown or terminating. 2145 */ 2146 final WorkQueue submissionQueue() { 2147 int r; 2148 if ((r = ThreadLocalRandom.getProbe()) == 0) { 2149 ThreadLocalRandom.localInit(); // initialize caller's probe 2150 r = ThreadLocalRandom.getProbe(); 2151 } 2152 for (int id = r << 1;;) { // even indices only 2153 int md = mode, n, i; WorkQueue q; ReentrantLock lock; 2154 WorkQueue[] qs = queues; 2155 if ((md & SHUTDOWN) != 0 || qs == null || (n = qs.length) <= 0) 2156 return null; 2157 else if ((q = qs[i = (n - 1) & id]) == null) { 2158 if ((lock = registrationLock) != null) { 2159 WorkQueue w = new WorkQueue(id | SRC); 2160 lock.lock(); // install under lock 2161 if (qs[i] == null) 2162 qs[i] = w; // else lost race; discard 2163 lock.unlock(); 2164 } 2165 } 2166 else if (!q.tryLock()) // move and restart 2167 id = (r = ThreadLocalRandom.advanceProbe(r)) << 1; 2168 else 2169 return q; 2170 } 2171 } 2172 2173 /** 2174 * Adds the given task to an external submission queue, or throws 2175 * exception if shutdown or terminating. 2176 * 2177 * @param task the task. Caller must ensure non-null. 2178 */ 2179 final void externalPush(ForkJoinTask<?> task) { 2180 WorkQueue q; 2181 if ((q = submissionQueue()) == null) 2182 throw new RejectedExecutionException(); // shutdown or disabled 2183 else if (q.lockedPush(task)) 2184 signalWork(); 2185 } 2186 2187 /** 2188 * Pushes a possibly-external submission. 2189 */ 2190 private <T> ForkJoinTask<T> externalSubmit(ForkJoinTask<T> task) { 2191 Thread t; ForkJoinWorkerThread wt; WorkQueue q; 2192 if (task == null) 2193 throw new NullPointerException(); 2194 if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) && 2195 (q = (wt = (ForkJoinWorkerThread)t).workQueue) != null && 2196 wt.pool == this) 2197 q.push(task, this); 2198 else 2199 externalPush(task); 2200 return task; 2201 } 2202 2203 /** 2204 * Returns common pool queue for an external thread that has 2205 * possibly ever submitted a common pool task (nonzero probe), or 2206 * null if none. 2207 */ 2208 static WorkQueue commonQueue() { 2209 ForkJoinPool p; WorkQueue[] qs; 2210 int r = ThreadLocalRandom.getProbe(), n; 2211 return ((p = common) != null && (qs = p.queues) != null && 2212 (n = qs.length) > 0 && r != 0) ? 2213 qs[(n - 1) & (r << 1)] : null; 2214 } 2215 2216 /** 2217 * Returns queue for an external thread, if one exists 2218 */ 2219 final WorkQueue externalQueue() { 2220 WorkQueue[] qs; 2221 int r = ThreadLocalRandom.getProbe(), n; 2222 return ((qs = queues) != null && (n = qs.length) > 0 && r != 0) ? 2223 qs[(n - 1) & (r << 1)] : null; 2224 } 2225 2226 /** 2227 * If the given executor is a ForkJoinPool, poll and execute 2228 * AsynchronousCompletionTasks from worker's queue until none are 2229 * available or blocker is released. 2230 */ 2231 static void helpAsyncBlocker(Executor e, ManagedBlocker blocker) { 2232 WorkQueue w = null; Thread t; ForkJoinWorkerThread wt; 2233 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) { 2234 if ((wt = (ForkJoinWorkerThread)t).pool == e) 2235 w = wt.workQueue; 2236 } 2237 else if (e instanceof ForkJoinPool) 2238 w = ((ForkJoinPool)e).externalQueue(); 2239 if (w != null) 2240 w.helpAsyncBlocker(blocker); 2241 } 2242 2243 /** 2244 * Returns a cheap heuristic guide for task partitioning when 2245 * programmers, frameworks, tools, or languages have little or no 2246 * idea about task granularity. In essence, by offering this 2247 * method, we ask users only about tradeoffs in overhead vs 2248 * expected throughput and its variance, rather than how finely to 2249 * partition tasks. 2250 * 2251 * In a steady state strict (tree-structured) computation, each 2252 * thread makes available for stealing enough tasks for other 2253 * threads to remain active. Inductively, if all threads play by 2254 * the same rules, each thread should make available only a 2255 * constant number of tasks. 2256 * 2257 * The minimum useful constant is just 1. But using a value of 1 2258 * would require immediate replenishment upon each steal to 2259 * maintain enough tasks, which is infeasible. Further, 2260 * partitionings/granularities of offered tasks should minimize 2261 * steal rates, which in general means that threads nearer the top 2262 * of computation tree should generate more than those nearer the 2263 * bottom. In perfect steady state, each thread is at 2264 * approximately the same level of computation tree. However, 2265 * producing extra tasks amortizes the uncertainty of progress and 2266 * diffusion assumptions. 2267 * 2268 * So, users will want to use values larger (but not much larger) 2269 * than 1 to both smooth over transient shortages and hedge 2270 * against uneven progress; as traded off against the cost of 2271 * extra task overhead. We leave the user to pick a threshold 2272 * value to compare with the results of this call to guide 2273 * decisions, but recommend values such as 3. 2274 * 2275 * When all threads are active, it is on average OK to estimate 2276 * surplus strictly locally. In steady-state, if one thread is 2277 * maintaining say 2 surplus tasks, then so are others. So we can 2278 * just use estimated queue length. However, this strategy alone 2279 * leads to serious mis-estimates in some non-steady-state 2280 * conditions (ramp-up, ramp-down, other stalls). We can detect 2281 * many of these by further considering the number of "idle" 2282 * threads, that are known to have zero queued tasks, so 2283 * compensate by a factor of (#idle/#active) threads. 2284 */ 2285 static int getSurplusQueuedTaskCount() { 2286 Thread t; ForkJoinWorkerThread wt; ForkJoinPool pool; WorkQueue q; 2287 if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) && 2288 (pool = (wt = (ForkJoinWorkerThread)t).pool) != null && 2289 (q = wt.workQueue) != null) { 2290 int p = pool.mode & SMASK; 2291 int a = p + (int)(pool.ctl >> RC_SHIFT); 2292 int n = q.top - q.base; 2293 return n - (a > (p >>>= 1) ? 0 : 2294 a > (p >>>= 1) ? 1 : 2295 a > (p >>>= 1) ? 2 : 2296 a > (p >>>= 1) ? 4 : 2297 8); 2298 } 2299 return 0; 2300 } 2301 2302 // Termination 2303 2304 /** 2305 * Possibly initiates and/or completes termination. 2306 * 2307 * @param now if true, unconditionally terminate, else only 2308 * if no work and no active workers 2309 * @param enable if true, terminate when next possible 2310 * @return true if terminating or terminated 2311 */ 2312 private boolean tryTerminate(boolean now, boolean enable) { 2313 int md; // try to set SHUTDOWN, then STOP, then help terminate 2314 if (((md = mode) & SHUTDOWN) == 0) { 2315 if (!enable) 2316 return false; 2317 md = getAndBitwiseOrMode(SHUTDOWN); 2318 } 2319 if ((md & STOP) == 0) { 2320 if (!now && !canStop()) 2321 return false; 2322 md = getAndBitwiseOrMode(STOP); 2323 } 2324 for (boolean rescan = true;;) { // repeat until no changes 2325 boolean changed = false; 2326 for (ForkJoinTask<?> t; (t = pollScan(false)) != null; ) { 2327 changed = true; 2328 ForkJoinTask.cancelIgnoringExceptions(t); // help cancel 2329 } 2330 WorkQueue[] qs; int n; WorkQueue q; Thread thread; 2331 if ((qs = queues) != null && (n = qs.length) > 0) { 2332 for (int j = 1; j < n; j += 2) { // unblock other workers 2333 if ((q = qs[j]) != null && (thread = q.owner) != null && 2334 !thread.isInterrupted()) { 2335 changed = true; 2336 try { 2337 thread.interrupt(); 2338 } catch (Throwable ignore) { 2339 } 2340 } 2341 } 2342 } 2343 ReentrantLock lock; Condition cond; // signal when no workers 2344 if (((md = mode) & TERMINATED) == 0 && 2345 (md & SMASK) + (short)(ctl >>> TC_SHIFT) <= 0 && 2346 (getAndBitwiseOrMode(TERMINATED) & TERMINATED) == 0 && 2347 (lock = registrationLock) != null) { 2348 lock.lock(); 2349 if ((cond = termination) != null) 2350 cond.signalAll(); 2351 lock.unlock(); 2352 } 2353 if (changed) 2354 rescan = true; 2355 else if (rescan) 2356 rescan = false; 2357 else 2358 break; 2359 } 2360 return true; 2361 } 2362 2363 // Exported methods 2364 2365 // Constructors 2366 2367 /** 2368 * Creates a {@code ForkJoinPool} with parallelism equal to {@link 2369 * java.lang.Runtime#availableProcessors}, using defaults for all 2370 * other parameters (see {@link #ForkJoinPool(int, 2371 * ForkJoinWorkerThreadFactory, UncaughtExceptionHandler, boolean, 2372 * int, int, int, Predicate, long, TimeUnit)}). 2373 * 2374 * @throws SecurityException if a security manager exists and 2375 * the caller is not permitted to modify threads 2376 * because it does not hold {@link 2377 * java.lang.RuntimePermission}{@code ("modifyThread")} 2378 */ 2379 public ForkJoinPool() { 2380 this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()), 2381 defaultForkJoinWorkerThreadFactory, null, false, 2382 0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS); 2383 } 2384 2385 /** 2386 * Creates a {@code ForkJoinPool} with the indicated parallelism 2387 * level, using defaults for all other parameters (see {@link 2388 * #ForkJoinPool(int, ForkJoinWorkerThreadFactory, 2389 * UncaughtExceptionHandler, boolean, int, int, int, Predicate, 2390 * long, TimeUnit)}). 2391 * 2392 * @param parallelism the parallelism level 2393 * @throws IllegalArgumentException if parallelism less than or 2394 * equal to zero, or greater than implementation limit 2395 * @throws SecurityException if a security manager exists and 2396 * the caller is not permitted to modify threads 2397 * because it does not hold {@link 2398 * java.lang.RuntimePermission}{@code ("modifyThread")} 2399 */ 2400 public ForkJoinPool(int parallelism) { 2401 this(parallelism, defaultForkJoinWorkerThreadFactory, null, false, 2402 0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS); 2403 } 2404 2405 /** 2406 * Creates a {@code ForkJoinPool} with the given parameters (using 2407 * defaults for others -- see {@link #ForkJoinPool(int, 2408 * ForkJoinWorkerThreadFactory, UncaughtExceptionHandler, boolean, 2409 * int, int, int, Predicate, long, TimeUnit)}). 2410 * 2411 * @param parallelism the parallelism level. For default value, 2412 * use {@link java.lang.Runtime#availableProcessors}. 2413 * @param factory the factory for creating new threads. For default value, 2414 * use {@link #defaultForkJoinWorkerThreadFactory}. 2415 * @param handler the handler for internal worker threads that 2416 * terminate due to unrecoverable errors encountered while executing 2417 * tasks. For default value, use {@code null}. 2418 * @param asyncMode if true, 2419 * establishes local first-in-first-out scheduling mode for forked 2420 * tasks that are never joined. This mode may be more appropriate 2421 * than default locally stack-based mode in applications in which 2422 * worker threads only process event-style asynchronous tasks. 2423 * For default value, use {@code false}. 2424 * @throws IllegalArgumentException if parallelism less than or 2425 * equal to zero, or greater than implementation limit 2426 * @throws NullPointerException if the factory is null 2427 * @throws SecurityException if a security manager exists and 2428 * the caller is not permitted to modify threads 2429 * because it does not hold {@link 2430 * java.lang.RuntimePermission}{@code ("modifyThread")} 2431 */ 2432 public ForkJoinPool(int parallelism, 2433 ForkJoinWorkerThreadFactory factory, 2434 UncaughtExceptionHandler handler, 2435 boolean asyncMode) { 2436 this(parallelism, factory, handler, asyncMode, 2437 0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS); 2438 } 2439 2440 /** 2441 * Creates a {@code ForkJoinPool} with the given parameters. 2442 * 2443 * @param parallelism the parallelism level. For default value, 2444 * use {@link java.lang.Runtime#availableProcessors}. 2445 * 2446 * @param factory the factory for creating new threads. For 2447 * default value, use {@link #defaultForkJoinWorkerThreadFactory}. 2448 * 2449 * @param handler the handler for internal worker threads that 2450 * terminate due to unrecoverable errors encountered while 2451 * executing tasks. For default value, use {@code null}. 2452 * 2453 * @param asyncMode if true, establishes local first-in-first-out 2454 * scheduling mode for forked tasks that are never joined. This 2455 * mode may be more appropriate than default locally stack-based 2456 * mode in applications in which worker threads only process 2457 * event-style asynchronous tasks. For default value, use {@code 2458 * false}. 2459 * 2460 * @param corePoolSize the number of threads to keep in the pool 2461 * (unless timed out after an elapsed keep-alive). Normally (and 2462 * by default) this is the same value as the parallelism level, 2463 * but may be set to a larger value to reduce dynamic overhead if 2464 * tasks regularly block. Using a smaller value (for example 2465 * {@code 0}) has the same effect as the default. 2466 * 2467 * @param maximumPoolSize the maximum number of threads allowed. 2468 * When the maximum is reached, attempts to replace blocked 2469 * threads fail. (However, because creation and termination of 2470 * different threads may overlap, and may be managed by the given 2471 * thread factory, this value may be transiently exceeded.) To 2472 * arrange the same value as is used by default for the common 2473 * pool, use {@code 256} plus the {@code parallelism} level. (By 2474 * default, the common pool allows a maximum of 256 spare 2475 * threads.) Using a value (for example {@code 2476 * Integer.MAX_VALUE}) larger than the implementation's total 2477 * thread limit has the same effect as using this limit (which is 2478 * the default). 2479 * 2480 * @param minimumRunnable the minimum allowed number of core 2481 * threads not blocked by a join or {@link ManagedBlocker}. To 2482 * ensure progress, when too few unblocked threads exist and 2483 * unexecuted tasks may exist, new threads are constructed, up to 2484 * the given maximumPoolSize. For the default value, use {@code 2485 * 1}, that ensures liveness. A larger value might improve 2486 * throughput in the presence of blocked activities, but might 2487 * not, due to increased overhead. A value of zero may be 2488 * acceptable when submitted tasks cannot have dependencies 2489 * requiring additional threads. 2490 * 2491 * @param saturate if non-null, a predicate invoked upon attempts 2492 * to create more than the maximum total allowed threads. By 2493 * default, when a thread is about to block on a join or {@link 2494 * ManagedBlocker}, but cannot be replaced because the 2495 * maximumPoolSize would be exceeded, a {@link 2496 * RejectedExecutionException} is thrown. But if this predicate 2497 * returns {@code true}, then no exception is thrown, so the pool 2498 * continues to operate with fewer than the target number of 2499 * runnable threads, which might not ensure progress. 2500 * 2501 * @param keepAliveTime the elapsed time since last use before 2502 * a thread is terminated (and then later replaced if needed). 2503 * For the default value, use {@code 60, TimeUnit.SECONDS}. 2504 * 2505 * @param unit the time unit for the {@code keepAliveTime} argument 2506 * 2507 * @throws IllegalArgumentException if parallelism is less than or 2508 * equal to zero, or is greater than implementation limit, 2509 * or if maximumPoolSize is less than parallelism, 2510 * of if the keepAliveTime is less than or equal to zero. 2511 * @throws NullPointerException if the factory is null 2512 * @throws SecurityException if a security manager exists and 2513 * the caller is not permitted to modify threads 2514 * because it does not hold {@link 2515 * java.lang.RuntimePermission}{@code ("modifyThread")} 2516 * @since 9 2517 */ 2518 public ForkJoinPool(int parallelism, 2519 ForkJoinWorkerThreadFactory factory, 2520 UncaughtExceptionHandler handler, 2521 boolean asyncMode, 2522 int corePoolSize, 2523 int maximumPoolSize, 2524 int minimumRunnable, 2525 Predicate<? super ForkJoinPool> saturate, 2526 long keepAliveTime, 2527 TimeUnit unit) { 2528 checkPermission(); 2529 int p = parallelism; 2530 if (p <= 0 || p > MAX_CAP || p > maximumPoolSize || keepAliveTime <= 0L) 2531 throw new IllegalArgumentException(); 2532 if (factory == null || unit == null) 2533 throw new NullPointerException(); 2534 this.factory = factory; 2535 this.ueh = handler; 2536 this.saturate = saturate; 2537 this.keepAlive = Math.max(unit.toMillis(keepAliveTime), TIMEOUT_SLOP); 2538 int size = 1 << (33 - Integer.numberOfLeadingZeros(p - 1)); 2539 int corep = Math.min(Math.max(corePoolSize, p), MAX_CAP); 2540 int maxSpares = Math.min(maximumPoolSize, MAX_CAP) - p; 2541 int minAvail = Math.min(Math.max(minimumRunnable, 0), MAX_CAP); 2542 this.bounds = ((minAvail - p) & SMASK) | (maxSpares << SWIDTH); 2543 this.mode = p | (asyncMode ? FIFO : 0); 2544 this.ctl = ((((long)(-corep) << TC_SHIFT) & TC_MASK) | 2545 (((long)(-p) << RC_SHIFT) & RC_MASK)); 2546 this.registrationLock = new ReentrantLock(); 2547 this.queues = new WorkQueue[size]; 2548 String pid = Integer.toString(getAndAddPoolIds(1) + 1); 2549 this.workerNamePrefix = "ForkJoinPool-" + pid + "-worker-"; 2550 } 2551 2552 // helper method for commonPool constructor 2553 private static Object newInstanceFromSystemProperty(String property) 2554 throws ReflectiveOperationException { 2555 String className = System.getProperty(property); 2556 return (className == null) 2557 ? null 2558 : ClassLoader.getSystemClassLoader().loadClass(className) 2559 .getConstructor().newInstance(); 2560 } 2561 2562 /** 2563 * Constructor for common pool using parameters possibly 2564 * overridden by system properties 2565 */ 2566 private ForkJoinPool(byte forCommonPoolOnly) { 2567 int parallelism = Math.max(1, Runtime.getRuntime().availableProcessors() - 1); 2568 ForkJoinWorkerThreadFactory fac = null; 2569 UncaughtExceptionHandler handler = null; 2570 try { // ignore exceptions in accessing/parsing properties 2571 fac = (ForkJoinWorkerThreadFactory) newInstanceFromSystemProperty( 2572 "java.util.concurrent.ForkJoinPool.common.threadFactory"); 2573 handler = (UncaughtExceptionHandler) newInstanceFromSystemProperty( 2574 "java.util.concurrent.ForkJoinPool.common.exceptionHandler"); 2575 String pp = System.getProperty 2576 ("java.util.concurrent.ForkJoinPool.common.parallelism"); 2577 if (pp != null) 2578 parallelism = Integer.parseInt(pp); 2579 } catch (Exception ignore) { 2580 } 2581 this.ueh = handler; 2582 this.keepAlive = DEFAULT_KEEPALIVE; 2583 this.saturate = null; 2584 this.workerNamePrefix = null; 2585 int p = Math.min(Math.max(parallelism, 0), MAX_CAP), size; 2586 this.mode = p; 2587 if (p > 0) { 2588 size = 1 << (33 - Integer.numberOfLeadingZeros(p - 1)); 2589 this.bounds = ((1 - p) & SMASK) | (COMMON_MAX_SPARES << SWIDTH); 2590 this.ctl = ((((long)(-p) << TC_SHIFT) & TC_MASK) | 2591 (((long)(-p) << RC_SHIFT) & RC_MASK)); 2592 } else { // zero min, max, spare counts, 1 slot 2593 size = 1; 2594 this.bounds = 0; 2595 this.ctl = 0L; 2596 } 2597 this.factory = (fac != null) ? fac : 2598 new DefaultCommonPoolForkJoinWorkerThreadFactory(); 2599 this.queues = new WorkQueue[size]; 2600 this.registrationLock = new ReentrantLock(); 2601 } 2602 2603 /** 2604 * Returns the common pool instance. This pool is statically 2605 * constructed; its run state is unaffected by attempts to {@link 2606 * #shutdown} or {@link #shutdownNow}. However this pool and any 2607 * ongoing processing are automatically terminated upon program 2608 * {@link System#exit}. Any program that relies on asynchronous 2609 * task processing to complete before program termination should 2610 * invoke {@code commonPool().}{@link #awaitQuiescence awaitQuiescence}, 2611 * before exit. 2612 * 2613 * @return the common pool instance 2614 * @since 1.8 2615 */ 2616 public static ForkJoinPool commonPool() { 2617 // assert common != null : "static init error"; 2618 return common; 2619 } 2620 2621 // Execution methods 2622 2623 /** 2624 * Performs the given task, returning its result upon completion. 2625 * If the computation encounters an unchecked Exception or Error, 2626 * it is rethrown as the outcome of this invocation. Rethrown 2627 * exceptions behave in the same way as regular exceptions, but, 2628 * when possible, contain stack traces (as displayed for example 2629 * using {@code ex.printStackTrace()}) of both the current thread 2630 * as well as the thread actually encountering the exception; 2631 * minimally only the latter. 2632 * 2633 * @param task the task 2634 * @param <T> the type of the task's result 2635 * @return the task's result 2636 * @throws NullPointerException if the task is null 2637 * @throws RejectedExecutionException if the task cannot be 2638 * scheduled for execution 2639 */ 2640 public <T> T invoke(ForkJoinTask<T> task) { 2641 externalSubmit(task); 2642 return task.joinForPoolInvoke(this); 2643 } 2644 2645 /** 2646 * Arranges for (asynchronous) execution of the given task. 2647 * 2648 * @param task the task 2649 * @throws NullPointerException if the task is null 2650 * @throws RejectedExecutionException if the task cannot be 2651 * scheduled for execution 2652 */ 2653 public void execute(ForkJoinTask<?> task) { 2654 externalSubmit(task); 2655 } 2656 2657 // AbstractExecutorService methods 2658 2659 /** 2660 * @throws NullPointerException if the task is null 2661 * @throws RejectedExecutionException if the task cannot be 2662 * scheduled for execution 2663 */ 2664 @Override 2665 @SuppressWarnings("unchecked") 2666 public void execute(Runnable task) { 2667 externalSubmit((task instanceof ForkJoinTask<?>) 2668 ? (ForkJoinTask<Void>) task // avoid re-wrap 2669 : new ForkJoinTask.RunnableExecuteAction(task)); 2670 } 2671 2672 /** 2673 * Submits a ForkJoinTask for execution. 2674 * 2675 * @param task the task to submit 2676 * @param <T> the type of the task's result 2677 * @return the task 2678 * @throws NullPointerException if the task is null 2679 * @throws RejectedExecutionException if the task cannot be 2680 * scheduled for execution 2681 */ 2682 public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) { 2683 return externalSubmit(task); 2684 } 2685 2686 /** 2687 * @throws NullPointerException if the task is null 2688 * @throws RejectedExecutionException if the task cannot be 2689 * scheduled for execution 2690 */ 2691 @Override 2692 public <T> ForkJoinTask<T> submit(Callable<T> task) { 2693 return externalSubmit(new ForkJoinTask.AdaptedCallable<T>(task)); 2694 } 2695 2696 /** 2697 * @throws NullPointerException if the task is null 2698 * @throws RejectedExecutionException if the task cannot be 2699 * scheduled for execution 2700 */ 2701 @Override 2702 public <T> ForkJoinTask<T> submit(Runnable task, T result) { 2703 return externalSubmit(new ForkJoinTask.AdaptedRunnable<T>(task, result)); 2704 } 2705 2706 /** 2707 * @throws NullPointerException if the task is null 2708 * @throws RejectedExecutionException if the task cannot be 2709 * scheduled for execution 2710 */ 2711 @Override 2712 @SuppressWarnings("unchecked") 2713 public ForkJoinTask<?> submit(Runnable task) { 2714 return externalSubmit((task instanceof ForkJoinTask<?>) 2715 ? (ForkJoinTask<Void>) task // avoid re-wrap 2716 : new ForkJoinTask.AdaptedRunnableAction(task)); 2717 } 2718 2719 /** 2720 * @throws NullPointerException {@inheritDoc} 2721 * @throws RejectedExecutionException {@inheritDoc} 2722 */ 2723 @Override 2724 public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) { 2725 ArrayList<Future<T>> futures = new ArrayList<>(tasks.size()); 2726 try { 2727 for (Callable<T> t : tasks) { 2728 ForkJoinTask<T> f = 2729 new ForkJoinTask.AdaptedInterruptibleCallable<T>(t); 2730 futures.add(f); 2731 externalSubmit(f); 2732 } 2733 for (int i = futures.size() - 1; i >= 0; --i) 2734 ((ForkJoinTask<?>)futures.get(i)).awaitPoolInvoke(this); 2735 return futures; 2736 } catch (Throwable t) { 2737 for (Future<T> e : futures) 2738 ForkJoinTask.cancelIgnoringExceptions(e); 2739 throw t; 2740 } 2741 } 2742 2743 @Override 2744 public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, 2745 long timeout, TimeUnit unit) 2746 throws InterruptedException { 2747 long nanos = unit.toNanos(timeout); 2748 ArrayList<Future<T>> futures = new ArrayList<>(tasks.size()); 2749 try { 2750 for (Callable<T> t : tasks) { 2751 ForkJoinTask<T> f = 2752 new ForkJoinTask.AdaptedInterruptibleCallable<T>(t); 2753 futures.add(f); 2754 externalSubmit(f); 2755 } 2756 long startTime = System.nanoTime(), ns = nanos; 2757 boolean timedOut = (ns < 0L); 2758 for (int i = futures.size() - 1; i >= 0; --i) { 2759 Future<T> f = futures.get(i); 2760 if (!f.isDone()) { 2761 if (timedOut) 2762 ForkJoinTask.cancelIgnoringExceptions(f); 2763 else { 2764 ((ForkJoinTask<T>)f).awaitPoolInvoke(this, ns); 2765 if ((ns = nanos - (System.nanoTime() - startTime)) < 0L) 2766 timedOut = true; 2767 } 2768 } 2769 } 2770 return futures; 2771 } catch (Throwable t) { 2772 for (Future<T> e : futures) 2773 ForkJoinTask.cancelIgnoringExceptions(e); 2774 throw t; 2775 } 2776 } 2777 2778 // Task to hold results from InvokeAnyTasks 2779 static final class InvokeAnyRoot<E> extends ForkJoinTask<E> { 2780 private static final long serialVersionUID = 2838392045355241008L; 2781 @SuppressWarnings("serial") // Conditionally serializable 2782 volatile E result; 2783 final AtomicInteger count; // in case all throw 2784 final ForkJoinPool pool; // to check shutdown while collecting 2785 InvokeAnyRoot(int n, ForkJoinPool p) { 2786 pool = p; 2787 count = new AtomicInteger(n); 2788 } 2789 final void tryComplete(Callable<E> c) { // called by InvokeAnyTasks 2790 Throwable ex = null; 2791 boolean failed; 2792 if (c == null || Thread.interrupted() || 2793 (pool != null && pool.mode < 0)) 2794 failed = true; 2795 else if (isDone()) 2796 failed = false; 2797 else { 2798 try { 2799 complete(c.call()); 2800 failed = false; 2801 } catch (Throwable tx) { 2802 ex = tx; 2803 failed = true; 2804 } 2805 } 2806 if ((pool != null && pool.mode < 0) || 2807 (failed && count.getAndDecrement() <= 1)) 2808 trySetThrown(ex != null ? ex : new CancellationException()); 2809 } 2810 public final boolean exec() { return false; } // never forked 2811 public final E getRawResult() { return result; } 2812 public final void setRawResult(E v) { result = v; } 2813 } 2814 2815 // Variant of AdaptedInterruptibleCallable with results in InvokeAnyRoot 2816 static final class InvokeAnyTask<E> extends ForkJoinTask<E> { 2817 private static final long serialVersionUID = 2838392045355241008L; 2818 final InvokeAnyRoot<E> root; 2819 @SuppressWarnings("serial") // Conditionally serializable 2820 final Callable<E> callable; 2821 transient volatile Thread runner; 2822 InvokeAnyTask(InvokeAnyRoot<E> root, Callable<E> callable) { 2823 this.root = root; 2824 this.callable = callable; 2825 } 2826 public final boolean exec() { 2827 Thread.interrupted(); 2828 runner = Thread.currentThread(); 2829 root.tryComplete(callable); 2830 runner = null; 2831 Thread.interrupted(); 2832 return true; 2833 } 2834 public final boolean cancel(boolean mayInterruptIfRunning) { 2835 Thread t; 2836 boolean stat = super.cancel(false); 2837 if (mayInterruptIfRunning && (t = runner) != null) { 2838 try { 2839 t.interrupt(); 2840 } catch (Throwable ignore) { 2841 } 2842 } 2843 return stat; 2844 } 2845 public final void setRawResult(E v) {} // unused 2846 public final E getRawResult() { return null; } 2847 } 2848 2849 @Override 2850 public <T> T invokeAny(Collection<? extends Callable<T>> tasks) 2851 throws InterruptedException, ExecutionException { 2852 int n = tasks.size(); 2853 if (n <= 0) 2854 throw new IllegalArgumentException(); 2855 InvokeAnyRoot<T> root = new InvokeAnyRoot<T>(n, this); 2856 ArrayList<InvokeAnyTask<T>> fs = new ArrayList<>(n); 2857 try { 2858 for (Callable<T> c : tasks) { 2859 if (c == null) 2860 throw new NullPointerException(); 2861 InvokeAnyTask<T> f = new InvokeAnyTask<T>(root, c); 2862 fs.add(f); 2863 externalSubmit(f); 2864 if (root.isDone()) 2865 break; 2866 } 2867 return root.getForPoolInvoke(this); 2868 } finally { 2869 for (InvokeAnyTask<T> f : fs) 2870 ForkJoinTask.cancelIgnoringExceptions(f); 2871 } 2872 } 2873 2874 @Override 2875 public <T> T invokeAny(Collection<? extends Callable<T>> tasks, 2876 long timeout, TimeUnit unit) 2877 throws InterruptedException, ExecutionException, TimeoutException { 2878 long nanos = unit.toNanos(timeout); 2879 int n = tasks.size(); 2880 if (n <= 0) 2881 throw new IllegalArgumentException(); 2882 InvokeAnyRoot<T> root = new InvokeAnyRoot<T>(n, this); 2883 ArrayList<InvokeAnyTask<T>> fs = new ArrayList<>(n); 2884 try { 2885 for (Callable<T> c : tasks) { 2886 if (c == null) 2887 throw new NullPointerException(); 2888 InvokeAnyTask<T> f = new InvokeAnyTask<T>(root, c); 2889 fs.add(f); 2890 externalSubmit(f); 2891 if (root.isDone()) 2892 break; 2893 } 2894 return root.getForPoolInvoke(this, nanos); 2895 } finally { 2896 for (InvokeAnyTask<T> f : fs) 2897 ForkJoinTask.cancelIgnoringExceptions(f); 2898 } 2899 } 2900 2901 /** 2902 * Returns the factory used for constructing new workers. 2903 * 2904 * @return the factory used for constructing new workers 2905 */ 2906 public ForkJoinWorkerThreadFactory getFactory() { 2907 return factory; 2908 } 2909 2910 /** 2911 * Returns the handler for internal worker threads that terminate 2912 * due to unrecoverable errors encountered while executing tasks. 2913 * 2914 * @return the handler, or {@code null} if none 2915 */ 2916 public UncaughtExceptionHandler getUncaughtExceptionHandler() { 2917 return ueh; 2918 } 2919 2920 /** 2921 * Returns the targeted parallelism level of this pool. 2922 * 2923 * @return the targeted parallelism level of this pool 2924 */ 2925 public int getParallelism() { 2926 int par = mode & SMASK; 2927 return (par > 0) ? par : 1; 2928 } 2929 2930 /** 2931 * Returns the targeted parallelism level of the common pool. 2932 * 2933 * @return the targeted parallelism level of the common pool 2934 * @since 1.8 2935 */ 2936 public static int getCommonPoolParallelism() { 2937 return COMMON_PARALLELISM; 2938 } 2939 2940 /** 2941 * Returns the number of worker threads that have started but not 2942 * yet terminated. The result returned by this method may differ 2943 * from {@link #getParallelism} when threads are created to 2944 * maintain parallelism when others are cooperatively blocked. 2945 * 2946 * @return the number of worker threads 2947 */ 2948 public int getPoolSize() { 2949 return ((mode & SMASK) + (short)(ctl >>> TC_SHIFT)); 2950 } 2951 2952 /** 2953 * Returns {@code true} if this pool uses local first-in-first-out 2954 * scheduling mode for forked tasks that are never joined. 2955 * 2956 * @return {@code true} if this pool uses async mode 2957 */ 2958 public boolean getAsyncMode() { 2959 return (mode & FIFO) != 0; 2960 } 2961 2962 /** 2963 * Returns an estimate of the number of worker threads that are 2964 * not blocked waiting to join tasks or for other managed 2965 * synchronization. This method may overestimate the 2966 * number of running threads. 2967 * 2968 * @return the number of worker threads 2969 */ 2970 public int getRunningThreadCount() { 2971 VarHandle.acquireFence(); 2972 WorkQueue[] qs; WorkQueue q; 2973 int rc = 0; 2974 if ((qs = queues) != null) { 2975 for (int i = 1; i < qs.length; i += 2) { 2976 if ((q = qs[i]) != null && q.isApparentlyUnblocked()) 2977 ++rc; 2978 } 2979 } 2980 return rc; 2981 } 2982 2983 /** 2984 * Returns an estimate of the number of threads that are currently 2985 * stealing or executing tasks. This method may overestimate the 2986 * number of active threads. 2987 * 2988 * @return the number of active threads 2989 */ 2990 public int getActiveThreadCount() { 2991 int r = (mode & SMASK) + (int)(ctl >> RC_SHIFT); 2992 return (r <= 0) ? 0 : r; // suppress momentarily negative values 2993 } 2994 2995 /** 2996 * Returns {@code true} if all worker threads are currently idle. 2997 * An idle worker is one that cannot obtain a task to execute 2998 * because none are available to steal from other threads, and 2999 * there are no pending submissions to the pool. This method is 3000 * conservative; it might not return {@code true} immediately upon 3001 * idleness of all threads, but will eventually become true if 3002 * threads remain inactive. 3003 * 3004 * @return {@code true} if all threads are currently idle 3005 */ 3006 public boolean isQuiescent() { 3007 return canStop(); 3008 } 3009 3010 /** 3011 * Returns an estimate of the total number of completed tasks that 3012 * were executed by a thread other than their submitter. The 3013 * reported value underestimates the actual total number of steals 3014 * when the pool is not quiescent. This value may be useful for 3015 * monitoring and tuning fork/join programs: in general, steal 3016 * counts should be high enough to keep threads busy, but low 3017 * enough to avoid overhead and contention across threads. 3018 * 3019 * @return the number of steals 3020 */ 3021 public long getStealCount() { 3022 long count = stealCount; 3023 WorkQueue[] qs; WorkQueue q; 3024 if ((qs = queues) != null) { 3025 for (int i = 1; i < qs.length; i += 2) { 3026 if ((q = qs[i]) != null) 3027 count += (long)q.nsteals & 0xffffffffL; 3028 } 3029 } 3030 return count; 3031 } 3032 3033 /** 3034 * Returns an estimate of the total number of tasks currently held 3035 * in queues by worker threads (but not including tasks submitted 3036 * to the pool that have not begun executing). This value is only 3037 * an approximation, obtained by iterating across all threads in 3038 * the pool. This method may be useful for tuning task 3039 * granularities. 3040 * 3041 * @return the number of queued tasks 3042 */ 3043 public long getQueuedTaskCount() { 3044 VarHandle.acquireFence(); 3045 WorkQueue[] qs; WorkQueue q; 3046 int count = 0; 3047 if ((qs = queues) != null) { 3048 for (int i = 1; i < qs.length; i += 2) { 3049 if ((q = qs[i]) != null) 3050 count += q.queueSize(); 3051 } 3052 } 3053 return count; 3054 } 3055 3056 /** 3057 * Returns an estimate of the number of tasks submitted to this 3058 * pool that have not yet begun executing. This method may take 3059 * time proportional to the number of submissions. 3060 * 3061 * @return the number of queued submissions 3062 */ 3063 public int getQueuedSubmissionCount() { 3064 VarHandle.acquireFence(); 3065 WorkQueue[] qs; WorkQueue q; 3066 int count = 0; 3067 if ((qs = queues) != null) { 3068 for (int i = 0; i < qs.length; i += 2) { 3069 if ((q = qs[i]) != null) 3070 count += q.queueSize(); 3071 } 3072 } 3073 return count; 3074 } 3075 3076 /** 3077 * Returns {@code true} if there are any tasks submitted to this 3078 * pool that have not yet begun executing. 3079 * 3080 * @return {@code true} if there are any queued submissions 3081 */ 3082 public boolean hasQueuedSubmissions() { 3083 VarHandle.acquireFence(); 3084 WorkQueue[] qs; WorkQueue q; 3085 if ((qs = queues) != null) { 3086 for (int i = 0; i < qs.length; i += 2) { 3087 if ((q = qs[i]) != null && !q.isEmpty()) 3088 return true; 3089 } 3090 } 3091 return false; 3092 } 3093 3094 /** 3095 * Removes and returns the next unexecuted submission if one is 3096 * available. This method may be useful in extensions to this 3097 * class that re-assign work in systems with multiple pools. 3098 * 3099 * @return the next submission, or {@code null} if none 3100 */ 3101 protected ForkJoinTask<?> pollSubmission() { 3102 return pollScan(true); 3103 } 3104 3105 /** 3106 * Removes all available unexecuted submitted and forked tasks 3107 * from scheduling queues and adds them to the given collection, 3108 * without altering their execution status. These may include 3109 * artificially generated or wrapped tasks. This method is 3110 * designed to be invoked only when the pool is known to be 3111 * quiescent. Invocations at other times may not remove all 3112 * tasks. A failure encountered while attempting to add elements 3113 * to collection {@code c} may result in elements being in 3114 * neither, either or both collections when the associated 3115 * exception is thrown. The behavior of this operation is 3116 * undefined if the specified collection is modified while the 3117 * operation is in progress. 3118 * 3119 * @param c the collection to transfer elements into 3120 * @return the number of elements transferred 3121 */ 3122 protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) { 3123 int count = 0; 3124 for (ForkJoinTask<?> t; (t = pollScan(false)) != null; ) { 3125 c.add(t); 3126 ++count; 3127 } 3128 return count; 3129 } 3130 3131 /** 3132 * Returns a string identifying this pool, as well as its state, 3133 * including indications of run state, parallelism level, and 3134 * worker and task counts. 3135 * 3136 * @return a string identifying this pool, as well as its state 3137 */ 3138 public String toString() { 3139 // Use a single pass through queues to collect counts 3140 int md = mode; // read volatile fields first 3141 long c = ctl; 3142 long st = stealCount; 3143 long qt = 0L, ss = 0L; int rc = 0; 3144 WorkQueue[] qs; WorkQueue q; 3145 if ((qs = queues) != null) { 3146 for (int i = 0; i < qs.length; ++i) { 3147 if ((q = qs[i]) != null) { 3148 int size = q.queueSize(); 3149 if ((i & 1) == 0) 3150 ss += size; 3151 else { 3152 qt += size; 3153 st += (long)q.nsteals & 0xffffffffL; 3154 if (q.isApparentlyUnblocked()) 3155 ++rc; 3156 } 3157 } 3158 } 3159 } 3160 3161 int pc = (md & SMASK); 3162 int tc = pc + (short)(c >>> TC_SHIFT); 3163 int ac = pc + (int)(c >> RC_SHIFT); 3164 if (ac < 0) // ignore transient negative 3165 ac = 0; 3166 String level = ((md & TERMINATED) != 0 ? "Terminated" : 3167 (md & STOP) != 0 ? "Terminating" : 3168 (md & SHUTDOWN) != 0 ? "Shutting down" : 3169 "Running"); 3170 return super.toString() + 3171 "[" + level + 3172 ", parallelism = " + pc + 3173 ", size = " + tc + 3174 ", active = " + ac + 3175 ", running = " + rc + 3176 ", steals = " + st + 3177 ", tasks = " + qt + 3178 ", submissions = " + ss + 3179 "]"; 3180 } 3181 3182 /** 3183 * Possibly initiates an orderly shutdown in which previously 3184 * submitted tasks are executed, but no new tasks will be 3185 * accepted. Invocation has no effect on execution state if this 3186 * is the {@link #commonPool()}, and no additional effect if 3187 * already shut down. Tasks that are in the process of being 3188 * submitted concurrently during the course of this method may or 3189 * may not be rejected. 3190 * 3191 * @throws SecurityException if a security manager exists and 3192 * the caller is not permitted to modify threads 3193 * because it does not hold {@link 3194 * java.lang.RuntimePermission}{@code ("modifyThread")} 3195 */ 3196 public void shutdown() { 3197 checkPermission(); 3198 if (this != common) 3199 tryTerminate(false, true); 3200 } 3201 3202 /** 3203 * Possibly attempts to cancel and/or stop all tasks, and reject 3204 * all subsequently submitted tasks. Invocation has no effect on 3205 * execution state if this is the {@link #commonPool()}, and no 3206 * additional effect if already shut down. Otherwise, tasks that 3207 * are in the process of being submitted or executed concurrently 3208 * during the course of this method may or may not be 3209 * rejected. This method cancels both existing and unexecuted 3210 * tasks, in order to permit termination in the presence of task 3211 * dependencies. So the method always returns an empty list 3212 * (unlike the case for some other Executors). 3213 * 3214 * @return an empty list 3215 * @throws SecurityException if a security manager exists and 3216 * the caller is not permitted to modify threads 3217 * because it does not hold {@link 3218 * java.lang.RuntimePermission}{@code ("modifyThread")} 3219 */ 3220 public List<Runnable> shutdownNow() { 3221 checkPermission(); 3222 if (this != common) 3223 tryTerminate(true, true); 3224 return Collections.emptyList(); 3225 } 3226 3227 /** 3228 * Returns {@code true} if all tasks have completed following shut down. 3229 * 3230 * @return {@code true} if all tasks have completed following shut down 3231 */ 3232 public boolean isTerminated() { 3233 return (mode & TERMINATED) != 0; 3234 } 3235 3236 /** 3237 * Returns {@code true} if the process of termination has 3238 * commenced but not yet completed. This method may be useful for 3239 * debugging. A return of {@code true} reported a sufficient 3240 * period after shutdown may indicate that submitted tasks have 3241 * ignored or suppressed interruption, or are waiting for I/O, 3242 * causing this executor not to properly terminate. (See the 3243 * advisory notes for class {@link ForkJoinTask} stating that 3244 * tasks should not normally entail blocking operations. But if 3245 * they do, they must abort them on interrupt.) 3246 * 3247 * @return {@code true} if terminating but not yet terminated 3248 */ 3249 public boolean isTerminating() { 3250 return (mode & (STOP | TERMINATED)) == STOP; 3251 } 3252 3253 /** 3254 * Returns {@code true} if this pool has been shut down. 3255 * 3256 * @return {@code true} if this pool has been shut down 3257 */ 3258 public boolean isShutdown() { 3259 return (mode & SHUTDOWN) != 0; 3260 } 3261 3262 /** 3263 * Blocks until all tasks have completed execution after a 3264 * shutdown request, or the timeout occurs, or the current thread 3265 * is interrupted, whichever happens first. Because the {@link 3266 * #commonPool()} never terminates until program shutdown, when 3267 * applied to the common pool, this method is equivalent to {@link 3268 * #awaitQuiescence(long, TimeUnit)} but always returns {@code false}. 3269 * 3270 * @param timeout the maximum time to wait 3271 * @param unit the time unit of the timeout argument 3272 * @return {@code true} if this executor terminated and 3273 * {@code false} if the timeout elapsed before termination 3274 * @throws InterruptedException if interrupted while waiting 3275 */ 3276 public boolean awaitTermination(long timeout, TimeUnit unit) 3277 throws InterruptedException { 3278 ReentrantLock lock; Condition cond; 3279 long nanos = unit.toNanos(timeout); 3280 boolean terminated = false; 3281 if (this == common) { 3282 Thread t; ForkJoinWorkerThread wt; int q; 3283 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread && 3284 (wt = (ForkJoinWorkerThread)t).pool == this) 3285 q = helpQuiescePool(wt.workQueue, nanos, true); 3286 else 3287 q = externalHelpQuiescePool(nanos, true); 3288 if (q < 0) 3289 throw new InterruptedException(); 3290 } 3291 else if (!(terminated = ((mode & TERMINATED) != 0)) && 3292 (lock = registrationLock) != null) { 3293 lock.lock(); 3294 try { 3295 if ((cond = termination) == null) 3296 termination = cond = lock.newCondition(); 3297 while (!(terminated = ((mode & TERMINATED) != 0)) && nanos > 0L) 3298 nanos = cond.awaitNanos(nanos); 3299 } finally { 3300 lock.unlock(); 3301 } 3302 } 3303 return terminated; 3304 } 3305 3306 /** 3307 * If called by a ForkJoinTask operating in this pool, equivalent 3308 * in effect to {@link ForkJoinTask#helpQuiesce}. Otherwise, 3309 * waits and/or attempts to assist performing tasks until this 3310 * pool {@link #isQuiescent} or the indicated timeout elapses. 3311 * 3312 * @param timeout the maximum time to wait 3313 * @param unit the time unit of the timeout argument 3314 * @return {@code true} if quiescent; {@code false} if the 3315 * timeout elapsed. 3316 */ 3317 public boolean awaitQuiescence(long timeout, TimeUnit unit) { 3318 Thread t; ForkJoinWorkerThread wt; int q; 3319 long nanos = unit.toNanos(timeout); 3320 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread && 3321 (wt = (ForkJoinWorkerThread)t).pool == this) 3322 q = helpQuiescePool(wt.workQueue, nanos, false); 3323 else 3324 q = externalHelpQuiescePool(nanos, false); 3325 return (q > 0); 3326 } 3327 3328 /** 3329 * Interface for extending managed parallelism for tasks running 3330 * in {@link ForkJoinPool}s. 3331 * 3332 * <p>A {@code ManagedBlocker} provides two methods. Method 3333 * {@link #isReleasable} must return {@code true} if blocking is 3334 * not necessary. Method {@link #block} blocks the current thread 3335 * if necessary (perhaps internally invoking {@code isReleasable} 3336 * before actually blocking). These actions are performed by any 3337 * thread invoking {@link 3338 * ForkJoinPool#managedBlock(ManagedBlocker)}. The unusual 3339 * methods in this API accommodate synchronizers that may, but 3340 * don't usually, block for long periods. Similarly, they allow 3341 * more efficient internal handling of cases in which additional 3342 * workers may be, but usually are not, needed to ensure 3343 * sufficient parallelism. Toward this end, implementations of 3344 * method {@code isReleasable} must be amenable to repeated 3345 * invocation. Neither method is invoked after a prior invocation 3346 * of {@code isReleasable} or {@code block} returns {@code true}. 3347 * 3348 * <p>For example, here is a ManagedBlocker based on a 3349 * ReentrantLock: 3350 * <pre> {@code 3351 * class ManagedLocker implements ManagedBlocker { 3352 * final ReentrantLock lock; 3353 * boolean hasLock = false; 3354 * ManagedLocker(ReentrantLock lock) { this.lock = lock; } 3355 * public boolean block() { 3356 * if (!hasLock) 3357 * lock.lock(); 3358 * return true; 3359 * } 3360 * public boolean isReleasable() { 3361 * return hasLock || (hasLock = lock.tryLock()); 3362 * } 3363 * }}</pre> 3364 * 3365 * <p>Here is a class that possibly blocks waiting for an 3366 * item on a given queue: 3367 * <pre> {@code 3368 * class QueueTaker<E> implements ManagedBlocker { 3369 * final BlockingQueue<E> queue; 3370 * volatile E item = null; 3371 * QueueTaker(BlockingQueue<E> q) { this.queue = q; } 3372 * public boolean block() throws InterruptedException { 3373 * if (item == null) 3374 * item = queue.take(); 3375 * return true; 3376 * } 3377 * public boolean isReleasable() { 3378 * return item != null || (item = queue.poll()) != null; 3379 * } 3380 * public E getItem() { // call after pool.managedBlock completes 3381 * return item; 3382 * } 3383 * }}</pre> 3384 */ 3385 public static interface ManagedBlocker { 3386 /** 3387 * Possibly blocks the current thread, for example waiting for 3388 * a lock or condition. 3389 * 3390 * @return {@code true} if no additional blocking is necessary 3391 * (i.e., if isReleasable would return true) 3392 * @throws InterruptedException if interrupted while waiting 3393 * (the method is not required to do so, but is allowed to) 3394 */ 3395 boolean block() throws InterruptedException; 3396 3397 /** 3398 * Returns {@code true} if blocking is unnecessary. 3399 * @return {@code true} if blocking is unnecessary 3400 */ 3401 boolean isReleasable(); 3402 } 3403 3404 /** 3405 * Runs the given possibly blocking task. When {@linkplain 3406 * ForkJoinTask#inForkJoinPool() running in a ForkJoinPool}, this 3407 * method possibly arranges for a spare thread to be activated if 3408 * necessary to ensure sufficient parallelism while the current 3409 * thread is blocked in {@link ManagedBlocker#block blocker.block()}. 3410 * 3411 * <p>This method repeatedly calls {@code blocker.isReleasable()} and 3412 * {@code blocker.block()} until either method returns {@code true}. 3413 * Every call to {@code blocker.block()} is preceded by a call to 3414 * {@code blocker.isReleasable()} that returned {@code false}. 3415 * 3416 * <p>If not running in a ForkJoinPool, this method is 3417 * behaviorally equivalent to 3418 * <pre> {@code 3419 * while (!blocker.isReleasable()) 3420 * if (blocker.block()) 3421 * break;}</pre> 3422 * 3423 * If running in a ForkJoinPool, the pool may first be expanded to 3424 * ensure sufficient parallelism available during the call to 3425 * {@code blocker.block()}. 3426 * 3427 * @param blocker the blocker task 3428 * @throws InterruptedException if {@code blocker.block()} did so 3429 */ 3430 public static void managedBlock(ManagedBlocker blocker) 3431 throws InterruptedException { 3432 Thread t; ForkJoinPool p; 3433 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread && 3434 (p = ((ForkJoinWorkerThread)t).pool) != null) 3435 p.compensatedBlock(blocker); 3436 else 3437 unmanagedBlock(blocker); 3438 } 3439 3440 /** ManagedBlock for ForkJoinWorkerThreads */ 3441 private void compensatedBlock(ManagedBlocker blocker) 3442 throws InterruptedException { 3443 if (blocker == null) throw new NullPointerException(); 3444 for (;;) { 3445 int comp; boolean done; 3446 long c = ctl; 3447 if (blocker.isReleasable()) 3448 break; 3449 if ((comp = tryCompensate(c)) >= 0) { 3450 long post = (comp == 0) ? 0L : RC_UNIT; 3451 try { 3452 done = blocker.block(); 3453 } finally { 3454 getAndAddCtl(post); 3455 } 3456 if (done) 3457 break; 3458 } 3459 } 3460 } 3461 3462 /** ManagedBlock for external threads */ 3463 private static void unmanagedBlock(ManagedBlocker blocker) 3464 throws InterruptedException { 3465 if (blocker == null) throw new NullPointerException(); 3466 do {} while (!blocker.isReleasable() && !blocker.block()); 3467 } 3468 3469 // AbstractExecutorService.newTaskFor overrides rely on 3470 // undocumented fact that ForkJoinTask.adapt returns ForkJoinTasks 3471 // that also implement RunnableFuture. 3472 3473 @Override 3474 protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { 3475 return new ForkJoinTask.AdaptedRunnable<T>(runnable, value); 3476 } 3477 3478 @Override 3479 protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { 3480 return new ForkJoinTask.AdaptedCallable<T>(callable); 3481 } 3482 3483 static { 3484 try { 3485 MethodHandles.Lookup l = MethodHandles.lookup(); 3486 CTL = l.findVarHandle(ForkJoinPool.class, "ctl", long.class); 3487 MODE = l.findVarHandle(ForkJoinPool.class, "mode", int.class); 3488 THREADIDS = l.findVarHandle(ForkJoinPool.class, "threadIds", int.class); 3489 POOLIDS = l.findStaticVarHandle(ForkJoinPool.class, "poolIds", int.class); 3490 } catch (ReflectiveOperationException e) { 3491 throw new ExceptionInInitializerError(e); 3492 } 3493 3494 // Reduce the risk of rare disastrous classloading in first call to 3495 // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773 3496 Class<?> ensureLoaded = LockSupport.class; 3497 3498 int commonMaxSpares = DEFAULT_COMMON_MAX_SPARES; 3499 try { 3500 String p = System.getProperty 3501 ("java.util.concurrent.ForkJoinPool.common.maximumSpares"); 3502 if (p != null) 3503 commonMaxSpares = Integer.parseInt(p); 3504 } catch (Exception ignore) {} 3505 COMMON_MAX_SPARES = commonMaxSpares; 3506 3507 defaultForkJoinWorkerThreadFactory = 3508 new DefaultForkJoinWorkerThreadFactory(); 3509 modifyThreadPermission = new RuntimePermission("modifyThread"); 3510 @SuppressWarnings("removal") 3511 ForkJoinPool tmp = AccessController.doPrivileged(new PrivilegedAction<>() { 3512 public ForkJoinPool run() { 3513 return new ForkJoinPool((byte)0); }}); 3514 common = tmp; 3515 3516 COMMON_PARALLELISM = Math.max(common.mode & SMASK, 1); 3517 } 3518 } 3519