ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/ForkJoinPool.java
Revision: 1.364
Committed: Sat Jan 18 12:30:04 2020 UTC (4 years, 4 months ago) by dl
Branch: MAIN
Changes since 1.363: +14 -15 lines
Log Message:
more consistent parallelism-0 handling; doc touchups

File Contents

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