ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jsr166y/ForkJoinPool.java
Revision: 1.70
Committed: Sat Sep 4 11:33:53 2010 UTC (13 years, 8 months ago) by dl
Branch: MAIN
Changes since 1.69: +0 -1 lines
Log Message:
Sync with j.u.c versions

File Contents

# User Rev Content
1 dl 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     * http://creativecommons.org/licenses/publicdomain
5     */
6    
7     package jsr166y;
8 jsr166 1.22
9 dl 1.1 import java.util.concurrent.*;
10 jsr166 1.22 import java.util.ArrayList;
11     import java.util.Arrays;
12     import java.util.Collection;
13     import java.util.Collections;
14     import java.util.List;
15     import java.util.concurrent.locks.LockSupport;
16     import java.util.concurrent.locks.ReentrantLock;
17     import java.util.concurrent.atomic.AtomicInteger;
18 dl 1.53 import java.util.concurrent.CountDownLatch;
19 dl 1.1
20     /**
21 jsr166 1.29 * An {@link ExecutorService} for running {@link ForkJoinTask}s.
22 jsr166 1.39 * A {@code ForkJoinPool} provides the entry point for submissions
23 dl 1.57 * from non-{@code ForkJoinTask} clients, as well as management and
24 jsr166 1.48 * monitoring operations.
25 dl 1.1 *
26 dl 1.42 * <p>A {@code ForkJoinPool} differs from other kinds of {@link
27     * ExecutorService} mainly by virtue of employing
28     * <em>work-stealing</em>: all threads in the pool attempt to find and
29     * execute subtasks created by other active tasks (eventually blocking
30     * waiting for work if none exist). This enables efficient processing
31     * when most tasks spawn other subtasks (as do most {@code
32 dl 1.57 * ForkJoinTask}s). When setting <em>asyncMode</em> to true in
33     * constructors, {@code ForkJoinPool}s may also be appropriate for use
34     * with event-style tasks that are never joined.
35 dl 1.1 *
36 dl 1.42 * <p>A {@code ForkJoinPool} is constructed with a given target
37     * parallelism level; by default, equal to the number of available
38 dl 1.57 * processors. The pool attempts to maintain enough active (or
39     * available) threads by dynamically adding, suspending, or resuming
40     * internal worker threads, even if some tasks are stalled waiting to
41     * join others. However, no such adjustments are guaranteed in the
42     * face of blocked IO or other unmanaged synchronization. The nested
43     * {@link ManagedBlocker} interface enables extension of the kinds of
44     * synchronization accommodated.
45 dl 1.1 *
46     * <p>In addition to execution and lifecycle control methods, this
47     * class provides status check methods (for example
48 jsr166 1.29 * {@link #getStealCount}) that are intended to aid in developing,
49 dl 1.1 * tuning, and monitoring fork/join applications. Also, method
50 jsr166 1.29 * {@link #toString} returns indications of pool state in a
51 dl 1.2 * convenient form for informal monitoring.
52 dl 1.1 *
53 dl 1.57 * <p> As is the case with other ExecutorServices, there are three
54 dl 1.60 * main task execution methods summarized in the following
55 dl 1.57 * table. These are designed to be used by clients not already engaged
56     * in fork/join computations in the current pool. The main forms of
57     * these methods accept instances of {@code ForkJoinTask}, but
58     * overloaded forms also allow mixed execution of plain {@code
59     * Runnable}- or {@code Callable}- based activities as well. However,
60     * tasks that are already executing in a pool should normally
61     * <em>NOT</em> use these pool execution methods, but instead use the
62 dl 1.59 * within-computation forms listed in the table.
63 dl 1.57 *
64     * <table BORDER CELLPADDING=3 CELLSPACING=1>
65     * <tr>
66     * <td></td>
67     * <td ALIGN=CENTER> <b>Call from non-fork/join clients</b></td>
68     * <td ALIGN=CENTER> <b>Call from within fork/join computations</b></td>
69     * </tr>
70     * <tr>
71 jsr166 1.67 * <td> <b>Arrange async execution</td>
72 dl 1.57 * <td> {@link #execute(ForkJoinTask)}</td>
73     * <td> {@link ForkJoinTask#fork}</td>
74     * </tr>
75     * <tr>
76     * <td> <b>Await and obtain result</td>
77     * <td> {@link #invoke(ForkJoinTask)}</td>
78     * <td> {@link ForkJoinTask#invoke}</td>
79     * </tr>
80     * <tr>
81     * <td> <b>Arrange exec and obtain Future</td>
82     * <td> {@link #submit(ForkJoinTask)}</td>
83     * <td> {@link ForkJoinTask#fork} (ForkJoinTasks <em>are</em> Futures)</td>
84     * </tr>
85     * </table>
86 dl 1.59 *
87 dl 1.42 * <p><b>Sample Usage.</b> Normally a single {@code ForkJoinPool} is
88     * used for all parallel task execution in a program or subsystem.
89     * Otherwise, use would not usually outweigh the construction and
90     * bookkeeping overhead of creating a large set of threads. For
91 jsr166 1.43 * example, a common pool could be used for the {@code SortTasks}
92 dl 1.42 * illustrated in {@link RecursiveAction}. Because {@code
93     * ForkJoinPool} uses threads in {@linkplain java.lang.Thread#isDaemon
94 jsr166 1.43 * daemon} mode, there is typically no need to explicitly {@link
95 dl 1.42 * #shutdown} such a pool upon program exit.
96     *
97     * <pre>
98     * static final ForkJoinPool mainPool = new ForkJoinPool();
99     * ...
100     * public void sort(long[] array) {
101     * mainPool.invoke(new SortTask(array, 0, array.length));
102     * }
103     * </pre>
104     *
105 dl 1.1 * <p><b>Implementation notes</b>: This implementation restricts the
106 dl 1.2 * maximum number of running threads to 32767. Attempts to create
107 jsr166 1.48 * pools with greater than the maximum number result in
108 jsr166 1.39 * {@code IllegalArgumentException}.
109 jsr166 1.16 *
110 jsr166 1.48 * <p>This implementation rejects submitted tasks (that is, by throwing
111 dl 1.58 * {@link RejectedExecutionException}) only when the pool is shut down
112 dl 1.62 * or internal resources have been exhausted.
113 jsr166 1.48 *
114 jsr166 1.16 * @since 1.7
115     * @author Doug Lea
116 dl 1.1 */
117 dl 1.2 public class ForkJoinPool extends AbstractExecutorService {
118 dl 1.1
119     /*
120 dl 1.53 * Implementation Overview
121     *
122     * This class provides the central bookkeeping and control for a
123     * set of worker threads: Submissions from non-FJ threads enter
124     * into a submission queue. Workers take these tasks and typically
125     * split them into subtasks that may be stolen by other workers.
126     * The main work-stealing mechanics implemented in class
127     * ForkJoinWorkerThread give first priority to processing tasks
128     * from their own queues (LIFO or FIFO, depending on mode), then
129     * to randomized FIFO steals of tasks in other worker queues, and
130     * lastly to new submissions. These mechanics do not consider
131     * affinities, loads, cache localities, etc, so rarely provide the
132     * best possible performance on a given machine, but portably
133     * provide good throughput by averaging over these factors.
134     * (Further, even if we did try to use such information, we do not
135     * usually have a basis for exploiting it. For example, some sets
136     * of tasks profit from cache affinities, but others are harmed by
137     * cache pollution effects.)
138     *
139 dl 1.58 * Beyond work-stealing support and essential bookkeeping, the
140 dl 1.60 * main responsibility of this framework is to take actions when
141     * one worker is waiting to join a task stolen (or always held by)
142 jsr166 1.67 * another. Because we are multiplexing many tasks on to a pool
143 dl 1.60 * of workers, we can't just let them block (as in Thread.join).
144     * We also cannot just reassign the joiner's run-time stack with
145     * another and replace it later, which would be a form of
146     * "continuation", that even if possible is not necessarily a good
147     * idea. Given that the creation costs of most threads on most
148     * systems mainly surrounds setting up runtime stacks, thread
149     * creation and switching is usually not much more expensive than
150     * stack creation and switching, and is more flexible). Instead we
151     * combine two tactics:
152 dl 1.58 *
153 dl 1.60 * Helping: Arranging for the joiner to execute some task that it
154 dl 1.58 * would be running if the steal had not occurred. Method
155     * ForkJoinWorkerThread.helpJoinTask tracks joining->stealing
156     * links to try to find such a task.
157     *
158 dl 1.61 * Compensating: Unless there are already enough live threads,
159 jsr166 1.68 * method helpMaintainParallelism() may create or
160 dl 1.61 * re-activate a spare thread to compensate for blocked
161     * joiners until they unblock.
162 dl 1.58 *
163 dl 1.66 * It is impossible to keep exactly the target (parallelism)
164     * number of threads running at any given time. Determining
165     * existence of conservatively safe helping targets, the
166     * availability of already-created spares, and the apparent need
167     * to create new spares are all racy and require heuristic
168     * guidance, so we rely on multiple retries of each. Compensation
169     * occurs in slow-motion. It is triggered only upon timeouts of
170     * Object.wait used for joins. This reduces poor decisions that
171     * would otherwise be made when threads are waiting for others
172     * that are stalled because of unrelated activities such as
173     * garbage collection.
174 dl 1.58 *
175 dl 1.61 * The ManagedBlocker extension API can't use helping so relies
176     * only on compensation in method awaitBlocker.
177 dl 1.58 *
178 dl 1.53 * The main throughput advantages of work-stealing stem from
179     * decentralized control -- workers mostly steal tasks from each
180     * other. We do not want to negate this by creating bottlenecks
181 dl 1.58 * implementing other management responsibilities. So we use a
182     * collection of techniques that avoid, reduce, or cope well with
183     * contention. These entail several instances of bit-packing into
184     * CASable fields to maintain only the minimally required
185     * atomicity. To enable such packing, we restrict maximum
186     * parallelism to (1<<15)-1 (enabling twice this (to accommodate
187     * unbalanced increments and decrements) to fit into a 16 bit
188     * field, which is far in excess of normal operating range. Even
189     * though updates to some of these bookkeeping fields do sometimes
190     * contend with each other, they don't normally cache-contend with
191     * updates to others enough to warrant memory padding or
192     * isolation. So they are all held as fields of ForkJoinPool
193     * objects. The main capabilities are as follows:
194 dl 1.53 *
195     * 1. Creating and removing workers. Workers are recorded in the
196     * "workers" array. This is an array as opposed to some other data
197     * structure to support index-based random steals by workers.
198     * Updates to the array recording new workers and unrecording
199     * terminated ones are protected from each other by a lock
200     * (workerLock) but the array is otherwise concurrently readable,
201     * and accessed directly by workers. To simplify index-based
202     * operations, the array size is always a power of two, and all
203 dl 1.56 * readers must tolerate null slots. Currently, all worker thread
204     * creation is on-demand, triggered by task submissions,
205     * replacement of terminated workers, and/or compensation for
206     * blocked workers. However, all other support code is set up to
207     * work with other policies.
208 dl 1.53 *
209 dl 1.61 * To ensure that we do not hold on to worker references that
210     * would prevent GC, ALL accesses to workers are via indices into
211     * the workers array (which is one source of some of the unusual
212     * code constructions here). In essence, the workers array serves
213     * as a WeakReference mechanism. Thus for example the event queue
214     * stores worker indices, not worker references. Access to the
215     * workers in associated methods (for example releaseEventWaiters)
216     * must both index-check and null-check the IDs. All such accesses
217     * ignore bad IDs by returning out early from what they are doing,
218     * since this can only be associated with shutdown, in which case
219     * it is OK to give up. On termination, we just clobber these
220     * data structures without trying to use them.
221     *
222 dl 1.53 * 2. Bookkeeping for dynamically adding and removing workers. We
223 dl 1.57 * aim to approximately maintain the given level of parallelism.
224     * When some workers are known to be blocked (on joins or via
225 dl 1.53 * ManagedBlocker), we may create or resume others to take their
226     * place until they unblock (see below). Implementing this
227     * requires counts of the number of "running" threads (i.e., those
228 jsr166 1.67 * that are neither blocked nor artificially suspended) as well as
229 dl 1.53 * the total number. These two values are packed into one field,
230     * "workerCounts" because we need accurate snapshots when deciding
231 dl 1.58 * to create, resume or suspend. Note however that the
232 jsr166 1.67 * correspondence of these counts to reality is not guaranteed. In
233 dl 1.58 * particular updates for unblocked threads may lag until they
234     * actually wake up.
235 dl 1.53 *
236     * 3. Maintaining global run state. The run state of the pool
237     * consists of a runLevel (SHUTDOWN, TERMINATING, etc) similar to
238     * those in other Executor implementations, as well as a count of
239     * "active" workers -- those that are, or soon will be, or
240     * recently were executing tasks. The runLevel and active count
241     * are packed together in order to correctly trigger shutdown and
242     * termination. Without care, active counts can be subject to very
243     * high contention. We substantially reduce this contention by
244     * relaxing update rules. A worker must claim active status
245     * prospectively, by activating if it sees that a submitted or
246     * stealable task exists (it may find after activating that the
247     * task no longer exists). It stays active while processing this
248     * task (if it exists) and any other local subtasks it produces,
249     * until it cannot find any other tasks. It then tries
250     * inactivating (see method preStep), but upon update contention
251     * instead scans for more tasks, later retrying inactivation if it
252     * doesn't find any.
253     *
254     * 4. Managing idle workers waiting for tasks. We cannot let
255     * workers spin indefinitely scanning for tasks when none are
256     * available. On the other hand, we must quickly prod them into
257     * action when new tasks are submitted or generated. We
258     * park/unpark these idle workers using an event-count scheme.
259     * Field eventCount is incremented upon events that may enable
260     * workers that previously could not find a task to now find one:
261     * Submission of a new task to the pool, or another worker pushing
262     * a task onto a previously empty queue. (We also use this
263 dl 1.64 * mechanism for configuration and termination actions that
264     * require wakeups of idle workers). Each worker maintains its
265     * last known event count, and blocks when a scan for work did not
266     * find a task AND its lastEventCount matches the current
267     * eventCount. Waiting idle workers are recorded in a variant of
268     * Treiber stack headed by field eventWaiters which, when nonzero,
269     * encodes the thread index and count awaited for by the worker
270     * thread most recently calling eventSync. This thread in turn has
271     * a record (field nextEventWaiter) for the next waiting worker.
272     * In addition to allowing simpler decisions about need for
273     * wakeup, the event count bits in eventWaiters serve the role of
274     * tags to avoid ABA errors in Treiber stacks. Upon any wakeup,
275 dl 1.66 * released threads also try to release at most two others. The
276     * net effect is a tree-like diffusion of signals, where released
277     * threads (and possibly others) help with unparks. To further
278     * reduce contention effects a bit, failed CASes to increment
279     * field eventCount are tolerated without retries in signalWork.
280 dl 1.53 * Conceptually they are merged into the same event, which is OK
281     * when their only purpose is to enable workers to scan for work.
282     *
283 dl 1.66 * 5. Managing suspension of extra workers. When a worker notices
284     * (usually upon timeout of a wait()) that there are too few
285     * running threads, we may create a new thread to maintain
286     * parallelism level, or at least avoid starvation. Usually, extra
287     * threads are needed for only very short periods, yet join
288     * dependencies are such that we sometimes need them in
289     * bursts. Rather than create new threads each time this happens,
290     * we suspend no-longer-needed extra ones as "spares". For most
291     * purposes, we don't distinguish "extra" spare threads from
292     * normal "core" threads: On each call to preStep (the only point
293     * at which we can do this) a worker checks to see if there are
294     * now too many running workers, and if so, suspends itself.
295     * Method helpMaintainParallelism looks for suspended threads to
296     * resume before considering creating a new replacement. The
297     * spares themselves are encoded on another variant of a Treiber
298     * Stack, headed at field "spareWaiters". Note that the use of
299     * spares is intrinsically racy. One thread may become a spare at
300     * about the same time as another is needlessly being created. We
301     * counteract this and related slop in part by requiring resumed
302     * spares to immediately recheck (in preStep) to see whether they
303     * they should re-suspend.
304     *
305     * 6. Killing off unneeded workers. A timeout mechanism is used to
306     * shed unused workers: The oldest (first) event queue waiter uses
307     * a timed rather than hard wait. When this wait times out without
308     * a normal wakeup, it tries to shutdown any one (for convenience
309     * the newest) other spare or event waiter via
310     * tryShutdownUnusedWorker. This eventually reduces the number of
311     * worker threads to a minimum of one after a long enough period
312     * without use.
313 dl 1.64 *
314     * 7. Deciding when to create new workers. The main dynamic
315 dl 1.61 * control in this class is deciding when to create extra threads
316     * in method helpMaintainParallelism. We would like to keep
317 jsr166 1.67 * exactly #parallelism threads running, which is an impossible
318 dl 1.61 * task. We always need to create one when the number of running
319     * threads would become zero and all workers are busy. Beyond
320 jsr166 1.68 * this, we must rely on heuristics that work well in the
321     * presence of transient phenomena such as GC stalls, dynamic
322 dl 1.61 * compilation, and wake-up lags. These transients are extremely
323     * common -- we are normally trying to fully saturate the CPUs on
324     * a machine, so almost any activity other than running tasks
325 dl 1.66 * impedes accuracy. Our main defense is to allow parallelism to
326     * lapse for a while during joins, and use a timeout to see if,
327     * after the resulting settling, there is still a need for
328     * additional workers. This also better copes with the fact that
329     * some of the methods in this class tend to never become compiled
330     * (but are interpreted), so some components of the entire set of
331     * controls might execute 100 times faster than others. And
332     * similarly for cases where the apparent lack of work is just due
333     * to GC stalls and other transient system activity.
334 dl 1.53 *
335     * Beware that there is a lot of representation-level coupling
336     * among classes ForkJoinPool, ForkJoinWorkerThread, and
337     * ForkJoinTask. For example, direct access to "workers" array by
338     * workers, and direct access to ForkJoinTask.status by both
339     * ForkJoinPool and ForkJoinWorkerThread. There is little point
340     * trying to reduce this, since any associated future changes in
341     * representations will need to be accompanied by algorithmic
342     * changes anyway.
343     *
344     * Style notes: There are lots of inline assignments (of form
345     * "while ((local = field) != 0)") which are usually the simplest
346 dl 1.61 * way to ensure the required read orderings (which are sometimes
347     * critical). Also several occurrences of the unusual "do {}
348 jsr166 1.69 * while (!cas...)" which is the simplest way to force an update of
349 dl 1.61 * a CAS'ed variable. There are also other coding oddities that
350     * help some methods perform reasonably even when interpreted (not
351     * compiled), at the expense of some messy constructions that
352     * reduce byte code counts.
353 dl 1.53 *
354     * The order of declarations in this file is: (1) statics (2)
355     * fields (along with constants used when unpacking some of them)
356     * (3) internal control methods (4) callbacks and other support
357     * for ForkJoinTask and ForkJoinWorkerThread classes, (5) exported
358     * methods (plus a few little helpers).
359 dl 1.1 */
360    
361     /**
362 jsr166 1.35 * Factory for creating new {@link ForkJoinWorkerThread}s.
363     * A {@code ForkJoinWorkerThreadFactory} must be defined and used
364     * for {@code ForkJoinWorkerThread} subclasses that extend base
365     * functionality or initialize threads with different contexts.
366 dl 1.1 */
367     public static interface ForkJoinWorkerThreadFactory {
368     /**
369     * Returns a new worker thread operating in the given pool.
370     *
371     * @param pool the pool this thread works in
372 jsr166 1.48 * @throws NullPointerException if the pool is null
373 dl 1.1 */
374     public ForkJoinWorkerThread newThread(ForkJoinPool pool);
375     }
376    
377     /**
378 jsr166 1.17 * Default ForkJoinWorkerThreadFactory implementation; creates a
379 dl 1.1 * new ForkJoinWorkerThread.
380     */
381 dl 1.57 static class DefaultForkJoinWorkerThreadFactory
382 dl 1.1 implements ForkJoinWorkerThreadFactory {
383     public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
384 dl 1.53 return new ForkJoinWorkerThread(pool);
385 dl 1.1 }
386     }
387    
388     /**
389 dl 1.2 * Creates a new ForkJoinWorkerThread. This factory is used unless
390     * overridden in ForkJoinPool constructors.
391 dl 1.1 */
392 dl 1.2 public static final ForkJoinWorkerThreadFactory
393 dl 1.1 defaultForkJoinWorkerThreadFactory =
394     new DefaultForkJoinWorkerThreadFactory();
395    
396     /**
397     * Permission required for callers of methods that may start or
398     * kill threads.
399     */
400     private static final RuntimePermission modifyThreadPermission =
401     new RuntimePermission("modifyThread");
402    
403     /**
404     * If there is a security manager, makes sure caller has
405     * permission to modify threads.
406     */
407     private static void checkPermission() {
408     SecurityManager security = System.getSecurityManager();
409     if (security != null)
410     security.checkPermission(modifyThreadPermission);
411     }
412    
413     /**
414     * Generator for assigning sequence numbers as pool names.
415     */
416     private static final AtomicInteger poolNumberGenerator =
417     new AtomicInteger();
418    
419     /**
420 dl 1.66 * The time to block in a join (see awaitJoin) before checking if
421     * a new worker should be (re)started to maintain parallelism
422 jsr166 1.67 * level. The value should be short enough to maintain global
423 dl 1.66 * responsiveness and progress but long enough to avoid
424     * counterproductive firings during GC stalls or unrelated system
425     * activity, and to not bog down systems with continual re-firings
426     * on GCs or legitimately long waits.
427     */
428     private static final long JOIN_TIMEOUT_MILLIS = 250L; // 4 per second
429    
430     /**
431 dl 1.64 * The wakeup interval (in nanoseconds) for the oldest worker
432 dl 1.66 * worker waiting for an event invokes tryShutdownUnusedWorker to shrink
433 dl 1.64 * the number of workers. The exact value does not matter too
434     * much, but should be long enough to slowly release resources
435     * during long periods without use without disrupting normal use.
436     */
437     private static final long SHRINK_RATE_NANOS =
438 dl 1.66 30L * 1000L * 1000L * 1000L; // 2 per minute
439 dl 1.64
440     /**
441 dl 1.61 * Absolute bound for parallelism level. Twice this number plus
442     * one (i.e., 0xfff) must fit into a 16bit field to enable
443     * word-packing for some counts and indices.
444 dl 1.53 */
445 dl 1.61 private static final int MAX_WORKERS = 0x7fff;
446 dl 1.53
447     /**
448     * Array holding all worker threads in the pool. Array size must
449     * be a power of two. Updates and replacements are protected by
450     * workerLock, but the array is always kept in a consistent enough
451     * state to be randomly accessed without locking by workers
452     * performing work-stealing, as well as other traversal-based
453     * methods in this class. All readers must tolerate that some
454     * array slots may be null.
455 dl 1.1 */
456     volatile ForkJoinWorkerThread[] workers;
457    
458     /**
459 dl 1.53 * Queue for external submissions.
460 dl 1.1 */
461 dl 1.53 private final LinkedTransferQueue<ForkJoinTask<?>> submissionQueue;
462 dl 1.1
463     /**
464 dl 1.53 * Lock protecting updates to workers array.
465 dl 1.1 */
466 dl 1.53 private final ReentrantLock workerLock;
467 dl 1.1
468     /**
469 dl 1.53 * Latch released upon termination.
470 dl 1.1 */
471 dl 1.57 private final Phaser termination;
472 dl 1.1
473     /**
474     * Creation factory for worker threads.
475     */
476     private final ForkJoinWorkerThreadFactory factory;
477    
478     /**
479 dl 1.53 * Sum of per-thread steal counts, updated only when threads are
480     * idle or terminating.
481 dl 1.1 */
482 dl 1.53 private volatile long stealCount;
483 dl 1.1
484     /**
485 jsr166 1.67 * Encoded record of top of Treiber stack of threads waiting for
486 dl 1.53 * events. The top 32 bits contain the count being waited for. The
487 dl 1.61 * bottom 16 bits contains one plus the pool index of waiting
488     * worker thread. (Bits 16-31 are unused.)
489 dl 1.1 */
490 dl 1.53 private volatile long eventWaiters;
491    
492     private static final int EVENT_COUNT_SHIFT = 32;
493 dl 1.61 private static final long WAITER_ID_MASK = (1L << 16) - 1L;
494 dl 1.1
495     /**
496 dl 1.53 * A counter for events that may wake up worker threads:
497     * - Submission of a new task to the pool
498     * - A worker pushing a task on an empty queue
499 dl 1.61 * - termination
500 dl 1.1 */
501 dl 1.53 private volatile int eventCount;
502    
503     /**
504 jsr166 1.67 * Encoded record of top of Treiber stack of spare threads waiting
505 dl 1.61 * for resumption. The top 16 bits contain an arbitrary count to
506     * avoid ABA effects. The bottom 16bits contains one plus the pool
507     * index of waiting worker thread.
508     */
509     private volatile int spareWaiters;
510    
511     private static final int SPARE_COUNT_SHIFT = 16;
512     private static final int SPARE_ID_MASK = (1 << 16) - 1;
513    
514     /**
515 dl 1.53 * Lifecycle control. The low word contains the number of workers
516     * that are (probably) executing tasks. This value is atomically
517     * incremented before a worker gets a task to run, and decremented
518     * when worker has no tasks and cannot find any. Bits 16-18
519     * contain runLevel value. When all are zero, the pool is
520     * running. Level transitions are monotonic (running -> shutdown
521     * -> terminating -> terminated) so each transition adds a bit.
522     * These are bundled together to ensure consistent read for
523     * termination checks (i.e., that runLevel is at least SHUTDOWN
524     * and active threads is zero).
525 dl 1.64 *
526     * Notes: Most direct CASes are dependent on these bitfield
527     * positions. Also, this field is non-private to enable direct
528     * performance-sensitive CASes in ForkJoinWorkerThread.
529 dl 1.53 */
530 dl 1.64 volatile int runState;
531 dl 1.53
532     // Note: The order among run level values matters.
533     private static final int RUNLEVEL_SHIFT = 16;
534     private static final int SHUTDOWN = 1 << RUNLEVEL_SHIFT;
535     private static final int TERMINATING = 1 << (RUNLEVEL_SHIFT + 1);
536     private static final int TERMINATED = 1 << (RUNLEVEL_SHIFT + 2);
537     private static final int ACTIVE_COUNT_MASK = (1 << RUNLEVEL_SHIFT) - 1;
538 dl 1.1
539     /**
540 dl 1.53 * Holds number of total (i.e., created and not yet terminated)
541     * and running (i.e., not blocked on joins or other managed sync)
542     * threads, packed together to ensure consistent snapshot when
543     * making decisions about creating and suspending spare
544     * threads. Updated only by CAS. Note that adding a new worker
545     * requires incrementing both counts, since workers start off in
546 dl 1.60 * running state.
547 dl 1.53 */
548     private volatile int workerCounts;
549    
550     private static final int TOTAL_COUNT_SHIFT = 16;
551     private static final int RUNNING_COUNT_MASK = (1 << TOTAL_COUNT_SHIFT) - 1;
552     private static final int ONE_RUNNING = 1;
553     private static final int ONE_TOTAL = 1 << TOTAL_COUNT_SHIFT;
554    
555 dl 1.1 /**
556 dl 1.53 * The target parallelism level.
557 dl 1.57 * Accessed directly by ForkJoinWorkerThreads.
558 dl 1.1 */
559 dl 1.57 final int parallelism;
560 dl 1.1
561     /**
562 dl 1.53 * True if use local fifo, not default lifo, for local polling
563 dl 1.57 * Read by, and replicated by ForkJoinWorkerThreads
564 dl 1.1 */
565 dl 1.57 final boolean locallyFifo;
566 dl 1.1
567     /**
568 dl 1.57 * The uncaught exception handler used when any worker abruptly
569     * terminates.
570 dl 1.1 */
571 dl 1.57 private final Thread.UncaughtExceptionHandler ueh;
572 dl 1.6
573     /**
574 dl 1.53 * Pool number, just for assigning useful names to worker threads
575 dl 1.1 */
576 dl 1.53 private final int poolNumber;
577 dl 1.1
578 dl 1.64 // Utilities for CASing fields. Note that most of these
579     // are usually manually inlined by callers
580 dl 1.1
581     /**
582 dl 1.61 * Increments running count part of workerCounts
583 dl 1.1 */
584 dl 1.57 final void incrementRunningCount() {
585     int c;
586 dl 1.53 do {} while (!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
587 dl 1.59 c = workerCounts,
588 dl 1.57 c + ONE_RUNNING));
589 dl 1.1 }
590 dl 1.58
591 dl 1.1 /**
592 dl 1.57 * Tries to decrement running count unless already zero
593 dl 1.56 */
594     final boolean tryDecrementRunningCount() {
595     int wc = workerCounts;
596     if ((wc & RUNNING_COUNT_MASK) == 0)
597     return false;
598     return UNSAFE.compareAndSwapInt(this, workerCountsOffset,
599     wc, wc - ONE_RUNNING);
600     }
601    
602     /**
603 dl 1.61 * Forces decrement of encoded workerCounts, awaiting nonzero if
604     * (rarely) necessary when other count updates lag.
605     *
606     * @param dr -- either zero or ONE_RUNNING
607     * @param dt == either zero or ONE_TOTAL
608 dl 1.58 */
609 dl 1.61 private void decrementWorkerCounts(int dr, int dt) {
610     for (;;) {
611     int wc = workerCounts;
612     if ((wc & RUNNING_COUNT_MASK) - dr < 0 ||
613 dl 1.64 (wc >>> TOTAL_COUNT_SHIFT) - dt < 0) {
614     if ((runState & TERMINATED) != 0)
615     return; // lagging termination on a backout
616 dl 1.61 Thread.yield();
617 dl 1.64 }
618 dl 1.61 if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
619     wc, wc - (dr + dt)))
620     return;
621     }
622     }
623    
624     /**
625 jsr166 1.16 * Tries decrementing active count; fails on contention.
626 dl 1.53 * Called when workers cannot find tasks to run.
627     */
628     final boolean tryDecrementActiveCount() {
629     int c;
630     return UNSAFE.compareAndSwapInt(this, runStateOffset,
631 dl 1.64 c = runState, c - 1);
632 dl 1.53 }
633    
634     /**
635     * Advances to at least the given level. Returns true if not
636     * already in at least the given level.
637     */
638     private boolean advanceRunLevel(int level) {
639     for (;;) {
640     int s = runState;
641     if ((s & level) != 0)
642     return false;
643     if (UNSAFE.compareAndSwapInt(this, runStateOffset, s, s | level))
644     return true;
645     }
646     }
647    
648     // workers array maintenance
649    
650     /**
651     * Records and returns a workers array index for new worker.
652     */
653     private int recordWorker(ForkJoinWorkerThread w) {
654     // Try using slot totalCount-1. If not available, scan and/or resize
655     int k = (workerCounts >>> TOTAL_COUNT_SHIFT) - 1;
656     final ReentrantLock lock = this.workerLock;
657     lock.lock();
658     try {
659     ForkJoinWorkerThread[] ws = workers;
660 dl 1.61 int n = ws.length;
661     if (k < 0 || k >= n || ws[k] != null) {
662     for (k = 0; k < n && ws[k] != null; ++k)
663 dl 1.53 ;
664 dl 1.61 if (k == n)
665     ws = Arrays.copyOf(ws, n << 1);
666 dl 1.53 }
667     ws[k] = w;
668     workers = ws; // volatile array write ensures slot visibility
669     } finally {
670     lock.unlock();
671     }
672     return k;
673     }
674    
675     /**
676     * Nulls out record of worker in workers array
677     */
678     private void forgetWorker(ForkJoinWorkerThread w) {
679     int idx = w.poolIndex;
680 jsr166 1.67 // Locking helps method recordWorker avoid unnecessary expansion
681 dl 1.53 final ReentrantLock lock = this.workerLock;
682     lock.lock();
683     try {
684     ForkJoinWorkerThread[] ws = workers;
685     if (idx >= 0 && idx < ws.length && ws[idx] == w) // verify
686     ws[idx] = null;
687     } finally {
688     lock.unlock();
689     }
690     }
691    
692     /**
693     * Final callback from terminating worker. Removes record of
694     * worker from array, and adjusts counts. If pool is shutting
695 jsr166 1.67 * down, tries to complete termination.
696 dl 1.53 *
697     * @param w the worker
698     */
699     final void workerTerminated(ForkJoinWorkerThread w) {
700     forgetWorker(w);
701 dl 1.61 decrementWorkerCounts(w.isTrimmed()? 0 : ONE_RUNNING, ONE_TOTAL);
702     while (w.stealCount != 0) // collect final count
703     tryAccumulateStealCount(w);
704     tryTerminate(false);
705 dl 1.53 }
706    
707     // Waiting for and signalling events
708    
709     /**
710     * Releases workers blocked on a count not equal to current count.
711 dl 1.61 * Normally called after precheck that eventWaiters isn't zero to
712 dl 1.64 * avoid wasted array checks. Gives up upon a change in count or
713 dl 1.66 * upon releasing two workers, letting others take over.
714 dl 1.53 */
715 dl 1.64 private void releaseEventWaiters() {
716 dl 1.61 ForkJoinWorkerThread[] ws = workers;
717     int n = ws.length;
718 dl 1.64 long h = eventWaiters;
719     int ec = eventCount;
720 dl 1.66 boolean releasedOne = false;
721 dl 1.64 ForkJoinWorkerThread w; int id;
722 dl 1.66 while ((id = ((int)(h & WAITER_ID_MASK)) - 1) >= 0 &&
723     (int)(h >>> EVENT_COUNT_SHIFT) != ec &&
724     id < n && (w = ws[id]) != null) {
725     if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
726     h, w.nextWaiter)) {
727     LockSupport.unpark(w);
728     if (releasedOne) // exit on second release
729     break;
730     releasedOne = true;
731     }
732     if (eventCount != ec)
733 dl 1.61 break;
734 dl 1.66 h = eventWaiters;
735 dl 1.53 }
736     }
737    
738     /**
739 dl 1.61 * Tries to advance eventCount and releases waiters. Called only
740     * from workers.
741 dl 1.53 */
742     final void signalWork() {
743 dl 1.61 int c; // try to increment event count -- CAS failure OK
744     UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
745     if (eventWaiters != 0L)
746 dl 1.64 releaseEventWaiters();
747 dl 1.53 }
748    
749     /**
750 dl 1.64 * Adds the given worker to event queue and blocks until
751 dl 1.66 * terminating or event count advances from the given value
752 dl 1.53 *
753     * @param w the calling worker thread
754 dl 1.66 * @param ec the count
755 dl 1.53 */
756 dl 1.66 private void eventSync(ForkJoinWorkerThread w, int ec) {
757 dl 1.64 long nh = (((long)ec) << EVENT_COUNT_SHIFT) | ((long)(w.poolIndex+1));
758 dl 1.61 long h;
759 dl 1.60 while ((runState < SHUTDOWN || !tryTerminate(false)) &&
760 dl 1.64 (((int)((h = eventWaiters) & WAITER_ID_MASK)) == 0 ||
761     (int)(h >>> EVENT_COUNT_SHIFT) == ec) &&
762     eventCount == ec) {
763 dl 1.60 if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
764 dl 1.61 w.nextWaiter = h, nh)) {
765 dl 1.64 awaitEvent(w, ec);
766     break;
767     }
768     }
769     }
770    
771     /**
772     * Blocks the given worker (that has already been entered as an
773     * event waiter) until terminating or event count advances from
774     * the given value. The oldest (first) waiter uses a timed wait to
775     * occasionally one-by-one shrink the number of workers (to a
776 dl 1.66 * minimum of one) if the pool has not been used for extended
777 dl 1.64 * periods.
778     *
779     * @param w the calling worker thread
780     * @param ec the count
781     */
782     private void awaitEvent(ForkJoinWorkerThread w, int ec) {
783     while (eventCount == ec) {
784     if (tryAccumulateStealCount(w)) { // transfer while idle
785     boolean untimed = (w.nextWaiter != 0L ||
786     (workerCounts & RUNNING_COUNT_MASK) <= 1);
787     long startTime = untimed? 0 : System.nanoTime();
788     Thread.interrupted(); // clear/ignore interrupt
789 dl 1.66 if (eventCount != ec || w.runState != 0 ||
790 dl 1.64 runState >= TERMINATING) // recheck after clear
791     break;
792     if (untimed)
793     LockSupport.park(w);
794     else {
795     LockSupport.parkNanos(w, SHRINK_RATE_NANOS);
796 dl 1.66 if (eventCount != ec || w.runState != 0 ||
797 dl 1.64 runState >= TERMINATING)
798 dl 1.61 break;
799 dl 1.64 if (System.nanoTime() - startTime >= SHRINK_RATE_NANOS)
800 dl 1.66 tryShutdownUnusedWorker(ec);
801 dl 1.61 }
802 dl 1.53 }
803     }
804 dl 1.64 }
805    
806 dl 1.66 // Maintaining parallelism
807 dl 1.61
808     /**
809     * Pushes worker onto the spare stack
810     */
811     final void pushSpare(ForkJoinWorkerThread w) {
812 dl 1.64 int ns = (++w.spareCount << SPARE_COUNT_SHIFT) | (w.poolIndex + 1);
813 dl 1.61 do {} while (!UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
814     w.nextSpare = spareWaiters,ns));
815     }
816    
817     /**
818 dl 1.66 * Tries (once) to resume a spare if the number of running
819     * threads is less than target.
820 dl 1.61 */
821 dl 1.66 private void tryResumeSpare() {
822 dl 1.61 int sw, id;
823 dl 1.66 ForkJoinWorkerThread[] ws = workers;
824     int n = ws.length;
825 dl 1.61 ForkJoinWorkerThread w;
826 dl 1.66 if ((sw = spareWaiters) != 0 &&
827     (id = (sw & SPARE_ID_MASK) - 1) >= 0 &&
828     id < n && (w = ws[id]) != null &&
829     (workerCounts & RUNNING_COUNT_MASK) < parallelism &&
830     spareWaiters == sw &&
831 dl 1.61 UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
832 dl 1.64 sw, w.nextSpare)) {
833 dl 1.66 int c; // increment running count before resume
834 jsr166 1.69 do {} while (!UNSAFE.compareAndSwapInt
835     (this, workerCountsOffset,
836     c = workerCounts, c + ONE_RUNNING));
837 dl 1.66 if (w.tryUnsuspend())
838     LockSupport.unpark(w);
839     else // back out if w was shutdown
840     decrementWorkerCounts(ONE_RUNNING, 0);
841 dl 1.64 }
842     }
843    
844     /**
845 dl 1.66 * Tries to increase the number of running workers if below target
846     * parallelism: If a spare exists tries to resume it via
847     * tryResumeSpare. Otherwise, if not enough total workers or all
848 jsr166 1.67 * existing workers are busy, adds a new worker. In all cases also
849 dl 1.66 * helps wake up releasable workers waiting for work.
850 dl 1.64 */
851 dl 1.66 private void helpMaintainParallelism() {
852 dl 1.64 int pc = parallelism;
853 dl 1.66 int wc, rs, tc;
854     while (((wc = workerCounts) & RUNNING_COUNT_MASK) < pc &&
855     (rs = runState) < TERMINATING) {
856     if (spareWaiters != 0)
857     tryResumeSpare();
858     else if ((tc = wc >>> TOTAL_COUNT_SHIFT) >= MAX_WORKERS ||
859     (tc >= pc && (rs & ACTIVE_COUNT_MASK) != tc))
860     break; // enough total
861     else if (runState == rs && workerCounts == wc &&
862     UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
863     wc + (ONE_RUNNING|ONE_TOTAL))) {
864     ForkJoinWorkerThread w = null;
865     try {
866     w = factory.newThread(this);
867     } finally { // adjust on null or exceptional factory return
868     if (w == null) {
869     decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
870     tryTerminate(false); // handle failure during shutdown
871     }
872     }
873     if (w == null)
874 dl 1.64 break;
875 dl 1.66 w.start(recordWorker(w), ueh);
876     if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc) {
877     int c; // advance event count
878     UNSAFE.compareAndSwapInt(this, eventCountOffset,
879     c = eventCount, c+1);
880     break; // add at most one unless total below target
881     }
882 dl 1.64 }
883     }
884 dl 1.66 if (eventWaiters != 0L)
885     releaseEventWaiters();
886 dl 1.64 }
887    
888     /**
889 dl 1.66 * Callback from the oldest waiter in awaitEvent waking up after a
890     * period of non-use. If all workers are idle, tries (once) to
891     * shutdown an event waiter or a spare, if one exists. Note that
892     * we don't need CAS or locks here because the method is called
893     * only from one thread occasionally waking (and even misfires are
894     * OK). Note that until the shutdown worker fully terminates,
895     * workerCounts will overestimate total count, which is tolerable.
896 dl 1.64 *
897 dl 1.66 * @param ec the event count waited on by caller (to abort
898     * attempt if count has since changed).
899 dl 1.64 */
900 dl 1.66 private void tryShutdownUnusedWorker(int ec) {
901     if (runState == 0 && eventCount == ec) { // only trigger if all idle
902     ForkJoinWorkerThread[] ws = workers;
903     int n = ws.length;
904     ForkJoinWorkerThread w = null;
905     boolean shutdown = false;
906     int sw;
907     long h;
908     if ((sw = spareWaiters) != 0) { // prefer killing spares
909     int id = (sw & SPARE_ID_MASK) - 1;
910     if (id >= 0 && id < n && (w = ws[id]) != null &&
911     UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
912     sw, w.nextSpare))
913     shutdown = true;
914     }
915     else if ((h = eventWaiters) != 0L) {
916     long nh;
917     int id = ((int)(h & WAITER_ID_MASK)) - 1;
918     if (id >= 0 && id < n && (w = ws[id]) != null &&
919     (nh = w.nextWaiter) != 0L && // keep at least one worker
920     UNSAFE.compareAndSwapLong(this, eventWaitersOffset, h, nh))
921     shutdown = true;
922     }
923     if (w != null && shutdown) {
924     w.shutdown();
925     LockSupport.unpark(w);
926     }
927 dl 1.61 }
928 dl 1.66 releaseEventWaiters(); // in case of interference
929 dl 1.61 }
930    
931 dl 1.53 /**
932     * Callback from workers invoked upon each top-level action (i.e.,
933 dl 1.64 * stealing a task or taking a submission and running it).
934     * Performs one or more of the following:
935 dl 1.61 *
936 dl 1.66 * 1. If the worker is active and either did not run a task
937     * or there are too many workers, try to set its active status
938     * to inactive and update activeCount. On contention, we may
939     * try again in this or a subsequent call.
940     *
941     * 2. If not enough total workers, help create some.
942     *
943     * 3. If there are too many running workers, suspend this worker
944     * (first forcing inactive if necessary). If it is not needed,
945     * it may be shutdown while suspended (via
946     * tryShutdownUnusedWorker). Otherwise, upon resume it
947     * rechecks running thread count and need for event sync.
948     *
949     * 4. If worker did not run a task, await the next task event via
950     * eventSync if necessary (first forcing inactivation), upon
951     * which the worker may be shutdown via
952     * tryShutdownUnusedWorker. Otherwise, help release any
953     * existing event waiters that are now releasable,
954 dl 1.53 *
955     * @param w the worker
956 dl 1.66 * @param ran true if worker ran a task since last call to this method
957 dl 1.53 */
958 dl 1.66 final void preStep(ForkJoinWorkerThread w, boolean ran) {
959     int wec = w.lastEventCount;
960 dl 1.53 boolean active = w.active;
961 dl 1.66 boolean inactivate = false;
962 dl 1.61 int pc = parallelism;
963 dl 1.66 int rs;
964     while (w.runState == 0 && (rs = runState) < TERMINATING) {
965     if ((inactivate || (active && (rs & ACTIVE_COUNT_MASK) >= pc)) &&
966     UNSAFE.compareAndSwapInt(this, runStateOffset, rs, rs - 1))
967     inactivate = active = w.active = false;
968     int wc = workerCounts;
969     if ((wc & RUNNING_COUNT_MASK) > pc) {
970     if (!(inactivate |= active) && // must inactivate to suspend
971 dl 1.64 workerCounts == wc && // try to suspend as spare
972 dl 1.61 UNSAFE.compareAndSwapInt(this, workerCountsOffset,
973 dl 1.66 wc, wc - ONE_RUNNING))
974 dl 1.64 w.suspendAsSpare();
975 dl 1.61 }
976 dl 1.66 else if ((wc >>> TOTAL_COUNT_SHIFT) < pc)
977     helpMaintainParallelism(); // not enough workers
978     else if (!ran) {
979     long h = eventWaiters;
980     int ec = eventCount;
981     if (h != 0L && (int)(h >>> EVENT_COUNT_SHIFT) != ec)
982     releaseEventWaiters(); // release others before waiting
983     else if (ec != wec) {
984     w.lastEventCount = ec; // no need to wait
985     break;
986 dl 1.64 }
987 jsr166 1.68 else if (!(inactivate |= active))
988 dl 1.66 eventSync(w, wec); // must inactivate before sync
989 dl 1.53 }
990 dl 1.66 else
991     break;
992 dl 1.53 }
993     }
994    
995     /**
996 dl 1.61 * Helps and/or blocks awaiting join of the given task.
997 dl 1.66 * See above for explanation.
998 dl 1.56 *
999     * @param joinMe the task to join
1000 dl 1.66 * @param worker the current worker thread
1001 dl 1.53 */
1002 dl 1.61 final void awaitJoin(ForkJoinTask<?> joinMe, ForkJoinWorkerThread worker) {
1003 dl 1.66 int retries = 2 + (parallelism >> 2); // #helpJoins before blocking
1004 dl 1.61 while (joinMe.status >= 0) {
1005 dl 1.66 int wc;
1006 dl 1.61 worker.helpJoinTask(joinMe);
1007     if (joinMe.status < 0)
1008     break;
1009 dl 1.66 else if (retries > 0)
1010     --retries;
1011     else if (((wc = workerCounts) & RUNNING_COUNT_MASK) != 0 &&
1012     UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1013     wc, wc - ONE_RUNNING)) {
1014     int stat, c; long h;
1015     while ((stat = joinMe.status) >= 0 &&
1016     (h = eventWaiters) != 0L && // help release others
1017     (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
1018     releaseEventWaiters();
1019     if (stat >= 0 &&
1020     ((workerCounts & RUNNING_COUNT_MASK) == 0 ||
1021     (stat =
1022     joinMe.internalAwaitDone(JOIN_TIMEOUT_MILLIS)) >= 0))
1023     helpMaintainParallelism(); // timeout or no running workers
1024 dl 1.61 do {} while (!UNSAFE.compareAndSwapInt
1025     (this, workerCountsOffset,
1026     c = workerCounts, c + ONE_RUNNING));
1027 dl 1.66 if (stat < 0)
1028     break; // else restart
1029 dl 1.53 }
1030     }
1031     }
1032    
1033     /**
1034 dl 1.66 * Same idea as awaitJoin, but no helping, retries, or timeouts.
1035 dl 1.53 */
1036 dl 1.57 final void awaitBlocker(ManagedBlocker blocker)
1037 dl 1.53 throws InterruptedException {
1038 dl 1.61 while (!blocker.isReleasable()) {
1039 dl 1.66 int wc = workerCounts;
1040     if ((wc & RUNNING_COUNT_MASK) != 0 &&
1041     UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1042     wc, wc - ONE_RUNNING)) {
1043 dl 1.61 try {
1044 dl 1.66 while (!blocker.isReleasable()) {
1045     long h = eventWaiters;
1046     if (h != 0L &&
1047     (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
1048     releaseEventWaiters();
1049     else if ((workerCounts & RUNNING_COUNT_MASK) == 0 &&
1050     runState < TERMINATING)
1051     helpMaintainParallelism();
1052     else if (blocker.block())
1053     break;
1054     }
1055 dl 1.61 } finally {
1056     int c;
1057     do {} while (!UNSAFE.compareAndSwapInt
1058     (this, workerCountsOffset,
1059     c = workerCounts, c + ONE_RUNNING));
1060 dl 1.58 }
1061 dl 1.60 break;
1062     }
1063 dl 1.53 }
1064 dl 1.59 }
1065 dl 1.54
1066     /**
1067 dl 1.53 * Possibly initiates and/or completes termination.
1068     *
1069     * @param now if true, unconditionally terminate, else only
1070     * if shutdown and empty queue and no active workers
1071     * @return true if now terminating or terminated
1072 dl 1.1 */
1073 dl 1.53 private boolean tryTerminate(boolean now) {
1074     if (now)
1075     advanceRunLevel(SHUTDOWN); // ensure at least SHUTDOWN
1076     else if (runState < SHUTDOWN ||
1077     !submissionQueue.isEmpty() ||
1078     (runState & ACTIVE_COUNT_MASK) != 0)
1079 dl 1.4 return false;
1080 dl 1.53
1081     if (advanceRunLevel(TERMINATING))
1082     startTerminating();
1083    
1084     // Finish now if all threads terminated; else in some subsequent call
1085     if ((workerCounts >>> TOTAL_COUNT_SHIFT) == 0) {
1086     advanceRunLevel(TERMINATED);
1087 dl 1.57 termination.arrive();
1088 dl 1.53 }
1089 dl 1.4 return true;
1090 dl 1.1 }
1091    
1092     /**
1093 dl 1.53 * Actions on transition to TERMINATING
1094 dl 1.61 *
1095     * Runs up to four passes through workers: (0) shutting down each
1096 dl 1.64 * (without waking up if parked) to quickly spread notifications
1097     * without unnecessary bouncing around event queues etc (1) wake
1098     * up and help cancel tasks (2) interrupt (3) mop up races with
1099     * interrupted workers
1100 dl 1.53 */
1101     private void startTerminating() {
1102 dl 1.61 cancelSubmissions();
1103     for (int passes = 0; passes < 4 && workerCounts != 0; ++passes) {
1104 dl 1.66 int c; // advance event count
1105     UNSAFE.compareAndSwapInt(this, eventCountOffset,
1106     c = eventCount, c+1);
1107 dl 1.61 eventWaiters = 0L; // clobber lists
1108     spareWaiters = 0;
1109     ForkJoinWorkerThread[] ws = workers;
1110     int n = ws.length;
1111     for (int i = 0; i < n; ++i) {
1112     ForkJoinWorkerThread w = ws[i];
1113     if (w != null) {
1114 dl 1.64 w.shutdown();
1115 dl 1.61 if (passes > 0 && !w.isTerminated()) {
1116     w.cancelTasks();
1117     LockSupport.unpark(w);
1118     if (passes > 1) {
1119     try {
1120     w.interrupt();
1121     } catch (SecurityException ignore) {
1122     }
1123     }
1124     }
1125     }
1126     }
1127 dl 1.56 }
1128     }
1129    
1130     /**
1131     * Clear out and cancel submissions, ignoring exceptions
1132     */
1133     private void cancelSubmissions() {
1134 dl 1.53 ForkJoinTask<?> task;
1135     while ((task = submissionQueue.poll()) != null) {
1136     try {
1137     task.cancel(false);
1138     } catch (Throwable ignore) {
1139     }
1140     }
1141 dl 1.56 }
1142    
1143 dl 1.53 // misc support for ForkJoinWorkerThread
1144    
1145     /**
1146     * Returns pool number
1147 dl 1.1 */
1148 dl 1.53 final int getPoolNumber() {
1149     return poolNumber;
1150 dl 1.1 }
1151    
1152     /**
1153 dl 1.61 * Tries to accumulates steal count from a worker, clearing
1154     * the worker's value.
1155     *
1156     * @return true if worker steal count now zero
1157 dl 1.1 */
1158 dl 1.61 final boolean tryAccumulateStealCount(ForkJoinWorkerThread w) {
1159 dl 1.53 int sc = w.stealCount;
1160 dl 1.61 long c = stealCount;
1161     // CAS even if zero, for fence effects
1162     if (UNSAFE.compareAndSwapLong(this, stealCountOffset, c, c + sc)) {
1163     if (sc != 0)
1164     w.stealCount = 0;
1165     return true;
1166 dl 1.1 }
1167 dl 1.61 return sc == 0;
1168 dl 1.1 }
1169    
1170     /**
1171 dl 1.53 * Returns the approximate (non-atomic) number of idle threads per
1172     * active thread.
1173     */
1174     final int idlePerActive() {
1175 dl 1.58 int pc = parallelism; // use parallelism, not rc
1176 jsr166 1.67 int ac = runState; // no mask -- artificially boosts during shutdown
1177 dl 1.53 // Use exact results for small values, saturate past 4
1178     return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
1179     }
1180    
1181     // Public and protected methods
1182 dl 1.1
1183     // Constructors
1184    
1185     /**
1186 dl 1.42 * Creates a {@code ForkJoinPool} with parallelism equal to {@link
1187 dl 1.57 * java.lang.Runtime#availableProcessors}, using the {@linkplain
1188     * #defaultForkJoinWorkerThreadFactory default thread factory},
1189     * no UncaughtExceptionHandler, and non-async LIFO processing mode.
1190 jsr166 1.17 *
1191 dl 1.1 * @throws SecurityException if a security manager exists and
1192     * the caller is not permitted to modify threads
1193     * because it does not hold {@link
1194 jsr166 1.17 * java.lang.RuntimePermission}{@code ("modifyThread")}
1195 dl 1.1 */
1196     public ForkJoinPool() {
1197     this(Runtime.getRuntime().availableProcessors(),
1198 dl 1.57 defaultForkJoinWorkerThreadFactory, null, false);
1199 dl 1.1 }
1200    
1201     /**
1202 dl 1.42 * Creates a {@code ForkJoinPool} with the indicated parallelism
1203 dl 1.57 * level, the {@linkplain
1204     * #defaultForkJoinWorkerThreadFactory default thread factory},
1205     * no UncaughtExceptionHandler, and non-async LIFO processing mode.
1206 jsr166 1.17 *
1207 dl 1.42 * @param parallelism the parallelism level
1208 dl 1.1 * @throws IllegalArgumentException if parallelism less than or
1209 jsr166 1.47 * equal to zero, or greater than implementation limit
1210 dl 1.1 * @throws SecurityException if a security manager exists and
1211     * the caller is not permitted to modify threads
1212     * because it does not hold {@link
1213 jsr166 1.17 * java.lang.RuntimePermission}{@code ("modifyThread")}
1214 dl 1.1 */
1215     public ForkJoinPool(int parallelism) {
1216 dl 1.57 this(parallelism, defaultForkJoinWorkerThreadFactory, null, false);
1217 dl 1.1 }
1218    
1219     /**
1220 dl 1.57 * Creates a {@code ForkJoinPool} with the given parameters.
1221 jsr166 1.17 *
1222 dl 1.57 * @param parallelism the parallelism level. For default value,
1223     * use {@link java.lang.Runtime#availableProcessors}.
1224     * @param factory the factory for creating new threads. For default value,
1225     * use {@link #defaultForkJoinWorkerThreadFactory}.
1226 dl 1.59 * @param handler the handler for internal worker threads that
1227     * terminate due to unrecoverable errors encountered while executing
1228 dl 1.57 * tasks. For default value, use <code>null</code>.
1229 dl 1.59 * @param asyncMode if true,
1230 dl 1.57 * establishes local first-in-first-out scheduling mode for forked
1231     * tasks that are never joined. This mode may be more appropriate
1232     * than default locally stack-based mode in applications in which
1233     * worker threads only process event-style asynchronous tasks.
1234     * For default value, use <code>false</code>.
1235 dl 1.1 * @throws IllegalArgumentException if parallelism less than or
1236 jsr166 1.47 * equal to zero, or greater than implementation limit
1237 jsr166 1.48 * @throws NullPointerException if the factory is null
1238 dl 1.1 * @throws SecurityException if a security manager exists and
1239     * the caller is not permitted to modify threads
1240     * because it does not hold {@link
1241 jsr166 1.17 * java.lang.RuntimePermission}{@code ("modifyThread")}
1242 dl 1.1 */
1243 dl 1.59 public ForkJoinPool(int parallelism,
1244 dl 1.57 ForkJoinWorkerThreadFactory factory,
1245     Thread.UncaughtExceptionHandler handler,
1246     boolean asyncMode) {
1247 dl 1.53 checkPermission();
1248     if (factory == null)
1249     throw new NullPointerException();
1250 dl 1.61 if (parallelism <= 0 || parallelism > MAX_WORKERS)
1251 dl 1.1 throw new IllegalArgumentException();
1252 dl 1.53 this.parallelism = parallelism;
1253 dl 1.1 this.factory = factory;
1254 dl 1.57 this.ueh = handler;
1255     this.locallyFifo = asyncMode;
1256     int arraySize = initialArraySizeFor(parallelism);
1257 dl 1.53 this.workers = new ForkJoinWorkerThread[arraySize];
1258     this.submissionQueue = new LinkedTransferQueue<ForkJoinTask<?>>();
1259 dl 1.1 this.workerLock = new ReentrantLock();
1260 dl 1.57 this.termination = new Phaser(1);
1261     this.poolNumber = poolNumberGenerator.incrementAndGet();
1262 dl 1.1 }
1263    
1264     /**
1265 dl 1.53 * Returns initial power of two size for workers array.
1266     * @param pc the initial parallelism level
1267     */
1268     private static int initialArraySizeFor(int pc) {
1269 dl 1.66 // If possible, initially allocate enough space for one spare
1270     int size = pc < MAX_WORKERS ? pc + 1 : MAX_WORKERS;
1271 dl 1.61 // See Hackers Delight, sec 3.2. We know MAX_WORKERS < (1 >>> 16)
1272 dl 1.53 size |= size >>> 1;
1273     size |= size >>> 2;
1274     size |= size >>> 4;
1275     size |= size >>> 8;
1276     return size + 1;
1277 dl 1.1 }
1278    
1279     // Execution methods
1280    
1281     /**
1282     * Common code for execute, invoke and submit
1283     */
1284     private <T> void doSubmit(ForkJoinTask<T> task) {
1285 dl 1.23 if (task == null)
1286     throw new NullPointerException();
1287 dl 1.53 if (runState >= SHUTDOWN)
1288 dl 1.1 throw new RejectedExecutionException();
1289 dl 1.58 submissionQueue.offer(task);
1290 dl 1.66 int c; // try to increment event count -- CAS failure OK
1291     UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
1292     helpMaintainParallelism(); // create, start, or resume some workers
1293 dl 1.1 }
1294    
1295     /**
1296 jsr166 1.17 * Performs the given task, returning its result upon completion.
1297     *
1298 dl 1.1 * @param task the task
1299     * @return the task's result
1300 jsr166 1.48 * @throws NullPointerException if the task is null
1301     * @throws RejectedExecutionException if the task cannot be
1302     * scheduled for execution
1303 dl 1.1 */
1304     public <T> T invoke(ForkJoinTask<T> task) {
1305     doSubmit(task);
1306     return task.join();
1307     }
1308    
1309     /**
1310     * Arranges for (asynchronous) execution of the given task.
1311 jsr166 1.17 *
1312 dl 1.1 * @param task the task
1313 jsr166 1.48 * @throws NullPointerException if the task is null
1314     * @throws RejectedExecutionException if the task cannot be
1315     * scheduled for execution
1316 dl 1.1 */
1317 dl 1.37 public void execute(ForkJoinTask<?> task) {
1318 dl 1.1 doSubmit(task);
1319     }
1320    
1321     // AbstractExecutorService methods
1322    
1323 jsr166 1.48 /**
1324     * @throws NullPointerException if the task is null
1325     * @throws RejectedExecutionException if the task cannot be
1326     * scheduled for execution
1327     */
1328 dl 1.1 public void execute(Runnable task) {
1329 dl 1.23 ForkJoinTask<?> job;
1330 jsr166 1.26 if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1331     job = (ForkJoinTask<?>) task;
1332 dl 1.23 else
1333 dl 1.33 job = ForkJoinTask.adapt(task, null);
1334 dl 1.23 doSubmit(job);
1335 dl 1.1 }
1336    
1337 jsr166 1.48 /**
1338 dl 1.57 * Submits a ForkJoinTask for execution.
1339     *
1340     * @param task the task to submit
1341     * @return the task
1342     * @throws NullPointerException if the task is null
1343     * @throws RejectedExecutionException if the task cannot be
1344     * scheduled for execution
1345     */
1346     public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
1347     doSubmit(task);
1348     return task;
1349     }
1350    
1351     /**
1352 jsr166 1.48 * @throws NullPointerException if the task is null
1353     * @throws RejectedExecutionException if the task cannot be
1354     * scheduled for execution
1355     */
1356 dl 1.1 public <T> ForkJoinTask<T> submit(Callable<T> task) {
1357 dl 1.33 ForkJoinTask<T> job = ForkJoinTask.adapt(task);
1358 dl 1.1 doSubmit(job);
1359     return job;
1360     }
1361    
1362 jsr166 1.48 /**
1363     * @throws NullPointerException if the task is null
1364     * @throws RejectedExecutionException if the task cannot be
1365     * scheduled for execution
1366     */
1367 dl 1.1 public <T> ForkJoinTask<T> submit(Runnable task, T result) {
1368 dl 1.33 ForkJoinTask<T> job = ForkJoinTask.adapt(task, result);
1369 dl 1.1 doSubmit(job);
1370     return job;
1371     }
1372    
1373 jsr166 1.48 /**
1374     * @throws NullPointerException if the task is null
1375     * @throws RejectedExecutionException if the task cannot be
1376     * scheduled for execution
1377     */
1378 dl 1.1 public ForkJoinTask<?> submit(Runnable task) {
1379 dl 1.23 ForkJoinTask<?> job;
1380 jsr166 1.26 if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1381     job = (ForkJoinTask<?>) task;
1382 dl 1.23 else
1383 dl 1.33 job = ForkJoinTask.adapt(task, null);
1384 dl 1.1 doSubmit(job);
1385     return job;
1386     }
1387    
1388     /**
1389 jsr166 1.48 * @throws NullPointerException {@inheritDoc}
1390     * @throws RejectedExecutionException {@inheritDoc}
1391     */
1392 dl 1.1 public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
1393 jsr166 1.20 ArrayList<ForkJoinTask<T>> forkJoinTasks =
1394 dl 1.1 new ArrayList<ForkJoinTask<T>>(tasks.size());
1395 jsr166 1.20 for (Callable<T> task : tasks)
1396 dl 1.33 forkJoinTasks.add(ForkJoinTask.adapt(task));
1397 jsr166 1.20 invoke(new InvokeAll<T>(forkJoinTasks));
1398    
1399     @SuppressWarnings({"unchecked", "rawtypes"})
1400 dl 1.54 List<Future<T>> futures = (List<Future<T>>) (List) forkJoinTasks;
1401 jsr166 1.20 return futures;
1402 dl 1.1 }
1403    
1404     static final class InvokeAll<T> extends RecursiveAction {
1405     final ArrayList<ForkJoinTask<T>> tasks;
1406     InvokeAll(ArrayList<ForkJoinTask<T>> tasks) { this.tasks = tasks; }
1407     public void compute() {
1408 jsr166 1.17 try { invokeAll(tasks); }
1409     catch (Exception ignore) {}
1410 dl 1.1 }
1411 jsr166 1.18 private static final long serialVersionUID = -7914297376763021607L;
1412 dl 1.1 }
1413    
1414     /**
1415 jsr166 1.17 * Returns the factory used for constructing new workers.
1416 dl 1.1 *
1417     * @return the factory used for constructing new workers
1418     */
1419     public ForkJoinWorkerThreadFactory getFactory() {
1420     return factory;
1421     }
1422    
1423     /**
1424 dl 1.2 * Returns the handler for internal worker threads that terminate
1425     * due to unrecoverable errors encountered while executing tasks.
1426 jsr166 1.17 *
1427 jsr166 1.28 * @return the handler, or {@code null} if none
1428 dl 1.2 */
1429     public Thread.UncaughtExceptionHandler getUncaughtExceptionHandler() {
1430 dl 1.53 return ueh;
1431 dl 1.2 }
1432    
1433     /**
1434 dl 1.42 * Returns the targeted parallelism level of this pool.
1435 dl 1.1 *
1436 dl 1.42 * @return the targeted parallelism level of this pool
1437 dl 1.1 */
1438     public int getParallelism() {
1439     return parallelism;
1440     }
1441    
1442     /**
1443     * Returns the number of worker threads that have started but not
1444     * yet terminated. This result returned by this method may differ
1445 jsr166 1.29 * from {@link #getParallelism} when threads are created to
1446 dl 1.1 * maintain parallelism when others are cooperatively blocked.
1447     *
1448     * @return the number of worker threads
1449     */
1450     public int getPoolSize() {
1451 dl 1.53 return workerCounts >>> TOTAL_COUNT_SHIFT;
1452 dl 1.1 }
1453    
1454     /**
1455 jsr166 1.28 * Returns {@code true} if this pool uses local first-in-first-out
1456 jsr166 1.16 * scheduling mode for forked tasks that are never joined.
1457 dl 1.6 *
1458 jsr166 1.28 * @return {@code true} if this pool uses async mode
1459 dl 1.6 */
1460     public boolean getAsyncMode() {
1461     return locallyFifo;
1462     }
1463    
1464     /**
1465 dl 1.2 * Returns an estimate of the number of worker threads that are
1466     * not blocked waiting to join tasks or for other managed
1467 dl 1.53 * synchronization. This method may overestimate the
1468     * number of running threads.
1469 dl 1.1 *
1470     * @return the number of worker threads
1471     */
1472     public int getRunningThreadCount() {
1473 dl 1.53 return workerCounts & RUNNING_COUNT_MASK;
1474 dl 1.1 }
1475    
1476     /**
1477 dl 1.2 * Returns an estimate of the number of threads that are currently
1478 dl 1.1 * stealing or executing tasks. This method may overestimate the
1479     * number of active threads.
1480 jsr166 1.17 *
1481 jsr166 1.16 * @return the number of active threads
1482 dl 1.1 */
1483     public int getActiveThreadCount() {
1484 dl 1.53 return runState & ACTIVE_COUNT_MASK;
1485 dl 1.1 }
1486    
1487     /**
1488 jsr166 1.28 * Returns {@code true} if all worker threads are currently idle.
1489     * An idle worker is one that cannot obtain a task to execute
1490     * because none are available to steal from other threads, and
1491     * there are no pending submissions to the pool. This method is
1492     * conservative; it might not return {@code true} immediately upon
1493     * idleness of all threads, but will eventually become true if
1494     * threads remain inactive.
1495 jsr166 1.17 *
1496 jsr166 1.28 * @return {@code true} if all threads are currently idle
1497 dl 1.1 */
1498     public boolean isQuiescent() {
1499 dl 1.53 return (runState & ACTIVE_COUNT_MASK) == 0;
1500 dl 1.1 }
1501    
1502     /**
1503     * Returns an estimate of the total number of tasks stolen from
1504     * one thread's work queue by another. The reported value
1505     * underestimates the actual total number of steals when the pool
1506     * is not quiescent. This value may be useful for monitoring and
1507 jsr166 1.17 * tuning fork/join programs: in general, steal counts should be
1508 dl 1.1 * high enough to keep threads busy, but low enough to avoid
1509     * overhead and contention across threads.
1510 jsr166 1.17 *
1511 jsr166 1.16 * @return the number of steals
1512 dl 1.1 */
1513     public long getStealCount() {
1514 dl 1.53 return stealCount;
1515 dl 1.1 }
1516    
1517     /**
1518 dl 1.2 * Returns an estimate of the total number of tasks currently held
1519     * in queues by worker threads (but not including tasks submitted
1520     * to the pool that have not begun executing). This value is only
1521     * an approximation, obtained by iterating across all threads in
1522     * the pool. This method may be useful for tuning task
1523     * granularities.
1524 jsr166 1.17 *
1525 jsr166 1.16 * @return the number of queued tasks
1526 dl 1.1 */
1527     public long getQueuedTaskCount() {
1528     long count = 0;
1529 dl 1.56 ForkJoinWorkerThread[] ws = workers;
1530 dl 1.61 int n = ws.length;
1531     for (int i = 0; i < n; ++i) {
1532 dl 1.56 ForkJoinWorkerThread w = ws[i];
1533 dl 1.53 if (w != null)
1534     count += w.getQueueSize();
1535 dl 1.1 }
1536     return count;
1537     }
1538    
1539     /**
1540 jsr166 1.40 * Returns an estimate of the number of tasks submitted to this
1541     * pool that have not yet begun executing. This method takes time
1542 dl 1.1 * proportional to the number of submissions.
1543 jsr166 1.17 *
1544 jsr166 1.16 * @return the number of queued submissions
1545 dl 1.1 */
1546     public int getQueuedSubmissionCount() {
1547     return submissionQueue.size();
1548     }
1549    
1550     /**
1551 jsr166 1.28 * Returns {@code true} if there are any tasks submitted to this
1552     * pool that have not yet begun executing.
1553 jsr166 1.17 *
1554 jsr166 1.16 * @return {@code true} if there are any queued submissions
1555 dl 1.1 */
1556     public boolean hasQueuedSubmissions() {
1557     return !submissionQueue.isEmpty();
1558     }
1559    
1560     /**
1561     * Removes and returns the next unexecuted submission if one is
1562     * available. This method may be useful in extensions to this
1563     * class that re-assign work in systems with multiple pools.
1564 jsr166 1.17 *
1565 jsr166 1.28 * @return the next submission, or {@code null} if none
1566 dl 1.1 */
1567     protected ForkJoinTask<?> pollSubmission() {
1568     return submissionQueue.poll();
1569     }
1570    
1571     /**
1572 dl 1.6 * Removes all available unexecuted submitted and forked tasks
1573     * from scheduling queues and adds them to the given collection,
1574     * without altering their execution status. These may include
1575 jsr166 1.41 * artificially generated or wrapped tasks. This method is
1576     * designed to be invoked only when the pool is known to be
1577 dl 1.6 * quiescent. Invocations at other times may not remove all
1578     * tasks. A failure encountered while attempting to add elements
1579 jsr166 1.16 * to collection {@code c} may result in elements being in
1580 dl 1.6 * neither, either or both collections when the associated
1581     * exception is thrown. The behavior of this operation is
1582     * undefined if the specified collection is modified while the
1583     * operation is in progress.
1584 jsr166 1.17 *
1585 dl 1.6 * @param c the collection to transfer elements into
1586     * @return the number of elements transferred
1587     */
1588 dl 1.30 protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
1589 dl 1.61 int count = submissionQueue.drainTo(c);
1590 dl 1.57 ForkJoinWorkerThread[] ws = workers;
1591 dl 1.61 int n = ws.length;
1592     for (int i = 0; i < n; ++i) {
1593 dl 1.57 ForkJoinWorkerThread w = ws[i];
1594     if (w != null)
1595 dl 1.61 count += w.drainTasksTo(c);
1596 dl 1.57 }
1597     return count;
1598     }
1599    
1600     /**
1601 dl 1.1 * Returns a string identifying this pool, as well as its state,
1602     * including indications of run state, parallelism level, and
1603     * worker and task counts.
1604     *
1605     * @return a string identifying this pool, as well as its state
1606     */
1607     public String toString() {
1608     long st = getStealCount();
1609     long qt = getQueuedTaskCount();
1610     long qs = getQueuedSubmissionCount();
1611 dl 1.53 int wc = workerCounts;
1612     int tc = wc >>> TOTAL_COUNT_SHIFT;
1613     int rc = wc & RUNNING_COUNT_MASK;
1614     int pc = parallelism;
1615     int rs = runState;
1616     int ac = rs & ACTIVE_COUNT_MASK;
1617 dl 1.1 return super.toString() +
1618 dl 1.53 "[" + runLevelToString(rs) +
1619     ", parallelism = " + pc +
1620     ", size = " + tc +
1621     ", active = " + ac +
1622     ", running = " + rc +
1623 dl 1.1 ", steals = " + st +
1624     ", tasks = " + qt +
1625     ", submissions = " + qs +
1626     "]";
1627     }
1628    
1629 dl 1.53 private static String runLevelToString(int s) {
1630     return ((s & TERMINATED) != 0 ? "Terminated" :
1631     ((s & TERMINATING) != 0 ? "Terminating" :
1632     ((s & SHUTDOWN) != 0 ? "Shutting down" :
1633     "Running")));
1634 dl 1.1 }
1635    
1636     /**
1637     * Initiates an orderly shutdown in which previously submitted
1638     * tasks are executed, but no new tasks will be accepted.
1639     * Invocation has no additional effect if already shut down.
1640     * Tasks that are in the process of being submitted concurrently
1641     * during the course of this method may or may not be rejected.
1642 jsr166 1.17 *
1643 dl 1.1 * @throws SecurityException if a security manager exists and
1644     * the caller is not permitted to modify threads
1645     * because it does not hold {@link
1646 jsr166 1.17 * java.lang.RuntimePermission}{@code ("modifyThread")}
1647 dl 1.1 */
1648     public void shutdown() {
1649     checkPermission();
1650 dl 1.53 advanceRunLevel(SHUTDOWN);
1651     tryTerminate(false);
1652 dl 1.1 }
1653    
1654     /**
1655 dl 1.42 * Attempts to cancel and/or stop all tasks, and reject all
1656     * subsequently submitted tasks. Tasks that are in the process of
1657     * being submitted or executed concurrently during the course of
1658     * this method may or may not be rejected. This method cancels
1659     * both existing and unexecuted tasks, in order to permit
1660     * termination in the presence of task dependencies. So the method
1661     * always returns an empty list (unlike the case for some other
1662     * Executors).
1663 jsr166 1.17 *
1664 dl 1.1 * @return an empty list
1665     * @throws SecurityException if a security manager exists and
1666     * the caller is not permitted to modify threads
1667     * because it does not hold {@link
1668 jsr166 1.17 * java.lang.RuntimePermission}{@code ("modifyThread")}
1669 dl 1.1 */
1670     public List<Runnable> shutdownNow() {
1671     checkPermission();
1672 dl 1.53 tryTerminate(true);
1673 dl 1.1 return Collections.emptyList();
1674     }
1675    
1676     /**
1677 jsr166 1.16 * Returns {@code true} if all tasks have completed following shut down.
1678 dl 1.1 *
1679 jsr166 1.16 * @return {@code true} if all tasks have completed following shut down
1680 dl 1.1 */
1681     public boolean isTerminated() {
1682 dl 1.53 return runState >= TERMINATED;
1683 dl 1.1 }
1684    
1685     /**
1686 jsr166 1.16 * Returns {@code true} if the process of termination has
1687 dl 1.42 * commenced but not yet completed. This method may be useful for
1688     * debugging. A return of {@code true} reported a sufficient
1689     * period after shutdown may indicate that submitted tasks have
1690     * ignored or suppressed interruption, causing this executor not
1691     * to properly terminate.
1692 dl 1.1 *
1693 dl 1.42 * @return {@code true} if terminating but not yet terminated
1694 dl 1.1 */
1695     public boolean isTerminating() {
1696 dl 1.53 return (runState & (TERMINATING|TERMINATED)) == TERMINATING;
1697 dl 1.1 }
1698    
1699     /**
1700 jsr166 1.16 * Returns {@code true} if this pool has been shut down.
1701 dl 1.1 *
1702 jsr166 1.16 * @return {@code true} if this pool has been shut down
1703 dl 1.1 */
1704     public boolean isShutdown() {
1705 dl 1.53 return runState >= SHUTDOWN;
1706 dl 1.42 }
1707    
1708     /**
1709 dl 1.1 * Blocks until all tasks have completed execution after a shutdown
1710     * request, or the timeout occurs, or the current thread is
1711     * interrupted, whichever happens first.
1712     *
1713     * @param timeout the maximum time to wait
1714     * @param unit the time unit of the timeout argument
1715 jsr166 1.16 * @return {@code true} if this executor terminated and
1716     * {@code false} if the timeout elapsed before termination
1717 dl 1.1 * @throws InterruptedException if interrupted while waiting
1718     */
1719     public boolean awaitTermination(long timeout, TimeUnit unit)
1720     throws InterruptedException {
1721 dl 1.57 try {
1722     return termination.awaitAdvanceInterruptibly(0, timeout, unit) > 0;
1723 jsr166 1.69 } catch (TimeoutException ex) {
1724 dl 1.57 return false;
1725     }
1726 dl 1.1 }
1727    
1728     /**
1729     * Interface for extending managed parallelism for tasks running
1730 jsr166 1.35 * in {@link ForkJoinPool}s.
1731     *
1732 dl 1.61 * <p>A {@code ManagedBlocker} provides two methods. Method
1733     * {@code isReleasable} must return {@code true} if blocking is
1734     * not necessary. Method {@code block} blocks the current thread
1735     * if necessary (perhaps internally invoking {@code isReleasable}
1736     * before actually blocking). The unusual methods in this API
1737     * accommodate synchronizers that may, but don't usually, block
1738     * for long periods. Similarly, they allow more efficient internal
1739     * handling of cases in which additional workers may be, but
1740     * usually are not, needed to ensure sufficient parallelism.
1741     * Toward this end, implementations of method {@code isReleasable}
1742     * must be amenable to repeated invocation.
1743 jsr166 1.17 *
1744 dl 1.1 * <p>For example, here is a ManagedBlocker based on a
1745     * ReentrantLock:
1746 jsr166 1.17 * <pre> {@code
1747     * class ManagedLocker implements ManagedBlocker {
1748     * final ReentrantLock lock;
1749     * boolean hasLock = false;
1750     * ManagedLocker(ReentrantLock lock) { this.lock = lock; }
1751     * public boolean block() {
1752     * if (!hasLock)
1753     * lock.lock();
1754     * return true;
1755     * }
1756     * public boolean isReleasable() {
1757     * return hasLock || (hasLock = lock.tryLock());
1758 dl 1.1 * }
1759 jsr166 1.17 * }}</pre>
1760 dl 1.61 *
1761     * <p>Here is a class that possibly blocks waiting for an
1762     * item on a given queue:
1763     * <pre> {@code
1764     * class QueueTaker<E> implements ManagedBlocker {
1765     * final BlockingQueue<E> queue;
1766     * volatile E item = null;
1767     * QueueTaker(BlockingQueue<E> q) { this.queue = q; }
1768     * public boolean block() throws InterruptedException {
1769     * if (item == null)
1770 dl 1.65 * item = queue.take();
1771 dl 1.61 * return true;
1772     * }
1773     * public boolean isReleasable() {
1774 dl 1.65 * return item != null || (item = queue.poll()) != null;
1775 dl 1.61 * }
1776     * public E getItem() { // call after pool.managedBlock completes
1777     * return item;
1778     * }
1779     * }}</pre>
1780 dl 1.1 */
1781     public static interface ManagedBlocker {
1782     /**
1783     * Possibly blocks the current thread, for example waiting for
1784     * a lock or condition.
1785 jsr166 1.17 *
1786 jsr166 1.28 * @return {@code true} if no additional blocking is necessary
1787     * (i.e., if isReleasable would return true)
1788 dl 1.1 * @throws InterruptedException if interrupted while waiting
1789 jsr166 1.17 * (the method is not required to do so, but is allowed to)
1790 dl 1.1 */
1791     boolean block() throws InterruptedException;
1792    
1793     /**
1794 jsr166 1.28 * Returns {@code true} if blocking is unnecessary.
1795 dl 1.1 */
1796     boolean isReleasable();
1797     }
1798    
1799     /**
1800     * Blocks in accord with the given blocker. If the current thread
1801 jsr166 1.38 * is a {@link ForkJoinWorkerThread}, this method possibly
1802     * arranges for a spare thread to be activated if necessary to
1803 dl 1.57 * ensure sufficient parallelism while the current thread is blocked.
1804 jsr166 1.38 *
1805     * <p>If the caller is not a {@link ForkJoinTask}, this method is
1806     * behaviorally equivalent to
1807 jsr166 1.17 * <pre> {@code
1808     * while (!blocker.isReleasable())
1809     * if (blocker.block())
1810     * return;
1811     * }</pre>
1812 jsr166 1.38 *
1813     * If the caller is a {@code ForkJoinTask}, then the pool may
1814     * first be expanded to ensure parallelism, and later adjusted.
1815 dl 1.1 *
1816     * @param blocker the blocker
1817 jsr166 1.16 * @throws InterruptedException if blocker.block did so
1818 dl 1.1 */
1819 dl 1.57 public static void managedBlock(ManagedBlocker blocker)
1820 dl 1.1 throws InterruptedException {
1821     Thread t = Thread.currentThread();
1822 dl 1.61 if (t instanceof ForkJoinWorkerThread) {
1823     ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
1824     w.pool.awaitBlocker(blocker);
1825     }
1826 dl 1.57 else {
1827     do {} while (!blocker.isReleasable() && !blocker.block());
1828     }
1829 dl 1.1 }
1830    
1831 dl 1.33 // AbstractExecutorService overrides. These rely on undocumented
1832     // fact that ForkJoinTask.adapt returns ForkJoinTasks that also
1833     // implement RunnableFuture.
1834 dl 1.2
1835     protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
1836 jsr166 1.34 return (RunnableFuture<T>) ForkJoinTask.adapt(runnable, value);
1837 dl 1.2 }
1838    
1839     protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
1840 jsr166 1.34 return (RunnableFuture<T>) ForkJoinTask.adapt(callable);
1841 dl 1.2 }
1842    
1843 jsr166 1.27 // Unsafe mechanics
1844 dl 1.1
1845 jsr166 1.21 private static final sun.misc.Unsafe UNSAFE = getUnsafe();
1846 dl 1.53 private static final long workerCountsOffset =
1847     objectFieldOffset("workerCounts", ForkJoinPool.class);
1848     private static final long runStateOffset =
1849     objectFieldOffset("runState", ForkJoinPool.class);
1850 jsr166 1.25 private static final long eventCountOffset =
1851 jsr166 1.27 objectFieldOffset("eventCount", ForkJoinPool.class);
1852 dl 1.53 private static final long eventWaitersOffset =
1853     objectFieldOffset("eventWaiters",ForkJoinPool.class);
1854     private static final long stealCountOffset =
1855     objectFieldOffset("stealCount",ForkJoinPool.class);
1856 dl 1.61 private static final long spareWaitersOffset =
1857     objectFieldOffset("spareWaiters",ForkJoinPool.class);
1858 dl 1.53
1859 jsr166 1.27 private static long objectFieldOffset(String field, Class<?> klazz) {
1860     try {
1861     return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
1862     } catch (NoSuchFieldException e) {
1863     // Convert Exception to corresponding Error
1864     NoSuchFieldError error = new NoSuchFieldError(field);
1865     error.initCause(e);
1866     throw error;
1867     }
1868     }
1869    
1870     /**
1871     * Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package.
1872     * Replace with a simple call to Unsafe.getUnsafe when integrating
1873     * into a jdk.
1874     *
1875     * @return a sun.misc.Unsafe
1876     */
1877     private static sun.misc.Unsafe getUnsafe() {
1878     try {
1879     return sun.misc.Unsafe.getUnsafe();
1880     } catch (SecurityException se) {
1881     try {
1882     return java.security.AccessController.doPrivileged
1883     (new java.security
1884     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
1885     public sun.misc.Unsafe run() throws Exception {
1886     java.lang.reflect.Field f = sun.misc
1887     .Unsafe.class.getDeclaredField("theUnsafe");
1888     f.setAccessible(true);
1889     return (sun.misc.Unsafe) f.get(null);
1890     }});
1891     } catch (java.security.PrivilegedActionException e) {
1892     throw new RuntimeException("Could not initialize intrinsics",
1893     e.getCause());
1894     }
1895     }
1896     }
1897 dl 1.1 }