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