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.reflect.Field; 40 import java.util.ArrayList; 41 import java.util.Collection; 42 import java.util.Collections; 43 import java.util.List; 44 import java.util.Objects; 45 import java.util.function.Consumer; 46 import java.util.function.Predicate; 47 import java.util.concurrent.CountDownLatch; 48 import java.util.concurrent.locks.LockSupport; 49 import jdk.internal.access.JavaLangAccess; 50 import jdk.internal.access.JavaUtilConcurrentFJPAccess; 51 import jdk.internal.access.SharedSecrets; 52 import jdk.internal.misc.Unsafe; 53 import jdk.internal.vm.SharedThreadContainer; 54 import static java.util.concurrent.DelayScheduler.ScheduledForkJoinTask; 55 56 /** 57 * An {@link ExecutorService} for running {@link ForkJoinTask}s. 58 * A {@code ForkJoinPool} provides the entry point for submissions 59 * from non-{@code ForkJoinTask} clients, as well as management and 60 * monitoring operations. 61 * 62 * <p>A {@code ForkJoinPool} differs from other kinds of {@link 63 * ExecutorService} mainly by virtue of employing 64 * <em>work-stealing</em>: all threads in the pool attempt to find and 65 * execute tasks submitted to the pool and/or created by other active 66 * tasks (eventually blocking waiting for work if none exist). This 67 * enables efficient processing when most tasks spawn other subtasks 68 * (as do most {@code ForkJoinTask}s), as well as when many small 69 * tasks are submitted to the pool from external clients. Especially 70 * when setting <em>asyncMode</em> to true in constructors, {@code 71 * ForkJoinPool}s may also be appropriate for use with event-style 72 * tasks that are never joined. All worker threads are initialized 73 * with {@link Thread#isDaemon} set {@code true}. 74 * 75 * <p>A static {@link #commonPool()} is available and appropriate for 76 * most applications. The common pool is used by any ForkJoinTask that 77 * is not explicitly submitted to a specified pool. Using the common 78 * pool normally reduces resource usage (its threads are slowly 79 * reclaimed during periods of non-use, and reinstated upon subsequent 80 * use). 81 * 82 * <p>For applications that require separate or custom pools, a {@code 83 * ForkJoinPool} may be constructed with a given target parallelism 84 * level; by default, equal to the number of available processors. 85 * The pool attempts to maintain enough active (or available) threads 86 * by dynamically adding, suspending, or resuming internal worker 87 * threads, even if some tasks are stalled waiting to join others. 88 * However, no such adjustments are guaranteed in the face of blocked 89 * I/O or other unmanaged synchronization. The nested {@link 90 * ManagedBlocker} interface enables extension of the kinds of 91 * synchronization accommodated. The default policies may be 92 * overridden using a constructor with parameters corresponding to 93 * those documented in class {@link ThreadPoolExecutor}. 94 * 95 * <p>In addition to execution and lifecycle control methods, this 96 * class provides status check methods (for example 97 * {@link #getStealCount}) that are intended to aid in developing, 98 * tuning, and monitoring fork/join applications. Also, method 99 * {@link #toString} returns indications of pool state in a 100 * convenient form for informal monitoring. 101 * 102 * <p>As is the case with other ExecutorServices, there are three 103 * main task execution methods summarized in the following table. 104 * These are designed to be used primarily by clients not already 105 * engaged in fork/join computations in the current pool. The main 106 * forms of these methods accept instances of {@code ForkJoinTask}, 107 * but overloaded forms also allow mixed execution of plain {@code 108 * Runnable}- or {@code Callable}- based activities as well. However, 109 * tasks that are already executing in a pool should normally instead 110 * use the within-computation forms listed in the table unless using 111 * async event-style tasks that are not usually joined, in which case 112 * there is little difference among choice of methods. 113 * 114 * <table class="plain"> 115 * <caption>Summary of task execution methods</caption> 116 * <tr> 117 * <td></td> 118 * <th scope="col"> Call from non-fork/join clients</th> 119 * <th scope="col"> Call from within fork/join computations</th> 120 * </tr> 121 * <tr> 122 * <th scope="row" style="text-align:left"> Arrange async execution</th> 123 * <td> {@link #execute(ForkJoinTask)}</td> 124 * <td> {@link ForkJoinTask#fork}</td> 125 * </tr> 126 * <tr> 127 * <th scope="row" style="text-align:left"> Await and obtain result</th> 128 * <td> {@link #invoke(ForkJoinTask)}</td> 129 * <td> {@link ForkJoinTask#invoke}</td> 130 * </tr> 131 * <tr> 132 * <th scope="row" style="text-align:left"> Arrange exec and obtain Future</th> 133 * <td> {@link #submit(ForkJoinTask)}</td> 134 * <td> {@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td> 135 * </tr> 136 * </table> 137 * 138 * <p>Additionally, this class supports {@link 139 * ScheduledExecutorService} methods to delay or periodically execute 140 * tasks, as well as method {@link #submitWithTimeout} to cancel tasks 141 * that take too long. The scheduled functions or actions may create 142 * and invoke other {@linkplain ForkJoinTask ForkJoinTasks}. Delayed 143 * actions become enabled for execution and behave as ordinary submitted 144 * tasks when their delays elapse. Scheduling methods return 145 * {@linkplain ForkJoinTask ForkJoinTasks} that implement the {@link 146 * ScheduledFuture} interface. Resource exhaustion encountered after 147 * initial submission results in task cancellation. When time-based 148 * methods are used, shutdown policies match the default policies of 149 * class {@link ScheduledThreadPoolExecutor}: upon {@link #shutdown}, 150 * existing periodic tasks will not re-execute, and the pool 151 * terminates when quiescent and existing delayed tasks 152 * complete. Method {@link #cancelDelayedTasksOnShutdown} may be used 153 * to disable all delayed tasks upon shutdown, and method {@link 154 * #shutdownNow} may be used to instead unconditionally initiate pool 155 * termination. Monitoring methods such as {@link #getQueuedTaskCount} 156 * do not include scheduled tasks that are not yet enabled for execution, 157 * which are reported separately by method {@link 158 * #getDelayedTaskCount}. 159 * 160 * <p>The parameters used to construct the common pool may be controlled by 161 * setting the following {@linkplain System#getProperty system properties}: 162 * <ul> 163 * <li>{@systemProperty java.util.concurrent.ForkJoinPool.common.parallelism} 164 * - the parallelism level, a non-negative integer. Usage is discouraged. 165 * Use {@link #setParallelism} instead. 166 * <li>{@systemProperty java.util.concurrent.ForkJoinPool.common.threadFactory} 167 * - the class name of a {@link ForkJoinWorkerThreadFactory}. 168 * The {@linkplain ClassLoader#getSystemClassLoader() system class loader} 169 * is used to load this class. 170 * <li>{@systemProperty java.util.concurrent.ForkJoinPool.common.exceptionHandler} 171 * - the class name of a {@link UncaughtExceptionHandler}. 172 * The {@linkplain ClassLoader#getSystemClassLoader() system class loader} 173 * is used to load this class. 174 * <li>{@systemProperty java.util.concurrent.ForkJoinPool.common.maximumSpares} 175 * - the maximum number of allowed extra threads to maintain target 176 * parallelism (default 256). 177 * </ul> 178 * If no thread factory is supplied via a system property, then the 179 * common pool uses a factory that uses the system class loader as the 180 * {@linkplain Thread#getContextClassLoader() thread context class loader}. 181 * 182 * Upon any error in establishing these settings, default parameters 183 * are used. It is possible to disable use of threads by using a 184 * factory that may return {@code null}, in which case some tasks may 185 * never execute. While possible, it is strongly discouraged to set 186 * the parallelism property to zero, which may be internally 187 * overridden in the presence of intrinsically async tasks. 188 * 189 * @implNote This implementation restricts the maximum number of 190 * running threads to 32767. Attempts to create pools with greater 191 * than the maximum number result in {@code 192 * IllegalArgumentException}. Also, this implementation rejects 193 * submitted tasks (that is, by throwing {@link 194 * RejectedExecutionException}) only when the pool is shut down or 195 * internal resources have been exhausted. 196 * 197 * @since 1.7 198 * @author Doug Lea 199 */ 200 public class ForkJoinPool extends AbstractExecutorService 201 implements ScheduledExecutorService { 202 203 /* 204 * Implementation Overview 205 * 206 * This class and its nested classes provide the main 207 * functionality and control for a set of worker threads. Because 208 * most internal methods and nested classes are interrelated, 209 * their main rationale and descriptions are presented here; 210 * individual methods and nested classes contain only brief 211 * comments about details. Broadly: submissions from non-FJ 212 * threads enter into submission queues. Workers take these tasks 213 * and typically split them into subtasks that may be stolen by 214 * other workers. Work-stealing based on randomized scans 215 * generally leads to better throughput than "work dealing" in 216 * which producers assign tasks to idle threads, in part because 217 * threads that have finished other tasks before the signalled 218 * thread wakes up (which can be a long time) can take the task 219 * instead. Preference rules give first priority to processing 220 * tasks from their own queues (LIFO or FIFO, depending on mode), 221 * then to randomized FIFO steals of tasks in other queues. 222 * 223 * This framework began as vehicle for supporting structured 224 * parallelism using work-stealing, designed to work best when 225 * tasks are dag-structured (wrt completion dependencies), nested 226 * (generated using recursion or completions), of reasonable 227 * granularity, independent (wrt memory and resources) and where 228 * callers participate in task execution. These are properties 229 * that anyone aiming for efficient parallel multicore execution 230 * should design for. Over time, the scalability advantages of 231 * this framework led to extensions to better support more diverse 232 * usage contexts, amounting to weakenings or violations of each 233 * of these properties. Accommodating them may compromise 234 * performance, but mechanics discussed below include tradeoffs 235 * attempting to arrange that no single performance issue dominates. 236 * 237 * Here's a brief history of major revisions, each also with other 238 * minor features and changes. 239 * 240 * 1. Only handle recursively structured computational tasks 241 * 2. Async (FIFO) mode and striped submission queues 242 * 3. Completion-based tasks (mainly CountedCompleters) 243 * 4. CommonPool and parallelStream support 244 * 5. InterruptibleTasks for externally submitted tasks 245 * 6. Support ScheduledExecutorService methods 246 * 247 * Most changes involve adaptions of base algorithms using 248 * combinations of static and dynamic bitwise mode settings (both 249 * here and in ForkJoinTask), and subclassing of ForkJoinTask. 250 * There are a fair number of odd code constructions and design 251 * decisions for components that reside at the edge of Java vs JVM 252 * functionality. 253 * 254 * WorkQueues 255 * ========== 256 * 257 * Most operations occur within work-stealing queues (in nested 258 * class WorkQueue). These are special forms of Deques that 259 * support only three of the four possible end-operations -- push, 260 * pop, and poll (aka steal), under the further constraints that 261 * push and pop are called only from the owning thread (or, as 262 * extended here, under a lock), while poll may be called from 263 * other threads. (If you are unfamiliar with them, you probably 264 * want to read Herlihy and Shavit's book "The Art of 265 * Multiprocessor programming", chapter 16 describing these in 266 * more detail before proceeding.) The main work-stealing queue 267 * design is roughly similar to those in the papers "Dynamic 268 * Circular Work-Stealing Deque" by Chase and Lev, SPAA 2005 269 * (http://research.sun.com/scalable/pubs/index.html) and 270 * "Idempotent work stealing" by Michael, Saraswat, and Vechev, 271 * PPoPP 2009 (http://portal.acm.org/citation.cfm?id=1504186). 272 * The main differences ultimately stem from GC requirements that 273 * we null out taken slots as soon as we can, to maintain as small 274 * a footprint as possible even in programs generating huge 275 * numbers of tasks. To accomplish this, we shift the CAS 276 * arbitrating pop vs poll (steal) from being on the indices 277 * ("base" and "top") to the slots themselves. These provide the 278 * primary required memory ordering -- see "Correct and Efficient 279 * Work-Stealing for Weak Memory Models" by Le, Pop, Cohen, and 280 * Nardelli, PPoPP 2013 281 * (http://www.di.ens.fr/~zappa/readings/ppopp13.pdf) for an 282 * analysis of memory ordering requirements in work-stealing 283 * algorithms similar to the one used here. We use per-operation 284 * ordered writes of various kinds for accesses when required. 285 * 286 * We also support a user mode in which local task processing is 287 * in FIFO, not LIFO order, simply by using a local version of 288 * poll rather than pop. This can be useful in message-passing 289 * frameworks in which tasks are never joined, although with 290 * increased contention among task producers and consumers. Also, 291 * the same data structure (and class) is used for "submission 292 * queues" (described below) holding externally submitted tasks, 293 * that differ only in that a lock (using field "phase"; see below) is 294 * required by external callers to push and pop tasks. 295 * 296 * Adding tasks then takes the form of a classic array push(task) 297 * in a circular buffer: 298 * q.array[q.top++ % length] = task; 299 * 300 * The actual code needs to null-check and size-check the array, 301 * uses masking, not mod, for indexing a power-of-two-sized array, 302 * enforces memory ordering, supports resizing, and possibly 303 * signals waiting workers to start scanning (described below), 304 * which requires stronger forms of order accesses. 305 * 306 * The pop operation (always performed by owner) is of the form: 307 * if ((task = getAndSet(q.array, (q.top-1) % length, null)) != null) 308 * decrement top and return task; 309 * If this fails, the queue is empty. This operation is one part 310 * of the nextLocalTask method, that instead does a local-poll 311 * in FIFO mode. 312 * 313 * The poll operation is, basically: 314 * if (CAS nonnull task t = q.array[k = q.base % length] to null) 315 * increment base and return task; 316 * 317 * However, there are several more cases that must be dealt with. 318 * Some of them are just due to asynchrony; others reflect 319 * contention and stealing policies. Stepping through them 320 * illustrates some of the implementation decisions in this class. 321 * 322 * * Slot k must be read with an acquiring read, which it must 323 * anyway to dereference and run the task if the (acquiring) 324 * CAS succeeds. 325 * 326 * * q.base may change between reading and using its value to 327 * index the slot. To avoid trying to use the wrong t, the 328 * index and slot must be reread (not necessarily immediately) 329 * until consistent, unless this is a local poll by owner, in 330 * which case this form of inconsistency can only appear as t 331 * being null, below. 332 * 333 * * Similarly, q.array may change (due to a resize), unless this 334 * is a local poll by owner. Otherwise, when t is present, this 335 * only needs consideration on CAS failure (since a CAS 336 * confirms the non-resized case.) 337 * 338 * * t may appear null because a previous poll operation has not 339 * yet incremented q.base, so the read is from an already-taken 340 * index. This form of stall reflects the non-lock-freedom of 341 * the poll operation. Stalls can be detected by observing that 342 * q.base doesn't change on repeated reads of null t and when 343 * no other alternatives apply, spin-wait for it to settle. To 344 * reduce producing these kinds of stalls by other stealers, we 345 * encourage timely writes to indices using otherwise 346 * unnecessarily strong writes. 347 * 348 * * The CAS may fail, in which case we may want to retry unless 349 * there is too much contention. One goal is to balance and 350 * spread out the many forms of contention that may be 351 * encountered across polling and other operations to avoid 352 * sustained performance degradations. Across all cases where 353 * alternatives exist, a bounded number of CAS misses or stalls 354 * are tolerated (for slots, ctl, and elsewhere described 355 * below) before taking alternative action. These may move 356 * contention or retries elsewhere, which is still preferable 357 * to single-point bottlenecks. 358 * 359 * * Even though the check "top == base" is quiescently accurate 360 * to determine whether a queue is empty, it is not of much use 361 * when deciding whether to try to poll or repoll after a 362 * failure. Both top and base may move independently, and both 363 * lag updates to the underlying array. To reduce memory 364 * contention, non-owners avoid reading the "top" when 365 * possible, by using one-ahead reads to check whether to 366 * repoll, relying on the fact that a non-empty queue does not 367 * have two null slots in a row, except in cases (resizes and 368 * shifts) that can be detected with a secondary recheck that 369 * is less likely to conflict with owner writes. 370 * 371 * The poll operations in q.poll(), runWorker(), helpJoin(), and 372 * elsewhere differ with respect to whether other queues are 373 * available to try, and the presence or nature of screening steps 374 * when only some kinds of tasks can be taken. When alternatives 375 * (or failing) is an option, they uniformly give up after 376 * bounded numbers of stalls and/or CAS failures, which reduces 377 * contention when too many workers are polling too few tasks. 378 * Overall, in the aggregate, we ensure probabilistic 379 * non-blockingness of work-stealing at least until checking 380 * quiescence (which is intrinsically blocking): If an attempted 381 * steal fails in these ways, a scanning thief chooses a different 382 * target to try next. In contexts where alternatives aren't 383 * available, and when progress conditions can be isolated to 384 * values of a single variable, simple spinloops (using 385 * Thread.onSpinWait) are used to reduce memory traffic. 386 * 387 * WorkQueues are also used in a similar way for tasks submitted 388 * to the pool. We cannot mix these tasks in the same queues used 389 * by workers. Instead, we randomly associate submission queues 390 * with submitting threads (or carriers when using VirtualThreads) 391 * using a form of hashing. The ThreadLocalRandom probe value 392 * serves as a hash code for choosing existing queues, and may be 393 * randomly repositioned upon contention with other submitters. 394 * In essence, submitters act like workers except that they are 395 * restricted to executing local tasks that they submitted (or 396 * when known, subtasks thereof). Insertion of tasks in shared 397 * mode requires a lock. We use only a simple spinlock (as one 398 * role of field "phase") because submitters encountering a busy 399 * queue move to a different position to use or create other 400 * queues. They (spin) block when registering new queues, or 401 * indirectly elsewhere, by revisiting later. 402 * 403 * Management 404 * ========== 405 * 406 * The main throughput advantages of work-stealing stem from 407 * decentralized control -- workers mostly take tasks from 408 * themselves or each other, at rates that can exceed a billion 409 * per second. Most non-atomic control is performed by some form 410 * of scanning across or within queues. The pool itself creates, 411 * activates (enables scanning for and running tasks), 412 * deactivates, blocks, and terminates threads, all with minimal 413 * central information. There are only a few properties that we 414 * can globally track or maintain, so we pack them into a small 415 * number of variables, often maintaining atomicity without 416 * blocking or locking. Nearly all essentially atomic control 417 * state is held in a few variables that are by far most often 418 * read (not written) as status and consistency checks. We pack as 419 * much information into them as we can. 420 * 421 * Field "ctl" contains 64 bits holding information needed to 422 * atomically decide to add, enqueue (on an event queue), and 423 * dequeue and release workers. To enable this packing, we 424 * restrict maximum parallelism to (1<<15)-1 (which is far in 425 * excess of normal operating range) to allow ids, counts, and 426 * their negations (used for thresholding) to fit into 16bit 427 * subfields. 428 * 429 * Field "runState" and per-WorkQueue field "phase" play similar 430 * roles, as lockable, versioned counters. Field runState also 431 * includes monotonic event bits: 432 * * SHUTDOWN: no more external tasks accepted; STOP when quiescent 433 * * STOP: no more tasks run, and deregister all workers 434 * * CLEANED: all unexecuted tasks have been cancelled 435 * * TERMINATED: all workers deregistered and all queues cleaned 436 * The version tags enable detection of state changes (by 437 * comparing two reads) modulo bit wraparound. The bit range in 438 * each case suffices for purposes of determining quiescence, 439 * termination, avoiding ABA-like errors, and signal control, most 440 * of which are ultimately based on at most 15bit ranges (due to 441 * 32767 max total workers). RunState updates do not need to be 442 * atomic with respect to ctl updates, but because they are not, 443 * some care is required to avoid stalls. The seqLock properties 444 * detect changes and conditionally upgrade to coordinate with 445 * updates. It is typically held for less than a dozen 446 * instructions unless the queue array is being resized, during 447 * which contention is rare. To be conservative, lockRunState is 448 * implemented as a spin/sleep loop. Here and elsewhere spin 449 * constants are short enough to apply even on systems with few 450 * available processors. In addition to checking pool status, 451 * reads of runState sometimes serve as acquire fences before 452 * reading other fields. 453 * 454 * Field "parallelism" holds the target parallelism (normally 455 * corresponding to pool size). Users can dynamically reset target 456 * parallelism, but is only accessed when signalling or awaiting 457 * work, so only slowly has an effect in creating threads or 458 * letting them time out and terminate when idle. 459 * 460 * Array "queues" holds references to WorkQueues. It is updated 461 * (only during worker creation and termination) under the 462 * runState lock. It is otherwise concurrently readable but reads 463 * for use in scans (see below) are always prefaced by a volatile 464 * read of runState (or equivalent constructions), ensuring that 465 * its state is current at the point it is used (which is all we 466 * require). To simplify index-based operations, the array size is 467 * always a power of two, and all readers must tolerate null 468 * slots. Worker queues are at odd indices. Worker phase ids 469 * masked with SMASK match their index. Shared (submission) queues 470 * are at even indices. Grouping them together in this way aids in 471 * task scanning: At top-level, both kinds of queues should be 472 * sampled with approximately the same probability, which is 473 * simpler if they are all in the same array. But we also need to 474 * identify what kind they are without looking at them, leading to 475 * this odd/even scheme. One disadvantage is that there are 476 * usually many fewer submission queues, so there can be many 477 * wasted probes (null slots). But this is still cheaper than 478 * alternatives. Other loops over the queues array vary in origin 479 * and stride depending on whether they cover only submission 480 * (even) or worker (odd) queues or both, and whether they require 481 * randomness (in which case cyclically exhaustive strides may be 482 * used). 483 * 484 * All worker thread creation is on-demand, triggered by task 485 * submissions, replacement of terminated workers, and/or 486 * compensation for blocked workers. However, all other support 487 * code is set up to work with other policies. To ensure that we 488 * do not hold on to worker or task references that would prevent 489 * GC, all accesses to workQueues in waiting, signalling, and 490 * control methods are via indices into the queues array (which is 491 * one source of some of the messy code constructions here). In 492 * essence, the queues array serves as a weak reference 493 * mechanism. In particular, the stack top subfield of ctl stores 494 * indices, not references. Operations on queues obtained from 495 * these indices remain valid (with at most some unnecessary extra 496 * work) even if an underlying worker failed and was replaced by 497 * another at the same index. During termination, worker queue 498 * array updates are disabled. 499 * 500 * Queuing Idle Workers. Unlike HPC work-stealing frameworks, we 501 * cannot let workers spin indefinitely scanning for tasks when 502 * none can be found immediately, and we cannot start/resume 503 * workers unless there appear to be tasks available. On the 504 * other hand, we must quickly prod them into action when new 505 * tasks are submitted or generated. These latencies are mainly a 506 * function of JVM park/unpark (and underlying OS) performance, 507 * which can be slow and variable (even though usages are 508 * streamlined as much as possible). In many usages, ramp-up time 509 * is the main limiting factor in overall performance, which is 510 * compounded at program start-up by JIT compilation and 511 * allocation. On the other hand, throughput degrades when too 512 * many threads poll for too few tasks. (See below.) 513 * 514 * The "ctl" field atomically maintains total and "released" 515 * worker counts, plus the head of the available worker queue 516 * (actually stack, represented by the lower 32bit subfield of 517 * ctl). Released workers are those known to be scanning for 518 * and/or running tasks (we cannot accurately determine 519 * which). Unreleased ("available") workers are recorded in the 520 * ctl stack. These workers are made eligible for signalling by 521 * enqueuing in ctl (see method deactivate). This "queue" is a 522 * form of Treiber stack. This is ideal for activating threads in 523 * most-recently used order, and improves performance and 524 * locality, outweighing the disadvantages of being prone to 525 * contention and inability to release a worker unless it is 526 * topmost on stack. The top stack state holds the value of the 527 * "phase" field of the worker: its index and status, plus a 528 * version counter that, in addition to the count subfields (also 529 * serving as version stamps) provide protection against Treiber 530 * stack ABA effects. 531 * 532 * Creating workers. To create a worker, we pre-increment counts 533 * (serving as a reservation), and attempt to construct a 534 * ForkJoinWorkerThread via its factory. On starting, the new 535 * thread first invokes registerWorker, where it is assigned an 536 * index in the queues array (expanding the array if necessary). 537 * Upon any exception across these steps, or null return from 538 * factory, deregisterWorker adjusts counts and records 539 * accordingly. If a null return, the pool continues running with 540 * fewer than the target number workers. If exceptional, the 541 * exception is propagated, generally to some external caller. 542 * 543 * WorkQueue field "phase" encodes the queue array id in lower 544 * bits, and otherwise acts similarly to the pool runState field: 545 * The "IDLE" bit is clear while active (either a released worker 546 * or a locked external queue), with other bits serving as a 547 * version counter to distinguish changes across multiple reads. 548 * Note that phase field updates lag queue CAS releases; seeing a 549 * non-idle phase does not guarantee that the worker is available 550 * (and so is never checked in this way). 551 * 552 * The ctl field also serves as the basis for memory 553 * synchronization surrounding activation. This uses a more 554 * efficient version of a Dekker-like rule that task producers and 555 * consumers sync with each other by both writing/CASing ctl (even 556 * if to its current value). However, rather than CASing ctl to 557 * its current value in the common case where no action is 558 * required, we reduce write contention by ensuring that 559 * signalWork invocations are prefaced with a fully fenced memory 560 * access (which is usually needed anyway). 561 * 562 * Signalling. Signals (in signalWork) cause new or reactivated 563 * workers to scan for tasks. SignalWork is invoked in two cases: 564 * (1) When a task is pushed onto an empty queue, and (2) When a 565 * worker takes a top-level task from a queue that has additional 566 * tasks. Together, these suffice in O(log(#threads)) steps to 567 * fully activate with at least enough workers, and ideally no 568 * more than required. This ideal is unobtainable: Callers do not 569 * know whether another worker will finish its current task and 570 * poll for others without need of a signal (which is otherwise an 571 * advantage of work-stealing vs other schemes), and also must 572 * conservatively estimate the triggering conditions of emptiness 573 * or non-emptiness; all of which usually cause more activations 574 * than necessary (see below). (Method signalWork is also used as 575 * failsafe in case of Thread failures in deregisterWorker.) 576 * 577 * Top-Level scheduling 578 * ==================== 579 * 580 * Scanning. Method runWorker performs top-level scanning for (and 581 * execution of) tasks by polling a pseudo-random permutation of 582 * the array (by starting at a given index, and using a constant 583 * cyclically exhaustive stride.) It uses the same basic polling 584 * method as WorkQueue.poll(), but restarts with a different 585 * permutation on each rescan. The pseudorandom generator need 586 * not have high-quality statistical properties in the long 587 * term. We use Marsaglia XorShifts, seeded with the Weyl sequence 588 * from ThreadLocalRandom probes, which are cheap and suffice. 589 * 590 * Deactivation. When no tasks are found by a worker in runWorker, 591 * it invokes awaitWork, that first deactivates (to an IDLE 592 * phase). Avoiding missed signals during deactivation requires a 593 * (conservative) rescan, reactivating if there may be tasks to 594 * poll. Because idle workers are often not yet blocked (parked), 595 * we use a WorkQueue field to advertise that a waiter actually 596 * needs unparking upon signal. 597 * 598 * When tasks are constructed as (recursive) dags, top-level 599 * scanning is usually infrequent, and doesn't encounter most 600 * of the following problems addressed by runWorker and awaitWork: 601 * 602 * Locality. Polls are organized into "runs", continuing until 603 * empty or contended, while also minimizing interference by 604 * postponing bookeeping to ends of runs. This may reduce 605 * fairness, which is partially counteracted by the following. 606 * 607 * Contention. When many workers try to poll few queues, they 608 * often collide, generating CAS failures and disrupting locality 609 * of workers already running their tasks. This also leads to 610 * stalls when tasks cannot be taken because other workers have 611 * not finished poll operations, which is detected by reading 612 * ahead in queue arrays. In both caes, workers restart scans in a 613 * way that approximates randomized backoff. 614 * 615 * Oversignalling. When many short top-level tasks are present in 616 * a small number of queues, the above signalling strategy may 617 * activate many more workers than needed, worsening locality and 618 * contention problems, while also generating more global 619 * contention (field is CASed on every activation and 620 * deactivation). We filter out (both in runWorker and 621 * signalWork) attempted signals that are surely not needed 622 * because the signalled tasks are already taken. 623 * 624 * Shutdown and Quiescence 625 * ======================= 626 * 627 * Quiescence. Workers scan looking for work, giving up when they 628 * don't find any, without being sure that none are available. 629 * However, some required functionality relies on consensus about 630 * quiescence (also termination, discussed below). The count 631 * fields in ctl allow accurate discovery of states in which all 632 * workers are idle. However, because external (asynchronous) 633 * submitters are not part of this vote, these mechanisms 634 * themselves do not guarantee that the pool is in a quiescent 635 * state with respect to methods isQuiescent, shutdown (which 636 * begins termination when quiescent), helpQuiesce, and indirectly 637 * others including tryCompensate. Method quiescent() is used in 638 * all of these contexts. It provides checks that all workers are 639 * idle and there are no submissions that they could poll if they 640 * were not idle, retrying on inconsistent reads of queues and 641 * using the runState seqLock to retry on queue array updates. 642 * (It also reports quiescence if the pool is terminating.) A true 643 * report means only that there was a moment at which quiescence 644 * held. False negatives are inevitable (for example when queues 645 * indices lag updates, as described above), which is accommodated 646 * when (tentatively) idle by scanning for work etc, and then 647 * re-invoking. This includes cases in which the final unparked 648 * thread (in deactivate()) uses quiescent() to check for tasks 649 * that could have been added during a race window that would not 650 * be accompanied by a signal, in which case re-activating itself 651 * (or any other worker) to rescan. Method helpQuiesce acts 652 * similarly but cannot rely on ctl counts to determine that all 653 * workers are inactive because the caller and any others 654 * executing helpQuiesce are not included in counts. 655 * 656 * Termination. Termination is initiated by setting STOP in one of 657 * three ways (via methods tryTerminate and quiescent): 658 * * A call to shutdownNow, in which case all workers are 659 * interrupted, first ensuring that the queues array is stable, 660 * to avoid missing any workers. 661 * * A call to shutdown when quiescent, in which case method 662 * releaseWaiters is used to dequeue them, at which point they notice 663 * STOP state and return from runWorker to deregister(); 664 * * The pool becomes quiescent() sometime after shutdown has 665 * been called, in which case releaseWaiters is also used to 666 * propagate as they deregister. 667 * Upon STOP, each worker, as well as external callers to 668 * tryTerminate (via close() etc) race to set CLEANED, indicating 669 * that all tasks have been cancelled. The implementation (method 670 * cleanQueues) balances cases in which there may be many tasks to 671 * cancel (benefitting from parallelism) versus contention and 672 * interference when many threads try to poll remaining queues, 673 * while also avoiding unnecessary rechecks, by using 674 * pseudorandom scans and giving up upon interference. This may be 675 * retried by the same caller only when there are no more 676 * registered workers, using the same criteria as method 677 * quiescent. When CLEANED and all workers have deregistered, 678 * TERMINATED is set, also signalling any caller of 679 * awaitTermination or close. Because shutdownNow-based 680 * termination relies on interrupts, there is no guarantee that 681 * workers will stop if their tasks ignore interrupts. Class 682 * InterruptibleTask (see below) further arranges runState checks 683 * before executing task bodies, and ensures interrupts while 684 * terminating. Even so, there are no guarantees because tasks may 685 * internally enter unbounded loops. 686 * 687 * Trimming workers. To release resources after periods of lack of 688 * use, a worker starting to wait when the pool is quiescent will 689 * time out and terminate if the pool has remained quiescent for 690 * period given by field keepAlive (default 60sec), which applies 691 * to the first timeout of a quiescent pool. Subsequent cases use 692 * minimal delays such that, if still quiescent, all will be 693 * released soon thereafter. This is checked by setting the 694 * "source" field of signallee to an invalid value, that will 695 * remain invalid only if it did not process any tasks. 696 * 697 * Joining Tasks 698 * ============= 699 * 700 * The "Join" part of ForkJoinPools consists of a set of 701 * mechanisms that sometimes or always (depending on the kind of 702 * task) avoid context switching or adding worker threads when one 703 * task would otherwise be blocked waiting for completion of 704 * another, basically, just by running that task or one of its 705 * subtasks if not already taken. These mechanics are disabled for 706 * InterruptibleTasks, that guarantee that callers do not execute 707 * submitted tasks. 708 * 709 * The basic structure of joining is an extended spin/block scheme 710 * in which workers check for task completions status between 711 * steps to find other work, until relevant pool state stabilizes 712 * enough to believe that no such tasks are available, at which 713 * point blocking. This is usually a good choice of when to block 714 * that would otherwise be harder to approximate. 715 * 716 * These forms of helping may increase stack space usage, but that 717 * space is bounded in tree/dag structured procedurally parallel 718 * designs to be no more than that if a task were executed only by 719 * the joining thread. This is arranged by associated task 720 * subclasses that also help detect and control the ways in which 721 * this may occur. 722 * 723 * Normally, the first option when joining a task that is not done 724 * is to try to take it from the local queue and run it. Method 725 * tryRemoveAndExec tries to do so. For tasks with any form of 726 * subtasks that must be completed first, we try to locate these 727 * subtasks and run them as well. This is easy when local, but 728 * when stolen, steal-backs are restricted to the same rules as 729 * stealing (polling), which requires additional bookkeeping and 730 * scanning. This cost is still very much worthwhile because of 731 * its impact on task scheduling and resource control. 732 * 733 * The two methods for finding and executing subtasks vary in 734 * details. The algorithm in helpJoin entails a form of "linear 735 * helping". Each worker records (in field "source") the index of 736 * the internal queue from which it last stole a task. (Note: 737 * because chains cannot include even-numbered external queues, 738 * they are ignored, and 0 is an OK default. However, the source 739 * field is set anyway, or eventually to DROPPED, to ensure 740 * volatile memory synchronization effects.) The scan in method 741 * helpJoin uses these markers to try to find a worker to help 742 * (i.e., steal back a task from and execute it) that could make 743 * progress toward completion of the actively joined task. Thus, 744 * the joiner executes a task that would be on its own local deque 745 * if the to-be-joined task had not been stolen. This is a 746 * conservative variant of the approach described in Wagner & 747 * Calder "Leapfrogging: a portable technique for implementing 748 * efficient futures" SIGPLAN Notices, 1993 749 * (http://portal.acm.org/citation.cfm?id=155354). It differs 750 * mainly in that we only record queues, not full dependency 751 * links. This requires a linear scan of the queues to locate 752 * stealers, but isolates cost to when it is needed, rather than 753 * adding to per-task overhead. For CountedCompleters, the 754 * analogous method helpComplete doesn't need stealer-tracking, 755 * but requires a similar (but simpler) check of completion 756 * chains. 757 * 758 * In either case, searches can fail to locate stealers when 759 * stalls delay recording sources or issuing subtasks. We avoid 760 * some of these cases by using snapshotted values of ctl as a 761 * check that the numbers of workers are not changing, along with 762 * rescans to deal with contention and stalls. But even when 763 * accurately identified, stealers might not ever produce a task 764 * that the joiner can in turn help with. 765 * 766 * Related method helpAsyncBlocker does not directly rely on 767 * subtask structure, but instead avoids or postpones blocking of 768 * tagged tasks (CompletableFuture.AsynchronousCompletionTask) by 769 * executing other asyncs that can be processed in any order. 770 * This is currently invoked only in non-join-based blocking 771 * contexts from classes CompletableFuture and 772 * SubmissionPublisher, that could be further generalized. 773 * 774 * When any of the above fail to avoid blocking, we rely on 775 * "compensation" -- an indirect form of context switching that 776 * either activates an existing worker to take the place of the 777 * blocked one, or expands the number of workers. 778 * 779 * Compensation does not by default aim to keep exactly the target 780 * parallelism number of unblocked threads running at any given 781 * time. Some previous versions of this class employed immediate 782 * compensations for any blocked join. However, in practice, the 783 * vast majority of blockages are transient byproducts of GC and 784 * other JVM or OS activities that are made worse by replacement 785 * by causing longer-term oversubscription. These are inevitable 786 * without (unobtainably) perfect information about whether worker 787 * creation is actually necessary. False alarms are common enough 788 * to negatively impact performance, so compensation is by default 789 * attempted only when it appears possible that the pool could 790 * stall due to lack of any unblocked workers. However, we allow 791 * users to override defaults using the long form of the 792 * ForkJoinPool constructor. The compensation mechanism may also 793 * be bounded. Bounds for the commonPool better enable JVMs to 794 * cope with programming errors and abuse before running out of 795 * resources to do so. 796 * 797 * The ManagedBlocker extension API can't use helping so relies 798 * only on compensation in method awaitBlocker. This API was 799 * designed to highlight the uncertainty of compensation decisions 800 * by requiring implementation of method isReleasable to abort 801 * compensation during attempts to obtain a stable snapshot. But 802 * users now rely upon the fact that if isReleasable always 803 * returns false, the API can be used to obtain precautionary 804 * compensation, which is sometimes the only reasonable option 805 * when running unknown code in tasks; which is now supported more 806 * simply (see method beginCompensatedBlock). 807 * 808 * Common Pool 809 * =========== 810 * 811 * The static common pool always exists after static 812 * initialization. Since it (or any other created pool) need 813 * never be used, we minimize initial construction overhead and 814 * footprint to the setup of about a dozen fields, although with 815 * some System property parsing properties are set. The common pool is 816 * distinguished by having a null workerNamePrefix (which is an 817 * odd convention, but avoids the need to decode status in factory 818 * classes). It also has PRESET_SIZE config set if parallelism 819 * was configured by system property. 820 * 821 * When external threads use the common pool, they can perform 822 * subtask processing (see helpComplete and related methods) upon 823 * joins, unless they are submitted using ExecutorService 824 * submission methods, which implicitly disallow this. This 825 * caller-helps policy makes it sensible to set common pool 826 * parallelism level to one (or more) less than the total number 827 * of available cores, or even zero for pure caller-runs. External 828 * threads waiting for joins first check the common pool for their 829 * task, which fails quickly if the caller did not fork to common 830 * pool. 831 * 832 * Guarantees for common pool parallelism zero are limited to 833 * tasks that are joined by their callers in a tree-structured 834 * fashion or use CountedCompleters (as is true for jdk 835 * parallelStreams). Support infiltrates several methods, 836 * including those that retry helping steps until we are sure that 837 * none apply if there are no workers. To deal with conflicting 838 * requirements, uses of the commonPool that require async because 839 * caller-runs need not apply, ensure threads are enabled (by 840 * setting parallelism) via method asyncCommonPool before 841 * proceeding. (In principle, overriding zero parallelism needs to 842 * ensure at least one worker, but due to other backward 843 * compatibility contraints, ensures two.) 844 * 845 * As a more appropriate default in managed environments, unless 846 * overridden by system properties, we use workers of subclass 847 * InnocuousForkJoinWorkerThread for the commonPool. These 848 * workers do not belong to any user-defined ThreadGroup, and 849 * clear all ThreadLocals and reset the ContextClassLoader before 850 * (re)activating to execute top-level tasks. The associated 851 * mechanics may be JVM-dependent and must access particular 852 * Thread class fields to achieve this effect. 853 * 854 * InterruptibleTasks 855 * ==================== 856 * 857 * Regular ForkJoinTasks manage task cancellation (method cancel) 858 * independently from the interrupt status of threads running 859 * tasks. Interrupts are issued internally only while 860 * terminating, to wake up workers and cancel queued tasks. By 861 * default, interrupts are cleared only when necessary to ensure 862 * that calls to LockSupport.park do not loop indefinitely (park 863 * returns immediately if the current thread is interrupted). 864 * 865 * To comply with ExecutorService specs, we use subclasses of 866 * abstract class InterruptibleTask for tasks that require 867 * stronger interruption and cancellation guarantees. External 868 * submitters never run these tasks, even if in the common pool 869 * (as indicated by ForkJoinTask.noUserHelp status bit). 870 * InterruptibleTasks include a "runner" field (implemented 871 * similarly to FutureTask) to support cancel(true). Upon pool 872 * shutdown, runners are interrupted so they can cancel. Since 873 * external joining callers never run these tasks, they must await 874 * cancellation by others, which can occur along several different 875 * paths. 876 * 877 * Across these APIs, rules for reporting exceptions for tasks 878 * with results accessed via join() differ from those via get(), 879 * which differ from those invoked using pool submit methods by 880 * non-workers (which comply with Future.get() specs). Internal 881 * usages of ForkJoinTasks ignore interrupt status when executing 882 * or awaiting completion. Otherwise, reporting task results or 883 * exceptions is preferred to throwing InterruptedExceptions, 884 * which are in turn preferred to timeouts. Similarly, completion 885 * status is preferred to reporting cancellation. Cancellation is 886 * reported as an unchecked exception by join(), and by worker 887 * calls to get(), but is otherwise wrapped in a (checked) 888 * ExecutionException. 889 * 890 * Worker Threads cannot be VirtualThreads, as enforced by 891 * requiring ForkJoinWorkerThreads in factories. There are 892 * several constructions relying on this. However as of this 893 * writing, virtual thread bodies are by default run as some form 894 * of InterruptibleTask. 895 * 896 * DelayScheduler 897 * ================ 898 * 899 * This class supports ScheduledExecutorService methods by 900 * creating and starting a DelayScheduler on first use of these 901 * methods (via startDelayScheduler). The scheduler operates 902 * independently in its own thread, relaying tasks to the pool to 903 * execute when their delays elapse (see method 904 * executeEnabledScheduledTask). The only other interactions with 905 * the delayScheduler are to control shutdown and maintain 906 * shutdown-related policies in methods quiescent() and 907 * tryTerminate(). In particular, processing must deal with cases 908 * in which tasks are submitted before shutdown, but not enabled 909 * until afterwards, in which case they must bypass some screening 910 * to be allowed to run. Conversely, the DelayScheduler checks 911 * runState status and when enabled, completes termination, using 912 * only methods shutdownStatus and tryStopIfShutdown. All of these 913 * methods are final and have signatures referencing 914 * DelaySchedulers, so cannot conflict with those of any existing 915 * FJP subclasses. 916 * 917 * Memory placement 918 * ================ 919 * 920 * Performance is very sensitive to placement of instances of 921 * ForkJoinPool and WorkQueues and their queue arrays, as well as 922 * the placement of their fields. Caches misses and contention due 923 * to false-sharing have been observed to slow down some programs 924 * by more than a factor of four. Effects may vary across initial 925 * memory configuarations, applications, and different garbage 926 * collectors and GC settings, so there is no perfect solution. 927 * Too much isolation may generate more cache misses in common 928 * cases (because some fields snd slots are usually read at the 929 * same time). The @Contended annotation provides only rough 930 * control (for good reason). Similarly for relying on fields 931 * being placed in size-sorted declaration order. 932 * 933 * We isolate the ForkJoinPool.ctl field that otherwise causes the 934 * most false-sharing misses with respect to other fields. Also, 935 * ForkJoinPool fields are ordered such that fields less prone to 936 * contention effects are first, offsetting those that otherwise 937 * would be, while also reducing total footprint vs using 938 * multiple @Contended regions, which tends to slow down 939 * less-contended applications. To help arrange this, some 940 * non-reference fields are declared as "long" even when ints or 941 * shorts would suffice. For class WorkQueue, an 942 * embedded @Contended region segregates fields most heavily 943 * updated by owners from those most commonly read by stealers or 944 * other management. 945 * 946 * Initial sizing and resizing of WorkQueue arrays is an even more 947 * delicate tradeoff because the best strategy systematically 948 * varies across garbage collectors. Small arrays are better for 949 * locality and reduce GC scan time, but large arrays reduce both 950 * direct false-sharing and indirect cases due to GC bookkeeping 951 * (cardmarks etc), and reduce the number of resizes, which are 952 * not especially fast because they require atomic transfers. 953 * Currently, arrays are initialized to be just large enough to 954 * avoid resizing in most tree-structured tasks, but grow rapidly 955 * until large. (Maintenance note: any changes in fields, queues, 956 * or their uses, or JVM layout policies, must be accompanied by 957 * re-evaluation of these placement and sizing decisions.) 958 * 959 * Style notes 960 * =========== 961 * 962 * Memory ordering relies mainly on atomic operations (CAS, 963 * getAndSet, getAndAdd) along with moded accesses. These use 964 * jdk-internal Unsafe for atomics and special memory modes, 965 * rather than VarHandles, to avoid initialization dependencies in 966 * other jdk components that require early parallelism. This can 967 * be awkward and ugly, but also reflects the need to control 968 * outcomes across the unusual cases that arise in very racy code 969 * with very few invariants. All atomic task slot updates use 970 * Unsafe operations requiring offset positions, not indices, as 971 * computed by method slotOffset. All fields are read into locals 972 * before use, and null-checked if they are references, even if 973 * they can never be null under current usages. Usually, 974 * computations (held in local variables) are defined as soon as 975 * logically enabled, sometimes to convince compilers that they 976 * may be performed despite memory ordering constraints. Array 977 * accesses using masked indices include checks (that are always 978 * true) that the array length is non-zero to avoid compilers 979 * inserting more expensive traps. This is usually done in a 980 * "C"-like style of listing declarations at the heads of methods 981 * or blocks, and using inline assignments on first encounter. 982 * Nearly all explicit checks lead to bypass/return, not exception 983 * throws, because they may legitimately arise during shutdown. A 984 * few unusual loop constructions encourage (with varying 985 * effectiveness) JVMs about where (not) to place safepoints. All 986 * public methods screen arguments (mainly null checks) before 987 * creating or executing tasks. 988 * 989 * There is a lot of representation-level coupling among classes 990 * ForkJoinPool, ForkJoinWorkerThread, and ForkJoinTask. The 991 * fields of WorkQueue maintain data structures managed by 992 * ForkJoinPool, so are directly accessed. There is little point 993 * trying to reduce this, since any associated future changes in 994 * representations will need to be accompanied by algorithmic 995 * changes anyway. Several methods intrinsically sprawl because 996 * they must accumulate sets of consistent reads of fields held in 997 * local variables. Some others are artificially broken up to 998 * reduce producer/consumer imbalances due to dynamic compilation. 999 * There are also other coding oddities (including several 1000 * unnecessary-looking hoisted null checks) that help some methods 1001 * perform reasonably even when interpreted (not compiled). 1002 * 1003 * The order of declarations in this file is (with a few exceptions): 1004 * (1) Static configuration constants 1005 * (2) Static utility functions 1006 * (3) Nested (static) classes 1007 * (4) Fields, along with constants used when unpacking some of them 1008 * (5) Internal control methods 1009 * (6) Callbacks and other support for ForkJoinTask methods 1010 * (7) Exported methods 1011 * (8) Static block initializing statics in minimally dependent order 1012 * 1013 */ 1014 1015 // static configuration constants 1016 1017 /** 1018 * Default idle timeout value (in milliseconds) for idle threads 1019 * to park waiting for new work before terminating. 1020 */ 1021 static final long DEFAULT_KEEPALIVE = 60_000L; 1022 1023 /** 1024 * Undershoot tolerance for idle timeouts, also serving as the 1025 * minimum allowed timeout value. 1026 */ 1027 static final long TIMEOUT_SLOP = 20L; 1028 1029 /** 1030 * The default value for common pool maxSpares. Overridable using 1031 * the "java.util.concurrent.ForkJoinPool.common.maximumSpares" 1032 * system property. The default value is far in excess of normal 1033 * requirements, but also far short of maximum capacity and typical OS 1034 * thread limits, so allows JVMs to catch misuse/abuse before 1035 * running out of resources needed to do so. 1036 */ 1037 static final int DEFAULT_COMMON_MAX_SPARES = 256; 1038 1039 /** 1040 * Initial capacity of work-stealing queue array. 1041 * Must be a power of two, at least 2. See above. 1042 */ 1043 static final int INITIAL_QUEUE_CAPACITY = 1 << 6; 1044 1045 // conversions among short, int, long 1046 static final int SMASK = 0xffff; // (unsigned) short bits 1047 static final long LMASK = 0xffffffffL; // lower 32 bits of long 1048 static final long UMASK = ~LMASK; // upper 32 bits 1049 1050 // masks and sentinels for queue indices 1051 static final int MAX_CAP = 0x7fff; // max # workers 1052 static final int EXTERNAL_ID_MASK = 0x3ffe; // max external queue id 1053 static final int INVALID_ID = 0x4000; // unused external queue id 1054 1055 // pool.runState bits 1056 static final long STOP = 1L << 0; // terminating 1057 static final long SHUTDOWN = 1L << 1; // terminate when quiescent 1058 static final long CLEANED = 1L << 2; // stopped and queues cleared 1059 static final long TERMINATED = 1L << 3; // only set if STOP also set 1060 static final long RS_LOCK = 1L << 4; // lowest seqlock bit 1061 1062 // spin/sleep limits for runState locking and elsewhere 1063 static final int SPIN_WAITS = 1 << 7; // max calls to onSpinWait 1064 static final int MIN_SLEEP = 1 << 10; // approx 1 usec as nanos 1065 static final int MAX_SLEEP = 1 << 20; // approx 1 sec as nanos 1066 1067 // {pool, workQueue} config bits 1068 static final int FIFO = 1 << 0; // fifo queue or access mode 1069 static final int CLEAR_TLS = 1 << 1; // set for Innocuous workers 1070 static final int PRESET_SIZE = 1 << 2; // size was set by property 1071 1072 // others 1073 static final int DROPPED = 1 << 16; // removed from ctl counts 1074 static final int UNCOMPENSATE = 1 << 16; // tryCompensate return 1075 static final int IDLE = 1 << 16; // phase seqlock/version count 1076 static final int MIN_QUEUES_SIZE = 1 << 4; // ensure external slots 1077 1078 /* 1079 * Bits and masks for ctl and bounds are packed with 4 16 bit subfields: 1080 * RC: Number of released (unqueued) workers 1081 * TC: Number of total workers 1082 * SS: version count and status of top waiting thread 1083 * ID: poolIndex of top of Treiber stack of waiters 1084 * 1085 * When convenient, we can extract the lower 32 stack top bits 1086 * (including version bits) as sp=(int)ctl. When sp is non-zero, 1087 * there are waiting workers. Count fields may be transiently 1088 * negative during termination because of out-of-order updates. 1089 * To deal with this, we use casts in and out of "short" and/or 1090 * signed shifts to maintain signedness. Because it occupies 1091 * uppermost bits, we can add one release count using getAndAdd of 1092 * RC_UNIT, rather than CAS, when returning from a blocked join. 1093 * Other updates of multiple subfields require CAS. 1094 */ 1095 1096 // Release counts 1097 static final int RC_SHIFT = 48; 1098 static final long RC_UNIT = 0x0001L << RC_SHIFT; 1099 static final long RC_MASK = 0xffffL << RC_SHIFT; 1100 // Total counts 1101 static final int TC_SHIFT = 32; 1102 static final long TC_UNIT = 0x0001L << TC_SHIFT; 1103 static final long TC_MASK = 0xffffL << TC_SHIFT; 1104 1105 /* 1106 * All atomic operations on task arrays (queues) use Unsafe 1107 * operations that take array offsets versus indices, based on 1108 * array base and shift constants established during static 1109 * initialization. 1110 */ 1111 static final long ABASE; 1112 static final int ASHIFT; 1113 1114 // Static utilities 1115 1116 /** 1117 * Returns the array offset corresponding to the given index for 1118 * Unsafe task queue operations 1119 */ 1120 static long slotOffset(int index) { 1121 return ((long)index << ASHIFT) + ABASE; 1122 } 1123 1124 // Nested classes 1125 1126 /** 1127 * Factory for creating new {@link ForkJoinWorkerThread}s. 1128 * A {@code ForkJoinWorkerThreadFactory} must be defined and used 1129 * for {@code ForkJoinWorkerThread} subclasses that extend base 1130 * functionality or initialize threads with different contexts. 1131 */ 1132 public static interface ForkJoinWorkerThreadFactory { 1133 /** 1134 * Returns a new worker thread operating in the given pool. 1135 * Returning null or throwing an exception may result in tasks 1136 * never being executed. If this method throws an exception, 1137 * it is relayed to the caller of the method (for example 1138 * {@code execute}) causing attempted thread creation. If this 1139 * method returns null or throws an exception, it is not 1140 * retried until the next attempted creation (for example 1141 * another call to {@code execute}). 1142 * 1143 * @param pool the pool this thread works in 1144 * @return the new worker thread, or {@code null} if the request 1145 * to create a thread is rejected 1146 * @throws NullPointerException if the pool is null 1147 */ 1148 public ForkJoinWorkerThread newThread(ForkJoinPool pool); 1149 } 1150 1151 /** 1152 * Default ForkJoinWorkerThreadFactory implementation; creates a 1153 * new ForkJoinWorkerThread using the system class loader as the 1154 * thread context class loader. 1155 */ 1156 static final class DefaultForkJoinWorkerThreadFactory 1157 implements ForkJoinWorkerThreadFactory { 1158 public final ForkJoinWorkerThread newThread(ForkJoinPool pool) { 1159 return ((pool.workerNamePrefix == null) ? // is commonPool 1160 new ForkJoinWorkerThread.InnocuousForkJoinWorkerThread(pool) : 1161 new ForkJoinWorkerThread(null, pool, true, false)); 1162 } 1163 } 1164 1165 /** 1166 * Queues supporting work-stealing as well as external task 1167 * submission. See above for descriptions and algorithms. 1168 */ 1169 static final class WorkQueue { 1170 // fields declared in order of their likely layout on most VMs 1171 final ForkJoinWorkerThread owner; // null if shared 1172 ForkJoinTask<?>[] array; // the queued tasks; power of 2 size 1173 int base; // index of next slot for poll 1174 final int config; // mode bits 1175 1176 // fields otherwise causing more unnecessary false-sharing cache misses 1177 @jdk.internal.vm.annotation.Contended("w") 1178 int top; // index of next slot for push 1179 @jdk.internal.vm.annotation.Contended("w") 1180 volatile int phase; // versioned active status 1181 @jdk.internal.vm.annotation.Contended("w") 1182 int stackPred; // pool stack (ctl) predecessor link 1183 @jdk.internal.vm.annotation.Contended("w") 1184 volatile int source; // source queue id (or DROPPED) 1185 @jdk.internal.vm.annotation.Contended("w") 1186 int nsteals; // number of steals from other queues 1187 @jdk.internal.vm.annotation.Contended("w") 1188 volatile int parking; // nonzero if parked in awaitWork 1189 1190 // Support for atomic operations 1191 private static final Unsafe U; 1192 private static final long PHASE; 1193 private static final long BASE; 1194 private static final long TOP; 1195 private static final long ARRAY; 1196 1197 final void updateBase(int v) { 1198 U.putIntVolatile(this, BASE, v); 1199 } 1200 final void updateTop(int v) { 1201 U.putIntOpaque(this, TOP, v); 1202 } 1203 final void updateArray(ForkJoinTask<?>[] a) { 1204 U.getAndSetReference(this, ARRAY, a); 1205 } 1206 final void unlockPhase() { 1207 U.getAndAddInt(this, PHASE, IDLE); 1208 } 1209 final boolean tryLockPhase() { // seqlock acquire 1210 int p; 1211 return (((p = phase) & IDLE) != 0 && 1212 U.compareAndSetInt(this, PHASE, p, p + IDLE)); 1213 } 1214 1215 /** 1216 * Constructor. For internal queues, most fields are initialized 1217 * upon thread start in pool.registerWorker. 1218 */ 1219 WorkQueue(ForkJoinWorkerThread owner, int id, int cfg, 1220 boolean clearThreadLocals) { 1221 this.config = (clearThreadLocals) ? cfg | CLEAR_TLS : cfg; 1222 if ((this.owner = owner) == null) { 1223 phase = id | IDLE; 1224 array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY]; 1225 } 1226 } 1227 1228 /** 1229 * Returns an exportable index (used by ForkJoinWorkerThread). 1230 */ 1231 final int getPoolIndex() { 1232 return (phase & 0xffff) >>> 1; // ignore odd/even tag bit 1233 } 1234 1235 /** 1236 * Returns the approximate number of tasks in the queue. 1237 */ 1238 final int queueSize() { 1239 int unused = phase; // for ordering effect 1240 return Math.max(top - base, 0); // ignore transient negative 1241 } 1242 1243 /** 1244 * Pushes a task. Called only by owner or if already locked 1245 * 1246 * @param task the task; no-op if null 1247 * @param pool the pool to signal if was previously empty, else null 1248 * @param internal if caller owns this queue 1249 * @throws RejectedExecutionException if array could not be resized 1250 */ 1251 final void push(ForkJoinTask<?> task, ForkJoinPool pool, boolean internal) { 1252 int s = top, b = base, m, cap, room; ForkJoinTask<?>[] a; 1253 if ((a = array) != null && (cap = a.length) > 0) { // else disabled 1254 if ((room = (m = cap - 1) - (s - b)) >= 0) { 1255 top = s + 1; 1256 long pos = slotOffset(m & s); 1257 if (!internal) 1258 U.putReference(a, pos, task); // inside lock 1259 else 1260 U.getAndSetReference(a, pos, task); // fully fenced 1261 if (room == 0 && (a = growArray(a, cap, s)) != null) 1262 m = a.length - 1; // resize 1263 } 1264 if (!internal) 1265 unlockPhase(); 1266 if (room < 0) 1267 throw new RejectedExecutionException("Queue capacity exceeded"); 1268 if (pool != null && a != null && 1269 U.getReferenceAcquire(a, slotOffset(m & (s - 1))) == null) 1270 pool.signalWork(a, m & s); // may have appeared empty 1271 } 1272 } 1273 1274 /** 1275 * Resizes the queue array unless out of memory. 1276 * @param a old array 1277 * @param cap old array capacity 1278 * @param s current top 1279 * @return new array, or null on failure 1280 */ 1281 private ForkJoinTask<?>[] growArray(ForkJoinTask<?>[] a, int cap, int s) { 1282 int newCap = (cap >= 1 << 16) ? cap << 1 : cap << 2; 1283 ForkJoinTask<?>[] newArray = null; 1284 if (a != null && a.length == cap && cap > 0 && newCap > 0) { 1285 try { 1286 newArray = new ForkJoinTask<?>[newCap]; 1287 } catch (OutOfMemoryError ex) { 1288 } 1289 if (newArray != null) { // else throw on next push 1290 int mask = cap - 1, newMask = newCap - 1; 1291 for (int k = s, j = cap; j > 0; --j, --k) { 1292 ForkJoinTask<?> u; // poll old, push to new 1293 if ((u = (ForkJoinTask<?>)U.getAndSetReference( 1294 a, slotOffset(k & mask), null)) == null) 1295 break; // lost to pollers 1296 newArray[k & newMask] = u; 1297 } 1298 updateArray(newArray); // fully fenced 1299 } 1300 } 1301 return newArray; 1302 } 1303 1304 /** 1305 * Takes next task, if one exists, in order specified by mode, 1306 * so acts as either local-pop or local-poll. Called only by owner. 1307 * @param fifo nonzero if FIFO mode 1308 */ 1309 private ForkJoinTask<?> nextLocalTask(int fifo) { 1310 ForkJoinTask<?> t = null; 1311 ForkJoinTask<?>[] a = array; 1312 int b = base, p = top, cap; 1313 if (p - b > 0 && a != null && (cap = a.length) > 0) { 1314 for (int m = cap - 1, s, nb;;) { 1315 if (fifo == 0 || (nb = b + 1) == p) { 1316 if ((t = (ForkJoinTask<?>)U.getAndSetReference( 1317 a, slotOffset(m & (s = p - 1)), null)) != null) 1318 updateTop(s); // else lost race for only task 1319 break; 1320 } 1321 if ((t = (ForkJoinTask<?>)U.getAndSetReference( 1322 a, slotOffset(m & b), null)) != null) { 1323 updateBase(nb); 1324 break; 1325 } 1326 while (b == (b = U.getIntAcquire(this, BASE))) 1327 Thread.onSpinWait(); // spin to reduce memory traffic 1328 if (p - b <= 0) 1329 break; 1330 } 1331 } 1332 return t; 1333 } 1334 1335 /** 1336 * Takes next task, if one exists, using configured mode. 1337 * (Always internal, never called for Common pool.) 1338 */ 1339 final ForkJoinTask<?> nextLocalTask() { 1340 return nextLocalTask(config & FIFO); 1341 } 1342 1343 /** 1344 * Pops the given task only if it is at the current top. 1345 * @param task the task. Caller must ensure non-null. 1346 * @param internal if caller owns this queue 1347 */ 1348 final boolean tryUnpush(ForkJoinTask<?> task, boolean internal) { 1349 boolean taken = false; 1350 ForkJoinTask<?>[] a = array; 1351 int p = top, s = p - 1, cap; long k; 1352 if (a != null && (cap = a.length) > 0 && 1353 U.getReference(a, k = slotOffset((cap - 1) & s)) == task && 1354 (internal || tryLockPhase())) { 1355 if (top == p && U.compareAndSetReference(a, k, task, null)) { 1356 taken = true; 1357 updateTop(s); 1358 } 1359 if (!internal) 1360 unlockPhase(); 1361 } 1362 return taken; 1363 } 1364 1365 /** 1366 * Returns next task, if one exists, in order specified by mode. 1367 */ 1368 final ForkJoinTask<?> peek() { 1369 ForkJoinTask<?>[] a = array; 1370 int b = base, cfg = config, p = top, cap; 1371 if (p != b && a != null && (cap = a.length) > 0) { 1372 if ((cfg & FIFO) == 0) 1373 return a[(cap - 1) & (p - 1)]; 1374 else { // skip over in-progress removals 1375 ForkJoinTask<?> t; 1376 for ( ; p - b > 0; ++b) { 1377 if ((t = a[(cap - 1) & b]) != null) 1378 return t; 1379 } 1380 } 1381 } 1382 return null; 1383 } 1384 1385 /** 1386 * Polls for a task. Used only by non-owners. 1387 */ 1388 final ForkJoinTask<?> poll() { 1389 for (int pb = -1, b; ; pb = b) { // track progress 1390 ForkJoinTask<?> t; int cap, nb; long k; ForkJoinTask<?>[] a; 1391 if ((a = array) == null || (cap = a.length) <= 0) 1392 break; 1393 t = (ForkJoinTask<?>)U.getReferenceAcquire( 1394 a, k = slotOffset((cap - 1) & (b = base))); 1395 Object u = U.getReference( // next slot 1396 a, slotOffset((cap - 1) & (nb = b + 1))); 1397 if (base != b) // inconsistent 1398 ; 1399 else if (t == null) { 1400 if (u == null && top - b <= 0) 1401 break; // empty 1402 if (pb == b) 1403 Thread.onSpinWait(); // stalled 1404 } 1405 else if (U.compareAndSetReference(a, k, t, null)) { 1406 updateBase(nb); 1407 return t; 1408 } 1409 } 1410 return null; 1411 } 1412 1413 // specialized execution methods 1414 1415 /** 1416 * Runs the given task, as well as remaining local tasks. 1417 */ 1418 final void topLevelExec(ForkJoinTask<?> task, int fifo) { 1419 while (task != null) { 1420 task.doExec(); 1421 task = nextLocalTask(fifo); 1422 } 1423 } 1424 1425 /** 1426 * Deep form of tryUnpush: Traverses from top and removes and 1427 * runs task if present. 1428 */ 1429 final void tryRemoveAndExec(ForkJoinTask<?> task, boolean internal) { 1430 ForkJoinTask<?>[] a = array; 1431 int b = base, p = top, s = p - 1, d = p - b, cap; 1432 if (a != null && (cap = a.length) > 0) { 1433 for (int m = cap - 1, i = s; d > 0; --i, --d) { 1434 long k; boolean taken; 1435 ForkJoinTask<?> t = (ForkJoinTask<?>)U.getReference( 1436 a, k = slotOffset(i & m)); 1437 if (t == null) 1438 break; 1439 if (t == task) { 1440 if (!internal && !tryLockPhase()) 1441 break; // fail if locked 1442 if (taken = 1443 (top == p && 1444 U.compareAndSetReference(a, k, task, null))) { 1445 if (i == s) // act as pop 1446 updateTop(s); 1447 else if (i == base) // act as poll 1448 updateBase(i + 1); 1449 else { // swap with top 1450 U.putReferenceVolatile( 1451 a, k, (ForkJoinTask<?>) 1452 U.getAndSetReference( 1453 a, slotOffset(s & m), null)); 1454 updateTop(s); 1455 } 1456 } 1457 if (!internal) 1458 unlockPhase(); 1459 if (taken) 1460 task.doExec(); 1461 break; 1462 } 1463 } 1464 } 1465 } 1466 1467 /** 1468 * Tries to pop and run tasks within the target's computation 1469 * until done, not found, or limit exceeded. 1470 * 1471 * @param task root of computation 1472 * @param limit max runs, or zero for no limit 1473 * @return task status if known to be done 1474 */ 1475 final int helpComplete(ForkJoinTask<?> task, boolean internal, int limit) { 1476 int status = 0; 1477 if (task != null) { 1478 outer: for (;;) { 1479 ForkJoinTask<?>[] a; boolean taken; Object o; 1480 int stat, p, s, cap; 1481 if ((stat = task.status) < 0) { 1482 status = stat; 1483 break; 1484 } 1485 if ((a = array) == null || (cap = a.length) <= 0) 1486 break; 1487 long k = slotOffset((cap - 1) & (s = (p = top) - 1)); 1488 if (!((o = U.getReference(a, k)) instanceof CountedCompleter)) 1489 break; 1490 CountedCompleter<?> t = (CountedCompleter<?>)o, f = t; 1491 for (int steps = cap;;) { // bound path 1492 if (f == task) 1493 break; 1494 if ((f = f.completer) == null || --steps == 0) 1495 break outer; 1496 } 1497 if (!internal && !tryLockPhase()) 1498 break; 1499 if (taken = 1500 (top == p && 1501 U.compareAndSetReference(a, k, t, null))) 1502 updateTop(s); 1503 if (!internal) 1504 unlockPhase(); 1505 if (!taken) 1506 break; 1507 t.doExec(); 1508 if (limit != 0 && --limit == 0) 1509 break; 1510 } 1511 } 1512 return status; 1513 } 1514 1515 /** 1516 * Tries to poll and run AsynchronousCompletionTasks until 1517 * none found or blocker is released 1518 * 1519 * @param blocker the blocker 1520 */ 1521 final void helpAsyncBlocker(ManagedBlocker blocker) { 1522 for (;;) { 1523 ForkJoinTask<?> t; ForkJoinTask<?>[] a; int b, cap; long k; 1524 if ((a = array) == null || (cap = a.length) <= 0) 1525 break; 1526 t = (ForkJoinTask<?>)U.getReferenceAcquire( 1527 a, k = slotOffset((cap - 1) & (b = base))); 1528 if (t == null) { 1529 if (top - b <= 0) 1530 break; 1531 } 1532 else if (!(t instanceof CompletableFuture 1533 .AsynchronousCompletionTask)) 1534 break; 1535 if (blocker != null && blocker.isReleasable()) 1536 break; 1537 if (base == b && t != null && 1538 U.compareAndSetReference(a, k, t, null)) { 1539 updateBase(b + 1); 1540 t.doExec(); 1541 } 1542 } 1543 } 1544 1545 // misc 1546 1547 /** 1548 * Cancels all local tasks. Called only by owner. 1549 */ 1550 final void cancelTasks() { 1551 for (ForkJoinTask<?> t; (t = nextLocalTask(0)) != null; ) { 1552 try { 1553 t.cancel(false); 1554 } catch (Throwable ignore) { 1555 } 1556 } 1557 } 1558 1559 /** 1560 * Returns true if internal and not known to be blocked. 1561 */ 1562 final boolean isApparentlyUnblocked() { 1563 Thread wt; Thread.State s; 1564 return ((wt = owner) != null && (phase & IDLE) != 0 && 1565 (s = wt.getState()) != Thread.State.BLOCKED && 1566 s != Thread.State.WAITING && 1567 s != Thread.State.TIMED_WAITING); 1568 } 1569 1570 static { 1571 U = Unsafe.getUnsafe(); 1572 Class<WorkQueue> klass = WorkQueue.class; 1573 PHASE = U.objectFieldOffset(klass, "phase"); 1574 BASE = U.objectFieldOffset(klass, "base"); 1575 TOP = U.objectFieldOffset(klass, "top"); 1576 ARRAY = U.objectFieldOffset(klass, "array"); 1577 } 1578 } 1579 1580 // static fields (initialized in static initializer below) 1581 1582 /** 1583 * Creates a new ForkJoinWorkerThread. This factory is used unless 1584 * overridden in ForkJoinPool constructors. 1585 */ 1586 public static final ForkJoinWorkerThreadFactory 1587 defaultForkJoinWorkerThreadFactory; 1588 1589 /** 1590 * Common (static) pool. Non-null for public use unless a static 1591 * construction exception, but internal usages null-check on use 1592 * to paranoically avoid potential initialization circularities 1593 * as well as to simplify generated code. 1594 */ 1595 static final ForkJoinPool common; 1596 1597 /** 1598 * Sequence number for creating worker names 1599 */ 1600 private static volatile int poolIds; 1601 1602 /** 1603 * For VirtualThread intrinsics 1604 */ 1605 private static final JavaLangAccess JLA; 1606 1607 // fields declared in order of their likely layout on most VMs 1608 volatile CountDownLatch termination; // lazily constructed 1609 final Predicate<? super ForkJoinPool> saturate; 1610 final ForkJoinWorkerThreadFactory factory; 1611 final UncaughtExceptionHandler ueh; // per-worker UEH 1612 final SharedThreadContainer container; 1613 final String workerNamePrefix; // null for common pool 1614 final String poolName; 1615 volatile DelayScheduler delayScheduler; // lazily constructed 1616 WorkQueue[] queues; // main registry 1617 volatile long runState; // versioned, lockable 1618 final long keepAlive; // milliseconds before dropping if idle 1619 final long config; // static configuration bits 1620 volatile long stealCount; // collects worker nsteals 1621 volatile long threadIds; // for worker thread names 1622 1623 @jdk.internal.vm.annotation.Contended("fjpctl") // segregate 1624 volatile long ctl; // main pool control 1625 @jdk.internal.vm.annotation.Contended("fjpctl") // colocate 1626 int parallelism; // target number of workers 1627 1628 // Support for atomic operations 1629 private static final Unsafe U; 1630 private static final long CTL; 1631 private static final long RUNSTATE; 1632 private static final long PARALLELISM; 1633 private static final long THREADIDS; 1634 private static final long TERMINATION; 1635 private static final Object POOLIDS_BASE; 1636 private static final long POOLIDS; 1637 1638 private boolean compareAndSetCtl(long c, long v) { 1639 return U.compareAndSetLong(this, CTL, c, v); 1640 } 1641 private long compareAndExchangeCtl(long c, long v) { 1642 return U.compareAndExchangeLong(this, CTL, c, v); 1643 } 1644 private long getAndAddCtl(long v) { 1645 return U.getAndAddLong(this, CTL, v); 1646 } 1647 private long incrementThreadIds() { 1648 return U.getAndAddLong(this, THREADIDS, 1L); 1649 } 1650 private static int getAndAddPoolIds(int x) { 1651 return U.getAndAddInt(POOLIDS_BASE, POOLIDS, x); 1652 } 1653 private int getAndSetParallelism(int v) { 1654 return U.getAndSetInt(this, PARALLELISM, v); 1655 } 1656 private int getParallelismOpaque() { 1657 return U.getIntOpaque(this, PARALLELISM); 1658 } 1659 private CountDownLatch cmpExTerminationSignal(CountDownLatch x) { 1660 return (CountDownLatch) 1661 U.compareAndExchangeReference(this, TERMINATION, null, x); 1662 } 1663 1664 // runState operations 1665 1666 private long getAndBitwiseOrRunState(long v) { // for status bits 1667 return U.getAndBitwiseOrLong(this, RUNSTATE, v); 1668 } 1669 private boolean casRunState(long c, long v) { 1670 return U.compareAndSetLong(this, RUNSTATE, c, v); 1671 } 1672 private void unlockRunState() { // increment lock bit 1673 U.getAndAddLong(this, RUNSTATE, RS_LOCK); 1674 } 1675 private long lockRunState() { // lock and return current state 1676 long s, u; // locked when RS_LOCK set 1677 if (((s = runState) & RS_LOCK) == 0L && casRunState(s, u = s + RS_LOCK)) 1678 return u; 1679 else 1680 return spinLockRunState(); 1681 } 1682 private long spinLockRunState() { // spin/sleep 1683 for (int waits = 0;;) { 1684 long s, u; 1685 if (((s = runState) & RS_LOCK) == 0L) { 1686 if (casRunState(s, u = s + RS_LOCK)) 1687 return u; 1688 waits = 0; 1689 } else if (waits < SPIN_WAITS) { 1690 ++waits; 1691 Thread.onSpinWait(); 1692 } else { 1693 if (waits < MIN_SLEEP) 1694 waits = MIN_SLEEP; 1695 LockSupport.parkNanos(this, (long)waits); 1696 if (waits < MAX_SLEEP) 1697 waits <<= 1; 1698 } 1699 } 1700 } 1701 1702 static boolean poolIsStopping(ForkJoinPool p) { // Used by ForkJoinTask 1703 return p != null && (p.runState & STOP) != 0L; 1704 } 1705 1706 // Creating, registering, and deregistering workers 1707 1708 /** 1709 * Tries to construct and start one worker. Assumes that total 1710 * count has already been incremented as a reservation. Invokes 1711 * deregisterWorker on any failure. 1712 * 1713 * @return true if successful 1714 */ 1715 private boolean createWorker() { 1716 ForkJoinWorkerThreadFactory fac = factory; 1717 SharedThreadContainer ctr = container; 1718 Throwable ex = null; 1719 ForkJoinWorkerThread wt = null; 1720 try { 1721 if ((runState & STOP) == 0L && // avoid construction if terminating 1722 fac != null && (wt = fac.newThread(this)) != null) { 1723 if (ctr != null) 1724 ctr.start(wt); 1725 else 1726 wt.start(); 1727 return true; 1728 } 1729 } catch (Throwable rex) { 1730 ex = rex; 1731 } 1732 deregisterWorker(wt, ex); 1733 return false; 1734 } 1735 1736 /** 1737 * Provides a name for ForkJoinWorkerThread constructor. 1738 */ 1739 final String nextWorkerThreadName() { 1740 String prefix = workerNamePrefix; 1741 long tid = incrementThreadIds() + 1L; 1742 if (prefix == null) // commonPool has no prefix 1743 prefix = "ForkJoinPool.commonPool-worker-"; 1744 return prefix.concat(Long.toString(tid)); 1745 } 1746 1747 /** 1748 * Finishes initializing and records internal queue. 1749 * 1750 * @param w caller's WorkQueue 1751 */ 1752 final void registerWorker(WorkQueue w) { 1753 if (w != null) { 1754 w.array = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY]; 1755 ThreadLocalRandom.localInit(); 1756 int seed = w.stackPred = ThreadLocalRandom.getProbe(); 1757 int phaseSeq = seed & ~((IDLE << 1) - 1); // initial phase tag 1758 int id = ((seed << 1) | 1) & SMASK; // base of linear-probe-like scan 1759 long stop = lockRunState() & STOP; 1760 try { 1761 WorkQueue[] qs; int n; 1762 if (stop == 0L && (qs = queues) != null && (n = qs.length) > 0) { 1763 for (int k = n, m = n - 1; ; id += 2) { 1764 if (qs[id &= m] == null) 1765 break; 1766 if ((k -= 2) <= 0) { 1767 id |= n; 1768 break; 1769 } 1770 } 1771 w.phase = id | phaseSeq; // now publishable 1772 if (id < n) 1773 qs[id] = w; 1774 else { // expand 1775 int an = n << 1, am = an - 1; 1776 WorkQueue[] as = new WorkQueue[an]; 1777 as[id & am] = w; 1778 for (int j = 1; j < n; j += 2) 1779 as[j] = qs[j]; 1780 for (int j = 0; j < n; j += 2) { 1781 WorkQueue q; // shared queues may move 1782 if ((q = qs[j]) != null) 1783 as[q.phase & EXTERNAL_ID_MASK & am] = q; 1784 } 1785 U.storeFence(); // fill before publish 1786 queues = as; 1787 } 1788 } 1789 } finally { 1790 unlockRunState(); 1791 } 1792 } 1793 } 1794 1795 /** 1796 * Final callback from terminating worker, as well as upon failure 1797 * to construct or start a worker. Removes record of worker from 1798 * array, and adjusts counts. If pool is shutting down, tries to 1799 * complete termination. 1800 * 1801 * @param wt the worker thread, or null if construction failed 1802 * @param ex the exception causing failure, or null if none 1803 */ 1804 final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) { 1805 WorkQueue w = null; // null if not created 1806 int phase = 0; // 0 if not registered 1807 if (wt != null && (w = wt.workQueue) != null && 1808 (phase = w.phase) != 0 && (phase & IDLE) != 0) 1809 releaseWaiters(); // ensure released 1810 if (w == null || w.source != DROPPED) { 1811 long c = ctl; // decrement counts 1812 do {} while (c != (c = compareAndExchangeCtl( 1813 c, ((RC_MASK & (c - RC_UNIT)) | 1814 (TC_MASK & (c - TC_UNIT)) | 1815 (LMASK & c))))); 1816 } 1817 if (phase != 0 && w != null) { // remove index unless terminating 1818 long ns = w.nsteals & 0xffffffffL; 1819 if ((runState & STOP) == 0L) { 1820 WorkQueue[] qs; int n, i; 1821 if ((lockRunState() & STOP) == 0L && 1822 (qs = queues) != null && (n = qs.length) > 0 && 1823 qs[i = phase & SMASK & (n - 1)] == w) { 1824 qs[i] = null; 1825 stealCount += ns; // accumulate steals 1826 } 1827 unlockRunState(); 1828 } 1829 } 1830 if ((tryTerminate(false, false) & STOP) == 0L && 1831 phase != 0 && w != null && w.source != DROPPED) { 1832 w.cancelTasks(); // clean queue 1833 signalWork(null, 0); // possibly replace 1834 } 1835 if (ex != null) 1836 ForkJoinTask.rethrow(ex); 1837 } 1838 1839 /** 1840 * Releases an idle worker, or creates one if not enough exist, 1841 * giving up if array a is nonnull and task at a[k] already taken. 1842 */ 1843 final void signalWork(ForkJoinTask<?>[] a, int k) { 1844 int pc = parallelism; 1845 for (long c = ctl;;) { 1846 WorkQueue[] qs = queues; 1847 long ac = (c + RC_UNIT) & RC_MASK, nc; 1848 int sp = (int)c, i = sp & SMASK; 1849 if ((short)(c >>> RC_SHIFT) >= pc) 1850 break; 1851 if (qs == null) 1852 break; 1853 if (qs.length <= i) 1854 break; 1855 WorkQueue w = qs[i], v = null; 1856 if (sp == 0) { 1857 if ((short)(c >>> TC_SHIFT) >= pc) 1858 break; 1859 nc = ((c + TC_UNIT) & TC_MASK) | ac; 1860 } 1861 else if ((v = w) == null) 1862 break; 1863 else 1864 nc = (v.stackPred & LMASK) | (c & TC_MASK) | ac; 1865 if (a != null && k < a.length && k >= 0 && a[k] == null) 1866 break; 1867 if (c == (c = ctl) && c == (c = compareAndExchangeCtl(c, nc))) { 1868 if (v == null) 1869 createWorker(); 1870 else { 1871 v.phase = sp; 1872 if (v.parking != 0) 1873 U.unpark(v.owner); 1874 } 1875 break; 1876 } 1877 } 1878 } 1879 1880 /** 1881 * Releases all waiting workers. Called only during shutdown. 1882 */ 1883 private void releaseWaiters() { 1884 for (long c = ctl;;) { 1885 WorkQueue[] qs; WorkQueue v; int sp, i; 1886 if ((sp = (int)c) == 0 || (qs = queues) == null || 1887 qs.length <= (i = sp & SMASK) || (v = qs[i]) == null) 1888 break; 1889 if (c == (c = compareAndExchangeCtl( 1890 c, ((UMASK & (c + RC_UNIT)) | (c & TC_MASK) | 1891 (v.stackPred & LMASK))))) { 1892 v.phase = sp; 1893 if (v.parking != 0) 1894 U.unpark(v.owner); 1895 } 1896 } 1897 } 1898 1899 /** 1900 * Internal version of isQuiescent and related functionality. 1901 * @return positive if stopping, nonnegative if terminating or all 1902 * workers are inactive and submission queues are empty and 1903 * unlocked; if so, setting STOP if shutdown is enabled 1904 */ 1905 private int quiescent() { 1906 for (;;) { 1907 long phaseSum = 0L; 1908 boolean swept = false; 1909 for (long e, prevRunState = 0L; ; prevRunState = e) { 1910 DelayScheduler ds; 1911 long c = ctl; 1912 if (((e = runState) & STOP) != 0L) 1913 return 1; // terminating 1914 else if ((c & RC_MASK) > 0L) 1915 return -1; // at least one active 1916 else if (!swept || e != prevRunState || (e & RS_LOCK) != 0) { 1917 long sum = c; 1918 WorkQueue[] qs = queues; 1919 int n = (qs == null) ? 0 : qs.length; 1920 for (int i = 0; i < n; ++i) { // scan queues 1921 WorkQueue q; 1922 if ((q = qs[i]) != null) { 1923 int p = q.phase, s = q.top, b = q.base; 1924 sum += (p & 0xffffffffL) | ((long)b << 32); 1925 if ((p & IDLE) == 0 || s - b > 0) 1926 return -1; 1927 } 1928 } 1929 swept = (phaseSum == (phaseSum = sum)); 1930 } 1931 else if ((e & SHUTDOWN) == 0) 1932 return 0; 1933 else if ((ds = delayScheduler) != null && !ds.canShutDown()) 1934 return 0; 1935 else if (compareAndSetCtl(c, c) && casRunState(e, e | STOP)) 1936 return 1; // enable termination 1937 else 1938 break; // restart 1939 } 1940 } 1941 } 1942 1943 /** 1944 * Top-level runloop for workers, called by ForkJoinWorkerThread.run. 1945 * See above for explanation. 1946 * 1947 * @param w caller's WorkQueue (may be null on failed initialization) 1948 */ 1949 final void runWorker(WorkQueue w) { 1950 if (w != null) { 1951 int phase = w.phase; 1952 int r = w.stackPred, origin = r; // seed from registerWorker 1953 int cfg = w.config, fifo = cfg & FIFO, clearLocals = cfg & CLEAR_TLS; 1954 int src = -1; // current source queue 1955 int taken = 0, ptaken = 0, staken = 0; // takes per phase and scan 1956 rescan: while ((runState & STOP) == 0L) { 1957 WorkQueue[] qs = queues; 1958 int n = (qs == null) ? 0 : qs.length; 1959 int i = origin, step = (r >>> 16) | 1; 1960 r ^= r << 13; r ^= r >>> 17; origin = r ^= r << 5; // xorshift 1961 for (int l = n; l > 0; --l, i += step) { // scan queues 1962 WorkQueue q; int j; 1963 if ((q = qs[j = i & (n - 1)]) != null) { 1964 for (;;) { // poll q 1965 ForkJoinTask<?>[] a; int cap, b, m, k; 1966 if ((a = q.array) == null || (cap = a.length) <= 0) 1967 break; 1968 long bp = slotOffset(k = (b = q.base) & (m = cap - 1)); 1969 int nb = b + 1, nk = nb & m; 1970 ForkJoinTask<?> t = (ForkJoinTask<?>) 1971 U.getReferenceAcquire(a, bp); 1972 if (q.array != a || q.base != b || a[k] != t) 1973 continue; // inconsistent 1974 if (t == null) { 1975 if (taken != staken) { 1976 staken = taken; 1977 continue rescan; // sweep until clean 1978 } 1979 if (a[nk] != null || a[(b + 2) & m] != null) 1980 continue rescan; // stalled; reorder scan 1981 break; // probably empty 1982 } 1983 if (U.compareAndSetReference(a, bp, t, null)) { 1984 q.base = nb; 1985 Object nt = U.getReferenceAcquire 1986 (a, slotOffset(nk)); // confirm below 1987 ++taken; 1988 if (src != j) 1989 w.source = src = j; 1990 if (nt != null && nt == a[nk]) 1991 signalWork(a, nk); // propagate 1992 w.topLevelExec(t, fifo); // run t & its subtasks 1993 } 1994 } 1995 } 1996 } 1997 if (taken != ptaken) { // end run 1998 ptaken = taken; 1999 origin = src; // hint for next run 2000 if (clearLocals != 0 && 2001 Thread.currentThread() instanceof ForkJoinWorkerThread wt) 2002 wt.resetThreadLocals(); 2003 w.nsteals = taken; 2004 } 2005 w.phase = phase += IDLE; // deactivate 2006 if ((phase = awaitWork(w, phase)) == IDLE) 2007 break; 2008 } 2009 } 2010 } 2011 2012 /** 2013 * Awaits signal or termination. 2014 * 2015 * @param w the work queue 2016 * @param p current phase (known to be idle 2017 * @return current phase or IDLE if worker should exit 2018 */ 2019 private int awaitWork(WorkQueue w, int p) { 2020 if (w == null) // never true; hoist checks 2021 return IDLE; 2022 int activePhase = p + IDLE; 2023 long ap = activePhase & LMASK, pc = ctl, qc; 2024 do { // enqueue 2025 qc = ap | ((pc - RC_UNIT) & UMASK); 2026 w.stackPred = (int)pc; // set ctl stack link 2027 } while (pc != (pc = compareAndExchangeCtl(pc, qc))); 2028 long psp = pc & LMASK; // reactivation stack prefix 2029 WorkQueue[] qs; int n; // missed signal check 2030 if ((runState & STOP) != 0 || (qs = queues) == null || (n = qs.length) <= 0) 2031 return IDLE; // already terminating 2032 for (int m = n - 1, origin = p + 1, i = 0; i < m; ++i) { 2033 WorkQueue q; long cc; // stagger origins 2034 if ((q = qs[(origin + i) & m]) != null && q.top - q.base > 0) { 2035 if ((p = w.phase) == activePhase) 2036 break; 2037 if ((int)(cc = ctl) == activePhase && 2038 compareAndSetCtl(cc, psp | ((cc + RC_UNIT) & UMASK))) { 2039 p = w.phase = activePhase; 2040 break; // reactivated 2041 } 2042 } 2043 } 2044 if (p != activePhase && (p = w.phase) != activePhase) { 2045 long deadline = 0L, c, e; // quiescence checks 2046 if (((e = runState) & STOP) != 0) 2047 return IDLE; 2048 else if ((int)(c = ctl) != activePhase || (c & RC_MASK) > 0L) { 2049 for (int spins = n; (p = w.phase) != activePhase && --spins > 0;) 2050 Thread.onSpinWait(); // spin unless possibly quiescent 2051 } 2052 else if ((e & SHUTDOWN) != 0L && quiescent() > 0) 2053 return IDLE; // quiescent termination 2054 else { // use trim timeout 2055 long d = ((w.source != INVALID_ID) ? keepAlive : 2056 TIMEOUT_SLOP) + System.currentTimeMillis(); 2057 deadline = (d == 0L)? 1L : d; // avoid zero 2058 p = w.phase; 2059 } 2060 if (p != activePhase) { // block 2061 LockSupport.setCurrentBlocker(this); 2062 w.parking = 1; // enable unpark 2063 while ((p = w.phase) != activePhase) { 2064 boolean trimmable = false; int trim; 2065 Thread.interrupted(); // clear status 2066 if ((runState & STOP) != 0L) 2067 break; 2068 if (deadline != 0L) { 2069 if ((trim = tryTrim(w, p, deadline)) > 0) 2070 break; 2071 else if (trim < 0) 2072 deadline = 0L; 2073 else 2074 trimmable = true; 2075 } 2076 U.park(trimmable, deadline); 2077 } 2078 w.parking = 0; 2079 LockSupport.setCurrentBlocker(null); 2080 if (p != activePhase) 2081 return IDLE; 2082 } 2083 } 2084 return activePhase; 2085 } 2086 2087 /** 2088 * Tries to remove and deregister worker after timeout, and release 2089 * another to do the same. 2090 * @return > 0: trimmed, < 0 : not trimmable, else 0 2091 */ 2092 private int tryTrim(WorkQueue w, int phase, long deadline) { 2093 long c, nc; int stat, activePhase, vp, i; WorkQueue[] vs; WorkQueue v; 2094 if ((activePhase = phase + IDLE) != (int)(c = ctl) || w == null) 2095 stat = -1; // no longer ctl top 2096 else if (deadline - System.currentTimeMillis() >= TIMEOUT_SLOP) 2097 stat = 0; // spurious wakeup 2098 else if (!compareAndSetCtl( 2099 c, nc = ((w.stackPred & LMASK) | (RC_MASK & c) | 2100 (TC_MASK & (c - TC_UNIT))))) 2101 stat = -1; // lost race to signaller 2102 else { 2103 stat = 1; 2104 w.source = DROPPED; 2105 w.phase = activePhase; 2106 if ((vp = (int)nc) != 0 && (vs = queues) != null && 2107 vs.length > (i = vp & SMASK) && (v = vs[i]) != null && 2108 compareAndSetCtl( // try to wake up next waiter 2109 nc, ((UMASK & (nc + RC_UNIT)) | 2110 (nc & TC_MASK) | (v.stackPred & LMASK)))) { 2111 v.source = INVALID_ID; // enable cascaded timeouts 2112 v.phase = vp; 2113 U.unpark(v.owner); 2114 } 2115 } 2116 return stat; 2117 } 2118 2119 /** 2120 * Scans for and returns a polled task, if available. Used only 2121 * for untracked polls. Begins scan at a random index to avoid 2122 * systematic unfairness. 2123 * 2124 * @param submissionsOnly if true, only scan submission queues 2125 */ 2126 private ForkJoinTask<?> pollScan(boolean submissionsOnly) { 2127 if ((runState & STOP) == 0L) { 2128 WorkQueue[] qs; int n; WorkQueue q; ForkJoinTask<?> t; 2129 int r = ThreadLocalRandom.nextSecondarySeed(); 2130 if (submissionsOnly) // even indices only 2131 r &= ~1; 2132 int step = (submissionsOnly) ? 2 : 1; 2133 if ((qs = queues) != null && (n = qs.length) > 0) { 2134 for (int i = n; i > 0; i -= step, r += step) { 2135 if ((q = qs[r & (n - 1)]) != null && 2136 (t = q.poll()) != null) 2137 return t; 2138 } 2139 } 2140 } 2141 return null; 2142 } 2143 2144 /** 2145 * Tries to decrement counts (sometimes implicitly) and possibly 2146 * arrange for a compensating worker in preparation for 2147 * blocking. May fail due to interference, in which case -1 is 2148 * returned so caller may retry. A zero return value indicates 2149 * that the caller doesn't need to re-adjust counts when later 2150 * unblocked. 2151 * 2152 * @param c incoming ctl value 2153 * @return UNCOMPENSATE: block then adjust, 0: block, -1 : retry 2154 */ 2155 private int tryCompensate(long c) { 2156 Predicate<? super ForkJoinPool> sat; 2157 long b = config; 2158 int pc = parallelism, // unpack fields 2159 minActive = (short)(b >>> RC_SHIFT), 2160 maxTotal = (short)(b >>> TC_SHIFT) + pc, 2161 active = (short)(c >>> RC_SHIFT), 2162 total = (short)(c >>> TC_SHIFT), 2163 sp = (int)c, 2164 stat = -1; // default retry return 2165 if (sp != 0 && active <= pc) { // activate idle worker 2166 WorkQueue[] qs; WorkQueue v; int i; 2167 if ((qs = queues) != null && qs.length > (i = sp & SMASK) && 2168 (v = qs[i]) != null && 2169 compareAndSetCtl(c, (c & UMASK) | (v.stackPred & LMASK))) { 2170 v.phase = sp; 2171 if (v.parking != 0) 2172 U.unpark(v.owner); 2173 stat = UNCOMPENSATE; 2174 } 2175 } 2176 else if (active > minActive && total >= pc) { // reduce active workers 2177 if (compareAndSetCtl(c, ((c - RC_UNIT) & RC_MASK) | (c & ~RC_MASK))) 2178 stat = UNCOMPENSATE; 2179 } 2180 else if (total < maxTotal && total < MAX_CAP) { // try to expand pool 2181 long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK); 2182 if ((runState & STOP) != 0L) // terminating 2183 stat = 0; 2184 else if (compareAndSetCtl(c, nc)) 2185 stat = createWorker() ? UNCOMPENSATE : 0; 2186 } 2187 else if (!compareAndSetCtl(c, c)) // validate 2188 ; 2189 else if ((sat = saturate) != null && sat.test(this)) 2190 stat = 0; 2191 else 2192 throw new RejectedExecutionException( 2193 "Thread limit exceeded replacing blocked worker"); 2194 return stat; 2195 } 2196 2197 /** 2198 * Readjusts RC count; called from ForkJoinTask after blocking. 2199 */ 2200 final void uncompensate() { 2201 getAndAddCtl(RC_UNIT); 2202 } 2203 2204 /** 2205 * Helps if possible until the given task is done. Processes 2206 * compatible local tasks and scans other queues for task produced 2207 * by w's stealers; returning compensated blocking sentinel if 2208 * none are found. 2209 * 2210 * @param task the task 2211 * @param w caller's WorkQueue 2212 * @param internal true if w is owned by a ForkJoinWorkerThread 2213 * @return task status on exit, or UNCOMPENSATE for compensated blocking 2214 */ 2215 final int helpJoin(ForkJoinTask<?> task, WorkQueue w, boolean internal) { 2216 if (w != null) 2217 w.tryRemoveAndExec(task, internal); 2218 int s = 0; 2219 if (task != null && (s = task.status) >= 0 && internal && w != null) { 2220 int wid = w.phase & SMASK, r = wid + 2, wsrc = w.source; 2221 long sctl = 0L; // track stability 2222 outer: for (boolean rescan = true;;) { 2223 if ((s = task.status) < 0) 2224 break; 2225 if (!rescan) { 2226 if ((runState & STOP) != 0L) 2227 break; 2228 if (sctl == (sctl = ctl) && (s = tryCompensate(sctl)) >= 0) 2229 break; 2230 } 2231 rescan = false; 2232 WorkQueue[] qs = queues; 2233 int n = (qs == null) ? 0 : qs.length; 2234 scan: for (int l = n >>> 1; l > 0; --l, r += 2) { 2235 int j; WorkQueue q; 2236 if ((q = qs[j = r & SMASK & (n - 1)]) != null) { 2237 for (;;) { 2238 ForkJoinTask<?> t; ForkJoinTask<?>[] a; 2239 boolean eligible = false; 2240 int sq = q.source, b, cap; long k; 2241 if ((a = q.array) == null || (cap = a.length) <= 0) 2242 break; 2243 t = (ForkJoinTask<?>)U.getReferenceAcquire( 2244 a, k = slotOffset((cap - 1) & (b = q.base))); 2245 if (t == task) 2246 eligible = true; 2247 else if (t != null) { // check steal chain 2248 for (int v = sq, d = cap;;) { 2249 WorkQueue p; 2250 if (v == wid) { 2251 eligible = true; 2252 break; 2253 } 2254 if ((v & 1) == 0 || // external or none 2255 --d < 0 || // bound depth 2256 (p = qs[v & (n - 1)]) == null) 2257 break; 2258 v = p.source; 2259 } 2260 } 2261 if ((s = task.status) < 0) 2262 break outer; // validate 2263 if (q.source == sq && q.base == b && 2264 U.getReference(a, k) == t) { 2265 if (!eligible) { // revisit if nonempty 2266 if (!rescan && t == null && q.top - b > 0) 2267 rescan = true; 2268 break; 2269 } 2270 if (U.compareAndSetReference(a, k, t, null)) { 2271 q.base = b + 1; 2272 w.source = j; // volatile write 2273 t.doExec(); 2274 w.source = wsrc; 2275 rescan = true; // restart at index r 2276 break scan; 2277 } 2278 } 2279 } 2280 } 2281 } 2282 } 2283 } 2284 return s; 2285 } 2286 2287 /** 2288 * Version of helpJoin for CountedCompleters. 2289 * 2290 * @param task root of computation (only called when a CountedCompleter) 2291 * @param w caller's WorkQueue 2292 * @param internal true if w is owned by a ForkJoinWorkerThread 2293 * @return task status on exit, or UNCOMPENSATE for compensated blocking 2294 */ 2295 final int helpComplete(ForkJoinTask<?> task, WorkQueue w, boolean internal) { 2296 int s = 0; 2297 if (task != null && (s = task.status) >= 0 && w != null) { 2298 int r = w.phase + 1; // for indexing 2299 long sctl = 0L; // track stability 2300 outer: for (boolean rescan = true, locals = true;;) { 2301 if (locals && (s = w.helpComplete(task, internal, 0)) < 0) 2302 break; 2303 if ((s = task.status) < 0) 2304 break; 2305 if (!rescan) { 2306 if ((runState & STOP) != 0L) 2307 break; 2308 if (sctl == (sctl = ctl) && 2309 (!internal || (s = tryCompensate(sctl)) >= 0)) 2310 break; 2311 } 2312 rescan = locals = false; 2313 WorkQueue[] qs = queues; 2314 int n = (qs == null) ? 0 : qs.length; 2315 scan: for (int l = n; l > 0; --l, ++r) { 2316 int j; WorkQueue q; 2317 if ((q = qs[j = r & SMASK & (n - 1)]) != null) { 2318 for (;;) { 2319 ForkJoinTask<?> t; ForkJoinTask<?>[] a; 2320 int b, cap, nb; long k; 2321 boolean eligible = false; 2322 if ((a = q.array) == null || (cap = a.length) <= 0) 2323 break; 2324 t = (ForkJoinTask<?>)U.getReferenceAcquire( 2325 a, k = slotOffset((cap - 1) & (b = q.base))); 2326 if (t instanceof CountedCompleter) { 2327 CountedCompleter<?> f = (CountedCompleter<?>)t; 2328 for (int steps = cap; steps > 0; --steps) { 2329 if (f == task) { 2330 eligible = true; 2331 break; 2332 } 2333 if ((f = f.completer) == null) 2334 break; 2335 } 2336 } 2337 if ((s = task.status) < 0) // validate 2338 break outer; 2339 if (q.base == b) { 2340 if (eligible) { 2341 if (U.compareAndSetReference( 2342 a, k, t, null)) { 2343 q.updateBase(b + 1); 2344 t.doExec(); 2345 locals = rescan = true; 2346 break scan; 2347 } 2348 } 2349 else if (U.getReference(a, k) == t) { 2350 if (!rescan && t == null && q.top - b > 0) 2351 rescan = true; // revisit 2352 break; 2353 } 2354 } 2355 } 2356 } 2357 } 2358 } 2359 } 2360 return s; 2361 } 2362 2363 /** 2364 * Runs tasks until all workers are inactive and no tasks are 2365 * found. Rather than blocking when tasks cannot be found, rescans 2366 * until all others cannot find tasks either. 2367 * 2368 * @param nanos max wait time (Long.MAX_VALUE if effectively untimed) 2369 * @param interruptible true if return on interrupt 2370 * @return positive if quiescent, negative if interrupted, else 0 2371 */ 2372 private int helpQuiesce(WorkQueue w, long nanos, boolean interruptible) { 2373 int phase; // w.phase inactive bit set when temporarily quiescent 2374 if (w == null || ((phase = w.phase) & IDLE) != 0) 2375 return 0; 2376 int wsrc = w.source; 2377 long startTime = System.nanoTime(); 2378 long maxSleep = Math.min(nanos >>> 8, MAX_SLEEP); // approx 1% nanos 2379 long prevSum = 0L; 2380 int activePhase = phase, inactivePhase = phase + IDLE; 2381 int r = phase + 1, waits = 0, returnStatus = 1; 2382 boolean locals = true; 2383 for (long e = runState;;) { 2384 if ((e & STOP) != 0L) 2385 break; // terminating 2386 if (interruptible && Thread.interrupted()) { 2387 returnStatus = -1; 2388 break; 2389 } 2390 if (locals) { // run local tasks before (re)polling 2391 locals = false; 2392 for (ForkJoinTask<?> u; (u = w.nextLocalTask()) != null;) 2393 u.doExec(); 2394 } 2395 WorkQueue[] qs = queues; 2396 int n = (qs == null) ? 0 : qs.length; 2397 long phaseSum = 0L; 2398 boolean rescan = false, busy = false; 2399 scan: for (int l = n; l > 0; --l, ++r) { 2400 int j; WorkQueue q; 2401 if ((q = qs[j = r & SMASK & (n - 1)]) != null && q != w) { 2402 for (;;) { 2403 ForkJoinTask<?> t; ForkJoinTask<?>[] a; 2404 int b, cap; long k; 2405 if ((a = q.array) == null || (cap = a.length) <= 0) 2406 break; 2407 t = (ForkJoinTask<?>)U.getReferenceAcquire( 2408 a, k = slotOffset((cap - 1) & (b = q.base))); 2409 if (t != null && phase == inactivePhase) // reactivate 2410 w.phase = phase = activePhase; 2411 if (q.base == b && U.getReference(a, k) == t) { 2412 int nb = b + 1; 2413 if (t == null) { 2414 if (!rescan) { 2415 int qp = q.phase, mq = qp & (IDLE | 1); 2416 phaseSum += qp; 2417 if (mq == 0 || q.top - b > 0) 2418 rescan = true; 2419 else if (mq == 1) 2420 busy = true; 2421 } 2422 break; 2423 } 2424 if (U.compareAndSetReference(a, k, t, null)) { 2425 q.base = nb; 2426 w.source = j; // volatile write 2427 t.doExec(); 2428 w.source = wsrc; 2429 rescan = locals = true; 2430 break scan; 2431 } 2432 } 2433 } 2434 } 2435 } 2436 if (e != (e = runState) || prevSum != (prevSum = phaseSum) || 2437 rescan || (e & RS_LOCK) != 0L) 2438 ; // inconsistent 2439 else if (!busy) 2440 break; 2441 else if (phase == activePhase) { 2442 waits = 0; // recheck, then sleep 2443 w.phase = phase = inactivePhase; 2444 } 2445 else if (System.nanoTime() - startTime > nanos) { 2446 returnStatus = 0; // timed out 2447 break; 2448 } 2449 else if (waits == 0) // same as spinLockRunState except 2450 waits = MIN_SLEEP; // with rescan instead of onSpinWait 2451 else { 2452 LockSupport.parkNanos(this, (long)waits); 2453 if (waits < maxSleep) 2454 waits <<= 1; 2455 } 2456 } 2457 w.phase = activePhase; 2458 return returnStatus; 2459 } 2460 2461 /** 2462 * Helps quiesce from external caller until done, interrupted, or timeout 2463 * 2464 * @param nanos max wait time (Long.MAX_VALUE if effectively untimed) 2465 * @param interruptible true if return on interrupt 2466 * @return positive if quiescent, negative if interrupted, else 0 2467 */ 2468 private int externalHelpQuiesce(long nanos, boolean interruptible) { 2469 if (quiescent() < 0) { 2470 long startTime = System.nanoTime(); 2471 long maxSleep = Math.min(nanos >>> 8, MAX_SLEEP); 2472 for (int waits = 0;;) { 2473 ForkJoinTask<?> t; 2474 if (interruptible && Thread.interrupted()) 2475 return -1; 2476 else if ((t = pollScan(false)) != null) { 2477 waits = 0; 2478 t.doExec(); 2479 } 2480 else if (quiescent() >= 0) 2481 break; 2482 else if (System.nanoTime() - startTime > nanos) 2483 return 0; 2484 else if (waits == 0) 2485 waits = MIN_SLEEP; 2486 else { 2487 LockSupport.parkNanos(this, (long)waits); 2488 if (waits < maxSleep) 2489 waits <<= 1; 2490 } 2491 } 2492 } 2493 return 1; 2494 } 2495 2496 /** 2497 * Helps quiesce from either internal or external caller 2498 * 2499 * @param pool the pool to use, or null if any 2500 * @param nanos max wait time (Long.MAX_VALUE if effectively untimed) 2501 * @param interruptible true if return on interrupt 2502 * @return positive if quiescent, negative if interrupted, else 0 2503 */ 2504 static final int helpQuiescePool(ForkJoinPool pool, long nanos, 2505 boolean interruptible) { 2506 Thread t; ForkJoinPool p; ForkJoinWorkerThread wt; 2507 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread && 2508 (p = (wt = (ForkJoinWorkerThread)t).pool) != null && 2509 (p == pool || pool == null)) 2510 return p.helpQuiesce(wt.workQueue, nanos, interruptible); 2511 else if ((p = pool) != null || (p = common) != null) 2512 return p.externalHelpQuiesce(nanos, interruptible); 2513 else 2514 return 0; 2515 } 2516 2517 /** 2518 * Gets and removes a local or stolen task for the given worker. 2519 * 2520 * @return a task, if available 2521 */ 2522 final ForkJoinTask<?> nextTaskFor(WorkQueue w) { 2523 ForkJoinTask<?> t; 2524 if (w == null || (t = w.nextLocalTask()) == null) 2525 t = pollScan(false); 2526 return t; 2527 } 2528 2529 // External operations 2530 2531 /** 2532 * Finds and locks a WorkQueue for an external submitter, or 2533 * throws RejectedExecutionException if shutdown 2534 * @param rejectOnShutdown true if RejectedExecutionException 2535 * should be thrown when shutdown 2536 */ 2537 final WorkQueue externalSubmissionQueue(boolean rejectOnShutdown) { 2538 int r; 2539 if ((r = ThreadLocalRandom.getProbe()) == 0) { 2540 ThreadLocalRandom.localInit(); // initialize caller's probe 2541 r = ThreadLocalRandom.getProbe(); 2542 } 2543 for (;;) { 2544 WorkQueue q; WorkQueue[] qs; int n, id, i; 2545 if ((qs = queues) == null || (n = qs.length) <= 0) 2546 break; 2547 if ((q = qs[i = (id = r & EXTERNAL_ID_MASK) & (n - 1)]) == null) { 2548 WorkQueue newq = new WorkQueue(null, id, 0, false); 2549 lockRunState(); 2550 if (qs[i] == null && queues == qs) 2551 q = qs[i] = newq; // else lost race to install 2552 unlockRunState(); 2553 } 2554 if (q != null && q.tryLockPhase()) { 2555 if (rejectOnShutdown && (runState & SHUTDOWN) != 0L) { 2556 q.unlockPhase(); // check while q lock held 2557 break; 2558 } 2559 return q; 2560 } 2561 r = ThreadLocalRandom.advanceProbe(r); // move 2562 } 2563 throw new RejectedExecutionException(); 2564 } 2565 2566 private <T> ForkJoinTask<T> poolSubmit(boolean signalIfEmpty, ForkJoinTask<T> task) { 2567 Thread t; ForkJoinWorkerThread wt; WorkQueue q; boolean internal; 2568 if (((t = JLA.currentCarrierThread()) instanceof ForkJoinWorkerThread) && 2569 (wt = (ForkJoinWorkerThread)t).pool == this) { 2570 internal = true; 2571 q = wt.workQueue; 2572 } 2573 else { // find and lock queue 2574 internal = false; 2575 q = externalSubmissionQueue(true); 2576 } 2577 q.push(task, signalIfEmpty ? this : null, internal); 2578 return task; 2579 } 2580 2581 /** 2582 * Returns queue for an external thread, if one exists that has 2583 * possibly ever submitted to the given pool (nonzero probe), or 2584 * null if none. 2585 */ 2586 static WorkQueue externalQueue(ForkJoinPool p) { 2587 WorkQueue[] qs; int n; 2588 int r = ThreadLocalRandom.getProbe(); 2589 return (p != null && (qs = p.queues) != null && 2590 (n = qs.length) > 0 && r != 0) ? 2591 qs[r & EXTERNAL_ID_MASK & (n - 1)] : null; 2592 } 2593 2594 /** 2595 * Returns external queue for common pool. 2596 */ 2597 static WorkQueue commonQueue() { 2598 return externalQueue(common); 2599 } 2600 2601 /** 2602 * If the given executor is a ForkJoinPool, poll and execute 2603 * AsynchronousCompletionTasks from worker's queue until none are 2604 * available or blocker is released. 2605 */ 2606 static void helpAsyncBlocker(Executor e, ManagedBlocker blocker) { 2607 WorkQueue w = null; Thread t; ForkJoinWorkerThread wt; 2608 if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) && 2609 (wt = (ForkJoinWorkerThread)t).pool == e) 2610 w = wt.workQueue; 2611 else if (e instanceof ForkJoinPool) 2612 w = externalQueue((ForkJoinPool)e); 2613 if (w != null) 2614 w.helpAsyncBlocker(blocker); 2615 } 2616 2617 /** 2618 * Returns a cheap heuristic guide for task partitioning when 2619 * programmers, frameworks, tools, or languages have little or no 2620 * idea about task granularity. In essence, by offering this 2621 * method, we ask users only about tradeoffs in overhead vs 2622 * expected throughput and its variance, rather than how finely to 2623 * partition tasks. 2624 * 2625 * In a steady state strict (tree-structured) computation, each 2626 * thread makes available for stealing enough tasks for other 2627 * threads to remain active. Inductively, if all threads play by 2628 * the same rules, each thread should make available only a 2629 * constant number of tasks. 2630 * 2631 * The minimum useful constant is just 1. But using a value of 1 2632 * would require immediate replenishment upon each steal to 2633 * maintain enough tasks, which is infeasible. Further, 2634 * partitionings/granularities of offered tasks should minimize 2635 * steal rates, which in general means that threads nearer the top 2636 * of computation tree should generate more than those nearer the 2637 * bottom. In perfect steady state, each thread is at 2638 * approximately the same level of computation tree. However, 2639 * producing extra tasks amortizes the uncertainty of progress and 2640 * diffusion assumptions. 2641 * 2642 * So, users will want to use values larger (but not much larger) 2643 * than 1 to both smooth over transient shortages and hedge 2644 * against uneven progress; as traded off against the cost of 2645 * extra task overhead. We leave the user to pick a threshold 2646 * value to compare with the results of this call to guide 2647 * decisions, but recommend values such as 3. 2648 * 2649 * When all threads are active, it is on average OK to estimate 2650 * surplus strictly locally. In steady-state, if one thread is 2651 * maintaining say 2 surplus tasks, then so are others. So we can 2652 * just use estimated queue length. However, this strategy alone 2653 * leads to serious mis-estimates in some non-steady-state 2654 * conditions (ramp-up, ramp-down, other stalls). We can detect 2655 * many of these by further considering the number of "idle" 2656 * threads, that are known to have zero queued tasks, so 2657 * compensate by a factor of (#idle/#active) threads. 2658 */ 2659 static int getSurplusQueuedTaskCount() { 2660 Thread t; ForkJoinWorkerThread wt; ForkJoinPool pool; WorkQueue q; 2661 if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) && 2662 (pool = (wt = (ForkJoinWorkerThread)t).pool) != null && 2663 (q = wt.workQueue) != null) { 2664 int n = q.top - q.base; 2665 int p = pool.parallelism; 2666 int a = (short)(pool.ctl >>> RC_SHIFT); 2667 return n - (a > (p >>>= 1) ? 0 : 2668 a > (p >>>= 1) ? 1 : 2669 a > (p >>>= 1) ? 2 : 2670 a > (p >>>= 1) ? 4 : 2671 8); 2672 } 2673 return 0; 2674 } 2675 2676 // Termination 2677 2678 /** 2679 * Possibly initiates and/or completes pool termination. 2680 * 2681 * @param now if true, unconditionally terminate, else only 2682 * if no work and no active workers 2683 * @param enable if true, terminate when next possible 2684 * @return runState on exit 2685 */ 2686 private long tryTerminate(boolean now, boolean enable) { 2687 long e, isShutdown, ps; 2688 if (((e = runState) & TERMINATED) != 0L) 2689 now = false; 2690 else if ((e & STOP) != 0L) 2691 now = true; 2692 else if (now) { 2693 if (((ps = getAndBitwiseOrRunState(SHUTDOWN|STOP) & STOP)) == 0L) { 2694 if ((ps & RS_LOCK) != 0L) { 2695 spinLockRunState(); // ensure queues array stable after stop 2696 unlockRunState(); 2697 } 2698 interruptAll(); 2699 } 2700 } 2701 else if ((isShutdown = (e & SHUTDOWN)) != 0L || enable) { 2702 long quiet; DelayScheduler ds; 2703 if (isShutdown == 0L) 2704 getAndBitwiseOrRunState(SHUTDOWN); 2705 if ((quiet = quiescent()) > 0) 2706 now = true; 2707 else if (quiet == 0 && (ds = delayScheduler) != null) 2708 ds.signal(); 2709 } 2710 2711 if (now) { 2712 DelayScheduler ds; 2713 releaseWaiters(); 2714 if ((ds = delayScheduler) != null) 2715 ds.signal(); 2716 for (;;) { 2717 if (((e = runState) & CLEANED) == 0L) { 2718 boolean clean = cleanQueues(); 2719 if (((e = runState) & CLEANED) == 0L && clean) 2720 e = getAndBitwiseOrRunState(CLEANED) | CLEANED; 2721 } 2722 if ((e & TERMINATED) != 0L) 2723 break; 2724 if (ctl != 0L) // else loop if didn't finish cleaning 2725 break; 2726 if ((ds = delayScheduler) != null && ds.signal() >= 0) 2727 break; 2728 if ((e & CLEANED) != 0L) { 2729 e |= TERMINATED; 2730 if ((getAndBitwiseOrRunState(TERMINATED) & TERMINATED) == 0L) { 2731 CountDownLatch done; SharedThreadContainer ctr; 2732 if ((done = termination) != null) 2733 done.countDown(); 2734 if ((ctr = container) != null) 2735 ctr.close(); 2736 } 2737 break; 2738 } 2739 } 2740 } 2741 return e; 2742 } 2743 2744 /** 2745 * Scans queues in a psuedorandom order based on thread id, 2746 * cancelling tasks until empty, or returning early upon 2747 * interference or still-active external queues, in which case 2748 * other calls will finish cancellation. 2749 * 2750 * @return true if all queues empty 2751 */ 2752 private boolean cleanQueues() { 2753 int r = (int)Thread.currentThread().threadId(); 2754 r ^= r << 13; r ^= r >>> 17; r ^= r << 5; // xorshift 2755 int step = (r >>> 16) | 1; // randomize traversals 2756 WorkQueue[] qs = queues; 2757 int n = (qs == null) ? 0 : qs.length; 2758 for (int l = n; l > 0; --l, r += step) { 2759 WorkQueue q; ForkJoinTask<?>[] a; int cap; 2760 if ((q = qs[r & (n - 1)]) != null && 2761 (a = q.array) != null && (cap = a.length) > 0) { 2762 for (;;) { 2763 ForkJoinTask<?> t; int b; long k; 2764 t = (ForkJoinTask<?>)U.getReferenceAcquire( 2765 a, k = slotOffset((cap - 1) & (b = q.base))); 2766 if (q.base == b && t != null && 2767 U.compareAndSetReference(a, k, t, null)) { 2768 q.updateBase(b + 1); 2769 try { 2770 t.cancel(false); 2771 } catch (Throwable ignore) { 2772 } 2773 } 2774 else if ((q.phase & (IDLE|1)) == 0 || // externally locked 2775 q.top - q.base > 0) 2776 return false; // incomplete 2777 else 2778 break; 2779 } 2780 } 2781 } 2782 return true; 2783 } 2784 2785 /** 2786 * Interrupts all workers 2787 */ 2788 private void interruptAll() { 2789 Thread current = Thread.currentThread(); 2790 WorkQueue[] qs = queues; 2791 int n = (qs == null) ? 0 : qs.length; 2792 for (int i = 1; i < n; i += 2) { 2793 WorkQueue q; Thread o; 2794 if ((q = qs[i]) != null && (o = q.owner) != null && o != current) { 2795 try { 2796 o.interrupt(); 2797 } catch (Throwable ignore) { 2798 } 2799 } 2800 } 2801 } 2802 2803 /** 2804 * Returns termination signal, constructing if necessary 2805 */ 2806 private CountDownLatch terminationSignal() { 2807 CountDownLatch signal, s, u; 2808 if ((signal = termination) == null) 2809 signal = ((u = cmpExTerminationSignal( 2810 s = new CountDownLatch(1))) == null) ? s : u; 2811 return signal; 2812 } 2813 2814 // Exported methods 2815 2816 // Constructors 2817 2818 /** 2819 * Creates a {@code ForkJoinPool} with parallelism equal to {@link 2820 * java.lang.Runtime#availableProcessors}, using defaults for all 2821 * other parameters (see {@link #ForkJoinPool(int, 2822 * ForkJoinWorkerThreadFactory, UncaughtExceptionHandler, boolean, 2823 * int, int, int, Predicate, long, TimeUnit)}). 2824 */ 2825 public ForkJoinPool() { 2826 this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()), 2827 defaultForkJoinWorkerThreadFactory, null, false, 2828 0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS); 2829 } 2830 2831 /** 2832 * Creates a {@code ForkJoinPool} with the indicated parallelism 2833 * level, using defaults for all other parameters (see {@link 2834 * #ForkJoinPool(int, ForkJoinWorkerThreadFactory, 2835 * UncaughtExceptionHandler, boolean, int, int, int, Predicate, 2836 * long, TimeUnit)}). 2837 * 2838 * @param parallelism the parallelism level 2839 * @throws IllegalArgumentException if parallelism less than or 2840 * equal to zero, or greater than implementation limit 2841 */ 2842 public ForkJoinPool(int parallelism) { 2843 this(parallelism, defaultForkJoinWorkerThreadFactory, null, false, 2844 0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS); 2845 } 2846 2847 /** 2848 * Creates a {@code ForkJoinPool} with the given parameters (using 2849 * defaults for others -- see {@link #ForkJoinPool(int, 2850 * ForkJoinWorkerThreadFactory, UncaughtExceptionHandler, boolean, 2851 * int, int, int, Predicate, long, TimeUnit)}). 2852 * 2853 * @param parallelism the parallelism level. For default value, 2854 * use {@link java.lang.Runtime#availableProcessors}. 2855 * @param factory the factory for creating new threads. For default value, 2856 * use {@link #defaultForkJoinWorkerThreadFactory}. 2857 * @param handler the handler for internal worker threads that 2858 * terminate due to unrecoverable errors encountered while executing 2859 * tasks. For default value, use {@code null}. 2860 * @param asyncMode if true, 2861 * establishes local first-in-first-out scheduling mode for forked 2862 * tasks that are never joined. This mode may be more appropriate 2863 * than default locally stack-based mode in applications in which 2864 * worker threads only process event-style asynchronous tasks. 2865 * For default value, use {@code false}. 2866 * @throws IllegalArgumentException if parallelism less than or 2867 * equal to zero, or greater than implementation limit 2868 * @throws NullPointerException if the factory is null 2869 */ 2870 public ForkJoinPool(int parallelism, 2871 ForkJoinWorkerThreadFactory factory, 2872 UncaughtExceptionHandler handler, 2873 boolean asyncMode) { 2874 this(parallelism, factory, handler, asyncMode, 2875 0, MAX_CAP, 1, null, DEFAULT_KEEPALIVE, TimeUnit.MILLISECONDS); 2876 } 2877 2878 /** 2879 * Creates a {@code ForkJoinPool} with the given parameters. 2880 * 2881 * @param parallelism the parallelism level. For default value, 2882 * use {@link java.lang.Runtime#availableProcessors}. 2883 * 2884 * @param factory the factory for creating new threads. For 2885 * default value, use {@link #defaultForkJoinWorkerThreadFactory}. 2886 * 2887 * @param handler the handler for internal worker threads that 2888 * terminate due to unrecoverable errors encountered while 2889 * executing tasks. For default value, use {@code null}. 2890 * 2891 * @param asyncMode if true, establishes local first-in-first-out 2892 * scheduling mode for forked tasks that are never joined. This 2893 * mode may be more appropriate than default locally stack-based 2894 * mode in applications in which worker threads only process 2895 * event-style asynchronous tasks. For default value, use {@code 2896 * false}. 2897 * 2898 * @param corePoolSize ignored: used in previous releases of this 2899 * class but no longer applicable. Using {@code 0} maintains 2900 * compatibility across releases. 2901 * 2902 * @param maximumPoolSize the maximum number of threads allowed. 2903 * When the maximum is reached, attempts to replace blocked 2904 * threads fail. (However, because creation and termination of 2905 * different threads may overlap, and may be managed by the given 2906 * thread factory, this value may be transiently exceeded.) To 2907 * arrange the same value as is used by default for the common 2908 * pool, use {@code 256} plus the {@code parallelism} level. (By 2909 * default, the common pool allows a maximum of 256 spare 2910 * threads.) Using a value (for example {@code 2911 * Integer.MAX_VALUE}) larger than the implementation's total 2912 * thread limit has the same effect as using this limit (which is 2913 * the default). 2914 * 2915 * @param minimumRunnable the minimum allowed number of core 2916 * threads not blocked by a join or {@link ManagedBlocker}. To 2917 * ensure progress, when too few unblocked threads exist and 2918 * unexecuted tasks may exist, new threads are constructed, up to 2919 * the given maximumPoolSize. For the default value, use {@code 2920 * 1}, that ensures liveness. A larger value might improve 2921 * throughput in the presence of blocked activities, but might 2922 * not, due to increased overhead. A value of zero may be 2923 * acceptable when submitted tasks cannot have dependencies 2924 * requiring additional threads. 2925 * 2926 * @param saturate if non-null, a predicate invoked upon attempts 2927 * to create more than the maximum total allowed threads. By 2928 * default, when a thread is about to block on a join or {@link 2929 * ManagedBlocker}, but cannot be replaced because the 2930 * maximumPoolSize would be exceeded, a {@link 2931 * RejectedExecutionException} is thrown. But if this predicate 2932 * returns {@code true}, then no exception is thrown, so the pool 2933 * continues to operate with fewer than the target number of 2934 * runnable threads, which might not ensure progress. 2935 * 2936 * @param keepAliveTime the elapsed time since last use before 2937 * a thread is terminated (and then later replaced if needed). 2938 * For the default value, use {@code 60, TimeUnit.SECONDS}. 2939 * 2940 * @param unit the time unit for the {@code keepAliveTime} argument 2941 * 2942 * @throws IllegalArgumentException if parallelism is less than or 2943 * equal to zero, or is greater than implementation limit, 2944 * or if maximumPoolSize is less than parallelism, 2945 * of if the keepAliveTime is less than or equal to zero. 2946 * @throws NullPointerException if the factory is null 2947 * @since 9 2948 */ 2949 public ForkJoinPool(int parallelism, 2950 ForkJoinWorkerThreadFactory factory, 2951 UncaughtExceptionHandler handler, 2952 boolean asyncMode, 2953 int corePoolSize, 2954 int maximumPoolSize, 2955 int minimumRunnable, 2956 Predicate<? super ForkJoinPool> saturate, 2957 long keepAliveTime, 2958 TimeUnit unit) { 2959 int p = parallelism; 2960 if (p <= 0 || p > MAX_CAP || p > maximumPoolSize || keepAliveTime <= 0L) 2961 throw new IllegalArgumentException(); 2962 if (factory == null || unit == null) 2963 throw new NullPointerException(); 2964 int size = Math.max(MIN_QUEUES_SIZE, 2965 1 << (33 - Integer.numberOfLeadingZeros(p - 1))); 2966 this.parallelism = p; 2967 this.factory = factory; 2968 this.ueh = handler; 2969 this.saturate = saturate; 2970 this.keepAlive = Math.max(unit.toMillis(keepAliveTime), TIMEOUT_SLOP); 2971 int maxSpares = Math.clamp(maximumPoolSize - p, 0, MAX_CAP); 2972 int minAvail = Math.clamp(minimumRunnable, 0, MAX_CAP); 2973 this.config = (((asyncMode ? FIFO : 0) & LMASK) | 2974 (((long)maxSpares) << TC_SHIFT) | 2975 (((long)minAvail) << RC_SHIFT)); 2976 this.queues = new WorkQueue[size]; 2977 String pid = Integer.toString(getAndAddPoolIds(1) + 1); 2978 String name = "ForkJoinPool-" + pid; 2979 this.poolName = name; 2980 this.workerNamePrefix = name + "-worker-"; 2981 this.container = SharedThreadContainer.create(name); 2982 } 2983 2984 /** 2985 * Constructor for common pool using parameters possibly 2986 * overridden by system properties 2987 */ 2988 private ForkJoinPool(byte forCommonPoolOnly) { 2989 String name = "ForkJoinPool.commonPool"; 2990 ForkJoinWorkerThreadFactory fac = defaultForkJoinWorkerThreadFactory; 2991 UncaughtExceptionHandler handler = null; 2992 int maxSpares = DEFAULT_COMMON_MAX_SPARES; 2993 int pc = 0, preset = 0; // nonzero if size set as property 2994 try { // ignore exceptions in accessing/parsing properties 2995 String pp = System.getProperty 2996 ("java.util.concurrent.ForkJoinPool.common.parallelism"); 2997 if (pp != null) { 2998 pc = Math.max(0, Integer.parseInt(pp)); 2999 preset = PRESET_SIZE; 3000 } 3001 String ms = System.getProperty 3002 ("java.util.concurrent.ForkJoinPool.common.maximumSpares"); 3003 if (ms != null) 3004 maxSpares = Math.clamp(Integer.parseInt(ms), 0, MAX_CAP); 3005 String sf = System.getProperty 3006 ("java.util.concurrent.ForkJoinPool.common.threadFactory"); 3007 String sh = System.getProperty 3008 ("java.util.concurrent.ForkJoinPool.common.exceptionHandler"); 3009 if (sf != null || sh != null) { 3010 ClassLoader ldr = ClassLoader.getSystemClassLoader(); 3011 if (sf != null) 3012 fac = (ForkJoinWorkerThreadFactory) 3013 ldr.loadClass(sf).getConstructor().newInstance(); 3014 if (sh != null) 3015 handler = (UncaughtExceptionHandler) 3016 ldr.loadClass(sh).getConstructor().newInstance(); 3017 } 3018 } catch (Exception ignore) { 3019 } 3020 if (preset == 0) 3021 pc = Math.max(1, Runtime.getRuntime().availableProcessors() - 1); 3022 int p = Math.min(pc, MAX_CAP); 3023 int size = Math.max(MIN_QUEUES_SIZE, 3024 (p == 0) ? 1 : 3025 1 << (33 - Integer.numberOfLeadingZeros(p-1))); 3026 this.parallelism = p; 3027 this.config = ((preset & LMASK) | (((long)maxSpares) << TC_SHIFT) | 3028 (1L << RC_SHIFT)); 3029 this.factory = fac; 3030 this.ueh = handler; 3031 this.keepAlive = DEFAULT_KEEPALIVE; 3032 this.saturate = null; 3033 this.workerNamePrefix = null; 3034 this.poolName = name; 3035 this.queues = new WorkQueue[size]; 3036 this.container = SharedThreadContainer.create(name); 3037 } 3038 3039 /** 3040 * Returns the common pool instance. This pool is statically 3041 * constructed; its run state is unaffected by attempts to {@link 3042 * #shutdown} or {@link #shutdownNow}. However this pool and any 3043 * ongoing processing are automatically terminated upon program 3044 * {@link System#exit}. Any program that relies on asynchronous 3045 * task processing to complete before program termination should 3046 * invoke {@code commonPool().}{@link #awaitQuiescence awaitQuiescence}, 3047 * before exit. 3048 * 3049 * @return the common pool instance 3050 * @since 1.8 3051 */ 3052 public static ForkJoinPool commonPool() { 3053 // assert common != null : "static init error"; 3054 return common; 3055 } 3056 3057 /** 3058 * Package-private access to commonPool overriding zero parallelism 3059 */ 3060 static ForkJoinPool asyncCommonPool() { 3061 ForkJoinPool cp; int p; 3062 if ((p = (cp = common).parallelism) == 0) 3063 U.compareAndSetInt(cp, PARALLELISM, 0, 2); 3064 return cp; 3065 } 3066 3067 // Execution methods 3068 3069 /** 3070 * Performs the given task, returning its result upon completion. 3071 * If the computation encounters an unchecked Exception or Error, 3072 * it is rethrown as the outcome of this invocation. Rethrown 3073 * exceptions behave in the same way as regular exceptions, but, 3074 * when possible, contain stack traces (as displayed for example 3075 * using {@code ex.printStackTrace()}) of both the current thread 3076 * as well as the thread actually encountering the exception; 3077 * minimally only the latter. 3078 * 3079 * @param task the task 3080 * @param <T> the type of the task's result 3081 * @return the task's result 3082 * @throws NullPointerException if the task is null 3083 * @throws RejectedExecutionException if the task cannot be 3084 * scheduled for execution 3085 */ 3086 public <T> T invoke(ForkJoinTask<T> task) { 3087 poolSubmit(true, Objects.requireNonNull(task)); 3088 try { 3089 return task.join(); 3090 } catch (RuntimeException | Error unchecked) { 3091 throw unchecked; 3092 } catch (Exception checked) { 3093 throw new RuntimeException(checked); 3094 } 3095 } 3096 3097 /** 3098 * Arranges for (asynchronous) execution of the given task. 3099 * 3100 * @param task the task 3101 * @throws NullPointerException if the task is null 3102 * @throws RejectedExecutionException if the task cannot be 3103 * scheduled for execution 3104 */ 3105 public void execute(ForkJoinTask<?> task) { 3106 poolSubmit(true, Objects.requireNonNull(task)); 3107 } 3108 3109 // AbstractExecutorService methods 3110 3111 /** 3112 * @throws NullPointerException if the task is null 3113 * @throws RejectedExecutionException if the task cannot be 3114 * scheduled for execution 3115 */ 3116 @Override 3117 @SuppressWarnings("unchecked") 3118 public void execute(Runnable task) { 3119 poolSubmit(true, (Objects.requireNonNull(task) instanceof ForkJoinTask<?>) 3120 ? (ForkJoinTask<Void>) task // avoid re-wrap 3121 : new ForkJoinTask.RunnableExecuteAction(task)); 3122 } 3123 3124 /** 3125 * Submits a ForkJoinTask for execution. 3126 * 3127 * @implSpec 3128 * This method is equivalent to {@link #externalSubmit(ForkJoinTask)} 3129 * when called from a thread that is not in this pool. 3130 * 3131 * @param task the task to submit 3132 * @param <T> the type of the task's result 3133 * @return the task 3134 * @throws NullPointerException if the task is null 3135 * @throws RejectedExecutionException if the task cannot be 3136 * scheduled for execution 3137 */ 3138 public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) { 3139 return poolSubmit(true, Objects.requireNonNull(task)); 3140 } 3141 3142 /** 3143 * @throws NullPointerException if the task is null 3144 * @throws RejectedExecutionException if the task cannot be 3145 * scheduled for execution 3146 */ 3147 @Override 3148 public <T> ForkJoinTask<T> submit(Callable<T> task) { 3149 Objects.requireNonNull(task); 3150 return poolSubmit( 3151 true, 3152 (Thread.currentThread() instanceof ForkJoinWorkerThread) ? 3153 new ForkJoinTask.AdaptedCallable<T>(task) : 3154 new ForkJoinTask.AdaptedInterruptibleCallable<T>(task)); 3155 } 3156 3157 /** 3158 * @throws NullPointerException if the task is null 3159 * @throws RejectedExecutionException if the task cannot be 3160 * scheduled for execution 3161 */ 3162 @Override 3163 public <T> ForkJoinTask<T> submit(Runnable task, T result) { 3164 Objects.requireNonNull(task); 3165 return poolSubmit( 3166 true, 3167 (Thread.currentThread() instanceof ForkJoinWorkerThread) ? 3168 new ForkJoinTask.AdaptedRunnable<T>(task, result) : 3169 new ForkJoinTask.AdaptedInterruptibleRunnable<T>(task, result)); 3170 } 3171 3172 /** 3173 * @throws NullPointerException if the task is null 3174 * @throws RejectedExecutionException if the task cannot be 3175 * scheduled for execution 3176 */ 3177 @Override 3178 @SuppressWarnings("unchecked") 3179 public ForkJoinTask<?> submit(Runnable task) { 3180 Objects.requireNonNull(task); 3181 return poolSubmit( 3182 true, 3183 (task instanceof ForkJoinTask<?>) ? 3184 (ForkJoinTask<Void>) task : // avoid re-wrap 3185 ((Thread.currentThread() instanceof ForkJoinWorkerThread) ? 3186 new ForkJoinTask.AdaptedRunnable<Void>(task, null) : 3187 new ForkJoinTask.AdaptedInterruptibleRunnable<Void>(task, null))); 3188 } 3189 3190 /** 3191 * Submits the given task as if submitted from a non-{@code ForkJoinTask} 3192 * client. The task is added to a scheduling queue for submissions to the 3193 * pool even when called from a thread in the pool. 3194 * 3195 * @implSpec 3196 * This method is equivalent to {@link #submit(ForkJoinTask)} when called 3197 * from a thread that is not in this pool. 3198 * 3199 * @return the task 3200 * @param task the task to submit 3201 * @param <T> the type of the task's result 3202 * @throws NullPointerException if the task is null 3203 * @throws RejectedExecutionException if the task cannot be 3204 * scheduled for execution 3205 * @since 20 3206 */ 3207 public <T> ForkJoinTask<T> externalSubmit(ForkJoinTask<T> task) { 3208 Objects.requireNonNull(task); 3209 externalSubmissionQueue(true).push(task, this, false); 3210 return task; 3211 } 3212 3213 /** 3214 * Submits the given task without guaranteeing that it will 3215 * eventually execute in the absence of available active threads. 3216 * In some contexts, this method may reduce contention and 3217 * overhead by relying on context-specific knowledge that existing 3218 * threads (possibly including the calling thread if operating in 3219 * this pool) will eventually be available to execute the task. 3220 * 3221 * @param task the task 3222 * @param <T> the type of the task's result 3223 * @return the task 3224 * @throws NullPointerException if the task is null 3225 * @throws RejectedExecutionException if the task cannot be 3226 * scheduled for execution 3227 * @since 19 3228 */ 3229 public <T> ForkJoinTask<T> lazySubmit(ForkJoinTask<T> task) { 3230 return poolSubmit(false, Objects.requireNonNull(task)); 3231 } 3232 3233 /** 3234 * Changes the target parallelism of this pool, controlling the 3235 * future creation, use, and termination of worker threads. 3236 * Applications include contexts in which the number of available 3237 * processors changes over time. 3238 * 3239 * @implNote This implementation restricts the maximum number of 3240 * running threads to 32767 3241 * 3242 * @param size the target parallelism level 3243 * @return the previous parallelism level. 3244 * @throws IllegalArgumentException if size is less than 1 or 3245 * greater than the maximum supported by this pool. 3246 * @throws UnsupportedOperationException this is the{@link 3247 * #commonPool()} and parallelism level was set by System 3248 * property {@systemProperty 3249 * java.util.concurrent.ForkJoinPool.common.parallelism}. 3250 * @since 19 3251 */ 3252 public int setParallelism(int size) { 3253 if (size < 1 || size > MAX_CAP) 3254 throw new IllegalArgumentException(); 3255 if ((config & PRESET_SIZE) != 0) 3256 throw new UnsupportedOperationException("Cannot override System property"); 3257 return getAndSetParallelism(size); 3258 } 3259 3260 /** 3261 * Uninterrupible version of {@code invokeAll}. Executes the given 3262 * tasks, returning a list of Futures holding their status and 3263 * results when all complete, ignoring interrupts. {@link 3264 * Future#isDone} is {@code true} for each element of the returned 3265 * list. Note that a <em>completed</em> task could have 3266 * terminated either normally or by throwing an exception. The 3267 * results of this method are undefined if the given collection is 3268 * modified while this operation is in progress. 3269 * 3270 * @apiNote This method supports usages that previously relied on an 3271 * incompatible override of 3272 * {@link ExecutorService#invokeAll(java.util.Collection)}. 3273 * 3274 * @param tasks the collection of tasks 3275 * @param <T> the type of the values returned from the tasks 3276 * @return a list of Futures representing the tasks, in the same 3277 * sequential order as produced by the iterator for the 3278 * given task list, each of which has completed 3279 * @throws NullPointerException if tasks or any of its elements are {@code null} 3280 * @throws RejectedExecutionException if any task cannot be 3281 * scheduled for execution 3282 * @since 22 3283 */ 3284 public <T> List<Future<T>> invokeAllUninterruptibly(Collection<? extends Callable<T>> tasks) { 3285 ArrayList<Future<T>> futures = new ArrayList<>(tasks.size()); 3286 try { 3287 for (Callable<T> t : tasks) { 3288 ForkJoinTask<T> f = ForkJoinTask.adapt(t); 3289 futures.add(f); 3290 poolSubmit(true, f); 3291 } 3292 for (int i = futures.size() - 1; i >= 0; --i) 3293 ((ForkJoinTask<?>)futures.get(i)).quietlyJoin(); 3294 return futures; 3295 } catch (Throwable t) { 3296 for (Future<T> e : futures) 3297 e.cancel(true); 3298 throw t; 3299 } 3300 } 3301 3302 /** 3303 * Common support for timed and untimed invokeAll 3304 */ 3305 private <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, 3306 long deadline) 3307 throws InterruptedException { 3308 ArrayList<Future<T>> futures = new ArrayList<>(tasks.size()); 3309 try { 3310 for (Callable<T> t : tasks) { 3311 ForkJoinTask<T> f = ForkJoinTask.adaptInterruptible(t); 3312 futures.add(f); 3313 poolSubmit(true, f); 3314 } 3315 for (int i = futures.size() - 1; i >= 0; --i) 3316 ((ForkJoinTask<?>)futures.get(i)) 3317 .quietlyJoinPoolInvokeAllTask(deadline); 3318 return futures; 3319 } catch (Throwable t) { 3320 for (Future<T> e : futures) 3321 e.cancel(true); 3322 throw t; 3323 } 3324 } 3325 3326 @Override 3327 public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) 3328 throws InterruptedException { 3329 return invokeAll(tasks, 0L); 3330 } 3331 // for jdk version < 22, replace with 3332 // /** 3333 // * @throws NullPointerException {@inheritDoc} 3334 // * @throws RejectedExecutionException {@inheritDoc} 3335 // */ 3336 // @Override 3337 // public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) { 3338 // return invokeAllUninterruptibly(tasks); 3339 // } 3340 3341 @Override 3342 public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, 3343 long timeout, TimeUnit unit) 3344 throws InterruptedException { 3345 return invokeAll(tasks, (System.nanoTime() + unit.toNanos(timeout)) | 1L); 3346 } 3347 3348 @Override 3349 public <T> T invokeAny(Collection<? extends Callable<T>> tasks) 3350 throws InterruptedException, ExecutionException { 3351 try { 3352 return new ForkJoinTask.InvokeAnyRoot<T>() 3353 .invokeAny(tasks, this, false, 0L); 3354 } catch (TimeoutException cannotHappen) { 3355 assert false; 3356 return null; 3357 } 3358 } 3359 3360 @Override 3361 public <T> T invokeAny(Collection<? extends Callable<T>> tasks, 3362 long timeout, TimeUnit unit) 3363 throws InterruptedException, ExecutionException, TimeoutException { 3364 return new ForkJoinTask.InvokeAnyRoot<T>() 3365 .invokeAny(tasks, this, true, unit.toNanos(timeout)); 3366 } 3367 3368 // Support for delayed tasks 3369 3370 /** 3371 * Returns STOP and SHUTDOWN status (zero if neither), masking or 3372 * truncating out other bits. 3373 */ 3374 final int shutdownStatus(DelayScheduler ds) { 3375 return (int)(runState & (SHUTDOWN | STOP)); 3376 } 3377 3378 /** 3379 * Tries to stop and possibly terminate if already enabled, return success. 3380 */ 3381 final boolean tryStopIfShutdown(DelayScheduler ds) { 3382 return (tryTerminate(false, false) & STOP) != 0L; 3383 } 3384 3385 /** 3386 * Creates and starts DelayScheduler 3387 */ 3388 private DelayScheduler startDelayScheduler() { 3389 DelayScheduler ds; 3390 if ((ds = delayScheduler) == null) { 3391 boolean start = false; 3392 String name = poolName + "-delayScheduler"; 3393 if (workerNamePrefix == null) 3394 asyncCommonPool(); // override common parallelism zero 3395 long isShutdown = lockRunState() & SHUTDOWN; 3396 try { 3397 if (isShutdown == 0L && (ds = delayScheduler) == null) { 3398 ds = delayScheduler = new DelayScheduler(this, name); 3399 start = true; 3400 } 3401 } finally { 3402 unlockRunState(); 3403 } 3404 if (start) { // start outside of lock 3405 SharedThreadContainer ctr; 3406 try { 3407 if ((ctr = container) != null) 3408 ctr.start(ds); 3409 else 3410 ds.start(); 3411 } catch (RuntimeException | Error ex) { // back out 3412 lockRunState(); 3413 ds = delayScheduler = null; 3414 unlockRunState(); 3415 tryTerminate(false, false); 3416 if (ex instanceof Error) 3417 throw ex; 3418 } 3419 } 3420 } 3421 return ds; 3422 } 3423 3424 /** 3425 * Arranges execution of a ScheduledForkJoinTask whose delay has 3426 * elapsed 3427 */ 3428 final void executeEnabledScheduledTask(ScheduledForkJoinTask<?> task) { 3429 externalSubmissionQueue(false).push(task, this, false); 3430 } 3431 3432 /** 3433 * Arranges delayed execution of a ScheduledForkJoinTask via the 3434 * DelayScheduler, creating and starting it if necessary. 3435 * @return the task 3436 */ 3437 final <T> ScheduledForkJoinTask<T> scheduleDelayedTask(ScheduledForkJoinTask<T> task) { 3438 DelayScheduler ds; 3439 if (((ds = delayScheduler) == null && 3440 (ds = startDelayScheduler()) == null) || 3441 (runState & SHUTDOWN) != 0L) 3442 throw new RejectedExecutionException(); 3443 ds.pend(task); 3444 return task; 3445 } 3446 3447 /** 3448 * Submits a one-shot task that becomes enabled for execution after the given 3449 * delay. At that point it will execute unless explicitly 3450 * cancelled, or fail to execute (eventually reporting 3451 * cancellation) when encountering resource exhaustion, or the 3452 * pool is {@link #shutdownNow}, or is {@link #shutdown} when 3453 * otherwise quiescent and {@link #cancelDelayedTasksOnShutdown} 3454 * is in effect. 3455 * 3456 * @param command the task to execute 3457 * @param delay the time from now to delay execution 3458 * @param unit the time unit of the delay parameter 3459 * @return a ForkJoinTask implementing the ScheduledFuture 3460 * interface, whose {@code get()} method will return 3461 * {@code null} upon normal completion. 3462 * @throws RejectedExecutionException if the pool is shutdown or 3463 * submission encounters resource exhaustion. 3464 * @throws NullPointerException if command or unit is null 3465 * @since 25 3466 */ 3467 public ScheduledFuture<?> schedule(Runnable command, 3468 long delay, TimeUnit unit) { 3469 return scheduleDelayedTask( 3470 new ScheduledForkJoinTask<Void>( 3471 unit.toNanos(delay), 0L, false, // implicit null check of unit 3472 Objects.requireNonNull(command), null, this)); 3473 } 3474 3475 /** 3476 * Submits a value-returning one-shot task that becomes enabled for execution 3477 * after the given delay. At that point it will execute unless 3478 * explicitly cancelled, or fail to execute (eventually reporting 3479 * cancellation) when encountering resource exhaustion, or the 3480 * pool is {@link #shutdownNow}, or is {@link #shutdown} when 3481 * otherwise quiescent and {@link #cancelDelayedTasksOnShutdown} 3482 * is in effect. 3483 * 3484 * @param callable the function to execute 3485 * @param delay the time from now to delay execution 3486 * @param unit the time unit of the delay parameter 3487 * @param <V> the type of the callable's result 3488 * @return a ForkJoinTask implementing the ScheduledFuture 3489 * interface, whose {@code get()} method will return the 3490 * value from the callable upon normal completion. 3491 * @throws RejectedExecutionException if the pool is shutdown or 3492 * submission encounters resource exhaustion. 3493 * @throws NullPointerException if command or unit is null 3494 * @since 25 3495 */ 3496 public <V> ScheduledFuture<V> schedule(Callable<V> callable, 3497 long delay, TimeUnit unit) { 3498 return scheduleDelayedTask( 3499 new ScheduledForkJoinTask<V>( 3500 unit.toNanos(delay), 0L, false, null, // implicit null check of unit 3501 Objects.requireNonNull(callable), this)); 3502 } 3503 3504 /** 3505 * Submits a periodic action that becomes enabled for execution first after the 3506 * given initial delay, and subsequently with the given period; 3507 * that is, executions will commence after 3508 * {@code initialDelay}, then {@code initialDelay + period}, then 3509 * {@code initialDelay + 2 * period}, and so on. 3510 * 3511 * <p>The sequence of task executions continues indefinitely until 3512 * one of the following exceptional completions occur: 3513 * <ul> 3514 * <li>The task is {@linkplain Future#cancel explicitly cancelled} 3515 * <li>Method {@link #shutdownNow} is called 3516 * <li>Method {@link #shutdown} is called and the pool is 3517 * otherwise quiescent, in which case existing executions continue 3518 * but subsequent executions do not. 3519 * <li>An execution or the task encounters resource exhaustion. 3520 * <li>An execution of the task throws an exception. In this case 3521 * calling {@link Future#get() get} on the returned future will throw 3522 * {@link ExecutionException}, holding the exception as its cause. 3523 * </ul> 3524 * Subsequent executions are suppressed. Subsequent calls to 3525 * {@link Future#isDone isDone()} on the returned future will 3526 * return {@code true}. 3527 * 3528 * <p>If any execution of this task takes longer than its period, then 3529 * subsequent executions may start late, but will not concurrently 3530 * execute. 3531 * @param command the task to execute 3532 * @param initialDelay the time to delay first execution 3533 * @param period the period between successive executions 3534 * @param unit the time unit of the initialDelay and period parameters 3535 * @return a ForkJoinTask implementing the ScheduledFuture 3536 * interface. The future's {@link Future#get() get()} 3537 * method will never return normally, and will throw an 3538 * exception upon task cancellation or abnormal 3539 * termination of a task execution. 3540 * @throws RejectedExecutionException if the pool is shutdown or 3541 * submission encounters resource exhaustion. 3542 * @throws NullPointerException if command or unit is null 3543 * @throws IllegalArgumentException if period less than or equal to zero 3544 * @since 25 3545 */ 3546 public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, 3547 long initialDelay, 3548 long period, TimeUnit unit) { 3549 if (period <= 0L) 3550 throw new IllegalArgumentException(); 3551 return scheduleDelayedTask( 3552 new ScheduledForkJoinTask<Void>( 3553 unit.toNanos(initialDelay), // implicit null check of unit 3554 unit.toNanos(period), false, 3555 Objects.requireNonNull(command), null, this)); 3556 } 3557 3558 /** 3559 * Submits a periodic action that becomes enabled for execution first after the 3560 * given initial delay, and subsequently with the given delay 3561 * between the termination of one execution and the commencement of 3562 * the next. 3563 * <p>The sequence of task executions continues indefinitely until 3564 * one of the following exceptional completions occur: 3565 * <ul> 3566 * <li>The task is {@linkplain Future#cancel explicitly cancelled} 3567 * <li>Method {@link #shutdownNow} is called 3568 * <li>Method {@link #shutdown} is called and the pool is 3569 * otherwise quiescent, in which case existing executions continue 3570 * but subsequent executions do not. 3571 * <li>An execution or the task encounters resource exhaustion. 3572 * <li>An execution of the task throws an exception. In this case 3573 * calling {@link Future#get() get} on the returned future will throw 3574 * {@link ExecutionException}, holding the exception as its cause. 3575 * </ul> 3576 * Subsequent executions are suppressed. Subsequent calls to 3577 * {@link Future#isDone isDone()} on the returned future will 3578 * return {@code true}. 3579 * @param command the task to execute 3580 * @param initialDelay the time to delay first execution 3581 * @param delay the delay between the termination of one 3582 * execution and the commencement of the next 3583 * @param unit the time unit of the initialDelay and delay parameters 3584 * @return a ForkJoinTask implementing the ScheduledFuture 3585 * interface. The future's {@link Future#get() get()} 3586 * method will never return normally, and will throw an 3587 * exception upon task cancellation or abnormal 3588 * termination of a task execution. 3589 * @throws RejectedExecutionException if the pool is shutdown or 3590 * submission encounters resource exhaustion. 3591 * @throws NullPointerException if command or unit is null 3592 * @throws IllegalArgumentException if delay less than or equal to zero 3593 * @since 25 3594 */ 3595 public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, 3596 long initialDelay, 3597 long delay, TimeUnit unit) { 3598 if (delay <= 0L) 3599 throw new IllegalArgumentException(); 3600 return scheduleDelayedTask( 3601 new ScheduledForkJoinTask<Void>( 3602 unit.toNanos(initialDelay), // implicit null check of unit 3603 -unit.toNanos(delay), false, // negative for fixed delay 3604 Objects.requireNonNull(command), null, this)); 3605 } 3606 3607 /** 3608 * Body of a task performed on timeout of another task 3609 */ 3610 static final class TimeoutAction<V> implements Runnable { 3611 // set after construction, nulled after use 3612 ForkJoinTask.CallableWithTimeout<V> task; 3613 Consumer<? super ForkJoinTask<V>> action; 3614 TimeoutAction(Consumer<? super ForkJoinTask<V>> action) { 3615 this.action = action; 3616 } 3617 public void run() { 3618 ForkJoinTask.CallableWithTimeout<V> t = task; 3619 Consumer<? super ForkJoinTask<V>> a = action; 3620 task = null; 3621 action = null; 3622 if (t != null && t.status >= 0) { 3623 if (a == null) 3624 t.cancel(true); 3625 else { 3626 a.accept(t); 3627 t.interruptIfRunning(true); 3628 } 3629 } 3630 } 3631 } 3632 3633 /** 3634 * Submits a task executing the given function, cancelling the 3635 * task or performing a given timeoutAction if not completed 3636 * within the given timeout period. If the optional {@code 3637 * timeoutAction} is null, the task is cancelled (via {@code 3638 * cancel(true)}. Otherwise, the action is applied and the task 3639 * may be interrupted if running. Actions may include {@link 3640 * ForkJoinTask#complete} to set a replacement value or {@link 3641 * ForkJoinTask#completeExceptionally} to throw an appropriate 3642 * exception. Note that these can succeed only if the task has 3643 * not already completed when the timeoutAction executes. 3644 * 3645 * @param callable the function to execute 3646 * @param <V> the type of the callable's result 3647 * @param timeout the time to wait before cancelling if not completed 3648 * @param timeoutAction if nonnull, an action to perform on 3649 * timeout, otherwise the default action is to cancel using 3650 * {@code cancel(true)}. 3651 * @param unit the time unit of the timeout parameter 3652 * @return a Future that can be used to extract result or cancel 3653 * @throws RejectedExecutionException if the task cannot be 3654 * scheduled for execution 3655 * @throws NullPointerException if callable or unit is null 3656 * @since 25 3657 */ 3658 public <V> ForkJoinTask<V> submitWithTimeout(Callable<V> callable, 3659 long timeout, TimeUnit unit, 3660 Consumer<? super ForkJoinTask<V>> timeoutAction) { 3661 ForkJoinTask.CallableWithTimeout<V> task; TimeoutAction<V> onTimeout; 3662 Objects.requireNonNull(callable); 3663 ScheduledForkJoinTask<Void> timeoutTask = 3664 new ScheduledForkJoinTask<Void>( 3665 unit.toNanos(timeout), 0L, true, 3666 onTimeout = new TimeoutAction<V>(timeoutAction), null, this); 3667 onTimeout.task = task = 3668 new ForkJoinTask.CallableWithTimeout<V>(callable, timeoutTask); 3669 scheduleDelayedTask(timeoutTask); 3670 return poolSubmit(true, task); 3671 } 3672 3673 /** 3674 * Arranges that scheduled tasks that are not executing and have 3675 * not already been enabled for execution will not be executed and 3676 * will be cancelled upon {@link #shutdown} (unless this pool is 3677 * the {@link #commonPool()} which never shuts down). This method 3678 * may be invoked either before {@link #shutdown} to take effect 3679 * upon the next call, or afterwards to cancel such tasks, which 3680 * may then allow termination. Note that subsequent executions of 3681 * periodic tasks are always disabled upon shutdown, so this 3682 * method applies meaningfully only to non-periodic tasks. 3683 * @since 25 3684 */ 3685 public void cancelDelayedTasksOnShutdown() { 3686 DelayScheduler ds; 3687 if ((ds = delayScheduler) != null || 3688 (ds = startDelayScheduler()) != null) 3689 ds.cancelDelayedTasksOnShutdown(); 3690 } 3691 3692 /** 3693 * Returns the factory used for constructing new workers. 3694 * 3695 * @return the factory used for constructing new workers 3696 */ 3697 public ForkJoinWorkerThreadFactory getFactory() { 3698 return factory; 3699 } 3700 3701 /** 3702 * Returns the handler for internal worker threads that terminate 3703 * due to unrecoverable errors encountered while executing tasks. 3704 * 3705 * @return the handler, or {@code null} if none 3706 */ 3707 public UncaughtExceptionHandler getUncaughtExceptionHandler() { 3708 return ueh; 3709 } 3710 3711 /** 3712 * Returns the targeted parallelism level of this pool. 3713 * 3714 * @return the targeted parallelism level of this pool 3715 */ 3716 public int getParallelism() { 3717 return Math.max(getParallelismOpaque(), 1); 3718 } 3719 3720 /** 3721 * Returns the targeted parallelism level of the common pool. 3722 * 3723 * @return the targeted parallelism level of the common pool 3724 * @since 1.8 3725 */ 3726 public static int getCommonPoolParallelism() { 3727 return common.getParallelism(); 3728 } 3729 3730 /** 3731 * Returns the number of worker threads that have started but not 3732 * yet terminated. The result returned by this method may differ 3733 * from {@link #getParallelism} when threads are created to 3734 * maintain parallelism when others are cooperatively blocked. 3735 * 3736 * @return the number of worker threads 3737 */ 3738 public int getPoolSize() { 3739 return (short)(ctl >>> TC_SHIFT); 3740 } 3741 3742 /** 3743 * Returns {@code true} if this pool uses local first-in-first-out 3744 * scheduling mode for forked tasks that are never joined. 3745 * 3746 * @return {@code true} if this pool uses async mode 3747 */ 3748 public boolean getAsyncMode() { 3749 return (config & FIFO) != 0; 3750 } 3751 3752 /** 3753 * Returns an estimate of the number of worker threads that are 3754 * not blocked waiting to join tasks or for other managed 3755 * synchronization. This method may overestimate the 3756 * number of running threads. 3757 * 3758 * @return the number of worker threads 3759 */ 3760 public int getRunningThreadCount() { 3761 WorkQueue[] qs; WorkQueue q; 3762 int rc = 0; 3763 if ((runState & TERMINATED) == 0L && (qs = queues) != null) { 3764 for (int i = 1; i < qs.length; i += 2) { 3765 if ((q = qs[i]) != null && q.isApparentlyUnblocked()) 3766 ++rc; 3767 } 3768 } 3769 return rc; 3770 } 3771 3772 /** 3773 * Returns an estimate of the number of threads that are currently 3774 * stealing or executing tasks. This method may overestimate the 3775 * number of active threads. 3776 * 3777 * @return the number of active threads 3778 */ 3779 public int getActiveThreadCount() { 3780 return Math.max((short)(ctl >>> RC_SHIFT), 0); 3781 } 3782 3783 /** 3784 * Returns {@code true} if all worker threads are currently idle. 3785 * An idle worker is one that cannot obtain a task to execute 3786 * because none are available to steal from other threads, and 3787 * there are no pending submissions to the pool. This method is 3788 * conservative; it might not return {@code true} immediately upon 3789 * idleness of all threads, but will eventually become true if 3790 * threads remain inactive. 3791 * 3792 * @return {@code true} if all threads are currently idle 3793 */ 3794 public boolean isQuiescent() { 3795 return quiescent() >= 0; 3796 } 3797 3798 /** 3799 * Returns an estimate of the total number of completed tasks that 3800 * were executed by a thread other than their submitter. The 3801 * reported value underestimates the actual total number of steals 3802 * when the pool is not quiescent. This value may be useful for 3803 * monitoring and tuning fork/join programs: in general, steal 3804 * counts should be high enough to keep threads busy, but low 3805 * enough to avoid overhead and contention across threads. 3806 * 3807 * @return the number of steals 3808 */ 3809 public long getStealCount() { 3810 long count = stealCount; 3811 WorkQueue[] qs; WorkQueue q; 3812 if ((qs = queues) != null) { 3813 for (int i = 1; i < qs.length; i += 2) { 3814 if ((q = qs[i]) != null) 3815 count += (long)q.nsteals & 0xffffffffL; 3816 } 3817 } 3818 return count; 3819 } 3820 3821 /** 3822 * Returns an estimate of the total number of tasks currently held 3823 * in queues by worker threads (but not including tasks submitted 3824 * to the pool that have not begun executing). This value is only 3825 * an approximation, obtained by iterating across all threads in 3826 * the pool. This method may be useful for tuning task 3827 * granularities.The returned count does not include scheduled 3828 * tasks that are not yet ready to execute, which are reported 3829 * separately by method {@link getDelayedTaskCount}. 3830 * 3831 * @return the number of queued tasks 3832 * @see ForkJoinWorkerThread#getQueuedTaskCount() 3833 */ 3834 public long getQueuedTaskCount() { 3835 WorkQueue[] qs; WorkQueue q; 3836 long count = 0; 3837 if ((runState & TERMINATED) == 0L && (qs = queues) != null) { 3838 for (int i = 1; i < qs.length; i += 2) { 3839 if ((q = qs[i]) != null) 3840 count += q.queueSize(); 3841 } 3842 } 3843 return count; 3844 } 3845 3846 /** 3847 * Returns an estimate of the number of tasks submitted to this 3848 * pool that have not yet begun executing. This method may take 3849 * time proportional to the number of submissions. 3850 * 3851 * @return the number of queued submissions 3852 */ 3853 public int getQueuedSubmissionCount() { 3854 WorkQueue[] qs; WorkQueue q; 3855 int count = 0; 3856 if ((runState & TERMINATED) == 0L && (qs = queues) != null) { 3857 for (int i = 0; i < qs.length; i += 2) { 3858 if ((q = qs[i]) != null) 3859 count += q.queueSize(); 3860 } 3861 } 3862 return count; 3863 } 3864 3865 /** 3866 * Returns an estimate of the number of delayed (including 3867 * periodic) tasks scheduled in this pool that are not yet ready 3868 * to submit for execution. The returned value is inaccurate while 3869 * delayed tasks are being processed. 3870 * 3871 * @return an estimate of the number of delayed tasks 3872 * @since 25 3873 */ 3874 public long getDelayedTaskCount() { 3875 DelayScheduler ds; 3876 return ((ds = delayScheduler) == null ? 0 : ds.lastStableSize()); 3877 } 3878 3879 /** 3880 * Returns {@code true} if there are any tasks submitted to this 3881 * pool that have not yet begun executing. 3882 * 3883 * @return {@code true} if there are any queued submissions 3884 */ 3885 public boolean hasQueuedSubmissions() { 3886 WorkQueue[] qs; WorkQueue q; 3887 if ((runState & STOP) == 0L && (qs = queues) != null) { 3888 for (int i = 0; i < qs.length; i += 2) { 3889 if ((q = qs[i]) != null && q.queueSize() > 0) 3890 return true; 3891 } 3892 } 3893 return false; 3894 } 3895 3896 /** 3897 * Removes and returns the next unexecuted submission if one is 3898 * available. This method may be useful in extensions to this 3899 * class that re-assign work in systems with multiple pools. 3900 * 3901 * @return the next submission, or {@code null} if none 3902 */ 3903 protected ForkJoinTask<?> pollSubmission() { 3904 return pollScan(true); 3905 } 3906 3907 /** 3908 * Removes all available unexecuted submitted and forked tasks 3909 * from scheduling queues and adds them to the given collection, 3910 * without altering their execution status. These may include 3911 * artificially generated or wrapped tasks. This method is 3912 * designed to be invoked only when the pool is known to be 3913 * quiescent. Invocations at other times may not remove all 3914 * tasks. A failure encountered while attempting to add elements 3915 * to collection {@code c} may result in elements being in 3916 * neither, either or both collections when the associated 3917 * exception is thrown. The behavior of this operation is 3918 * undefined if the specified collection is modified while the 3919 * operation is in progress. 3920 * 3921 * @param c the collection to transfer elements into 3922 * @return the number of elements transferred 3923 */ 3924 protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) { 3925 int count = 0; 3926 for (ForkJoinTask<?> t; (t = pollScan(false)) != null; ) { 3927 c.add(t); 3928 ++count; 3929 } 3930 return count; 3931 } 3932 3933 /** 3934 * Returns a string identifying this pool, as well as its state, 3935 * including indications of run state, parallelism level, and 3936 * worker and task counts. 3937 * 3938 * @return a string identifying this pool, as well as its state 3939 */ 3940 public String toString() { 3941 // Use a single pass through queues to collect counts 3942 DelayScheduler ds; 3943 long e = runState; 3944 long st = stealCount; 3945 long qt = 0L, ss = 0L; int rc = 0; 3946 WorkQueue[] qs; WorkQueue q; 3947 if ((qs = queues) != null) { 3948 for (int i = 0; i < qs.length; ++i) { 3949 if ((q = qs[i]) != null) { 3950 int size = q.queueSize(); 3951 if ((i & 1) == 0) 3952 ss += size; 3953 else { 3954 qt += size; 3955 st += (long)q.nsteals & 0xffffffffL; 3956 if (q.isApparentlyUnblocked()) 3957 ++rc; 3958 } 3959 } 3960 } 3961 } 3962 String delayed = ((ds = delayScheduler) == null ? "" : 3963 ", delayed = " + ds.lastStableSize()); 3964 int pc = parallelism; 3965 long c = ctl; 3966 int tc = (short)(c >>> TC_SHIFT); 3967 int ac = (short)(c >>> RC_SHIFT); 3968 if (ac < 0) // ignore transient negative 3969 ac = 0; 3970 String level = ((e & TERMINATED) != 0L ? "Terminated" : 3971 (e & STOP) != 0L ? "Terminating" : 3972 (e & SHUTDOWN) != 0L ? "Shutting down" : 3973 "Running"); 3974 return super.toString() + 3975 "[" + level + 3976 ", parallelism = " + pc + 3977 ", size = " + tc + 3978 ", active = " + ac + 3979 ", running = " + rc + 3980 ", steals = " + st + 3981 ", tasks = " + qt + 3982 ", submissions = " + ss + 3983 delayed + 3984 "]"; 3985 } 3986 3987 /** 3988 * Possibly initiates an orderly shutdown in which previously 3989 * submitted tasks are executed, but no new tasks will be 3990 * accepted. Invocation has no effect on execution state if this 3991 * is the {@link #commonPool()}, and no additional effect if 3992 * already shut down. Tasks that are in the process of being 3993 * submitted concurrently during the course of this method may or 3994 * may not be rejected. 3995 */ 3996 public void shutdown() { 3997 if (workerNamePrefix != null) // not common pool 3998 tryTerminate(false, true); 3999 } 4000 4001 /** 4002 * Possibly attempts to cancel and/or stop all tasks, and reject 4003 * all subsequently submitted tasks. Invocation has no effect on 4004 * execution state if this is the {@link #commonPool()}, and no 4005 * additional effect if already shut down. Otherwise, tasks that 4006 * are in the process of being submitted or executed concurrently 4007 * during the course of this method may or may not be 4008 * rejected. This method cancels both existing and unexecuted 4009 * tasks, in order to permit termination in the presence of task 4010 * dependencies. So the method always returns an empty list 4011 * (unlike the case for some other Executors). 4012 * 4013 * @return an empty list 4014 */ 4015 public List<Runnable> shutdownNow() { 4016 if (workerNamePrefix != null) // not common pool 4017 tryTerminate(true, true); 4018 return Collections.emptyList(); 4019 } 4020 4021 /** 4022 * Returns {@code true} if all tasks have completed following shut down. 4023 * 4024 * @return {@code true} if all tasks have completed following shut down 4025 */ 4026 public boolean isTerminated() { 4027 return (tryTerminate(false, false) & TERMINATED) != 0; 4028 } 4029 4030 /** 4031 * Returns {@code true} if the process of termination has 4032 * commenced but not yet completed. This method may be useful for 4033 * debugging. A return of {@code true} reported a sufficient 4034 * period after shutdown may indicate that submitted tasks have 4035 * ignored or suppressed interruption, or are waiting for I/O, 4036 * causing this executor not to properly terminate. (See the 4037 * advisory notes for class {@link ForkJoinTask} stating that 4038 * tasks should not normally entail blocking operations. But if 4039 * they do, they must abort them on interrupt.) 4040 * 4041 * @return {@code true} if terminating but not yet terminated 4042 */ 4043 public boolean isTerminating() { 4044 return (tryTerminate(false, false) & (STOP | TERMINATED)) == STOP; 4045 } 4046 4047 /** 4048 * Returns {@code true} if this pool has been shut down. 4049 * 4050 * @return {@code true} if this pool has been shut down 4051 */ 4052 public boolean isShutdown() { 4053 return (runState & SHUTDOWN) != 0L; 4054 } 4055 4056 /** 4057 * Blocks until all tasks have completed execution after a 4058 * shutdown request, or the timeout occurs, or the current thread 4059 * is interrupted, whichever happens first. Because the {@link 4060 * #commonPool()} never terminates until program shutdown, when 4061 * applied to the common pool, this method is equivalent to {@link 4062 * #awaitQuiescence(long, TimeUnit)} but always returns {@code false}. 4063 * 4064 * @param timeout the maximum time to wait 4065 * @param unit the time unit of the timeout argument 4066 * @return {@code true} if this executor terminated and 4067 * {@code false} if the timeout elapsed before termination 4068 * @throws InterruptedException if interrupted while waiting 4069 */ 4070 public boolean awaitTermination(long timeout, TimeUnit unit) 4071 throws InterruptedException { 4072 long nanos = unit.toNanos(timeout); 4073 CountDownLatch done; 4074 if (workerNamePrefix == null) { // is common pool 4075 if (helpQuiescePool(this, nanos, true) < 0) 4076 throw new InterruptedException(); 4077 return false; 4078 } 4079 else if ((tryTerminate(false, false) & TERMINATED) != 0 || 4080 (done = terminationSignal()) == null || 4081 (runState & TERMINATED) != 0L) 4082 return true; 4083 else 4084 return done.await(nanos, TimeUnit.NANOSECONDS); 4085 } 4086 4087 /** 4088 * If called by a ForkJoinTask operating in this pool, equivalent 4089 * in effect to {@link ForkJoinTask#helpQuiesce}. Otherwise, 4090 * waits and/or attempts to assist performing tasks until this 4091 * pool {@link #isQuiescent} or the indicated timeout elapses. 4092 * 4093 * @param timeout the maximum time to wait 4094 * @param unit the time unit of the timeout argument 4095 * @return {@code true} if quiescent; {@code false} if the 4096 * timeout elapsed. 4097 */ 4098 public boolean awaitQuiescence(long timeout, TimeUnit unit) { 4099 return (helpQuiescePool(this, unit.toNanos(timeout), false) > 0); 4100 } 4101 4102 /** 4103 * Unless this is the {@link #commonPool()}, initiates an orderly 4104 * shutdown in which previously submitted tasks are executed, but 4105 * no new tasks will be accepted, and waits until all tasks have 4106 * completed execution and the executor has terminated. 4107 * 4108 * <p> If already terminated, or this is the {@link 4109 * #commonPool()}, this method has no effect on execution, and 4110 * does not wait. Otherwise, if interrupted while waiting, this 4111 * method stops all executing tasks as if by invoking {@link 4112 * #shutdownNow()}. It then continues to wait until all actively 4113 * executing tasks have completed. Tasks that were awaiting 4114 * execution are not executed. The interrupt status will be 4115 * re-asserted before this method returns. 4116 * 4117 * @since 19 4118 */ 4119 @Override 4120 public void close() { 4121 if (workerNamePrefix != null) { 4122 CountDownLatch done = null; 4123 boolean interrupted = false; 4124 while ((tryTerminate(interrupted, true) & TERMINATED) == 0) { 4125 if (done == null) 4126 done = terminationSignal(); 4127 else { 4128 try { 4129 done.await(); 4130 break; 4131 } catch (InterruptedException ex) { 4132 interrupted = true; 4133 } 4134 } 4135 } 4136 if (interrupted) 4137 Thread.currentThread().interrupt(); 4138 } 4139 } 4140 4141 /** 4142 * Interface for extending managed parallelism for tasks running 4143 * in {@link ForkJoinPool}s. 4144 * 4145 * <p>A {@code ManagedBlocker} provides two methods. Method 4146 * {@link #isReleasable} must return {@code true} if blocking is 4147 * not necessary. Method {@link #block} blocks the current thread 4148 * if necessary (perhaps internally invoking {@code isReleasable} 4149 * before actually blocking). These actions are performed by any 4150 * thread invoking {@link 4151 * ForkJoinPool#managedBlock(ManagedBlocker)}. The unusual 4152 * methods in this API accommodate synchronizers that may, but 4153 * don't usually, block for long periods. Similarly, they allow 4154 * more efficient internal handling of cases in which additional 4155 * workers may be, but usually are not, needed to ensure 4156 * sufficient parallelism. Toward this end, implementations of 4157 * method {@code isReleasable} must be amenable to repeated 4158 * invocation. Neither method is invoked after a prior invocation 4159 * of {@code isReleasable} or {@code block} returns {@code true}. 4160 * 4161 * <p>For example, here is a ManagedBlocker based on a 4162 * ReentrantLock: 4163 * <pre> {@code 4164 * class ManagedLocker implements ManagedBlocker { 4165 * final ReentrantLock lock; 4166 * boolean hasLock = false; 4167 * ManagedLocker(ReentrantLock lock) { this.lock = lock; } 4168 * public boolean block() { 4169 * if (!hasLock) 4170 * lock.lock(); 4171 * return true; 4172 * } 4173 * public boolean isReleasable() { 4174 * return hasLock || (hasLock = lock.tryLock()); 4175 * } 4176 * }}</pre> 4177 * 4178 * <p>Here is a class that possibly blocks waiting for an 4179 * item on a given queue: 4180 * <pre> {@code 4181 * class QueueTaker<E> implements ManagedBlocker { 4182 * final BlockingQueue<E> queue; 4183 * volatile E item = null; 4184 * QueueTaker(BlockingQueue<E> q) { this.queue = q; } 4185 * public boolean block() throws InterruptedException { 4186 * if (item == null) 4187 * item = queue.take(); 4188 * return true; 4189 * } 4190 * public boolean isReleasable() { 4191 * return item != null || (item = queue.poll()) != null; 4192 * } 4193 * public E getItem() { // call after pool.managedBlock completes 4194 * return item; 4195 * } 4196 * }}</pre> 4197 */ 4198 public static interface ManagedBlocker { 4199 /** 4200 * Possibly blocks the current thread, for example waiting for 4201 * a lock or condition. 4202 * 4203 * @return {@code true} if no additional blocking is necessary 4204 * (i.e., if isReleasable would return true) 4205 * @throws InterruptedException if interrupted while waiting 4206 * (the method is not required to do so, but is allowed to) 4207 */ 4208 boolean block() throws InterruptedException; 4209 4210 /** 4211 * Returns {@code true} if blocking is unnecessary. 4212 * @return {@code true} if blocking is unnecessary 4213 */ 4214 boolean isReleasable(); 4215 } 4216 4217 /** 4218 * Runs the given possibly blocking task. When {@linkplain 4219 * ForkJoinTask#inForkJoinPool() running in a ForkJoinPool}, this 4220 * method possibly arranges for a spare thread to be activated if 4221 * necessary to ensure sufficient parallelism while the current 4222 * thread is blocked in {@link ManagedBlocker#block blocker.block()}. 4223 * 4224 * <p>This method repeatedly calls {@code blocker.isReleasable()} and 4225 * {@code blocker.block()} until either method returns {@code true}. 4226 * Every call to {@code blocker.block()} is preceded by a call to 4227 * {@code blocker.isReleasable()} that returned {@code false}. 4228 * 4229 * <p>If not running in a ForkJoinPool, this method is 4230 * behaviorally equivalent to 4231 * <pre> {@code 4232 * while (!blocker.isReleasable()) 4233 * if (blocker.block()) 4234 * break;}</pre> 4235 * 4236 * If running in a ForkJoinPool, the pool may first be expanded to 4237 * ensure sufficient parallelism available during the call to 4238 * {@code blocker.block()}. 4239 * 4240 * @param blocker the blocker task 4241 * @throws InterruptedException if {@code blocker.block()} did so 4242 */ 4243 public static void managedBlock(ManagedBlocker blocker) 4244 throws InterruptedException { 4245 Thread t; ForkJoinPool p; 4246 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread && 4247 (p = ((ForkJoinWorkerThread)t).pool) != null) 4248 p.compensatedBlock(blocker); 4249 else 4250 unmanagedBlock(blocker); 4251 } 4252 4253 /** ManagedBlock for ForkJoinWorkerThreads */ 4254 private void compensatedBlock(ManagedBlocker blocker) 4255 throws InterruptedException { 4256 Objects.requireNonNull(blocker); 4257 for (;;) { 4258 int comp; boolean done; 4259 long c = ctl; 4260 if (blocker.isReleasable()) 4261 break; 4262 if ((runState & STOP) != 0L) 4263 throw new InterruptedException(); 4264 if ((comp = tryCompensate(c)) >= 0) { 4265 try { 4266 done = blocker.block(); 4267 } finally { 4268 if (comp > 0) 4269 getAndAddCtl(RC_UNIT); 4270 } 4271 if (done) 4272 break; 4273 } 4274 } 4275 } 4276 4277 /** 4278 * Invokes tryCompensate to create or re-activate a spare thread to 4279 * compensate for a thread that performs a blocking operation. When the 4280 * blocking operation is done then endCompensatedBlock must be invoked 4281 * with the value returned by this method to re-adjust the parallelism. 4282 * @return value to use in endCompensatedBlock 4283 */ 4284 final long beginCompensatedBlock() { 4285 int c; 4286 do {} while ((c = tryCompensate(ctl)) < 0); 4287 return (c == 0) ? 0L : RC_UNIT; 4288 } 4289 4290 /** 4291 * Re-adjusts parallelism after a blocking operation completes. 4292 * @param post value from beginCompensatedBlock 4293 */ 4294 void endCompensatedBlock(long post) { 4295 if (post > 0L) { 4296 getAndAddCtl(post); 4297 } 4298 } 4299 4300 /** ManagedBlock for external threads */ 4301 private static void unmanagedBlock(ManagedBlocker blocker) 4302 throws InterruptedException { 4303 Objects.requireNonNull(blocker); 4304 do {} while (!blocker.isReleasable() && !blocker.block()); 4305 } 4306 4307 @Override 4308 protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { 4309 Objects.requireNonNull(runnable); 4310 return (Thread.currentThread() instanceof ForkJoinWorkerThread) ? 4311 new ForkJoinTask.AdaptedRunnable<T>(runnable, value) : 4312 new ForkJoinTask.AdaptedInterruptibleRunnable<T>(runnable, value); 4313 } 4314 4315 @Override 4316 protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { 4317 Objects.requireNonNull(callable); 4318 return (Thread.currentThread() instanceof ForkJoinWorkerThread) ? 4319 new ForkJoinTask.AdaptedCallable<T>(callable) : 4320 new ForkJoinTask.AdaptedInterruptibleCallable<T>(callable); 4321 } 4322 4323 static { 4324 U = Unsafe.getUnsafe(); 4325 Class<ForkJoinPool> klass = ForkJoinPool.class; 4326 try { 4327 Field poolIdsField = klass.getDeclaredField("poolIds"); 4328 POOLIDS_BASE = U.staticFieldBase(poolIdsField); 4329 POOLIDS = U.staticFieldOffset(poolIdsField); 4330 } catch (NoSuchFieldException e) { 4331 throw new ExceptionInInitializerError(e); 4332 } 4333 CTL = U.objectFieldOffset(klass, "ctl"); 4334 RUNSTATE = U.objectFieldOffset(klass, "runState"); 4335 PARALLELISM = U.objectFieldOffset(klass, "parallelism"); 4336 THREADIDS = U.objectFieldOffset(klass, "threadIds"); 4337 TERMINATION = U.objectFieldOffset(klass, "termination"); 4338 Class<ForkJoinTask[]> aklass = ForkJoinTask[].class; 4339 ABASE = U.arrayBaseOffset(aklass); 4340 int scale = U.arrayIndexScale(aklass); 4341 ASHIFT = 31 - Integer.numberOfLeadingZeros(scale); 4342 if ((scale & (scale - 1)) != 0) 4343 throw new Error("array index scale not a power of two"); 4344 4345 Class<?> dep = LockSupport.class; // ensure loaded 4346 // allow access to non-public methods 4347 JLA = SharedSecrets.getJavaLangAccess(); 4348 SharedSecrets.setJavaUtilConcurrentFJPAccess( 4349 new JavaUtilConcurrentFJPAccess() { 4350 @Override 4351 public long beginCompensatedBlock(ForkJoinPool pool) { 4352 return pool.beginCompensatedBlock(); 4353 } 4354 public void endCompensatedBlock(ForkJoinPool pool, long post) { 4355 pool.endCompensatedBlock(post); 4356 } 4357 }); 4358 defaultForkJoinWorkerThreadFactory = 4359 new DefaultForkJoinWorkerThreadFactory(); 4360 common = new ForkJoinPool((byte)0); 4361 } 4362 }