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