ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/ForkJoinPool.java
Revision: 1.363
Committed: Fri Jan 17 20:39:42 2020 UTC (4 years, 4 months ago) by jsr166
Branch: MAIN
Changes since 1.362: +1 -1 lines
Log Message:
typo

File Contents

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