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 int s = top, cap, k; ForkJoinTask<?>[] a; 1050 if ((a = array) != null && (cap = a.length) > 0 && 1051 a[k = (cap - 1) & (s - 1)] == task && tryLock()) { 1052 if (top == s && array == a && 1053 (taken = casSlotToNull(a, k, task))) 1054 top = s - 1; 1055 source = 0; // release lock 1056 } 1057 return taken; 1058 } 1059 1060 /** 1061 * Deep form of tryUnpush: Traverses from top and removes task if 1062 * present, shifting others to fill gap. 1063 */ 1064 final boolean tryRemove(ForkJoinTask<?> task, boolean owned) { 1065 boolean taken = false; 1066 int p = top, cap; ForkJoinTask<?>[] a; ForkJoinTask<?> t; 1067 if ((a = array) != null && task != null && (cap = a.length) > 0) { 1068 int m = cap - 1, s = p - 1, d = p - base; 1069 for (int i = s, k; d > 0; --i, --d) { 1070 if ((t = a[k = i & m]) == task) { 1071 if (owned || tryLock()) { 1072 if ((owned || (array == a && top == p)) && 1073 (taken = casSlotToNull(a, k, t))) { 1074 for (int j = i; j != s; ) // shift down 1075 a[j & m] = getAndClearSlot(a, ++j & m); 1076 top = s; 1077 } 1078 if (!owned) 1079 source = 0; 1080 } 1081 break; 1082 } 1083 } 1084 } 1085 return taken; 1086 } 1087 1088 // variants of poll 1089 1090 /** 1091 * Tries once to poll next task in FIFO order, failing on 1092 * inconsistency or contention. 1093 */ 1094 final ForkJoinTask<?> tryPoll() { 1095 int cap, b, k; ForkJoinTask<?>[] a; 1096 if ((a = array) != null && (cap = a.length) > 0) { 1097 ForkJoinTask<?> t = getSlot(a, k = (cap - 1) & (b = base)); 1098 if (base == b++ && t != null && casSlotToNull(a, k, t)) { 1099 setBaseOpaque(b); 1100 return t; 1101 } 1102 } 1103 return null; 1104 } 1105 1106 /** 1107 * Takes next task, if one exists, in order specified by mode. 1108 */ 1109 final ForkJoinTask<?> nextLocalTask(int cfg) { 1110 ForkJoinTask<?> t = null; 1111 int s = top, cap; ForkJoinTask<?>[] a; 1112 if ((a = array) != null && (cap = a.length) > 0) { 1113 for (int b, d;;) { 1114 if ((d = s - (b = base)) <= 0) 1115 break; 1116 if (d == 1 || (cfg & FIFO) == 0) { 1117 if ((t = getAndClearSlot(a, --s & (cap - 1))) != null) 1118 top = s; 1119 break; 1120 } 1121 if ((t = getAndClearSlot(a, b++ & (cap - 1))) != null) { 1122 setBaseOpaque(b); 1123 break; 1124 } 1125 } 1126 } 1127 return t; 1128 } 1129 1130 /** 1131 * Takes next task, if one exists, using configured mode. 1132 */ 1133 final ForkJoinTask<?> nextLocalTask() { 1134 return nextLocalTask(config); 1135 } 1136 1137 /** 1138 * Returns next task, if one exists, in order specified by mode. 1139 */ 1140 final ForkJoinTask<?> peek() { 1141 VarHandle.acquireFence(); 1142 int cap; ForkJoinTask<?>[] a; 1143 return ((a = array) != null && (cap = a.length) > 0) ? 1144 a[(cap - 1) & ((config & FIFO) != 0 ? base : top - 1)] : null; 1145 } 1146 1147 // specialized execution methods 1148 1149 /** 1150 * Runs the given (stolen) task if nonnull, as well as 1151 * remaining local tasks and/or others available from the 1152 * given queue. 1153 */ 1154 final void topLevelExec(ForkJoinTask<?> task, WorkQueue q) { 1155 int cfg = config, nstolen = 1; 1156 while (task != null) { 1157 task.doExec(); 1158 if ((task = nextLocalTask(cfg)) == null && 1159 q != null && (task = q.tryPoll()) != null) 1160 ++nstolen; 1161 } 1162 nsteals += nstolen; 1163 source = 0; 1164 if ((cfg & INNOCUOUS) != 0) 1165 ThreadLocalRandom.eraseThreadLocals(Thread.currentThread()); 1166 } 1167 1168 /** 1169 * Tries to pop and run tasks within the target's computation 1170 * until done, not found, or limit exceeded. 1171 * 1172 * @param task root of CountedCompleter computation 1173 * @param owned true if owned by a ForkJoinWorkerThread 1174 * @param limit max runs, or zero for no limit 1175 * @return task status on exit 1176 */ 1177 final int helpComplete(ForkJoinTask<?> task, boolean owned, int limit) { 1178 int status = 0, cap, k, p, s; ForkJoinTask<?>[] a; ForkJoinTask<?> t; 1179 while (task != null && (status = task.status) >= 0 && 1180 (a = array) != null && (cap = a.length) > 0 && 1181 (t = a[k = (cap - 1) & (s = (p = top) - 1)]) 1182 instanceof CountedCompleter) { 1183 CountedCompleter<?> f = (CountedCompleter<?>)t; 1184 boolean taken = false; 1185 for (;;) { // exec if root task is a completer of t 1186 if (f == task) { 1187 if (owned) { 1188 if ((taken = casSlotToNull(a, k, t))) 1189 top = s; 1190 } 1191 else if (tryLock()) { 1192 if (top == p && array == a && 1193 (taken = casSlotToNull(a, k, t))) 1194 top = s; 1195 source = 0; 1196 } 1197 break; 1198 } 1199 else if ((f = f.completer) == null) 1200 break; 1201 } 1202 if (!taken) 1203 break; 1204 t.doExec(); 1205 if (limit != 0 && --limit == 0) 1206 break; 1207 } 1208 return status; 1209 } 1210 1211 /** 1212 * Tries to poll and run AsynchronousCompletionTasks until 1213 * none found or blocker is released. 1214 * 1215 * @param blocker the blocker 1216 */ 1217 final void helpAsyncBlocker(ManagedBlocker blocker) { 1218 int cap, b, d, k; ForkJoinTask<?>[] a; ForkJoinTask<?> t; 1219 while (blocker != null && (d = top - (b = base)) > 0 && 1220 (a = array) != null && (cap = a.length) > 0 && 1221 (((t = getSlot(a, k = (cap - 1) & b)) == null && d > 1) || 1222 t instanceof 1223 CompletableFuture.AsynchronousCompletionTask) && 1224 !blocker.isReleasable()) { 1225 if (t != null && base == b++ && casSlotToNull(a, k, t)) { 1226 setBaseOpaque(b); 1227 t.doExec(); 1228 } 1229 } 1230 } 1231 1232 // misc 1233 1234 /** AccessControlContext for innocuous workers, created on 1st use. */ 1235 private static AccessControlContext INNOCUOUS_ACC; 1236 1237 /** 1238 * Initializes (upon registration) InnocuousForkJoinWorkerThreads. 1239 */ 1240 final void initializeInnocuousWorker() { 1241 AccessControlContext acc; // racy construction OK 1242 if ((acc = INNOCUOUS_ACC) == null) 1243 INNOCUOUS_ACC = acc = new AccessControlContext( 1244 new ProtectionDomain[] { new ProtectionDomain(null, null) }); 1245 Thread t = Thread.currentThread(); 1246 ThreadLocalRandom.setInheritedAccessControlContext(t, acc); 1247 ThreadLocalRandom.eraseThreadLocals(t); 1248 } 1249 1250 /** 1251 * Returns true if owned by a worker thread and not known to be blocked. 1252 */ 1253 final boolean isApparentlyUnblocked() { 1254 Thread wt; Thread.State s; 1255 return ((wt = owner) != null && 1256 (s = wt.getState()) != Thread.State.BLOCKED && 1257 s != Thread.State.WAITING && 1258 s != Thread.State.TIMED_WAITING); 1259 } 1260 1261 static { 1262 try { 1263 QA = MethodHandles.arrayElementVarHandle(ForkJoinTask[].class); 1264 MethodHandles.Lookup l = MethodHandles.lookup(); 1265 SOURCE = l.findVarHandle(WorkQueue.class, "source", int.class); 1266 BASE = l.findVarHandle(WorkQueue.class, "base", int.class); 1267 } catch (ReflectiveOperationException e) { 1268 throw new ExceptionInInitializerError(e); 1269 } 1270 } 1271 } 1272 1273 // static fields (initialized in static initializer below) 1274 1275 /** 1276 * Creates a new ForkJoinWorkerThread. This factory is used unless 1277 * overridden in ForkJoinPool constructors. 1278 */ 1279 public static final ForkJoinWorkerThreadFactory 1280 defaultForkJoinWorkerThreadFactory; 1281 1282 /** 1283 * Permission required for callers of methods that may start or 1284 * kill threads. 1285 */ 1286 static final RuntimePermission modifyThreadPermission; 1287 1288 /** 1289 * Common (static) pool. Non-null for public use unless a static 1290 * construction exception, but internal usages null-check on use 1291 * to paranoically avoid potential initialization circularities 1292 * as well as to simplify generated code. 1293 */ 1294 static final ForkJoinPool common; 1295 1296 /** 1297 * Common pool parallelism. To allow simpler use and management 1298 * when common pool threads are disabled, we allow the underlying 1299 * common.parallelism field to be zero, but in that case still report 1300 * parallelism as 1 to reflect resulting caller-runs mechanics. 1301 */ 1302 static final int COMMON_PARALLELISM; 1303 1304 /** 1305 * Limit on spare thread construction in tryCompensate. 1306 */ 1307 private static final int COMMON_MAX_SPARES; 1308 1309 /** 1310 * Sequence number for creating worker names 1311 */ 1312 private static volatile int poolIds; 1313 1314 // static configuration constants 1315 1316 /** 1317 * Default idle timeout value (in milliseconds) for the thread 1318 * triggering quiescence to park waiting for new work 1319 */ 1320 private static final long DEFAULT_KEEPALIVE = 60_000L; 1321 1322 /** 1323 * Undershoot tolerance for idle timeouts 1324 */ 1325 private static final long TIMEOUT_SLOP = 20L; 1326 1327 /** 1328 * The default value for COMMON_MAX_SPARES. Overridable using the 1329 * "java.util.concurrent.ForkJoinPool.common.maximumSpares" system 1330 * property. The default value is far in excess of normal 1331 * requirements, but also far short of MAX_CAP and typical OS 1332 * thread limits, so allows JVMs to catch misuse/abuse before 1333 * running out of resources needed to do so. 1334 */ 1335 private static final int DEFAULT_COMMON_MAX_SPARES = 256; 1336 1337 /* 1338 * Bits and masks for field ctl, packed with 4 16 bit subfields: 1339 * RC: Number of released (unqueued) workers minus target parallelism 1340 * TC: Number of total workers minus target parallelism 1341 * SS: version count and status of top waiting thread 1342 * ID: poolIndex of top of Treiber stack of waiters 1343 * 1344 * When convenient, we can extract the lower 32 stack top bits 1345 * (including version bits) as sp=(int)ctl. The offsets of counts 1346 * by the target parallelism and the positionings of fields makes 1347 * it possible to perform the most common checks via sign tests of 1348 * fields: When ac is negative, there are not enough unqueued 1349 * workers, when tc is negative, there are not enough total 1350 * workers. When sp is non-zero, there are waiting workers. To 1351 * deal with possibly negative fields, we use casts in and out of 1352 * "short" and/or signed shifts to maintain signedness. 1353 * 1354 * Because it occupies uppermost bits, we can add one release 1355 * count using getAndAdd of RC_UNIT, rather than CAS, when 1356 * returning from a blocked join. Other updates entail multiple 1357 * subfields and masking, requiring CAS. 1358 * 1359 * The limits packed in field "bounds" are also offset by the 1360 * parallelism level to make them comparable to the ctl rc and tc 1361 * fields. 1362 */ 1363 1364 // Lower and upper word masks 1365 private static final long SP_MASK = 0xffffffffL; 1366 private static final long UC_MASK = ~SP_MASK; 1367 1368 // Release counts 1369 private static final int RC_SHIFT = 48; 1370 private static final long RC_UNIT = 0x0001L << RC_SHIFT; 1371 private static final long RC_MASK = 0xffffL << RC_SHIFT; 1372 1373 // Total counts 1374 private static final int TC_SHIFT = 32; 1375 private static final long TC_UNIT = 0x0001L << TC_SHIFT; 1376 private static final long TC_MASK = 0xffffL << TC_SHIFT; 1377 private static final long ADD_WORKER = 0x0001L << (TC_SHIFT + 15); // sign 1378 1379 // Instance fields 1380 1381 final long keepAlive; // milliseconds before dropping if idle 1382 volatile long stealCount; // collects worker nsteals 1383 int scanRover; // advances across pollScan calls 1384 volatile int threadIds; // for worker thread names 1385 final int bounds; // min, max threads packed as shorts 1386 volatile int mode; // parallelism, runstate, queue mode 1387 WorkQueue[] queues; // main registry 1388 final ReentrantLock registrationLock; 1389 Condition termination; // lazily constructed 1390 final String workerNamePrefix; // null for common pool 1391 final ForkJoinWorkerThreadFactory factory; 1392 final UncaughtExceptionHandler ueh; // per-worker UEH 1393 final Predicate<? super ForkJoinPool> saturate; 1394 1395 @jdk.internal.vm.annotation.Contended("fjpctl") // segregate 1396 volatile long ctl; // main pool control 1397 1398 // Support for atomic operations 1399 private static final VarHandle CTL; 1400 private static final VarHandle MODE; 1401 private static final VarHandle THREADIDS; 1402 private static final VarHandle POOLIDS; 1403 private boolean compareAndSetCtl(long c, long v) { 1404 return CTL.compareAndSet(this, c, v); 1405 } 1406 private long compareAndExchangeCtl(long c, long v) { 1407 return (long)CTL.compareAndExchange(this, c, v); 1408 } 1409 private long getAndAddCtl(long v) { 1410 return (long)CTL.getAndAdd(this, v); 1411 } 1412 private int getAndBitwiseOrMode(int v) { 1413 return (int)MODE.getAndBitwiseOr(this, v); 1414 } 1415 private int getAndAddThreadIds(int x) { 1416 return (int)THREADIDS.getAndAdd(this, x); 1417 } 1418 private static int getAndAddPoolIds(int x) { 1419 return (int)POOLIDS.getAndAdd(x); 1420 } 1421 1422 // Creating, registering and deregistering workers 1423 1424 /** 1425 * Tries to construct and start one worker. Assumes that total 1426 * count has already been incremented as a reservation. Invokes 1427 * deregisterWorker on any failure. 1428 * 1429 * @return true if successful 1430 */ 1431 private boolean createWorker() { 1432 ForkJoinWorkerThreadFactory fac = factory; 1433 Throwable ex = null; 1434 ForkJoinWorkerThread wt = null; 1435 try { 1436 if (fac != null && (wt = fac.newThread(this)) != null) { 1437 wt.start(); 1438 return true; 1439 } 1440 } catch (Throwable rex) { 1441 ex = rex; 1442 } 1443 deregisterWorker(wt, ex); 1444 return false; 1445 } 1446 1447 /** 1448 * Provides a name for ForkJoinWorkerThread constructor. 1449 */ 1450 final String nextWorkerThreadName() { 1451 String prefix = workerNamePrefix; 1452 int tid = getAndAddThreadIds(1) + 1; 1453 if (prefix == null) // commonPool has no prefix 1454 prefix = "ForkJoinPool.commonPool-worker-"; 1455 return prefix.concat(Integer.toString(tid)); 1456 } 1457 1458 /** 1459 * Finishes initializing and records owned queue. 1460 * 1461 * @param w caller's WorkQueue 1462 */ 1463 final void registerWorker(WorkQueue w) { 1464 ReentrantLock lock = registrationLock; 1465 ThreadLocalRandom.localInit(); 1466 int seed = ThreadLocalRandom.getProbe(); 1467 if (w != null && lock != null) { 1468 int modebits = (mode & FIFO) | w.config; 1469 w.array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY]; 1470 w.stackPred = seed; // stash for runWorker 1471 if ((modebits & INNOCUOUS) != 0) 1472 w.initializeInnocuousWorker(); 1473 int id = (seed << 1) | 1; // initial index guess 1474 lock.lock(); 1475 try { 1476 WorkQueue[] qs; int n; // find queue index 1477 if ((qs = queues) != null && (n = qs.length) > 0) { 1478 int k = n, m = n - 1; 1479 for (; qs[id &= m] != null && k > 0; id -= 2, k -= 2); 1480 if (k == 0) 1481 id = n | 1; // resize below 1482 w.phase = w.config = id | modebits; // now publishable 1483 1484 if (id < n) 1485 qs[id] = w; 1486 else { // expand array 1487 int an = n << 1, am = an - 1; 1488 WorkQueue[] as = new WorkQueue[an]; 1489 as[id & am] = w; 1490 for (int j = 1; j < n; j += 2) 1491 as[j] = qs[j]; 1492 for (int j = 0; j < n; j += 2) { 1493 WorkQueue q; 1494 if ((q = qs[j]) != null) // shared queues may move 1495 as[q.config & am] = q; 1496 } 1497 VarHandle.releaseFence(); // fill before publish 1498 queues = as; 1499 } 1500 } 1501 } finally { 1502 lock.unlock(); 1503 } 1504 } 1505 } 1506 1507 /** 1508 * Final callback from terminating worker, as well as upon failure 1509 * to construct or start a worker. Removes record of worker from 1510 * array, and adjusts counts. If pool is shutting down, tries to 1511 * complete termination. 1512 * 1513 * @param wt the worker thread, or null if construction failed 1514 * @param ex the exception causing failure, or null if none 1515 */ 1516 final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) { 1517 ReentrantLock lock = registrationLock; 1518 WorkQueue w = null; 1519 int cfg = 0; 1520 if (wt != null && (w = wt.workQueue) != null && lock != null) { 1521 WorkQueue[] qs; int n, i; 1522 cfg = w.config; 1523 long ns = w.nsteals & 0xffffffffL; 1524 lock.lock(); // remove index from array 1525 if ((qs = queues) != null && (n = qs.length) > 0 && 1526 qs[i = cfg & (n - 1)] == w) 1527 qs[i] = null; 1528 stealCount += ns; // accumulate steals 1529 lock.unlock(); 1530 long c = ctl; 1531 if ((cfg & QUIET) == 0) // unless self-signalled, decrement counts 1532 do {} while (c != (c = compareAndExchangeCtl( 1533 c, ((RC_MASK & (c - RC_UNIT)) | 1534 (TC_MASK & (c - TC_UNIT)) | 1535 (SP_MASK & c))))); 1536 else if ((int)c == 0) // was dropped on timeout 1537 cfg = 0; // suppress signal if last 1538 for (ForkJoinTask<?> t; (t = w.pop()) != null; ) 1539 ForkJoinTask.cancelIgnoringExceptions(t); // cancel tasks 1540 } 1541 1542 if (!tryTerminate(false, false) && w != null && (cfg & SRC) != 0) 1543 signalWork(); // possibly replace worker 1544 if (ex != null) 1545 ForkJoinTask.rethrow(ex); 1546 } 1547 1548 /* 1549 * Tries to create or release a worker if too few are running. 1550 */ 1551 final void signalWork() { 1552 for (long c = ctl; c < 0L;) { 1553 int sp, i; WorkQueue[] qs; WorkQueue v; 1554 if ((sp = (int)c & ~UNSIGNALLED) == 0) { // no idle workers 1555 if ((c & ADD_WORKER) == 0L) // enough total workers 1556 break; 1557 if (c == (c = compareAndExchangeCtl( 1558 c, ((RC_MASK & (c + RC_UNIT)) | 1559 (TC_MASK & (c + TC_UNIT)))))) { 1560 createWorker(); 1561 break; 1562 } 1563 } 1564 else if ((qs = queues) == null) 1565 break; // unstarted/terminated 1566 else if (qs.length <= (i = sp & SMASK)) 1567 break; // terminated 1568 else if ((v = qs[i]) == null) 1569 break; // terminating 1570 else { 1571 long nc = (v.stackPred & SP_MASK) | (UC_MASK & (c + RC_UNIT)); 1572 Thread vt = v.owner; 1573 if (c == (c = compareAndExchangeCtl(c, nc))) { 1574 v.phase = sp; 1575 LockSupport.unpark(vt); // release idle worker 1576 break; 1577 } 1578 } 1579 } 1580 } 1581 1582 /** 1583 * Top-level runloop for workers, called by ForkJoinWorkerThread.run. 1584 * See above for explanation. 1585 * 1586 * @param w caller's WorkQueue (may be null on failed initialization) 1587 */ 1588 final void runWorker(WorkQueue w) { 1589 if (w != null) { // skip on failed init 1590 w.config |= SRC; // mark as valid source 1591 int r = w.stackPred, src = 0; // use seed from registerWorker 1592 do { 1593 r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift 1594 } while ((src = scan(w, src, r)) >= 0 || 1595 (src = awaitWork(w)) == 0); 1596 } 1597 } 1598 1599 /** 1600 * Scans for and if found executes top-level tasks: Tries to poll 1601 * each queue starting at a random index with random stride, 1602 * returning source id or retry indicator if contended or 1603 * inconsistent. 1604 * 1605 * @param w caller's WorkQueue 1606 * @param prevSrc the previous queue stolen from in current phase, or 0 1607 * @param r random seed 1608 * @return id of queue if taken, negative if none found, prevSrc for retry 1609 */ 1610 private int scan(WorkQueue w, int prevSrc, int r) { 1611 WorkQueue[] qs = queues; 1612 int n = (w == null || qs == null) ? 0 : qs.length; 1613 for (int step = (r >>> 16) | 1, i = n; i > 0; --i, r += step) { 1614 int j, cap, b; WorkQueue q; ForkJoinTask<?>[] a; 1615 if ((q = qs[j = r & (n - 1)]) != null && // poll at qs[j].array[k] 1616 (a = q.array) != null && (cap = a.length) > 0) { 1617 int k = (cap - 1) & (b = q.base), nextBase = b + 1; 1618 int nextIndex = (cap - 1) & nextBase, src = j | SRC; 1619 ForkJoinTask<?> t = WorkQueue.getSlot(a, k); 1620 if (q.base != b) // inconsistent 1621 return prevSrc; 1622 else if (t != null && WorkQueue.casSlotToNull(a, k, t)) { 1623 q.base = nextBase; 1624 ForkJoinTask<?> next = a[nextIndex]; 1625 if ((w.source = src) != prevSrc && next != null) 1626 signalWork(); // propagate 1627 w.topLevelExec(t, q); 1628 return src; 1629 } 1630 else if (a[nextIndex] != null) // revisit 1631 return prevSrc; 1632 } 1633 } 1634 return (queues != qs) ? prevSrc: -1; // possibly resized 1635 } 1636 1637 /** 1638 * Advances worker phase, pushes onto ctl stack, and awaits signal 1639 * or reports termination. 1640 * 1641 * @return negative if terminated, else 0 1642 */ 1643 private int awaitWork(WorkQueue w) { 1644 if (w == null) 1645 return -1; // already terminated 1646 int phase = (w.phase + SS_SEQ) & ~UNSIGNALLED; 1647 w.phase = phase | UNSIGNALLED; // advance phase 1648 long prevCtl = ctl, c; // enqueue 1649 do { 1650 w.stackPred = (int)prevCtl; 1651 c = ((prevCtl - RC_UNIT) & UC_MASK) | (phase & SP_MASK); 1652 } while (prevCtl != (prevCtl = compareAndExchangeCtl(prevCtl, c))); 1653 1654 Thread.interrupted(); // clear status 1655 LockSupport.setCurrentBlocker(this); // prepare to block (exit also OK) 1656 long deadline = 0L; // nonzero if possibly quiescent 1657 int ac = (int)(c >> RC_SHIFT), md; 1658 if ((md = mode) < 0) // pool is terminating 1659 return -1; 1660 else if ((md & SMASK) + ac <= 0) { 1661 boolean checkTermination = (md & SHUTDOWN) != 0; 1662 if ((deadline = System.currentTimeMillis() + keepAlive) == 0L) 1663 deadline = 1L; // avoid zero 1664 WorkQueue[] qs = queues; // check for racing submission 1665 int n = (qs == null) ? 0 : qs.length; 1666 for (int i = 0; i < n; i += 2) { 1667 WorkQueue q; ForkJoinTask<?>[] a; int cap, b; 1668 if (ctl != c) { // already signalled 1669 checkTermination = false; 1670 break; 1671 } 1672 else if ((q = qs[i]) != null && 1673 (a = q.array) != null && (cap = a.length) > 0 && 1674 ((b = q.base) != q.top || a[(cap - 1) & b] != null || 1675 q.source != 0)) { 1676 if (compareAndSetCtl(c, prevCtl)) 1677 w.phase = phase; // self-signal 1678 checkTermination = false; 1679 break; 1680 } 1681 } 1682 if (checkTermination && tryTerminate(false, false)) 1683 return -1; // trigger quiescent termination 1684 } 1685 1686 for (boolean alt = false;;) { // await activation or termination 1687 if (w.phase >= 0) 1688 break; 1689 else if (mode < 0) 1690 return -1; 1691 else if ((c = ctl) == prevCtl) 1692 Thread.onSpinWait(); // signal in progress 1693 else if (!(alt = !alt)) // check between park calls 1694 Thread.interrupted(); 1695 else if (deadline == 0L) 1696 LockSupport.park(); 1697 else if (deadline - System.currentTimeMillis() > TIMEOUT_SLOP) 1698 LockSupport.parkUntil(deadline); 1699 else if (((int)c & SMASK) == (w.config & SMASK) && 1700 compareAndSetCtl(c, ((UC_MASK & (c - TC_UNIT)) | 1701 (prevCtl & SP_MASK)))) { 1702 w.config |= QUIET; // sentinel for deregisterWorker 1703 return -1; // drop on timeout 1704 } 1705 else if ((deadline += keepAlive) == 0L) 1706 deadline = 1L; // not at head; restart timer 1707 } 1708 return 0; 1709 } 1710 1711 // Utilities used by ForkJoinTask 1712 1713 /** 1714 * Returns true if all workers are busy, possibly creating one if allowed 1715 */ 1716 final boolean isSaturated() { 1717 int maxTotal = bounds >>> SWIDTH; 1718 for (long c;;) { 1719 if (((int)(c = ctl) & ~UNSIGNALLED) != 0) 1720 return false; 1721 if ((short)(c >>> TC_SHIFT) >= maxTotal) 1722 return true; 1723 long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK); 1724 if (compareAndSetCtl(c, nc)) 1725 return !createWorker(); 1726 } 1727 } 1728 1729 /** 1730 * Returns true if can start terminating if enabled, or already terminated 1731 */ 1732 final boolean canStop() { 1733 outer: for (long oldSum = 0L;;) { // repeat until stable 1734 int md; WorkQueue[] qs; long c; 1735 if ((qs = queues) == null || ((md = mode) & STOP) != 0) 1736 return true; 1737 if ((md & SMASK) + (int)((c = ctl) >> RC_SHIFT) > 0) 1738 break; 1739 long checkSum = c; 1740 for (int i = 1; i < qs.length; i += 2) { // scan submitters 1741 WorkQueue q; ForkJoinTask<?>[] a; int s = 0, cap; 1742 if ((q = qs[i]) != null && (a = q.array) != null && 1743 (cap = a.length) > 0 && 1744 ((s = q.top) != q.base || a[(cap - 1) & s] != null || 1745 q.source != 0)) 1746 break outer; 1747 checkSum += (((long)i) << 32) ^ s; 1748 } 1749 if (oldSum == (oldSum = checkSum) && queues == qs) 1750 return true; 1751 } 1752 return (mode & STOP) != 0; // recheck mode on false return 1753 } 1754 1755 /** 1756 * Tries to decrement counts (sometimes implicitly) and possibly 1757 * arrange for a compensating worker in preparation for 1758 * blocking. May fail due to interference, in which case -1 is 1759 * returned so caller may retry. A zero return value indicates 1760 * that the caller doesn't need to re-adjust counts when later 1761 * unblocked. 1762 * 1763 * @param c incoming ctl value 1764 * @return UNCOMPENSATE: block then adjust, 0: block, -1 : retry 1765 */ 1766 private int tryCompensate(long c) { 1767 Predicate<? super ForkJoinPool> sat; 1768 int b = bounds; // counts are signed; centered at parallelism level == 0 1769 int minActive = (short)(b & SMASK), 1770 maxTotal = b >>> SWIDTH, 1771 active = (int)(c >> RC_SHIFT), 1772 total = (short)(c >>> TC_SHIFT), 1773 sp = (int)c & ~UNSIGNALLED; 1774 if (total >= 0) { 1775 if (sp != 0) { // activate idle worker 1776 WorkQueue[] qs; int n; WorkQueue v; 1777 if ((qs = queues) != null && (n = qs.length) > 0 && 1778 (v = qs[sp & (n - 1)]) != null) { 1779 Thread vt = v.owner; 1780 long nc = ((long)v.stackPred & SP_MASK) | (UC_MASK & c); 1781 if (compareAndSetCtl(c, nc)) { 1782 v.phase = sp; 1783 LockSupport.unpark(vt); 1784 return UNCOMPENSATE; 1785 } 1786 } 1787 return -1; // retry 1788 } 1789 else if (active > minActive) { // reduce parallelism 1790 long nc = ((RC_MASK & (c - RC_UNIT)) | (~RC_MASK & c)); 1791 return compareAndSetCtl(c, nc) ? UNCOMPENSATE : -1; 1792 } 1793 } 1794 if (total < maxTotal) { // expand pool 1795 long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK); 1796 return (!compareAndSetCtl(c, nc) ? -1 : 1797 !createWorker() ? 0 : UNCOMPENSATE); 1798 } 1799 else if (!compareAndSetCtl(c, c)) // validate 1800 return -1; 1801 else if ((sat = saturate) != null && sat.test(this)) 1802 return 0; 1803 else 1804 throw new RejectedExecutionException( 1805 "Thread limit exceeded replacing blocked worker"); 1806 } 1807 1808 /** 1809 * Readjusts RC count; called from ForkJoinTask after blocking. 1810 */ 1811 final void uncompensate() { 1812 getAndAddCtl(RC_UNIT); 1813 } 1814 1815 /** 1816 * Helps if possible until the given task is done. Scans other 1817 * queues for a task produced by one of w's stealers; returning 1818 * compensated blocking sentinel if none are found. 1819 * 1820 * @param task the task 1821 * @param w caller's WorkQueue 1822 * @return task status on exit, or UNCOMPENSATE for compensated blocking 1823 */ 1824 final int helpJoin(ForkJoinTask<?> task, WorkQueue w) { 1825 int s = 0; 1826 if (task != null && w != null) { 1827 int wsrc = w.source, wid = w.config & SMASK, r = wid + 2; 1828 boolean scan = true; 1829 long c = 0L; // track ctl stability 1830 outer: for (;;) { 1831 if ((s = task.status) < 0) 1832 break; 1833 else if (scan = !scan) { // previous scan was empty 1834 if (mode < 0) 1835 ForkJoinTask.cancelIgnoringExceptions(task); 1836 else if (c == (c = ctl) && (s = tryCompensate(c)) >= 0) 1837 break; // block 1838 } 1839 else { // scan for subtasks 1840 WorkQueue[] qs = queues; 1841 int n = (qs == null) ? 0 : qs.length, m = n - 1; 1842 for (int i = n; i > 0; i -= 2, r += 2) { 1843 int j; WorkQueue q, x, y; ForkJoinTask<?>[] a; 1844 if ((q = qs[j = r & m]) != null) { 1845 int sq = q.source & SMASK, cap, b; 1846 if ((a = q.array) != null && (cap = a.length) > 0) { 1847 int k = (cap - 1) & (b = q.base); 1848 int nextBase = b + 1, src = j | SRC, sx; 1849 ForkJoinTask<?> t = WorkQueue.getSlot(a, k); 1850 boolean eligible = sq == wid || 1851 ((x = qs[sq & m]) != null && // indirect 1852 ((sx = (x.source & SMASK)) == wid || 1853 ((y = qs[sx & m]) != null && // 2-indirect 1854 (y.source & SMASK) == wid))); 1855 if ((s = task.status) < 0) 1856 break outer; 1857 else if ((q.source & SMASK) != sq || 1858 q.base != b) 1859 scan = true; // inconsistent 1860 else if (t == null) 1861 scan |= (a[nextBase & (cap - 1)] != null || 1862 q.top != b); // lagging 1863 else if (eligible) { 1864 if (WorkQueue.casSlotToNull(a, k, t)) { 1865 q.base = nextBase; 1866 w.source = src; 1867 t.doExec(); 1868 w.source = wsrc; 1869 } 1870 scan = true; 1871 break; 1872 } 1873 } 1874 } 1875 } 1876 } 1877 } 1878 } 1879 return s; 1880 } 1881 1882 /** 1883 * Extra helpJoin steps for CountedCompleters. Scans for and runs 1884 * subtasks of the given root task, returning if none are found. 1885 * 1886 * @param task root of CountedCompleter computation 1887 * @param w caller's WorkQueue 1888 * @param owned true if owned by a ForkJoinWorkerThread 1889 * @return task status on exit 1890 */ 1891 final int helpComplete(ForkJoinTask<?> task, WorkQueue w, boolean owned) { 1892 int s = 0; 1893 if (task != null && w != null) { 1894 int r = w.config; 1895 boolean scan = true, locals = true; 1896 long c = 0L; 1897 outer: for (;;) { 1898 if (locals) { // try locals before scanning 1899 if ((s = w.helpComplete(task, owned, 0)) < 0) 1900 break; 1901 locals = false; 1902 } 1903 else if ((s = task.status) < 0) 1904 break; 1905 else if (scan = !scan) { 1906 if (c == (c = ctl)) 1907 break; 1908 } 1909 else { // scan for subtasks 1910 WorkQueue[] qs = queues; 1911 int n = (qs == null) ? 0 : qs.length; 1912 for (int i = n; i > 0; --i, ++r) { 1913 int j, cap, b; WorkQueue q; ForkJoinTask<?>[] a; 1914 boolean eligible = false; 1915 if ((q = qs[j = r & (n - 1)]) != null && 1916 (a = q.array) != null && (cap = a.length) > 0) { 1917 int k = (cap - 1) & (b = q.base), nextBase = b + 1; 1918 ForkJoinTask<?> t = WorkQueue.getSlot(a, k); 1919 if (t instanceof CountedCompleter) { 1920 CountedCompleter<?> f = (CountedCompleter<?>)t; 1921 do {} while (!(eligible = (f == task)) && 1922 (f = f.completer) != null); 1923 } 1924 if ((s = task.status) < 0) 1925 break outer; 1926 else if (q.base != b) 1927 scan = true; // inconsistent 1928 else if (t == null) 1929 scan |= (a[nextBase & (cap - 1)] != null || 1930 q.top != b); 1931 else if (eligible) { 1932 if (WorkQueue.casSlotToNull(a, k, t)) { 1933 q.setBaseOpaque(nextBase); 1934 t.doExec(); 1935 locals = true; 1936 } 1937 scan = true; 1938 break; 1939 } 1940 } 1941 } 1942 } 1943 } 1944 } 1945 return s; 1946 } 1947 1948 /** 1949 * Scans for and returns a polled task, if available. Used only 1950 * for untracked polls. Begins scan at an index (scanRover) 1951 * advanced on each call, to avoid systematic unfairness. 1952 * 1953 * @param submissionsOnly if true, only scan submission queues 1954 */ 1955 private ForkJoinTask<?> pollScan(boolean submissionsOnly) { 1956 VarHandle.acquireFence(); 1957 int r = scanRover += 0x61c88647; // Weyl increment; raciness OK 1958 if (submissionsOnly) // even indices only 1959 r &= ~1; 1960 int step = (submissionsOnly) ? 2 : 1; 1961 WorkQueue[] qs; int n; 1962 while ((qs = queues) != null && (n = qs.length) > 0) { 1963 boolean scan = false; 1964 for (int i = 0; i < n; i += step) { 1965 int j, cap, b; WorkQueue q; ForkJoinTask<?>[] a; 1966 if ((q = qs[j = (n - 1) & (r + i)]) != null && 1967 (a = q.array) != null && (cap = a.length) > 0) { 1968 int k = (cap - 1) & (b = q.base), nextBase = b + 1; 1969 ForkJoinTask<?> t = WorkQueue.getSlot(a, k); 1970 if (q.base != b) 1971 scan = true; 1972 else if (t == null) 1973 scan |= (q.top != b || a[nextBase & (cap - 1)] != null); 1974 else if (!WorkQueue.casSlotToNull(a, k, t)) 1975 scan = true; 1976 else { 1977 q.setBaseOpaque(nextBase); 1978 return t; 1979 } 1980 } 1981 } 1982 if (!scan && queues == qs) 1983 break; 1984 } 1985 return null; 1986 } 1987 1988 /** 1989 * Runs tasks until {@code isQuiescent()}. Rather than blocking 1990 * when tasks cannot be found, rescans until all others cannot 1991 * find tasks either. 1992 * 1993 * @param nanos max wait time (Long.MAX_VALUE if effectively untimed) 1994 * @param interruptible true if return on interrupt 1995 * @return positive if quiescent, negative if interrupted, else 0 1996 */ 1997 final int helpQuiescePool(WorkQueue w, long nanos, boolean interruptible) { 1998 if (w == null) 1999 return 0; 2000 long startTime = System.nanoTime(), parkTime = 0L; 2001 int prevSrc = w.source, wsrc = prevSrc, cfg = w.config, r = cfg + 1; 2002 for (boolean active = true, locals = true;;) { 2003 boolean busy = false, scan = false; 2004 if (locals) { // run local tasks before (re)polling 2005 locals = false; 2006 for (ForkJoinTask<?> u; (u = w.nextLocalTask(cfg)) != null;) 2007 u.doExec(); 2008 } 2009 WorkQueue[] qs = queues; 2010 int n = (qs == null) ? 0 : qs.length; 2011 for (int i = n; i > 0; --i, ++r) { 2012 int j, b, cap; WorkQueue q; ForkJoinTask<?>[] a; 2013 if ((q = qs[j = (n - 1) & r]) != null && q != w && 2014 (a = q.array) != null && (cap = a.length) > 0) { 2015 int k = (cap - 1) & (b = q.base); 2016 int nextBase = b + 1, src = j | SRC; 2017 ForkJoinTask<?> t = WorkQueue.getSlot(a, k); 2018 if (q.base != b) 2019 busy = scan = true; 2020 else if (t != null) { 2021 busy = scan = true; 2022 if (!active) { // increment before taking 2023 active = true; 2024 getAndAddCtl(RC_UNIT); 2025 } 2026 if (WorkQueue.casSlotToNull(a, k, t)) { 2027 q.base = nextBase; 2028 w.source = src; 2029 t.doExec(); 2030 w.source = wsrc = prevSrc; 2031 locals = true; 2032 } 2033 break; 2034 } 2035 else if (!busy) { 2036 if (q.top != b || a[nextBase & (cap - 1)] != null) 2037 busy = scan = true; 2038 else if (q.source != QUIET && q.phase >= 0) 2039 busy = true; 2040 } 2041 } 2042 } 2043 VarHandle.acquireFence(); 2044 if (!scan && queues == qs) { 2045 boolean interrupted; 2046 if (!busy) { 2047 w.source = prevSrc; 2048 if (!active) 2049 getAndAddCtl(RC_UNIT); 2050 return 1; 2051 } 2052 if (wsrc != QUIET) 2053 w.source = wsrc = QUIET; 2054 if (active) { // decrement 2055 active = false; 2056 parkTime = 0L; 2057 getAndAddCtl(RC_MASK & -RC_UNIT); 2058 } 2059 else if (parkTime == 0L) { 2060 parkTime = 1L << 10; // initially about 1 usec 2061 Thread.yield(); 2062 } 2063 else if ((interrupted = interruptible && Thread.interrupted()) || 2064 System.nanoTime() - startTime > nanos) { 2065 getAndAddCtl(RC_UNIT); 2066 return interrupted ? -1 : 0; 2067 } 2068 else { 2069 LockSupport.parkNanos(this, parkTime); 2070 if (parkTime < nanos >>> 8 && parkTime < 1L << 20) 2071 parkTime <<= 1; // max sleep approx 1 sec or 1% nanos 2072 } 2073 } 2074 } 2075 } 2076 2077 /** 2078 * Helps quiesce from external caller until done, interrupted, or timeout 2079 * 2080 * @param nanos max wait time (Long.MAX_VALUE if effectively untimed) 2081 * @param interruptible true if return on interrupt 2082 * @return positive if quiescent, negative if interrupted, else 0 2083 */ 2084 final int externalHelpQuiescePool(long nanos, boolean interruptible) { 2085 for (long startTime = System.nanoTime(), parkTime = 0L;;) { 2086 ForkJoinTask<?> t; 2087 if ((t = pollScan(false)) != null) { 2088 t.doExec(); 2089 parkTime = 0L; 2090 } 2091 else if (canStop()) 2092 return 1; 2093 else if (parkTime == 0L) { 2094 parkTime = 1L << 10; 2095 Thread.yield(); 2096 } 2097 else if ((System.nanoTime() - startTime) > nanos) 2098 return 0; 2099 else if (interruptible && Thread.interrupted()) 2100 return -1; 2101 else { 2102 LockSupport.parkNanos(this, parkTime); 2103 if (parkTime < nanos >>> 8 && parkTime < 1L << 20) 2104 parkTime <<= 1; 2105 } 2106 } 2107 } 2108 2109 /** 2110 * Gets and removes a local or stolen task for the given worker. 2111 * 2112 * @return a task, if available 2113 */ 2114 final ForkJoinTask<?> nextTaskFor(WorkQueue w) { 2115 ForkJoinTask<?> t; 2116 if (w == null || (t = w.nextLocalTask(w.config)) == null) 2117 t = pollScan(false); 2118 return t; 2119 } 2120 2121 // External operations 2122 2123 /** 2124 * Finds and locks a WorkQueue for an external submitter, or 2125 * returns null if shutdown or terminating. 2126 */ 2127 final WorkQueue submissionQueue() { 2128 int r; 2129 if ((r = ThreadLocalRandom.getProbe()) == 0) { 2130 ThreadLocalRandom.localInit(); // initialize caller's probe 2131 r = ThreadLocalRandom.getProbe(); 2132 } 2133 for (int id = r << 1;;) { // even indices only 2134 int md = mode, n, i; WorkQueue q; ReentrantLock lock; 2135 WorkQueue[] qs = queues; 2136 if ((md & SHUTDOWN) != 0 || qs == null || (n = qs.length) <= 0) 2137 return null; 2138 else if ((q = qs[i = (n - 1) & id]) == null) { 2139 if ((lock = registrationLock) != null) { 2140 WorkQueue w = new WorkQueue(id | SRC); 2141 lock.lock(); // install under lock 2142 if (qs[i] == null) 2143 qs[i] = w; // else lost race; discard 2144 lock.unlock(); 2145 } 2146 } 2147 else if (!q.tryLock()) // move and restart 2148 id = (r = ThreadLocalRandom.advanceProbe(r)) << 1; 2149 else 2150 return q; 2151 } 2152 } 2153 2154 /** 2155 * Adds the given task to an external submission queue, or throws 2156 * exception if shutdown or terminating. 2157 * 2158 * @param task the task. Caller must ensure non-null. 2159 */ 2160 final void externalPush(ForkJoinTask<?> task) { 2161 WorkQueue q; 2162 if ((q = submissionQueue()) == null) 2163 throw new RejectedExecutionException(); // shutdown or disabled 2164 else if (q.lockedPush(task)) 2165 signalWork(); 2166 } 2167 2168 /** 2169 * Pushes a possibly-external submission. 2170 */ 2171 private <T> ForkJoinTask<T> externalSubmit(ForkJoinTask<T> task) { 2172 Thread t; ForkJoinWorkerThread wt; WorkQueue q; 2173 if (task == null) 2174 throw new NullPointerException(); 2175 if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) && 2176 (q = (wt = (ForkJoinWorkerThread)t).workQueue) != null && 2177 wt.pool == this) 2178 q.push(task, this); 2179 else 2180 externalPush(task); 2181 return task; 2182 } 2183 2184 /** 2185 * Returns common pool queue for an external thread that has 2186 * possibly ever submitted a common pool task (nonzero probe), or 2187 * null if none. 2188 */ 2189 static WorkQueue commonQueue() { 2190 ForkJoinPool p; WorkQueue[] qs; 2191 int r = ThreadLocalRandom.getProbe(), n; 2192 return ((p = common) != null && (qs = p.queues) != null && 2193 (n = qs.length) > 0 && r != 0) ? 2194 qs[(n - 1) & (r << 1)] : null; 2195 } 2196 2197 /** 2198 * If the given executor is a ForkJoinPool, poll and execute 2199 * AsynchronousCompletionTasks from worker's queue until none are 2200 * available or blocker is released. 2201 */ 2202 static void helpAsyncBlocker(Executor e, ManagedBlocker blocker) { 2203 WorkQueue w = null; Thread t; ForkJoinWorkerThread wt; 2204 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) { 2205 if ((wt = (ForkJoinWorkerThread)t).pool == e) 2206 w = wt.workQueue; 2207 } 2208 else if (e == common) 2209 w = commonQueue(); 2210 if (w != null) 2211 w.helpAsyncBlocker(blocker); 2212 } 2213 2214 /** 2215 * Returns a cheap heuristic guide for task partitioning when 2216 * programmers, frameworks, tools, or languages have little or no 2217 * idea about task granularity. In essence, by offering this 2218 * method, we ask users only about tradeoffs in overhead vs 2219 * expected throughput and its variance, rather than how finely to 2220 * partition tasks. 2221 * 2222 * In a steady state strict (tree-structured) computation, each 2223 * thread makes available for stealing enough tasks for other 2224 * threads to remain active. Inductively, if all threads play by 2225 * the same rules, each thread should make available only a 2226 * constant number of tasks. 2227 * 2228 * The minimum useful constant is just 1. But using a value of 1 2229 * would require immediate replenishment upon each steal to 2230 * maintain enough tasks, which is infeasible. Further, 2231 * partitionings/granularities of offered tasks should minimize 2232 * steal rates, which in general means that threads nearer the top 2233 * of computation tree should generate more than those nearer the 2234 * bottom. In perfect steady state, each thread is at 2235 * approximately the same level of computation tree. However, 2236 * producing extra tasks amortizes the uncertainty of progress and 2237 * diffusion assumptions. 2238 * 2239 * So, users will want to use values larger (but not much larger) 2240 * than 1 to both smooth over transient shortages and hedge 2241 * against uneven progress; as traded off against the cost of 2242 * extra task overhead. We leave the user to pick a threshold 2243 * value to compare with the results of this call to guide 2244 * decisions, but recommend values such as 3. 2245 * 2246 * When all threads are active, it is on average OK to estimate 2247 * surplus strictly locally. In steady-state, if one thread is 2248 * maintaining say 2 surplus tasks, then so are others. So we can 2249 * just use estimated queue length. However, this strategy alone 2250 * leads to serious mis-estimates in some non-steady-state 2251 * conditions (ramp-up, ramp-down, other stalls). We can detect 2252 * many of these by further considering the number of "idle" 2253 * threads, that are known to have zero queued tasks, so 2254 * compensate by a factor of (#idle/#active) threads. 2255 */ 2256 static int getSurplusQueuedTaskCount() { 2257 Thread t; ForkJoinWorkerThread wt; ForkJoinPool pool; WorkQueue q; 2258 if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) && 2259 (pool = (wt = (ForkJoinWorkerThread)t).pool) != null && 2260 (q = wt.workQueue) != null) { 2261 int p = pool.mode & SMASK; 2262 int a = p + (int)(pool.ctl >> RC_SHIFT); 2263 int n = q.top - q.base; 2264 return n - (a > (p >>>= 1) ? 0 : 2265 a > (p >>>= 1) ? 1 : 2266 a > (p >>>= 1) ? 2 : 2267 a > (p >>>= 1) ? 4 : 2268 8); 2269 } 2270 return 0; 2271 } 2272 2273 // Termination 2274 2275 /** 2276 * Possibly initiates and/or completes termination. 2277 * 2278 * @param now if true, unconditionally terminate, else only 2279 * if no work and no active workers 2280 * @param enable if true, terminate when next possible 2281 * @return true if terminating or terminated 2282 */ 2283 private boolean tryTerminate(boolean now, boolean enable) { 2284 int md; // try to set SHUTDOWN, then STOP, then help terminate 2285 if (((md = mode) & SHUTDOWN) == 0) { 2286 if (!enable) 2287 return false; 2288 md = getAndBitwiseOrMode(SHUTDOWN); 2289 } 2290 if ((md & STOP) == 0) { 2291 if (!now && !canStop()) 2292 return false; 2293 md = getAndBitwiseOrMode(STOP); 2294 } 2295 for (int k = 0; k < 2; ++k) { // twice in case of lagging qs updates 2296 for (ForkJoinTask<?> t; (t = pollScan(false)) != null; ) 2297 ForkJoinTask.cancelIgnoringExceptions(t); // help cancel 2298 WorkQueue[] qs; int n; WorkQueue q; Thread thread; 2299 if ((qs = queues) != null && (n = qs.length) > 0) { 2300 for (int j = 1; j < n; j += 2) { // unblock other workers 2301 if ((q = qs[j]) != null && (thread = q.owner) != null && 2302 !thread.isInterrupted()) { 2303 try { 2304 thread.interrupt(); 2305 } catch (Throwable ignore) { 2306 } 2307 } 2308 } 2309 } 2310 ReentrantLock lock; Condition cond; // signal when no workers 2311 if (((md = mode) & TERMINATED) == 0 && 2312 (md & SMASK) + (short)(ctl >>> TC_SHIFT) <= 0 && 2313 (getAndBitwiseOrMode(TERMINATED) & TERMINATED) == 0 && 2314 (lock = registrationLock) != null) { 2315 lock.lock(); 2316 if ((cond = termination) != null) 2317 cond.signalAll(); 2318 lock.unlock(); 2319 } 2320 } 2321 return true; 2322 } 2323 2324 // Exported methods 2325 2326 // Constructors 2327 2328 /** 2329 * Creates a {@code ForkJoinPool} with parallelism equal to {@link 2330 * java.lang.Runtime#availableProcessors}, using defaults for all 2331 * other parameters (see {@link #ForkJoinPool(int, 2332 * ForkJoinWorkerThreadFactory, UncaughtExceptionHandler, boolean, 2333 * int, int, int, Predicate, long, TimeUnit)}). 2334 * 2335 * @throws SecurityException if a security manager exists and 2336 * the caller is not permitted to modify threads 2337 * because it does not hold {@link 2338 * java.lang.RuntimePermission}{@code ("modifyThread")} 2339 */ 2340 public ForkJoinPool() { 2341 this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()), 2342 defaultForkJoinWorkerThreadFactory, null, false, 2343 0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS); 2344 } 2345 2346 /** 2347 * Creates a {@code ForkJoinPool} with the indicated parallelism 2348 * level, using defaults for all other parameters (see {@link 2349 * #ForkJoinPool(int, ForkJoinWorkerThreadFactory, 2350 * UncaughtExceptionHandler, boolean, int, int, int, Predicate, 2351 * long, TimeUnit)}). 2352 * 2353 * @param parallelism the parallelism level 2354 * @throws IllegalArgumentException if parallelism less than or 2355 * equal to zero, or greater than implementation limit 2356 * @throws SecurityException if a security manager exists and 2357 * the caller is not permitted to modify threads 2358 * because it does not hold {@link 2359 * java.lang.RuntimePermission}{@code ("modifyThread")} 2360 */ 2361 public ForkJoinPool(int parallelism) { 2362 this(parallelism, defaultForkJoinWorkerThreadFactory, null, false, 2363 0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS); 2364 } 2365 2366 /** 2367 * Creates a {@code ForkJoinPool} with the given parameters (using 2368 * defaults for others -- see {@link #ForkJoinPool(int, 2369 * ForkJoinWorkerThreadFactory, UncaughtExceptionHandler, boolean, 2370 * int, int, int, Predicate, long, TimeUnit)}). 2371 * 2372 * @param parallelism the parallelism level. For default value, 2373 * use {@link java.lang.Runtime#availableProcessors}. 2374 * @param factory the factory for creating new threads. For default value, 2375 * use {@link #defaultForkJoinWorkerThreadFactory}. 2376 * @param handler the handler for internal worker threads that 2377 * terminate due to unrecoverable errors encountered while executing 2378 * tasks. For default value, use {@code null}. 2379 * @param asyncMode if true, 2380 * establishes local first-in-first-out scheduling mode for forked 2381 * tasks that are never joined. This mode may be more appropriate 2382 * than default locally stack-based mode in applications in which 2383 * worker threads only process event-style asynchronous tasks. 2384 * For default value, use {@code false}. 2385 * @throws IllegalArgumentException if parallelism less than or 2386 * equal to zero, or greater than implementation limit 2387 * @throws NullPointerException if the factory is null 2388 * @throws SecurityException if a security manager exists and 2389 * the caller is not permitted to modify threads 2390 * because it does not hold {@link 2391 * java.lang.RuntimePermission}{@code ("modifyThread")} 2392 */ 2393 public ForkJoinPool(int parallelism, 2394 ForkJoinWorkerThreadFactory factory, 2395 UncaughtExceptionHandler handler, 2396 boolean asyncMode) { 2397 this(parallelism, factory, handler, asyncMode, 2398 0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS); 2399 } 2400 2401 /** 2402 * Creates a {@code ForkJoinPool} with the given parameters. 2403 * 2404 * @param parallelism the parallelism level. For default value, 2405 * use {@link java.lang.Runtime#availableProcessors}. 2406 * 2407 * @param factory the factory for creating new threads. For 2408 * default value, use {@link #defaultForkJoinWorkerThreadFactory}. 2409 * 2410 * @param handler the handler for internal worker threads that 2411 * terminate due to unrecoverable errors encountered while 2412 * executing tasks. For default value, use {@code null}. 2413 * 2414 * @param asyncMode if true, establishes local first-in-first-out 2415 * scheduling mode for forked tasks that are never joined. This 2416 * mode may be more appropriate than default locally stack-based 2417 * mode in applications in which worker threads only process 2418 * event-style asynchronous tasks. For default value, use {@code 2419 * false}. 2420 * 2421 * @param corePoolSize the number of threads to keep in the pool 2422 * (unless timed out after an elapsed keep-alive). Normally (and 2423 * by default) this is the same value as the parallelism level, 2424 * but may be set to a larger value to reduce dynamic overhead if 2425 * tasks regularly block. Using a smaller value (for example 2426 * {@code 0}) has the same effect as the default. 2427 * 2428 * @param maximumPoolSize the maximum number of threads allowed. 2429 * When the maximum is reached, attempts to replace blocked 2430 * threads fail. (However, because creation and termination of 2431 * different threads may overlap, and may be managed by the given 2432 * thread factory, this value may be transiently exceeded.) To 2433 * arrange the same value as is used by default for the common 2434 * pool, use {@code 256} plus the {@code parallelism} level. (By 2435 * default, the common pool allows a maximum of 256 spare 2436 * threads.) Using a value (for example {@code 2437 * Integer.MAX_VALUE}) larger than the implementation's total 2438 * thread limit has the same effect as using this limit (which is 2439 * the default). 2440 * 2441 * @param minimumRunnable the minimum allowed number of core 2442 * threads not blocked by a join or {@link ManagedBlocker}. To 2443 * ensure progress, when too few unblocked threads exist and 2444 * unexecuted tasks may exist, new threads are constructed, up to 2445 * the given maximumPoolSize. For the default value, use {@code 2446 * 1}, that ensures liveness. A larger value might improve 2447 * throughput in the presence of blocked activities, but might 2448 * not, due to increased overhead. A value of zero may be 2449 * acceptable when submitted tasks cannot have dependencies 2450 * requiring additional threads. 2451 * 2452 * @param saturate if non-null, a predicate invoked upon attempts 2453 * to create more than the maximum total allowed threads. By 2454 * default, when a thread is about to block on a join or {@link 2455 * ManagedBlocker}, but cannot be replaced because the 2456 * maximumPoolSize would be exceeded, a {@link 2457 * RejectedExecutionException} is thrown. But if this predicate 2458 * returns {@code true}, then no exception is thrown, so the pool 2459 * continues to operate with fewer than the target number of 2460 * runnable threads, which might not ensure progress. 2461 * 2462 * @param keepAliveTime the elapsed time since last use before 2463 * a thread is terminated (and then later replaced if needed). 2464 * For the default value, use {@code 60, TimeUnit.SECONDS}. 2465 * 2466 * @param unit the time unit for the {@code keepAliveTime} argument 2467 * 2468 * @throws IllegalArgumentException if parallelism is less than or 2469 * equal to zero, or is greater than implementation limit, 2470 * or if maximumPoolSize is less than parallelism, 2471 * of if the keepAliveTime is less than or equal to zero. 2472 * @throws NullPointerException if the factory is null 2473 * @throws SecurityException if a security manager exists and 2474 * the caller is not permitted to modify threads 2475 * because it does not hold {@link 2476 * java.lang.RuntimePermission}{@code ("modifyThread")} 2477 * @since 9 2478 */ 2479 public ForkJoinPool(int parallelism, 2480 ForkJoinWorkerThreadFactory factory, 2481 UncaughtExceptionHandler handler, 2482 boolean asyncMode, 2483 int corePoolSize, 2484 int maximumPoolSize, 2485 int minimumRunnable, 2486 Predicate<? super ForkJoinPool> saturate, 2487 long keepAliveTime, 2488 TimeUnit unit) { 2489 checkPermission(); 2490 int p = parallelism; 2491 if (p <= 0 || p > MAX_CAP || p > maximumPoolSize || keepAliveTime <= 0L) 2492 throw new IllegalArgumentException(); 2493 if (factory == null || unit == null) 2494 throw new NullPointerException(); 2495 this.factory = factory; 2496 this.ueh = handler; 2497 this.saturate = saturate; 2498 this.keepAlive = Math.max(unit.toMillis(keepAliveTime), TIMEOUT_SLOP); 2499 int size = 1 << (33 - Integer.numberOfLeadingZeros(p - 1)); 2500 int corep = Math.min(Math.max(corePoolSize, p), MAX_CAP); 2501 int maxSpares = Math.min(maximumPoolSize, MAX_CAP) - p; 2502 int minAvail = Math.min(Math.max(minimumRunnable, 0), MAX_CAP); 2503 this.bounds = ((minAvail - p) & SMASK) | (maxSpares << SWIDTH); 2504 this.mode = p | (asyncMode ? FIFO : 0); 2505 this.ctl = ((((long)(-corep) << TC_SHIFT) & TC_MASK) | 2506 (((long)(-p) << RC_SHIFT) & RC_MASK)); 2507 this.registrationLock = new ReentrantLock(); 2508 this.queues = new WorkQueue[size]; 2509 String pid = Integer.toString(getAndAddPoolIds(1) + 1); 2510 this.workerNamePrefix = "ForkJoinPool-" + pid + "-worker-"; 2511 } 2512 2513 // helper method for commonPool constructor 2514 private static Object newInstanceFromSystemProperty(String property) 2515 throws ReflectiveOperationException { 2516 String className = System.getProperty(property); 2517 return (className == null) 2518 ? null 2519 : ClassLoader.getSystemClassLoader().loadClass(className) 2520 .getConstructor().newInstance(); 2521 } 2522 2523 /** 2524 * Constructor for common pool using parameters possibly 2525 * overridden by system properties 2526 */ 2527 private ForkJoinPool(byte forCommonPoolOnly) { 2528 int parallelism = Runtime.getRuntime().availableProcessors() - 1; 2529 ForkJoinWorkerThreadFactory fac = null; 2530 UncaughtExceptionHandler handler = null; 2531 try { // ignore exceptions in accessing/parsing properties 2532 fac = (ForkJoinWorkerThreadFactory) newInstanceFromSystemProperty( 2533 "java.util.concurrent.ForkJoinPool.common.threadFactory"); 2534 handler = (UncaughtExceptionHandler) newInstanceFromSystemProperty( 2535 "java.util.concurrent.ForkJoinPool.common.exceptionHandler"); 2536 String pp = System.getProperty 2537 ("java.util.concurrent.ForkJoinPool.common.parallelism"); 2538 if (pp != null) 2539 parallelism = Integer.parseInt(pp); 2540 } catch (Exception ignore) { 2541 } 2542 int p = this.mode = Math.min(Math.max(parallelism, 0), MAX_CAP); 2543 int size = 1 << (33 - Integer.numberOfLeadingZeros(p > 0 ? p - 1 : 1)); 2544 this.factory = (fac != null) ? fac : 2545 new DefaultCommonPoolForkJoinWorkerThreadFactory(); 2546 this.ueh = handler; 2547 this.keepAlive = DEFAULT_KEEPALIVE; 2548 this.saturate = null; 2549 this.workerNamePrefix = null; 2550 this.bounds = ((1 - p) & SMASK) | (COMMON_MAX_SPARES << SWIDTH); 2551 this.ctl = ((((long)(-p) << TC_SHIFT) & TC_MASK) | 2552 (((long)(-p) << RC_SHIFT) & RC_MASK)); 2553 this.queues = new WorkQueue[size]; 2554 this.registrationLock = new ReentrantLock(); 2555 } 2556 2557 /** 2558 * Returns the common pool instance. This pool is statically 2559 * constructed; its run state is unaffected by attempts to {@link 2560 * #shutdown} or {@link #shutdownNow}. However this pool and any 2561 * ongoing processing are automatically terminated upon program 2562 * {@link System#exit}. Any program that relies on asynchronous 2563 * task processing to complete before program termination should 2564 * invoke {@code commonPool().}{@link #awaitQuiescence awaitQuiescence}, 2565 * before exit. 2566 * 2567 * @return the common pool instance 2568 * @since 1.8 2569 */ 2570 public static ForkJoinPool commonPool() { 2571 // assert common != null : "static init error"; 2572 return common; 2573 } 2574 2575 // Execution methods 2576 2577 /** 2578 * Performs the given task, returning its result upon completion. 2579 * If the computation encounters an unchecked Exception or Error, 2580 * it is rethrown as the outcome of this invocation. Rethrown 2581 * exceptions behave in the same way as regular exceptions, but, 2582 * when possible, contain stack traces (as displayed for example 2583 * using {@code ex.printStackTrace()}) of both the current thread 2584 * as well as the thread actually encountering the exception; 2585 * minimally only the latter. 2586 * 2587 * @param task the task 2588 * @param <T> the type of the task's result 2589 * @return the task's result 2590 * @throws NullPointerException if the task is null 2591 * @throws RejectedExecutionException if the task cannot be 2592 * scheduled for execution 2593 */ 2594 public <T> T invoke(ForkJoinTask<T> task) { 2595 externalSubmit(task); 2596 return task.join(); 2597 } 2598 2599 /** 2600 * Arranges for (asynchronous) execution of the given task. 2601 * 2602 * @param task the task 2603 * @throws NullPointerException if the task is null 2604 * @throws RejectedExecutionException if the task cannot be 2605 * scheduled for execution 2606 */ 2607 public void execute(ForkJoinTask<?> task) { 2608 externalSubmit(task); 2609 } 2610 2611 // AbstractExecutorService methods 2612 2613 /** 2614 * @throws NullPointerException if the task is null 2615 * @throws RejectedExecutionException if the task cannot be 2616 * scheduled for execution 2617 */ 2618 @Override 2619 @SuppressWarnings("unchecked") 2620 public void execute(Runnable task) { 2621 externalSubmit((task instanceof ForkJoinTask<?>) 2622 ? (ForkJoinTask<Void>) task // avoid re-wrap 2623 : new ForkJoinTask.RunnableExecuteAction(task)); 2624 } 2625 2626 /** 2627 * Submits a ForkJoinTask for execution. 2628 * 2629 * @param task the task to submit 2630 * @param <T> the type of the task's result 2631 * @return the task 2632 * @throws NullPointerException if the task is null 2633 * @throws RejectedExecutionException if the task cannot be 2634 * scheduled for execution 2635 */ 2636 public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) { 2637 return externalSubmit(task); 2638 } 2639 2640 /** 2641 * @throws NullPointerException if the task is null 2642 * @throws RejectedExecutionException if the task cannot be 2643 * scheduled for execution 2644 */ 2645 @Override 2646 public <T> ForkJoinTask<T> submit(Callable<T> task) { 2647 return externalSubmit(new ForkJoinTask.AdaptedCallable<T>(task)); 2648 } 2649 2650 /** 2651 * @throws NullPointerException if the task is null 2652 * @throws RejectedExecutionException if the task cannot be 2653 * scheduled for execution 2654 */ 2655 @Override 2656 public <T> ForkJoinTask<T> submit(Runnable task, T result) { 2657 return externalSubmit(new ForkJoinTask.AdaptedRunnable<T>(task, result)); 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 @SuppressWarnings("unchecked") 2667 public ForkJoinTask<?> submit(Runnable task) { 2668 return externalSubmit((task instanceof ForkJoinTask<?>) 2669 ? (ForkJoinTask<Void>) task // avoid re-wrap 2670 : new ForkJoinTask.AdaptedRunnableAction(task)); 2671 } 2672 2673 /** 2674 * @throws NullPointerException {@inheritDoc} 2675 * @throws RejectedExecutionException {@inheritDoc} 2676 */ 2677 @Override 2678 public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) { 2679 ArrayList<Future<T>> futures = new ArrayList<>(tasks.size()); 2680 try { 2681 for (Callable<T> t : tasks) { 2682 ForkJoinTask<T> f = 2683 new ForkJoinTask.AdaptedInterruptibleCallable<T>(t); 2684 futures.add(f); 2685 externalSubmit(f); 2686 } 2687 for (int i = futures.size() - 1; i >= 0; --i) 2688 ((ForkJoinTask<?>)futures.get(i)).quietlyJoin(); 2689 return futures; 2690 } catch (Throwable t) { 2691 for (Future<T> e : futures) 2692 ForkJoinTask.cancelIgnoringExceptions(e); 2693 throw t; 2694 } 2695 } 2696 2697 @Override 2698 public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, 2699 long timeout, TimeUnit unit) 2700 throws InterruptedException { 2701 long nanos = unit.toNanos(timeout); 2702 ArrayList<Future<T>> futures = new ArrayList<>(tasks.size()); 2703 try { 2704 for (Callable<T> t : tasks) { 2705 ForkJoinTask<T> f = 2706 new ForkJoinTask.AdaptedInterruptibleCallable<T>(t); 2707 futures.add(f); 2708 externalSubmit(f); 2709 } 2710 long startTime = System.nanoTime(), ns = nanos; 2711 boolean timedOut = (ns < 0L); 2712 for (int i = futures.size() - 1; i >= 0; --i) { 2713 Future<T> f = futures.get(i); 2714 if (!f.isDone()) { 2715 if (timedOut) 2716 ForkJoinTask.cancelIgnoringExceptions(f); 2717 else { 2718 try { 2719 f.get(ns, TimeUnit.NANOSECONDS); 2720 } catch (CancellationException | TimeoutException | 2721 ExecutionException ok) { 2722 } 2723 if ((ns = nanos - (System.nanoTime() - startTime)) < 0L) 2724 timedOut = true; 2725 } 2726 } 2727 } 2728 return futures; 2729 } catch (Throwable t) { 2730 for (Future<T> e : futures) 2731 ForkJoinTask.cancelIgnoringExceptions(e); 2732 throw t; 2733 } 2734 } 2735 2736 // Task to hold results from InvokeAnyTasks 2737 static final class InvokeAnyRoot<E> extends ForkJoinTask<E> { 2738 private static final long serialVersionUID = 2838392045355241008L; 2739 @SuppressWarnings("serial") // Conditionally serializable 2740 volatile E result; 2741 final AtomicInteger count; // in case all throw 2742 final ForkJoinPool pool; // to check shutdown while collecting 2743 InvokeAnyRoot(int n, ForkJoinPool p) { 2744 pool = p; 2745 count = new AtomicInteger(n); 2746 } 2747 final void tryComplete(Callable<E> c) { // called by InvokeAnyTasks 2748 Throwable ex = null; 2749 boolean failed = (c == null || isCancelled() || 2750 (pool != null && pool.mode < 0)); 2751 if (!failed && !isDone()) { 2752 try { 2753 complete(c.call()); 2754 } catch (Throwable tx) { 2755 ex = tx; 2756 failed = true; 2757 } 2758 } 2759 if ((pool != null && pool.mode < 0) || 2760 (failed && count.getAndDecrement() <= 1)) 2761 trySetThrown(ex != null ? ex : new CancellationException()); 2762 } 2763 public final boolean exec() { return false; } // never forked 2764 public final E getRawResult() { return result; } 2765 public final void setRawResult(E v) { result = v; } 2766 } 2767 2768 // Variant of AdaptedInterruptibleCallable with results in InvokeAnyRoot 2769 static final class InvokeAnyTask<E> extends ForkJoinTask<E> { 2770 private static final long serialVersionUID = 2838392045355241008L; 2771 final InvokeAnyRoot<E> root; 2772 @SuppressWarnings("serial") // Conditionally serializable 2773 final Callable<E> callable; 2774 transient volatile Thread runner; 2775 InvokeAnyTask(InvokeAnyRoot<E> root, Callable<E> callable) { 2776 this.root = root; 2777 this.callable = callable; 2778 } 2779 public final boolean exec() { 2780 Thread.interrupted(); 2781 runner = Thread.currentThread(); 2782 root.tryComplete(callable); 2783 runner = null; 2784 Thread.interrupted(); 2785 return true; 2786 } 2787 public final boolean cancel(boolean mayInterruptIfRunning) { 2788 Thread t; 2789 boolean stat = super.cancel(false); 2790 if (mayInterruptIfRunning && (t = runner) != null) { 2791 try { 2792 t.interrupt(); 2793 } catch (Throwable ignore) { 2794 } 2795 } 2796 return stat; 2797 } 2798 public final void setRawResult(E v) {} // unused 2799 public final E getRawResult() { return null; } 2800 } 2801 2802 @Override 2803 public <T> T invokeAny(Collection<? extends Callable<T>> tasks) 2804 throws InterruptedException, ExecutionException { 2805 int n = tasks.size(); 2806 if (n <= 0) 2807 throw new IllegalArgumentException(); 2808 InvokeAnyRoot<T> root = new InvokeAnyRoot<T>(n, this); 2809 ArrayList<InvokeAnyTask<T>> fs = new ArrayList<>(n); 2810 try { 2811 for (Callable<T> c : tasks) { 2812 if (c == null) 2813 throw new NullPointerException(); 2814 InvokeAnyTask<T> f = new InvokeAnyTask<T>(root, c); 2815 fs.add(f); 2816 externalSubmit(f); 2817 if (root.isDone()) 2818 break; 2819 } 2820 return root.get(); 2821 } finally { 2822 for (InvokeAnyTask<T> f : fs) 2823 ForkJoinTask.cancelIgnoringExceptions(f); 2824 } 2825 } 2826 2827 @Override 2828 public <T> T invokeAny(Collection<? extends Callable<T>> tasks, 2829 long timeout, TimeUnit unit) 2830 throws InterruptedException, ExecutionException, TimeoutException { 2831 long nanos = unit.toNanos(timeout); 2832 int n = tasks.size(); 2833 if (n <= 0) 2834 throw new IllegalArgumentException(); 2835 InvokeAnyRoot<T> root = new InvokeAnyRoot<T>(n, this); 2836 ArrayList<InvokeAnyTask<T>> fs = new ArrayList<>(n); 2837 try { 2838 for (Callable<T> c : tasks) { 2839 if (c == null) 2840 throw new NullPointerException(); 2841 InvokeAnyTask<T> f = new InvokeAnyTask<T>(root, c); 2842 fs.add(f); 2843 externalSubmit(f); 2844 if (root.isDone()) 2845 break; 2846 } 2847 return root.get(nanos, TimeUnit.NANOSECONDS); 2848 } finally { 2849 for (InvokeAnyTask<T> f : fs) 2850 ForkJoinTask.cancelIgnoringExceptions(f); 2851 } 2852 } 2853 2854 /** 2855 * Returns the factory used for constructing new workers. 2856 * 2857 * @return the factory used for constructing new workers 2858 */ 2859 public ForkJoinWorkerThreadFactory getFactory() { 2860 return factory; 2861 } 2862 2863 /** 2864 * Returns the handler for internal worker threads that terminate 2865 * due to unrecoverable errors encountered while executing tasks. 2866 * 2867 * @return the handler, or {@code null} if none 2868 */ 2869 public UncaughtExceptionHandler getUncaughtExceptionHandler() { 2870 return ueh; 2871 } 2872 2873 /** 2874 * Returns the targeted parallelism level of this pool. 2875 * 2876 * @return the targeted parallelism level of this pool 2877 */ 2878 public int getParallelism() { 2879 int par = mode & SMASK; 2880 return (par > 0) ? par : 1; 2881 } 2882 2883 /** 2884 * Returns the targeted parallelism level of the common pool. 2885 * 2886 * @return the targeted parallelism level of the common pool 2887 * @since 1.8 2888 */ 2889 public static int getCommonPoolParallelism() { 2890 return COMMON_PARALLELISM; 2891 } 2892 2893 /** 2894 * Returns the number of worker threads that have started but not 2895 * yet terminated. The result returned by this method may differ 2896 * from {@link #getParallelism} when threads are created to 2897 * maintain parallelism when others are cooperatively blocked. 2898 * 2899 * @return the number of worker threads 2900 */ 2901 public int getPoolSize() { 2902 return ((mode & SMASK) + (short)(ctl >>> TC_SHIFT)); 2903 } 2904 2905 /** 2906 * Returns {@code true} if this pool uses local first-in-first-out 2907 * scheduling mode for forked tasks that are never joined. 2908 * 2909 * @return {@code true} if this pool uses async mode 2910 */ 2911 public boolean getAsyncMode() { 2912 return (mode & FIFO) != 0; 2913 } 2914 2915 /** 2916 * Returns an estimate of the number of worker threads that are 2917 * not blocked waiting to join tasks or for other managed 2918 * synchronization. This method may overestimate the 2919 * number of running threads. 2920 * 2921 * @return the number of worker threads 2922 */ 2923 public int getRunningThreadCount() { 2924 VarHandle.acquireFence(); 2925 WorkQueue[] qs; WorkQueue q; 2926 int rc = 0; 2927 if ((qs = queues) != null) { 2928 for (int i = 1; i < qs.length; i += 2) { 2929 if ((q = qs[i]) != null && q.isApparentlyUnblocked()) 2930 ++rc; 2931 } 2932 } 2933 return rc; 2934 } 2935 2936 /** 2937 * Returns an estimate of the number of threads that are currently 2938 * stealing or executing tasks. This method may overestimate the 2939 * number of active threads. 2940 * 2941 * @return the number of active threads 2942 */ 2943 public int getActiveThreadCount() { 2944 int r = (mode & SMASK) + (int)(ctl >> RC_SHIFT); 2945 return (r <= 0) ? 0 : r; // suppress momentarily negative values 2946 } 2947 2948 /** 2949 * Returns {@code true} if all worker threads are currently idle. 2950 * An idle worker is one that cannot obtain a task to execute 2951 * because none are available to steal from other threads, and 2952 * there are no pending submissions to the pool. This method is 2953 * conservative; it might not return {@code true} immediately upon 2954 * idleness of all threads, but will eventually become true if 2955 * threads remain inactive. 2956 * 2957 * @return {@code true} if all threads are currently idle 2958 */ 2959 public boolean isQuiescent() { 2960 return canStop(); 2961 } 2962 2963 /** 2964 * Returns an estimate of the total number of completed tasks that 2965 * were executed by a thread other than their submitter. The 2966 * reported value underestimates the actual total number of steals 2967 * when the pool is not quiescent. This value may be useful for 2968 * monitoring and tuning fork/join programs: in general, steal 2969 * counts should be high enough to keep threads busy, but low 2970 * enough to avoid overhead and contention across threads. 2971 * 2972 * @return the number of steals 2973 */ 2974 public long getStealCount() { 2975 long count = stealCount; 2976 WorkQueue[] qs; WorkQueue q; 2977 if ((qs = queues) != null) { 2978 for (int i = 1; i < qs.length; i += 2) { 2979 if ((q = qs[i]) != null) 2980 count += (long)q.nsteals & 0xffffffffL; 2981 } 2982 } 2983 return count; 2984 } 2985 2986 /** 2987 * Returns an estimate of the total number of tasks currently held 2988 * in queues by worker threads (but not including tasks submitted 2989 * to the pool that have not begun executing). This value is only 2990 * an approximation, obtained by iterating across all threads in 2991 * the pool. This method may be useful for tuning task 2992 * granularities. 2993 * 2994 * @return the number of queued tasks 2995 */ 2996 public long getQueuedTaskCount() { 2997 VarHandle.acquireFence(); 2998 WorkQueue[] qs; WorkQueue q; 2999 int count = 0; 3000 if ((qs = queues) != null) { 3001 for (int i = 1; i < qs.length; i += 2) { 3002 if ((q = qs[i]) != null) 3003 count += q.queueSize(); 3004 } 3005 } 3006 return count; 3007 } 3008 3009 /** 3010 * Returns an estimate of the number of tasks submitted to this 3011 * pool that have not yet begun executing. This method may take 3012 * time proportional to the number of submissions. 3013 * 3014 * @return the number of queued submissions 3015 */ 3016 public int getQueuedSubmissionCount() { 3017 VarHandle.acquireFence(); 3018 WorkQueue[] qs; WorkQueue q; 3019 int count = 0; 3020 if ((qs = queues) != null) { 3021 for (int i = 0; i < qs.length; i += 2) { 3022 if ((q = qs[i]) != null) 3023 count += q.queueSize(); 3024 } 3025 } 3026 return count; 3027 } 3028 3029 /** 3030 * Returns {@code true} if there are any tasks submitted to this 3031 * pool that have not yet begun executing. 3032 * 3033 * @return {@code true} if there are any queued submissions 3034 */ 3035 public boolean hasQueuedSubmissions() { 3036 VarHandle.acquireFence(); 3037 WorkQueue[] qs; WorkQueue q; 3038 if ((qs = queues) != null) { 3039 for (int i = 0; i < qs.length; i += 2) { 3040 if ((q = qs[i]) != null && !q.isEmpty()) 3041 return true; 3042 } 3043 } 3044 return false; 3045 } 3046 3047 /** 3048 * Removes and returns the next unexecuted submission if one is 3049 * available. This method may be useful in extensions to this 3050 * class that re-assign work in systems with multiple pools. 3051 * 3052 * @return the next submission, or {@code null} if none 3053 */ 3054 protected ForkJoinTask<?> pollSubmission() { 3055 return pollScan(true); 3056 } 3057 3058 /** 3059 * Removes all available unexecuted submitted and forked tasks 3060 * from scheduling queues and adds them to the given collection, 3061 * without altering their execution status. These may include 3062 * artificially generated or wrapped tasks. This method is 3063 * designed to be invoked only when the pool is known to be 3064 * quiescent. Invocations at other times may not remove all 3065 * tasks. A failure encountered while attempting to add elements 3066 * to collection {@code c} may result in elements being in 3067 * neither, either or both collections when the associated 3068 * exception is thrown. The behavior of this operation is 3069 * undefined if the specified collection is modified while the 3070 * operation is in progress. 3071 * 3072 * @param c the collection to transfer elements into 3073 * @return the number of elements transferred 3074 */ 3075 protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) { 3076 int count = 0; 3077 for (ForkJoinTask<?> t; (t = pollScan(false)) != null; ) { 3078 c.add(t); 3079 ++count; 3080 } 3081 return count; 3082 } 3083 3084 /** 3085 * Returns a string identifying this pool, as well as its state, 3086 * including indications of run state, parallelism level, and 3087 * worker and task counts. 3088 * 3089 * @return a string identifying this pool, as well as its state 3090 */ 3091 public String toString() { 3092 // Use a single pass through queues to collect counts 3093 int md = mode; // read volatile fields first 3094 long c = ctl; 3095 long st = stealCount; 3096 long qt = 0L, ss = 0L; int rc = 0; 3097 WorkQueue[] qs; WorkQueue q; 3098 if ((qs = queues) != null) { 3099 for (int i = 0; i < qs.length; ++i) { 3100 if ((q = qs[i]) != null) { 3101 int size = q.queueSize(); 3102 if ((i & 1) == 0) 3103 ss += size; 3104 else { 3105 qt += size; 3106 st += (long)q.nsteals & 0xffffffffL; 3107 if (q.isApparentlyUnblocked()) 3108 ++rc; 3109 } 3110 } 3111 } 3112 } 3113 3114 int pc = (md & SMASK); 3115 int tc = pc + (short)(c >>> TC_SHIFT); 3116 int ac = pc + (int)(c >> RC_SHIFT); 3117 if (ac < 0) // ignore transient negative 3118 ac = 0; 3119 String level = ((md & TERMINATED) != 0 ? "Terminated" : 3120 (md & STOP) != 0 ? "Terminating" : 3121 (md & SHUTDOWN) != 0 ? "Shutting down" : 3122 "Running"); 3123 return super.toString() + 3124 "[" + level + 3125 ", parallelism = " + pc + 3126 ", size = " + tc + 3127 ", active = " + ac + 3128 ", running = " + rc + 3129 ", steals = " + st + 3130 ", tasks = " + qt + 3131 ", submissions = " + ss + 3132 "]"; 3133 } 3134 3135 /** 3136 * Possibly initiates an orderly shutdown in which previously 3137 * submitted tasks are executed, but no new tasks will be 3138 * accepted. Invocation has no effect on execution state if this 3139 * is the {@link #commonPool()}, and no additional effect if 3140 * already shut down. Tasks that are in the process of being 3141 * submitted concurrently during the course of this method may or 3142 * may not be rejected. 3143 * 3144 * @throws SecurityException if a security manager exists and 3145 * the caller is not permitted to modify threads 3146 * because it does not hold {@link 3147 * java.lang.RuntimePermission}{@code ("modifyThread")} 3148 */ 3149 public void shutdown() { 3150 checkPermission(); 3151 if (this != common) 3152 tryTerminate(false, true); 3153 } 3154 3155 /** 3156 * Possibly attempts to cancel and/or stop all tasks, and reject 3157 * all subsequently submitted tasks. Invocation has no effect on 3158 * execution state if this is the {@link #commonPool()}, and no 3159 * additional effect if already shut down. Otherwise, tasks that 3160 * are in the process of being submitted or executed concurrently 3161 * during the course of this method may or may not be 3162 * rejected. This method cancels both existing and unexecuted 3163 * tasks, in order to permit termination in the presence of task 3164 * dependencies. So the method always returns an empty list 3165 * (unlike the case for some other Executors). 3166 * 3167 * @return an empty list 3168 * @throws SecurityException if a security manager exists and 3169 * the caller is not permitted to modify threads 3170 * because it does not hold {@link 3171 * java.lang.RuntimePermission}{@code ("modifyThread")} 3172 */ 3173 public List<Runnable> shutdownNow() { 3174 checkPermission(); 3175 if (this != common) 3176 tryTerminate(true, true); 3177 return Collections.emptyList(); 3178 } 3179 3180 /** 3181 * Returns {@code true} if all tasks have completed following shut down. 3182 * 3183 * @return {@code true} if all tasks have completed following shut down 3184 */ 3185 public boolean isTerminated() { 3186 return (mode & TERMINATED) != 0; 3187 } 3188 3189 /** 3190 * Returns {@code true} if the process of termination has 3191 * commenced but not yet completed. This method may be useful for 3192 * debugging. A return of {@code true} reported a sufficient 3193 * period after shutdown may indicate that submitted tasks have 3194 * ignored or suppressed interruption, or are waiting for I/O, 3195 * causing this executor not to properly terminate. (See the 3196 * advisory notes for class {@link ForkJoinTask} stating that 3197 * tasks should not normally entail blocking operations. But if 3198 * they do, they must abort them on interrupt.) 3199 * 3200 * @return {@code true} if terminating but not yet terminated 3201 */ 3202 public boolean isTerminating() { 3203 return (mode & (STOP | TERMINATED)) == STOP; 3204 } 3205 3206 /** 3207 * Returns {@code true} if this pool has been shut down. 3208 * 3209 * @return {@code true} if this pool has been shut down 3210 */ 3211 public boolean isShutdown() { 3212 return (mode & SHUTDOWN) != 0; 3213 } 3214 3215 /** 3216 * Blocks until all tasks have completed execution after a 3217 * shutdown request, or the timeout occurs, or the current thread 3218 * is interrupted, whichever happens first. Because the {@link 3219 * #commonPool()} never terminates until program shutdown, when 3220 * applied to the common pool, this method is equivalent to {@link 3221 * #awaitQuiescence(long, TimeUnit)} but always returns {@code false}. 3222 * 3223 * @param timeout the maximum time to wait 3224 * @param unit the time unit of the timeout argument 3225 * @return {@code true} if this executor terminated and 3226 * {@code false} if the timeout elapsed before termination 3227 * @throws InterruptedException if interrupted while waiting 3228 */ 3229 public boolean awaitTermination(long timeout, TimeUnit unit) 3230 throws InterruptedException { 3231 ReentrantLock lock; Condition cond; 3232 long nanos = unit.toNanos(timeout); 3233 boolean terminated = false; 3234 if (this == common) { 3235 Thread t; ForkJoinWorkerThread wt; int q; 3236 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread && 3237 (wt = (ForkJoinWorkerThread)t).pool == this) 3238 q = helpQuiescePool(wt.workQueue, nanos, true); 3239 else 3240 q = externalHelpQuiescePool(nanos, true); 3241 if (q < 0) 3242 throw new InterruptedException(); 3243 } 3244 else if (!(terminated = ((mode & TERMINATED) != 0)) && 3245 (lock = registrationLock) != null) { 3246 lock.lock(); 3247 try { 3248 if ((cond = termination) == null) 3249 termination = cond = lock.newCondition(); 3250 while (!(terminated = ((mode & TERMINATED) != 0)) && nanos > 0L) 3251 nanos = cond.awaitNanos(nanos); 3252 } finally { 3253 lock.unlock(); 3254 } 3255 } 3256 return terminated; 3257 } 3258 3259 /** 3260 * If called by a ForkJoinTask operating in this pool, equivalent 3261 * in effect to {@link ForkJoinTask#helpQuiesce}. Otherwise, 3262 * waits and/or attempts to assist performing tasks until this 3263 * pool {@link #isQuiescent} or the indicated timeout elapses. 3264 * 3265 * @param timeout the maximum time to wait 3266 * @param unit the time unit of the timeout argument 3267 * @return {@code true} if quiescent; {@code false} if the 3268 * timeout elapsed. 3269 */ 3270 public boolean awaitQuiescence(long timeout, TimeUnit unit) { 3271 Thread t; ForkJoinWorkerThread wt; int q; 3272 long nanos = unit.toNanos(timeout); 3273 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread && 3274 (wt = (ForkJoinWorkerThread)t).pool == this) 3275 q = helpQuiescePool(wt.workQueue, nanos, false); 3276 else 3277 q = externalHelpQuiescePool(nanos, false); 3278 return (q > 0); 3279 } 3280 3281 /** 3282 * Interface for extending managed parallelism for tasks running 3283 * in {@link ForkJoinPool}s. 3284 * 3285 * <p>A {@code ManagedBlocker} provides two methods. Method 3286 * {@link #isReleasable} must return {@code true} if blocking is 3287 * not necessary. Method {@link #block} blocks the current thread 3288 * if necessary (perhaps internally invoking {@code isReleasable} 3289 * before actually blocking). These actions are performed by any 3290 * thread invoking {@link 3291 * ForkJoinPool#managedBlock(ManagedBlocker)}. The unusual 3292 * methods in this API accommodate synchronizers that may, but 3293 * don't usually, block for long periods. Similarly, they allow 3294 * more efficient internal handling of cases in which additional 3295 * workers may be, but usually are not, needed to ensure 3296 * sufficient parallelism. Toward this end, implementations of 3297 * method {@code isReleasable} must be amenable to repeated 3298 * invocation. Neither method is invoked after a prior invocation 3299 * of {@code isReleasable} or {@code block} returns {@code true}. 3300 * 3301 * <p>For example, here is a ManagedBlocker based on a 3302 * ReentrantLock: 3303 * <pre> {@code 3304 * class ManagedLocker implements ManagedBlocker { 3305 * final ReentrantLock lock; 3306 * boolean hasLock = false; 3307 * ManagedLocker(ReentrantLock lock) { this.lock = lock; } 3308 * public boolean block() { 3309 * if (!hasLock) 3310 * lock.lock(); 3311 * return true; 3312 * } 3313 * public boolean isReleasable() { 3314 * return hasLock || (hasLock = lock.tryLock()); 3315 * } 3316 * }}</pre> 3317 * 3318 * <p>Here is a class that possibly blocks waiting for an 3319 * item on a given queue: 3320 * <pre> {@code 3321 * class QueueTaker<E> implements ManagedBlocker { 3322 * final BlockingQueue<E> queue; 3323 * volatile E item = null; 3324 * QueueTaker(BlockingQueue<E> q) { this.queue = q; } 3325 * public boolean block() throws InterruptedException { 3326 * if (item == null) 3327 * item = queue.take(); 3328 * return true; 3329 * } 3330 * public boolean isReleasable() { 3331 * return item != null || (item = queue.poll()) != null; 3332 * } 3333 * public E getItem() { // call after pool.managedBlock completes 3334 * return item; 3335 * } 3336 * }}</pre> 3337 */ 3338 public static interface ManagedBlocker { 3339 /** 3340 * Possibly blocks the current thread, for example waiting for 3341 * a lock or condition. 3342 * 3343 * @return {@code true} if no additional blocking is necessary 3344 * (i.e., if isReleasable would return true) 3345 * @throws InterruptedException if interrupted while waiting 3346 * (the method is not required to do so, but is allowed to) 3347 */ 3348 boolean block() throws InterruptedException; 3349 3350 /** 3351 * Returns {@code true} if blocking is unnecessary. 3352 * @return {@code true} if blocking is unnecessary 3353 */ 3354 boolean isReleasable(); 3355 } 3356 3357 /** 3358 * Runs the given possibly blocking task. When {@linkplain 3359 * ForkJoinTask#inForkJoinPool() running in a ForkJoinPool}, this 3360 * method possibly arranges for a spare thread to be activated if 3361 * necessary to ensure sufficient parallelism while the current 3362 * thread is blocked in {@link ManagedBlocker#block blocker.block()}. 3363 * 3364 * <p>This method repeatedly calls {@code blocker.isReleasable()} and 3365 * {@code blocker.block()} until either method returns {@code true}. 3366 * Every call to {@code blocker.block()} is preceded by a call to 3367 * {@code blocker.isReleasable()} that returned {@code false}. 3368 * 3369 * <p>If not running in a ForkJoinPool, this method is 3370 * behaviorally equivalent to 3371 * <pre> {@code 3372 * while (!blocker.isReleasable()) 3373 * if (blocker.block()) 3374 * break;}</pre> 3375 * 3376 * If running in a ForkJoinPool, the pool may first be expanded to 3377 * ensure sufficient parallelism available during the call to 3378 * {@code blocker.block()}. 3379 * 3380 * @param blocker the blocker task 3381 * @throws InterruptedException if {@code blocker.block()} did so 3382 */ 3383 public static void managedBlock(ManagedBlocker blocker) 3384 throws InterruptedException { 3385 Thread t; ForkJoinPool p; 3386 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread && 3387 (p = ((ForkJoinWorkerThread)t).pool) != null) 3388 p.compensatedBlock(blocker); 3389 else 3390 unmanagedBlock(blocker); 3391 } 3392 3393 /** ManagedBlock for ForkJoinWorkerThreads */ 3394 private void compensatedBlock(ManagedBlocker blocker) 3395 throws InterruptedException { 3396 if (blocker == null) throw new NullPointerException(); 3397 for (;;) { 3398 int comp; boolean done; 3399 long c = ctl; 3400 if (blocker.isReleasable()) 3401 break; 3402 if ((comp = tryCompensate(c)) >= 0) { 3403 long post = (comp == 0) ? 0L : RC_UNIT; 3404 try { 3405 done = blocker.block(); 3406 } finally { 3407 getAndAddCtl(post); 3408 } 3409 if (done) 3410 break; 3411 } 3412 } 3413 } 3414 3415 /** ManagedBlock for external threads */ 3416 private static void unmanagedBlock(ManagedBlocker blocker) 3417 throws InterruptedException { 3418 if (blocker == null) throw new NullPointerException(); 3419 do {} while (!blocker.isReleasable() && !blocker.block()); 3420 } 3421 3422 // AbstractExecutorService.newTaskFor overrides rely on 3423 // undocumented fact that ForkJoinTask.adapt returns ForkJoinTasks 3424 // that also implement RunnableFuture. 3425 3426 @Override 3427 protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { 3428 return new ForkJoinTask.AdaptedRunnable<T>(runnable, value); 3429 } 3430 3431 @Override 3432 protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { 3433 return new ForkJoinTask.AdaptedCallable<T>(callable); 3434 } 3435 3436 static { 3437 try { 3438 MethodHandles.Lookup l = MethodHandles.lookup(); 3439 CTL = l.findVarHandle(ForkJoinPool.class, "ctl", long.class); 3440 MODE = l.findVarHandle(ForkJoinPool.class, "mode", int.class); 3441 THREADIDS = l.findVarHandle(ForkJoinPool.class, "threadIds", int.class); 3442 POOLIDS = l.findStaticVarHandle(ForkJoinPool.class, "poolIds", int.class); 3443 } catch (ReflectiveOperationException e) { 3444 throw new ExceptionInInitializerError(e); 3445 } 3446 3447 // Reduce the risk of rare disastrous classloading in first call to 3448 // LockSupport.park: https://bugs.openjdk.java.net/browse/JDK-8074773 3449 Class<?> ensureLoaded = LockSupport.class; 3450 3451 int commonMaxSpares = DEFAULT_COMMON_MAX_SPARES; 3452 try { 3453 String p = System.getProperty 3454 ("java.util.concurrent.ForkJoinPool.common.maximumSpares"); 3455 if (p != null) 3456 commonMaxSpares = Integer.parseInt(p); 3457 } catch (Exception ignore) {} 3458 COMMON_MAX_SPARES = commonMaxSpares; 3459 3460 defaultForkJoinWorkerThreadFactory = 3461 new DefaultForkJoinWorkerThreadFactory(); 3462 modifyThreadPermission = new RuntimePermission("modifyThread"); 3463 common = AccessController.doPrivileged(new PrivilegedAction<>() { 3464 public ForkJoinPool run() { 3465 return new ForkJoinPool((byte)0); }}); 3466 3467 COMMON_PARALLELISM = Math.max(common.mode & SMASK, 1); 3468 } 3469 }