ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/ForkJoinTask.java
Revision: 1.152
Committed: Sun Feb 7 15:01:38 2021 UTC (3 years, 4 months ago) by dl
Branch: MAIN
Changes since 1.151: +121 -136 lines
Log Message:
Refactor last 2 changes

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.48 * http://creativecommons.org/publicdomain/zero/1.0/
5 jsr166 1.1 */
6    
7     package java.util.concurrent;
8    
9     import java.io.Serializable;
10 dl 1.109 import java.lang.invoke.MethodHandles;
11     import java.lang.invoke.VarHandle;
12 jsr166 1.90 import java.lang.reflect.Constructor;
13 jsr166 1.1 import java.util.Collection;
14     import java.util.List;
15 jsr166 1.7 import java.util.RandomAccess;
16 dl 1.124 import java.util.concurrent.locks.LockSupport;
17 jsr166 1.1
18     /**
19 jsr166 1.6 * Abstract base class for tasks that run within a {@link ForkJoinPool}.
20     * A {@code ForkJoinTask} is a thread-like entity that is much
21 jsr166 1.1 * lighter weight than a normal thread. Huge numbers of tasks and
22     * subtasks may be hosted by a small number of actual threads in a
23     * ForkJoinPool, at the price of some usage limitations.
24     *
25 dl 1.64 * <p>A "main" {@code ForkJoinTask} begins execution when it is
26     * explicitly submitted to a {@link ForkJoinPool}, or, if not already
27     * engaged in a ForkJoin computation, commenced in the {@link
28 dl 1.67 * ForkJoinPool#commonPool()} via {@link #fork}, {@link #invoke}, or
29 dl 1.64 * related methods. Once started, it will usually in turn start other
30     * subtasks. As indicated by the name of this class, many programs
31     * using {@code ForkJoinTask} employ only methods {@link #fork} and
32     * {@link #join}, or derivatives such as {@link
33 jsr166 1.27 * #invokeAll(ForkJoinTask...) invokeAll}. However, this class also
34     * provides a number of other methods that can come into play in
35 dl 1.64 * advanced usages, as well as extension mechanics that allow support
36     * of new forms of fork/join processing.
37 jsr166 1.1 *
38 jsr166 1.6 * <p>A {@code ForkJoinTask} is a lightweight form of {@link Future}.
39     * The efficiency of {@code ForkJoinTask}s stems from a set of
40     * restrictions (that are only partially statically enforceable)
41 dl 1.54 * reflecting their main use as computational tasks calculating pure
42     * functions or operating on purely isolated objects. The primary
43     * coordination mechanisms are {@link #fork}, that arranges
44 jsr166 1.6 * asynchronous execution, and {@link #join}, that doesn't proceed
45     * until the task's result has been computed. Computations should
46 dl 1.54 * ideally avoid {@code synchronized} methods or blocks, and should
47     * minimize other blocking synchronization apart from joining other
48     * tasks or using synchronizers such as Phasers that are advertised to
49     * cooperate with fork/join scheduling. Subdividable tasks should also
50 jsr166 1.68 * not perform blocking I/O, and should ideally access variables that
51 dl 1.54 * are completely independent of those accessed by other running
52     * tasks. These guidelines are loosely enforced by not permitting
53     * checked exceptions such as {@code IOExceptions} to be
54     * thrown. However, computations may still encounter unchecked
55     * exceptions, that are rethrown to callers attempting to join
56     * them. These exceptions may additionally include {@link
57     * RejectedExecutionException} stemming from internal resource
58     * exhaustion, such as failure to allocate internal task
59     * queues. Rethrown exceptions behave in the same way as regular
60     * exceptions, but, when possible, contain stack traces (as displayed
61     * for example using {@code ex.printStackTrace()}) of both the thread
62     * that initiated the computation as well as the thread actually
63     * encountering the exception; minimally only the latter.
64     *
65     * <p>It is possible to define and use ForkJoinTasks that may block,
66 jsr166 1.110 * but doing so requires three further considerations: (1) Completion
67 dl 1.54 * of few if any <em>other</em> tasks should be dependent on a task
68 jsr166 1.68 * that blocks on external synchronization or I/O. Event-style async
69 dl 1.63 * tasks that are never joined (for example, those subclassing {@link
70     * CountedCompleter}) often fall into this category. (2) To minimize
71     * resource impact, tasks should be small; ideally performing only the
72     * (possibly) blocking action. (3) Unless the {@link
73 dl 1.54 * ForkJoinPool.ManagedBlocker} API is used, or the number of possibly
74     * blocked tasks is known to be less than the pool's {@link
75     * ForkJoinPool#getParallelism} level, the pool cannot guarantee that
76     * enough threads will be available to ensure progress or good
77     * performance.
78 jsr166 1.1 *
79     * <p>The primary method for awaiting completion and extracting
80     * results of a task is {@link #join}, but there are several variants:
81     * The {@link Future#get} methods support interruptible and/or timed
82     * waits for completion and report results using {@code Future}
83 dl 1.16 * conventions. Method {@link #invoke} is semantically
84 jsr166 1.8 * equivalent to {@code fork(); join()} but always attempts to begin
85     * execution in the current thread. The "<em>quiet</em>" forms of
86     * these methods do not extract results or report exceptions. These
87 jsr166 1.1 * may be useful when a set of tasks are being executed, and you need
88     * to delay processing of results or exceptions until all complete.
89     * Method {@code invokeAll} (available in multiple versions)
90     * performs the most common form of parallel invocation: forking a set
91     * of tasks and joining them all.
92     *
93 jsr166 1.57 * <p>In the most typical usages, a fork-join pair act like a call
94 dl 1.54 * (fork) and return (join) from a parallel recursive function. As is
95     * the case with other forms of recursive calls, returns (joins)
96     * should be performed innermost-first. For example, {@code a.fork();
97     * b.fork(); b.join(); a.join();} is likely to be substantially more
98     * efficient than joining {@code a} before {@code b}.
99     *
100 jsr166 1.8 * <p>The execution status of tasks may be queried at several levels
101     * of detail: {@link #isDone} is true if a task completed in any way
102     * (including the case where a task was cancelled without executing);
103     * {@link #isCompletedNormally} is true if a task completed without
104 jsr166 1.10 * cancellation or encountering an exception; {@link #isCancelled} is
105     * true if the task was cancelled (in which case {@link #getException}
106 jsr166 1.115 * returns a {@link CancellationException}); and
107 jsr166 1.10 * {@link #isCompletedAbnormally} is true if a task was either
108     * cancelled or encountered an exception, in which case {@link
109     * #getException} will return either the encountered exception or
110 jsr166 1.115 * {@link CancellationException}.
111 jsr166 1.8 *
112 jsr166 1.6 * <p>The ForkJoinTask class is not usually directly subclassed.
113 jsr166 1.1 * Instead, you subclass one of the abstract classes that support a
114 jsr166 1.6 * particular style of fork/join processing, typically {@link
115 dl 1.62 * RecursiveAction} for most computations that do not return results,
116     * {@link RecursiveTask} for those that do, and {@link
117     * CountedCompleter} for those in which completed actions trigger
118     * other actions. Normally, a concrete ForkJoinTask subclass declares
119     * fields comprising its parameters, established in a constructor, and
120     * then defines a {@code compute} method that somehow uses the control
121 dl 1.64 * methods supplied by this base class.
122 jsr166 1.1 *
123 dl 1.38 * <p>Method {@link #join} and its variants are appropriate for use
124     * only when completion dependencies are acyclic; that is, the
125     * parallel computation can be described as a directed acyclic graph
126     * (DAG). Otherwise, executions may encounter a form of deadlock as
127     * tasks cyclically wait for each other. However, this framework
128     * supports other methods and techniques (for example the use of
129     * {@link Phaser}, {@link #helpQuiesce}, and {@link #complete}) that
130     * may be of use in constructing custom subclasses for problems that
131 jsr166 1.79 * are not statically structured as DAGs. To support such usages, a
132 dl 1.63 * ForkJoinTask may be atomically <em>tagged</em> with a {@code short}
133     * value using {@link #setForkJoinTaskTag} or {@link
134 dl 1.60 * #compareAndSetForkJoinTaskTag} and checked using {@link
135 dl 1.63 * #getForkJoinTaskTag}. The ForkJoinTask implementation does not use
136     * these {@code protected} methods or tags for any purpose, but they
137     * may be of use in the construction of specialized subclasses. For
138     * example, parallel graph traversals can use the supplied methods to
139     * avoid revisiting nodes/tasks that have already been processed.
140     * (Method names for tagging are bulky in part to encourage definition
141     * of methods that reflect their usage patterns.)
142 dl 1.38 *
143 jsr166 1.7 * <p>Most base support methods are {@code final}, to prevent
144     * overriding of implementations that are intrinsically tied to the
145     * underlying lightweight task scheduling framework. Developers
146     * creating new basic styles of fork/join processing should minimally
147     * implement {@code protected} methods {@link #exec}, {@link
148     * #setRawResult}, and {@link #getRawResult}, while also introducing
149     * an abstract computational method that can be implemented in its
150     * subclasses, possibly relying on other {@code protected} methods
151     * provided by this class.
152 jsr166 1.1 *
153     * <p>ForkJoinTasks should perform relatively small amounts of
154 jsr166 1.7 * computation. Large tasks should be split into smaller subtasks,
155     * usually via recursive decomposition. As a very rough rule of thumb,
156     * a task should perform more than 100 and less than 10000 basic
157 dl 1.40 * computational steps, and should avoid indefinite looping. If tasks
158     * are too big, then parallelism cannot improve throughput. If too
159     * small, then memory and internal task maintenance overhead may
160     * overwhelm processing.
161 jsr166 1.1 *
162 jsr166 1.8 * <p>This class provides {@code adapt} methods for {@link Runnable}
163     * and {@link Callable}, that may be of use when mixing execution of
164 dl 1.16 * {@code ForkJoinTasks} with other kinds of tasks. When all tasks are
165     * of this form, consider using a pool constructed in <em>asyncMode</em>.
166 jsr166 1.6 *
167 jsr166 1.7 * <p>ForkJoinTasks are {@code Serializable}, which enables them to be
168     * used in extensions such as remote execution frameworks. It is
169     * sensible to serialize tasks only before or after, but not during,
170     * execution. Serialization is not relied on during execution itself.
171 jsr166 1.1 *
172     * @since 1.7
173     * @author Doug Lea
174     */
175     public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
176    
177 dl 1.13 /*
178     * See the internal documentation of class ForkJoinPool for a
179     * general implementation overview. ForkJoinTasks are mainly
180     * responsible for maintaining their "status" field amidst relays
181 jsr166 1.51 * to methods in ForkJoinWorkerThread and ForkJoinPool.
182     *
183     * The methods of this class are more-or-less layered into
184     * (1) basic status maintenance
185     * (2) execution and awaiting completion
186     * (3) user-level methods that additionally report results.
187     * This is sometimes hard to see because this file orders exported
188     * methods in a way that flows well in javadocs.
189 dl 1.118 *
190 dl 1.124 * Revision notes: The use of "Aux" field replaces previous
191     * reliance on a table to hold exceptions and synchronized blocks
192     * and monitors to wait for completion.
193 dl 1.118 */
194 jsr166 1.1
195     /**
196 dl 1.124 * Nodes for threads waiting for completion, or holding a thrown
197     * exception (never both). Waiting threads prepend nodes
198     * Treiber-stack-style. Signallers detach and unpark
199     * waiters. Cancelled waiters try to unsplice.
200     */
201     static final class Aux {
202     final Thread thread;
203     final Throwable ex; // null if a waiter
204     Aux next; // accessed only via memory-acquire chains
205     Aux(Thread thread, Throwable ex) {
206     this.thread = thread;
207     this.ex = ex;
208     }
209     final boolean casNext(Aux c, Aux v) { // used only in cancellation
210     return NEXT.compareAndSet(this, c, v);
211     }
212     private static final VarHandle NEXT;
213     static {
214     try {
215     NEXT = MethodHandles.lookup()
216     .findVarHandle(Aux.class, "next", Aux.class);
217     } catch (ReflectiveOperationException e) {
218     throw new ExceptionInInitializerError(e);
219 dl 1.13 }
220     }
221 jsr166 1.1 }
222    
223 dl 1.124 /*
224     * The status field holds bits packed into a single int to ensure
225     * atomicity. Status is initially zero, and takes on nonnegative
226     * values until completed, upon which it holds (sign bit) DONE,
227     * possibly with ABNORMAL (cancelled or exceptional) and THROWN
228 dl 1.128 * (in which case an exception has been stored). A value of
229     * ABNORMAL without DONE signifies an interrupted wait. These
230     * control bits occupy only (some of) the upper half (16 bits) of
231     * status field. The lower bits are used for user-defined tags.
232 jsr166 1.1 */
233 dl 1.134 private static final int DONE = 1 << 31; // must be negative
234     private static final int ABNORMAL = 1 << 16;
235     private static final int THROWN = 1 << 17;
236     private static final int SMASK = 0xffff; // short bits for tags
237     private static final int UNCOMPENSATE = 1 << 16; // helpJoin return sentinel
238 dl 1.124
239     // Fields
240     volatile int status; // accessed directly by pool and workers
241     private transient volatile Aux aux; // either waiters or thrown Exception
242 dl 1.128
243 dl 1.124 // Support for atomic operations
244     private static final VarHandle STATUS;
245     private static final VarHandle AUX;
246     private int getAndBitwiseOrStatus(int v) {
247     return (int)STATUS.getAndBitwiseOr(this, v);
248     }
249     private boolean casStatus(int c, int v) {
250 dl 1.150 return STATUS.compareAndSet(this, c, v);
251 dl 1.124 }
252     private boolean casAux(Aux c, Aux v) {
253     return AUX.compareAndSet(this, c, v);
254     }
255    
256     /** Removes and unparks waiters */
257     private void signalWaiters() {
258     for (Aux a; (a = aux) != null && a.ex == null; ) {
259     if (casAux(a, null)) { // detach entire list
260     for (Thread t; a != null; a = a.next) {
261     if ((t = a.thread) != Thread.currentThread() && t != null)
262     LockSupport.unpark(t); // don't self-signal
263     }
264     break;
265 dl 1.54 }
266 jsr166 1.1 }
267     }
268    
269     /**
270 dl 1.124 * Sets DONE status and wakes up threads waiting to join this task.
271     * @return status on exit
272 jsr166 1.1 */
273 dl 1.124 private int setDone() {
274     int s = getAndBitwiseOrStatus(DONE) | DONE;
275     signalWaiters();
276 dl 1.45 return s;
277     }
278    
279     /**
280 dl 1.124 * Sets ABNORMAL DONE status unless already done, and wakes up threads
281     * waiting to join this task.
282     * @return status on exit
283 dl 1.118 */
284 dl 1.124 private int trySetCancelled() {
285 dl 1.118 int s;
286 dl 1.124 do {} while ((s = status) >= 0 && !casStatus(s, s |= (DONE | ABNORMAL)));
287     signalWaiters();
288     return s;
289 dl 1.118 }
290    
291     /**
292 dl 1.124 * Records exception and sets ABNORMAL THROWN DONE status unless
293     * already done, and wakes up threads waiting to join this task.
294     * If losing a race with setDone or trySetCancelled, the exception
295     * may be recorded but not reported.
296 dl 1.54 *
297 dl 1.124 * @return status on exit
298 dl 1.45 */
299 dl 1.124 final int trySetThrown(Throwable ex) {
300     Aux h = new Aux(Thread.currentThread(), ex), p = null;
301     boolean installed = false;
302     int s;
303     while ((s = status) >= 0) {
304     Aux a;
305     if (!installed && ((a = aux) == null || a.ex == null) &&
306     (installed = casAux(a, h)))
307     p = a; // list of waiters replaced by h
308     if (installed && casStatus(s, s |= (DONE | ABNORMAL | THROWN)))
309     break;
310     }
311     for (; p != null; p = p.next)
312     LockSupport.unpark(p.thread);
313     return s;
314 jsr166 1.1 }
315    
316     /**
317 dl 1.124 * Records exception unless already done. Overridable in subclasses.
318 dl 1.54 *
319 dl 1.124 * @return status on exit
320 jsr166 1.1 */
321 dl 1.124 int trySetException(Throwable ex) {
322     return trySetThrown(ex);
323 dl 1.45 }
324    
325 jsr166 1.139 /**
326     * Constructor for subclasses to call.
327     */
328     public ForkJoinTask() {}
329    
330 dl 1.124 static boolean isExceptionalStatus(int s) { // needed by subclasses
331     return (s & THROWN) != 0;
332 dl 1.45 }
333    
334     /**
335 dl 1.124 * Unless done, calls exec and records status if completed, but
336     * doesn't wait for completion otherwise.
337 dl 1.45 *
338 dl 1.124 * @return status on exit from this method
339 dl 1.45 */
340 dl 1.124 final int doExec() {
341     int s; boolean completed;
342 dl 1.62 if ((s = status) >= 0) {
343     try {
344 dl 1.124 completed = exec();
345     } catch (Throwable rex) {
346     s = trySetException(rex);
347     completed = false;
348 dl 1.45 }
349 dl 1.124 if (completed)
350     s = setDone();
351 dl 1.45 }
352 dl 1.62 return s;
353     }
354    
355     /**
356 dl 1.134 * Helps and/or waits for completion from join, get, or invoke;
357     * called from either internal or external threads.
358 dl 1.63 *
359 dl 1.152 * @param pool if nonnull, known submitted pool, else assumes current pool
360 dl 1.128 * @param ran true if task known to have been exec'd
361 dl 1.134 * @param interruptible true if park interruptibly when external
362     * @param timed true if use timed wait
363     * @param nanos if timed, timeout value
364     * @return ABNORMAL if interrupted, else status on exit
365     */
366 dl 1.152 private int awaitDone(ForkJoinPool pool, boolean ran,
367 dl 1.151 boolean interruptible, boolean timed,
368 dl 1.134 long nanos) {
369 dl 1.152 ForkJoinPool p; boolean internal; int s; Thread t;
370     ForkJoinPool.WorkQueue q = null;
371     if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
372     ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
373     p = wt.pool;
374     if (pool == null)
375     pool = p;
376     if (internal = (pool == p))
377     q = wt.workQueue;
378 dl 1.124 }
379 dl 1.126 else {
380 dl 1.152 internal = false;
381 dl 1.126 p = ForkJoinPool.common;
382 dl 1.152 if (pool == null)
383     pool = p;
384     if (pool == p && p != null)
385     q = p.externalQueue();
386 dl 1.124 }
387 dl 1.152 if (interruptible && Thread.interrupted())
388     return ABNORMAL;
389 dl 1.134 if ((s = status) < 0)
390     return s;
391     long deadline = 0L;
392     if (timed) {
393     if (nanos <= 0L)
394     return 0;
395     else if ((deadline = nanos + System.nanoTime()) == 0L)
396     deadline = 1L;
397 dl 1.128 }
398 dl 1.151 boolean uncompensate = false;
399 dl 1.152 if (q != null && p != null) { // try helping
400     // help even in timed mode if pool has no parallelism
401     boolean canHelp = !timed || (p.mode & SMASK) == 0;
402     if (canHelp) {
403     if ((this instanceof CountedCompleter) &&
404     (s = p.helpComplete(this, q, internal)) < 0)
405     return s;
406     if (!ran && ((!internal && q.externalTryUnpush(this)) ||
407     q.tryRemove(this, internal)) && (s = doExec()) < 0)
408     return s;
409     }
410     if (internal) {
411     if ((s = p.helpJoin(this, q, canHelp)) < 0)
412 dl 1.128 return s;
413 dl 1.134 if (s == UNCOMPENSATE)
414 dl 1.151 uncompensate = true;
415 dl 1.128 }
416     }
417 dl 1.152 // block until done or cancelled wait
418     boolean interrupted = false, queued = false;
419     boolean parked = false, fail = false;
420     Aux node = null;
421     while ((s = status) >= 0) {
422     Aux a; long ns;
423     if (fail || (fail = (pool != null && pool.mode < 0)))
424     casStatus(s, s | (DONE | ABNORMAL)); // try to cancel
425     else if (parked && Thread.interrupted()) {
426     if (interruptible) {
427     s = ABNORMAL;
428     break;
429     }
430     interrupted = true;
431     }
432     else if (queued) {
433     if (deadline != 0L) {
434     if ((ns = deadline - System.nanoTime()) <= 0L)
435     break;
436     LockSupport.parkNanos(ns);
437     }
438     else
439     LockSupport.park();
440     parked = true;
441     }
442     else if (node != null) {
443     if ((a = aux) != null && a.ex != null)
444     Thread.onSpinWait(); // exception in progress
445     else if (queued = casAux(node.next = a, node))
446     LockSupport.setCurrentBlocker(this);
447     }
448     else {
449     try {
450     node = new Aux(Thread.currentThread(), null);
451     } catch (Throwable ex) { // cannot create
452     fail = true;
453     }
454     }
455     }
456     if (pool != null && uncompensate)
457     pool.uncompensate();
458    
459     if (queued) {
460     LockSupport.setCurrentBlocker(null);
461     if (s >= 0) { // cancellation similar to AbstractQueuedSynchronizer
462     outer: for (Aux a; (a = aux) != null && a.ex == null; ) {
463     for (Aux trail = null;;) {
464     Aux next = a.next;
465     if (a == node) {
466     if (trail != null)
467     trail.casNext(trail, next);
468     else if (casAux(a, next))
469     break outer; // cannot be re-encountered
470     break; // restart
471     } else {
472     trail = a;
473     if ((a = next) == null)
474     break outer;
475     }
476     }
477     }
478     }
479     else {
480     signalWaiters(); // help clean or signal
481     if (interrupted)
482     Thread.currentThread().interrupt();
483     }
484     }
485     return s;
486 dl 1.45 }
487    
488     /**
489 dl 1.128 * Cancels, ignoring any exceptions thrown by cancel. Cancel is
490     * spec'ed not to throw any exceptions, but if it does anyway, we
491     * have no recourse, so guard against this case.
492 dl 1.54 */
493 dl 1.128 static final void cancelIgnoringExceptions(Future<?> t) {
494 dl 1.124 if (t != null) {
495 dl 1.54 try {
496 dl 1.129 t.cancel(true);
497 dl 1.54 } catch (Throwable ignore) {
498     }
499     }
500     }
501    
502     /**
503 jsr166 1.105 * Returns a rethrowable exception for this task, if available.
504     * To provide accurate stack traces, if the exception was not
505     * thrown by the current thread, we try to create a new exception
506     * of the same type as the one thrown, but with the recorded
507     * exception as its cause. If there is no such constructor, we
508     * instead try to use a no-arg constructor, followed by initCause,
509     * to the same effect. If none of these apply, or any fail due to
510     * other exceptions, we return the recorded exception, which is
511     * still correct, although it may contain a misleading stack
512     * trace.
513 dl 1.45 *
514     * @return the exception, or null if none
515     */
516     private Throwable getThrowableException() {
517 dl 1.124 Throwable ex; Aux a;
518     if ((a = aux) == null)
519     ex = null;
520     else if ((ex = a.ex) != null && a.thread != Thread.currentThread()) {
521 dl 1.45 try {
522 dl 1.124 Constructor<?> noArgCtor = null, oneArgCtor = null;
523 jsr166 1.106 for (Constructor<?> c : ex.getClass().getConstructors()) {
524 dl 1.45 Class<?>[] ps = c.getParameterTypes();
525     if (ps.length == 0)
526     noArgCtor = c;
527 dl 1.124 else if (ps.length == 1 && ps[0] == Throwable.class) {
528     oneArgCtor = c;
529     break;
530     }
531 dl 1.45 }
532 dl 1.124 if (oneArgCtor != null)
533     ex = (Throwable)oneArgCtor.newInstance(ex);
534     else if (noArgCtor != null) {
535     Throwable rx = (Throwable)noArgCtor.newInstance();
536     rx.initCause(ex);
537     ex = rx;
538 dl 1.45 }
539     } catch (Exception ignore) {
540     }
541     }
542     return ex;
543     }
544    
545     /**
546 dl 1.128 * Returns exception associated with the given status, or null if none.
547     */
548     private Throwable getException(int s) {
549     Throwable ex = null;
550     if ((s & ABNORMAL) != 0 &&
551     ((s & THROWN) == 0 || (ex = getThrowableException()) == null))
552     ex = new CancellationException();
553     return ex;
554     }
555    
556     /**
557 dl 1.124 * Throws exception associated with the given status, or
558     * CancellationException if none recorded.
559 dl 1.45 */
560 dl 1.124 private void reportException(int s) {
561     ForkJoinTask.<RuntimeException>uncheckedThrow(
562     (s & THROWN) != 0 ? getThrowableException() : null);
563 dl 1.45 }
564    
565     /**
566 dl 1.129 * Throws exception for (timed or untimed) get, wrapping if
567     * necessary in an ExecutionException.
568     */
569 dl 1.134 private void reportExecutionException(int s) {
570 dl 1.129 Throwable ex = null;
571     if (s == ABNORMAL)
572     ex = new InterruptedException();
573     else if (s >= 0)
574     ex = new TimeoutException();
575     else if ((s & THROWN) != 0 && (ex = getThrowableException()) != null)
576     ex = new ExecutionException(ex);
577     ForkJoinTask.<RuntimeException>uncheckedThrow(ex);
578     }
579    
580     /**
581 dl 1.124 * A version of "sneaky throw" to relay exceptions in other
582     * contexts.
583 dl 1.65 */
584 jsr166 1.78 static void rethrow(Throwable ex) {
585 dl 1.100 ForkJoinTask.<RuntimeException>uncheckedThrow(ex);
586 dl 1.65 }
587    
588     /**
589     * The sneaky part of sneaky throw, relying on generics
590     * limitations to evade compiler complaints about rethrowing
591 dl 1.124 * unchecked exceptions. If argument null, throws
592     * CancellationException.
593 dl 1.65 */
594     @SuppressWarnings("unchecked") static <T extends Throwable>
595 dl 1.100 void uncheckedThrow(Throwable t) throws T {
596 dl 1.124 if (t == null)
597     t = new CancellationException();
598     throw (T)t; // rely on vacuous cast
599 jsr166 1.1 }
600    
601     // public methods
602    
603     /**
604 dl 1.64 * Arranges to asynchronously execute this task in the pool the
605     * current task is running in, if applicable, or using the {@link
606 dl 1.67 * ForkJoinPool#commonPool()} if not {@link #inForkJoinPool}. While
607 dl 1.64 * it is not necessarily enforced, it is a usage error to fork a
608     * task more than once unless it has completed and been
609     * reinitialized. Subsequent modifications to the state of this
610     * task or any data it operates on are not necessarily
611     * consistently observable by any thread other than the one
612     * executing it unless preceded by a call to {@link #join} or
613     * related methods, or a call to {@link #isDone} returning {@code
614     * true}.
615 jsr166 1.2 *
616 jsr166 1.6 * @return {@code this}, to simplify usage
617 jsr166 1.1 */
618 jsr166 1.2 public final ForkJoinTask<V> fork() {
619 dl 1.124 Thread t; ForkJoinWorkerThread w;
620 dl 1.64 if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
621 dl 1.124 (w = (ForkJoinWorkerThread)t).workQueue.push(this, w.pool);
622 dl 1.64 else
623 dl 1.71 ForkJoinPool.common.externalPush(this);
624 jsr166 1.2 return this;
625 jsr166 1.1 }
626    
627     /**
628 jsr166 1.115 * Returns the result of the computation when it
629     * {@linkplain #isDone is done}.
630     * This method differs from {@link #get()} in that abnormal
631     * completion results in {@code RuntimeException} or {@code Error},
632     * not {@code ExecutionException}, and that interrupts of the
633     * calling thread do <em>not</em> cause the method to abruptly
634     * return by throwing {@code InterruptedException}.
635 jsr166 1.1 *
636     * @return the computed result
637     */
638     public final V join() {
639 dl 1.59 int s;
640 dl 1.124 if ((s = status) >= 0)
641 dl 1.152 s = awaitDone(null, false, false, false, 0L);
642 dl 1.124 if ((s & ABNORMAL) != 0)
643 dl 1.59 reportException(s);
644     return getRawResult();
645 jsr166 1.1 }
646    
647     /**
648     * Commences performing this task, awaits its completion if
649 jsr166 1.21 * necessary, and returns its result, or throws an (unchecked)
650 dl 1.20 * {@code RuntimeException} or {@code Error} if the underlying
651     * computation did so.
652 jsr166 1.1 *
653     * @return the computed result
654     */
655     public final V invoke() {
656 dl 1.59 int s;
657 dl 1.124 if ((s = doExec()) >= 0)
658 dl 1.152 s = awaitDone(null, true, false, false, 0L);
659 dl 1.124 if ((s & ABNORMAL) != 0)
660 dl 1.59 reportException(s);
661     return getRawResult();
662 jsr166 1.1 }
663    
664     /**
665 jsr166 1.8 * Forks the given tasks, returning when {@code isDone} holds for
666     * each task or an (unchecked) exception is encountered, in which
667 dl 1.20 * case the exception is rethrown. If more than one task
668     * encounters an exception, then this method throws any one of
669     * these exceptions. If any task encounters an exception, the
670     * other may be cancelled. However, the execution status of
671     * individual tasks is not guaranteed upon exceptional return. The
672     * status of each task may be obtained using {@link
673     * #getException()} and related methods to check if they have been
674     * cancelled, completed normally or exceptionally, or left
675     * unprocessed.
676 jsr166 1.6 *
677     * @param t1 the first task
678     * @param t2 the second task
679     * @throws NullPointerException if any task is null
680 jsr166 1.1 */
681 jsr166 1.6 public static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) {
682 dl 1.59 int s1, s2;
683 dl 1.124 if (t1 == null || t2 == null)
684     throw new NullPointerException();
685 jsr166 1.1 t2.fork();
686 dl 1.124 if ((s1 = t1.doExec()) >= 0)
687 dl 1.152 s1 = t1.awaitDone(null, true, false, false, 0L);
688 dl 1.128 if ((s1 & ABNORMAL) != 0) {
689     cancelIgnoringExceptions(t2);
690 dl 1.59 t1.reportException(s1);
691 dl 1.128 }
692 dl 1.152 else if (((s2 = t2.awaitDone(null, false, false, false, 0L)) & ABNORMAL) != 0)
693 dl 1.134 t2.reportException(s2);
694 jsr166 1.1 }
695    
696     /**
697 jsr166 1.6 * Forks the given tasks, returning when {@code isDone} holds for
698 jsr166 1.8 * each task or an (unchecked) exception is encountered, in which
699 dl 1.20 * case the exception is rethrown. If more than one task
700     * encounters an exception, then this method throws any one of
701     * these exceptions. If any task encounters an exception, others
702     * may be cancelled. However, the execution status of individual
703     * tasks is not guaranteed upon exceptional return. The status of
704     * each task may be obtained using {@link #getException()} and
705     * related methods to check if they have been cancelled, completed
706     * normally or exceptionally, or left unprocessed.
707 jsr166 1.6 *
708     * @param tasks the tasks
709 jsr166 1.8 * @throws NullPointerException if any task is null
710 jsr166 1.1 */
711     public static void invokeAll(ForkJoinTask<?>... tasks) {
712     Throwable ex = null;
713     int last = tasks.length - 1;
714 jsr166 1.144 for (int i = last; i >= 0; --i) {
715 dl 1.124 ForkJoinTask<?> t;
716     if ((t = tasks[i]) == null) {
717     ex = new NullPointerException();
718     break;
719 jsr166 1.1 }
720 dl 1.124 if (i == 0) {
721 jsr166 1.144 int s;
722 dl 1.124 if ((s = t.doExec()) >= 0)
723 dl 1.152 s = t.awaitDone(null, true, false, false, 0L);
724 dl 1.124 if ((s & ABNORMAL) != 0)
725 dl 1.128 ex = t.getException(s);
726 dl 1.124 break;
727 jsr166 1.1 }
728 dl 1.124 t.fork();
729 jsr166 1.1 }
730 dl 1.124 if (ex == null) {
731 jsr166 1.144 for (int i = 1; i <= last; ++i) {
732 dl 1.124 ForkJoinTask<?> t;
733     if ((t = tasks[i]) != null) {
734 jsr166 1.144 int s;
735 dl 1.124 if ((s = t.status) >= 0)
736 dl 1.152 s = t.awaitDone(null, false, false, false, 0L);
737 dl 1.128 if ((s & ABNORMAL) != 0 && (ex = t.getException(s)) != null)
738 dl 1.124 break;
739     }
740     }
741     }
742 dl 1.128 if (ex != null) {
743 jsr166 1.143 for (int i = 1; i <= last; ++i)
744 dl 1.128 cancelIgnoringExceptions(tasks[i]);
745 dl 1.65 rethrow(ex);
746 dl 1.128 }
747 jsr166 1.1 }
748    
749     /**
750 jsr166 1.7 * Forks all tasks in the specified collection, returning when
751 jsr166 1.8 * {@code isDone} holds for each task or an (unchecked) exception
752 dl 1.20 * is encountered, in which case the exception is rethrown. If
753     * more than one task encounters an exception, then this method
754     * throws any one of these exceptions. If any task encounters an
755     * exception, others may be cancelled. However, the execution
756     * status of individual tasks is not guaranteed upon exceptional
757     * return. The status of each task may be obtained using {@link
758     * #getException()} and related methods to check if they have been
759     * cancelled, completed normally or exceptionally, or left
760     * unprocessed.
761 jsr166 1.6 *
762 jsr166 1.1 * @param tasks the collection of tasks
763 jsr166 1.82 * @param <T> the type of the values returned from the tasks
764 jsr166 1.2 * @return the tasks argument, to simplify usage
765 jsr166 1.1 * @throws NullPointerException if tasks or any element are null
766     */
767 jsr166 1.2 public static <T extends ForkJoinTask<?>> Collection<T> invokeAll(Collection<T> tasks) {
768 jsr166 1.7 if (!(tasks instanceof RandomAccess) || !(tasks instanceof List<?>)) {
769 jsr166 1.121 invokeAll(tasks.toArray(new ForkJoinTask<?>[0]));
770 jsr166 1.2 return tasks;
771 jsr166 1.1 }
772     @SuppressWarnings("unchecked")
773     List<? extends ForkJoinTask<?>> ts =
774     (List<? extends ForkJoinTask<?>>) tasks;
775     Throwable ex = null;
776 dl 1.124 int last = ts.size() - 1; // nearly same as array version
777 jsr166 1.144 for (int i = last; i >= 0; --i) {
778 dl 1.124 ForkJoinTask<?> t;
779     if ((t = ts.get(i)) == null) {
780     ex = new NullPointerException();
781     break;
782 jsr166 1.1 }
783 dl 1.124 if (i == 0) {
784 jsr166 1.144 int s;
785 dl 1.124 if ((s = t.doExec()) >= 0)
786 dl 1.152 s = t.awaitDone(null, true, false, false, 0L);
787 dl 1.124 if ((s & ABNORMAL) != 0)
788 dl 1.128 ex = t.getException(s);
789 dl 1.124 break;
790     }
791     t.fork();
792     }
793     if (ex == null) {
794 jsr166 1.144 for (int i = 1; i <= last; ++i) {
795 dl 1.124 ForkJoinTask<?> t;
796     if ((t = ts.get(i)) != null) {
797 jsr166 1.144 int s;
798 dl 1.124 if ((s = t.status) >= 0)
799 dl 1.152 s = t.awaitDone(null, false, false, false, 0L);
800 dl 1.128 if ((s & ABNORMAL) != 0 && (ex = t.getException(s)) != null)
801 dl 1.124 break;
802     }
803 jsr166 1.1 }
804     }
805 dl 1.128 if (ex != null) {
806 jsr166 1.142 for (int i = 1; i <= last; ++i)
807 dl 1.128 cancelIgnoringExceptions(ts.get(i));
808 dl 1.65 rethrow(ex);
809 dl 1.128 }
810 jsr166 1.2 return tasks;
811 jsr166 1.1 }
812 jsr166 1.135
813 jsr166 1.1 /**
814 jsr166 1.7 * Attempts to cancel execution of this task. This attempt will
815 jsr166 1.36 * fail if the task has already completed or could not be
816     * cancelled for some other reason. If successful, and this task
817     * has not started when {@code cancel} is called, execution of
818 dl 1.38 * this task is suppressed. After this method returns
819     * successfully, unless there is an intervening call to {@link
820     * #reinitialize}, subsequent calls to {@link #isCancelled},
821     * {@link #isDone}, and {@code cancel} will return {@code true}
822     * and calls to {@link #join} and related methods will result in
823     * {@code CancellationException}.
824 jsr166 1.1 *
825     * <p>This method may be overridden in subclasses, but if so, must
826 dl 1.38 * still ensure that these properties hold. In particular, the
827     * {@code cancel} method itself must not throw exceptions.
828 jsr166 1.1 *
829 jsr166 1.6 * <p>This method is designed to be invoked by <em>other</em>
830 jsr166 1.1 * tasks. To terminate the current task, you can just return or
831     * throw an unchecked exception from its computation method, or
832 jsr166 1.74 * invoke {@link #completeExceptionally(Throwable)}.
833 jsr166 1.1 *
834 dl 1.38 * @param mayInterruptIfRunning this value has no effect in the
835     * default implementation because interrupts are not used to
836     * control cancellation.
837 jsr166 1.1 *
838 jsr166 1.4 * @return {@code true} if this task is now cancelled
839 jsr166 1.1 */
840     public boolean cancel(boolean mayInterruptIfRunning) {
841 dl 1.124 return (trySetCancelled() & (ABNORMAL | THROWN)) == ABNORMAL;
842 jsr166 1.1 }
843    
844 jsr166 1.8 public final boolean isDone() {
845     return status < 0;
846     }
847    
848     public final boolean isCancelled() {
849 dl 1.118 return (status & (ABNORMAL | THROWN)) == ABNORMAL;
850 jsr166 1.8 }
851    
852     /**
853 jsr166 1.4 * Returns {@code true} if this task threw an exception or was cancelled.
854 jsr166 1.1 *
855 jsr166 1.4 * @return {@code true} if this task threw an exception or was cancelled
856 jsr166 1.1 */
857     public final boolean isCompletedAbnormally() {
858 dl 1.118 return (status & ABNORMAL) != 0;
859 jsr166 1.1 }
860    
861     /**
862 jsr166 1.8 * Returns {@code true} if this task completed without throwing an
863     * exception and was not cancelled.
864     *
865     * @return {@code true} if this task completed without throwing an
866     * exception and was not cancelled
867     */
868     public final boolean isCompletedNormally() {
869 dl 1.118 return (status & (DONE | ABNORMAL)) == DONE;
870 jsr166 1.8 }
871    
872     /**
873 jsr166 1.1 * Returns the exception thrown by the base computation, or a
874 jsr166 1.6 * {@code CancellationException} if cancelled, or {@code null} if
875     * none or if the method has not yet completed.
876 jsr166 1.1 *
877 jsr166 1.4 * @return the exception, or {@code null} if none
878 jsr166 1.1 */
879     public final Throwable getException() {
880 dl 1.128 return getException(status);
881 jsr166 1.1 }
882    
883     /**
884     * Completes this task abnormally, and if not already aborted or
885     * cancelled, causes it to throw the given exception upon
886     * {@code join} and related operations. This method may be used
887     * to induce exceptions in asynchronous tasks, or to force
888     * completion of tasks that would not otherwise complete. Its use
889 jsr166 1.6 * in other situations is discouraged. This method is
890 jsr166 1.1 * overridable, but overridden versions must invoke {@code super}
891     * implementation to maintain guarantees.
892     *
893 jsr166 1.11 * @param ex the exception to throw. If this exception is not a
894     * {@code RuntimeException} or {@code Error}, the actual exception
895     * thrown will be a {@code RuntimeException} with cause {@code ex}.
896 jsr166 1.1 */
897     public void completeExceptionally(Throwable ex) {
898 dl 1.124 trySetException((ex instanceof RuntimeException) ||
899     (ex instanceof Error) ? ex :
900     new RuntimeException(ex));
901 jsr166 1.1 }
902    
903     /**
904     * Completes this task, and if not already aborted or cancelled,
905 dl 1.22 * returning the given value as the result of subsequent
906     * invocations of {@code join} and related operations. This method
907     * may be used to provide results for asynchronous tasks, or to
908     * provide alternative handling for tasks that would not otherwise
909     * complete normally. Its use in other situations is
910     * discouraged. This method is overridable, but overridden
911     * versions must invoke {@code super} implementation to maintain
912     * guarantees.
913 jsr166 1.1 *
914     * @param value the result value for this task
915     */
916     public void complete(V value) {
917     try {
918     setRawResult(value);
919     } catch (Throwable rex) {
920 dl 1.124 trySetException(rex);
921 jsr166 1.1 return;
922     }
923 dl 1.118 setDone();
924 jsr166 1.1 }
925    
926 jsr166 1.25 /**
927 dl 1.62 * Completes this task normally without setting a value. The most
928     * recent value established by {@link #setRawResult} (or {@code
929     * null} by default) will be returned as the result of subsequent
930     * invocations of {@code join} and related operations.
931     *
932     * @since 1.8
933 dl 1.60 */
934     public final void quietlyComplete() {
935 dl 1.118 setDone();
936 dl 1.60 }
937    
938     /**
939 dl 1.29 * Waits if necessary for the computation to complete, and then
940     * retrieves its result.
941     *
942     * @return the computed result
943     * @throws CancellationException if the computation was cancelled
944     * @throws ExecutionException if the computation threw an
945     * exception
946     * @throws InterruptedException if the current thread is not a
947     * member of a ForkJoinPool and was interrupted while waiting
948 jsr166 1.25 */
949 jsr166 1.1 public final V get() throws InterruptedException, ExecutionException {
950 dl 1.152 int s = awaitDone(null, false, true, false, 0L);
951 dl 1.151 if ((s & ABNORMAL) != 0)
952 dl 1.134 reportExecutionException(s);
953 dl 1.129 return getRawResult();
954 jsr166 1.1 }
955 dl 1.14
956 jsr166 1.25 /**
957 dl 1.29 * Waits if necessary for at most the given time for the computation
958     * to complete, and then retrieves its result, if available.
959     *
960     * @param timeout the maximum time to wait
961     * @param unit the time unit of the timeout argument
962     * @return the computed result
963     * @throws CancellationException if the computation was cancelled
964     * @throws ExecutionException if the computation threw an
965     * exception
966     * @throws InterruptedException if the current thread is not a
967     * member of a ForkJoinPool and was interrupted while waiting
968     * @throws TimeoutException if the wait timed out
969 jsr166 1.25 */
970 jsr166 1.1 public final V get(long timeout, TimeUnit unit)
971     throws InterruptedException, ExecutionException, TimeoutException {
972 dl 1.151 long nanos = unit.toNanos(timeout);
973 dl 1.152 int s = awaitDone(null, false, true, true, nanos);
974 dl 1.151 if (s >= 0 || (s & ABNORMAL) != 0)
975 dl 1.134 reportExecutionException(s);
976 dl 1.129 return getRawResult();
977 jsr166 1.1 }
978    
979     /**
980 dl 1.17 * Joins this task, without returning its result or throwing its
981 jsr166 1.1 * exception. This method may be useful when processing
982     * collections of tasks when some have been cancelled or otherwise
983     * known to have aborted.
984     */
985     public final void quietlyJoin() {
986 dl 1.124 if (status >= 0)
987 dl 1.152 awaitDone(null, false, false, false, 0L);
988 jsr166 1.1 }
989    
990 dl 1.151
991 jsr166 1.1 /**
992     * Commences performing this task and awaits its completion if
993 dl 1.17 * necessary, without returning its result or throwing its
994 dl 1.22 * exception.
995 jsr166 1.1 */
996     public final void quietlyInvoke() {
997 dl 1.124 if (doExec() >= 0)
998 dl 1.152 awaitDone(null, true, false, false, 0L);
999 dl 1.151 }
1000    
1001     // Versions of join/get for pool.invoke* methods that use external,
1002     // possibly-non-commonPool submits
1003    
1004 dl 1.152 final void awaitPoolInvoke(ForkJoinPool pool) {
1005     awaitDone(pool, false, false, false, 0L);
1006     }
1007     final void awaitPoolInvoke(ForkJoinPool pool, long nanos) {
1008     awaitDone(pool, false, true, true, nanos);
1009     }
1010 dl 1.151 final V joinForPoolInvoke(ForkJoinPool pool) {
1011 dl 1.152 int s = awaitDone(pool, false, false, false, 0L);
1012 dl 1.151 if ((s & ABNORMAL) != 0)
1013     reportException(s);
1014     return getRawResult();
1015     }
1016     final V getForPoolInvoke(ForkJoinPool pool)
1017     throws InterruptedException, ExecutionException {
1018 dl 1.152 int s = awaitDone(pool, false, true, false, 0L);
1019 dl 1.151 if ((s & ABNORMAL) != 0)
1020     reportExecutionException(s);
1021     return getRawResult();
1022     }
1023     final V getForPoolInvoke(ForkJoinPool pool, long nanos)
1024     throws InterruptedException, ExecutionException, TimeoutException {
1025 dl 1.152 int s = awaitDone(pool, false, true, true, nanos);
1026 dl 1.151 if (s >= 0 || (s & ABNORMAL) != 0)
1027     reportExecutionException(s);
1028     return getRawResult();
1029 jsr166 1.1 }
1030    
1031     /**
1032     * Possibly executes tasks until the pool hosting the current task
1033 jsr166 1.104 * {@linkplain ForkJoinPool#isQuiescent is quiescent}. This
1034     * method may be of use in designs in which many tasks are forked,
1035     * but none are explicitly joined, instead executing them until
1036     * all are processed.
1037 jsr166 1.1 */
1038     public static void helpQuiesce() {
1039 dl 1.124 Thread t; ForkJoinWorkerThread w; ForkJoinPool p;
1040     if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread &&
1041     (p = (w = (ForkJoinWorkerThread)t).pool) != null)
1042 dl 1.128 p.helpQuiescePool(w.workQueue, Long.MAX_VALUE, false);
1043 dl 1.64 else
1044 dl 1.128 ForkJoinPool.common.externalHelpQuiescePool(Long.MAX_VALUE, false);
1045 jsr166 1.1 }
1046    
1047     /**
1048     * Resets the internal bookkeeping state of this task, allowing a
1049     * subsequent {@code fork}. This method allows repeated reuse of
1050     * this task, but only if reuse occurs when this task has either
1051     * never been forked, or has been forked, then completed and all
1052     * outstanding joins of this task have also completed. Effects
1053 jsr166 1.6 * under any other usage conditions are not guaranteed.
1054     * This method may be useful when executing
1055 jsr166 1.1 * pre-constructed trees of subtasks in loops.
1056 jsr166 1.34 *
1057 dl 1.33 * <p>Upon completion of this method, {@code isDone()} reports
1058     * {@code false}, and {@code getException()} reports {@code
1059     * null}. However, the value returned by {@code getRawResult} is
1060     * unaffected. To clear this value, you can invoke {@code
1061     * setRawResult(null)}.
1062 jsr166 1.1 */
1063     public void reinitialize() {
1064 dl 1.124 aux = null;
1065     status = 0;
1066 jsr166 1.1 }
1067    
1068     /**
1069 jsr166 1.103 * Returns the pool hosting the current thread, or {@code null}
1070     * if the current thread is executing outside of any ForkJoinPool.
1071     *
1072     * <p>This method returns {@code null} if and only if {@link
1073     * #inForkJoinPool} returns {@code false}.
1074 jsr166 1.1 *
1075 jsr166 1.97 * @return the pool, or {@code null} if none
1076 jsr166 1.1 */
1077     public static ForkJoinPool getPool() {
1078 dl 1.124 Thread t;
1079     return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1080     ((ForkJoinWorkerThread) t).pool : null);
1081 jsr166 1.1 }
1082    
1083     /**
1084 dl 1.42 * Returns {@code true} if the current thread is a {@link
1085     * ForkJoinWorkerThread} executing as a ForkJoinPool computation.
1086 jsr166 1.1 *
1087 dl 1.42 * @return {@code true} if the current thread is a {@link
1088     * ForkJoinWorkerThread} executing as a ForkJoinPool computation,
1089     * or {@code false} otherwise
1090 jsr166 1.1 */
1091     public static boolean inForkJoinPool() {
1092     return Thread.currentThread() instanceof ForkJoinWorkerThread;
1093     }
1094    
1095     /**
1096     * Tries to unschedule this task for execution. This method will
1097 dl 1.64 * typically (but is not guaranteed to) succeed if this task is
1098     * the most recently forked task by the current thread, and has
1099     * not commenced executing in another thread. This method may be
1100     * useful when arranging alternative local processing of tasks
1101     * that could have been, but were not, stolen.
1102 jsr166 1.1 *
1103 jsr166 1.4 * @return {@code true} if unforked
1104 jsr166 1.1 */
1105     public boolean tryUnfork() {
1106 dl 1.134 Thread t; ForkJoinPool.WorkQueue q;
1107 jsr166 1.136 return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
1108     ? (q = ((ForkJoinWorkerThread)t).workQueue) != null
1109     && q.tryUnpush(this)
1110     : (q = ForkJoinPool.commonQueue()) != null
1111     && q.externalTryUnpush(this);
1112 jsr166 1.1 }
1113    
1114     /**
1115     * Returns an estimate of the number of tasks that have been
1116     * forked by the current worker thread but not yet executed. This
1117     * value may be useful for heuristic decisions about whether to
1118     * fork other tasks.
1119     *
1120     * @return the number of tasks
1121     */
1122     public static int getQueuedTaskCount() {
1123 dl 1.66 Thread t; ForkJoinPool.WorkQueue q;
1124     if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
1125     q = ((ForkJoinWorkerThread)t).workQueue;
1126     else
1127 dl 1.124 q = ForkJoinPool.commonQueue();
1128 dl 1.66 return (q == null) ? 0 : q.queueSize();
1129 jsr166 1.1 }
1130    
1131     /**
1132     * Returns an estimate of how many more locally queued tasks are
1133     * held by the current worker thread than there are other worker
1134 dl 1.64 * threads that might steal them, or zero if this thread is not
1135     * operating in a ForkJoinPool. This value may be useful for
1136 jsr166 1.1 * heuristic decisions about whether to fork other tasks. In many
1137     * usages of ForkJoinTasks, at steady state, each worker should
1138     * aim to maintain a small constant surplus (for example, 3) of
1139     * tasks, and to process computations locally if this threshold is
1140     * exceeded.
1141     *
1142     * @return the surplus number of tasks, which may be negative
1143     */
1144     public static int getSurplusQueuedTaskCount() {
1145 dl 1.66 return ForkJoinPool.getSurplusQueuedTaskCount();
1146 jsr166 1.1 }
1147    
1148     // Extension methods
1149    
1150     /**
1151 jsr166 1.4 * Returns the result that would be returned by {@link #join}, even
1152     * if this task completed abnormally, or {@code null} if this task
1153     * is not known to have been completed. This method is designed
1154     * to aid debugging, as well as to support extensions. Its use in
1155     * any other context is discouraged.
1156 jsr166 1.1 *
1157 jsr166 1.4 * @return the result, or {@code null} if not completed
1158 jsr166 1.1 */
1159     public abstract V getRawResult();
1160    
1161     /**
1162     * Forces the given value to be returned as a result. This method
1163     * is designed to support extensions, and should not in general be
1164     * called otherwise.
1165     *
1166     * @param value the value
1167     */
1168     protected abstract void setRawResult(V value);
1169    
1170     /**
1171 dl 1.62 * Immediately performs the base action of this task and returns
1172     * true if, upon return from this method, this task is guaranteed
1173 dl 1.122 * to have completed. This method may return false otherwise, to
1174     * indicate that this task is not necessarily complete (or is not
1175     * known to be complete), for example in asynchronous actions that
1176     * require explicit invocations of completion methods. This method
1177     * may also throw an (unchecked) exception to indicate abnormal
1178     * exit. This method is designed to support extensions, and should
1179     * not in general be called otherwise.
1180 jsr166 1.1 *
1181 dl 1.62 * @return {@code true} if this task is known to have completed normally
1182 jsr166 1.1 */
1183     protected abstract boolean exec();
1184    
1185     /**
1186 jsr166 1.5 * Returns, but does not unschedule or execute, a task queued by
1187     * the current thread but not yet executed, if one is immediately
1188 dl 1.66 * available. There is no guarantee that this task will actually
1189     * be polled or executed next. Conversely, this method may return
1190     * null even if a task exists but cannot be accessed without
1191     * contention with other threads. This method is designed
1192 jsr166 1.5 * primarily to support extensions, and is unlikely to be useful
1193 jsr166 1.6 * otherwise.
1194     *
1195 jsr166 1.4 * @return the next task, or {@code null} if none are available
1196 jsr166 1.1 */
1197     protected static ForkJoinTask<?> peekNextLocalTask() {
1198 dl 1.66 Thread t; ForkJoinPool.WorkQueue q;
1199     if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
1200     q = ((ForkJoinWorkerThread)t).workQueue;
1201     else
1202 dl 1.124 q = ForkJoinPool.commonQueue();
1203 dl 1.66 return (q == null) ? null : q.peek();
1204 jsr166 1.1 }
1205    
1206     /**
1207     * Unschedules and returns, without executing, the next task
1208 dl 1.64 * queued by the current thread but not yet executed, if the
1209     * current thread is operating in a ForkJoinPool. This method is
1210     * designed primarily to support extensions, and is unlikely to be
1211     * useful otherwise.
1212 jsr166 1.1 *
1213 jsr166 1.4 * @return the next task, or {@code null} if none are available
1214 jsr166 1.1 */
1215     protected static ForkJoinTask<?> pollNextLocalTask() {
1216 dl 1.64 Thread t;
1217 dl 1.124 return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1218     ((ForkJoinWorkerThread)t).workQueue.nextLocalTask() : null);
1219 jsr166 1.1 }
1220    
1221     /**
1222 dl 1.64 * If the current thread is operating in a ForkJoinPool,
1223     * unschedules and returns, without executing, the next task
1224 jsr166 1.1 * queued by the current thread but not yet executed, if one is
1225     * available, or if not available, a task that was forked by some
1226     * other thread, if available. Availability may be transient, so a
1227 dl 1.64 * {@code null} result does not necessarily imply quiescence of
1228     * the pool this task is operating in. This method is designed
1229 jsr166 1.1 * primarily to support extensions, and is unlikely to be useful
1230 jsr166 1.6 * otherwise.
1231     *
1232 jsr166 1.4 * @return a task, or {@code null} if none are available
1233 jsr166 1.1 */
1234     protected static ForkJoinTask<?> pollTask() {
1235 dl 1.124 Thread t; ForkJoinWorkerThread w;
1236     return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1237     (w = (ForkJoinWorkerThread)t).pool.nextTaskFor(w.workQueue) :
1238     null);
1239 dl 1.54 }
1240    
1241 dl 1.94 /**
1242     * If the current thread is operating in a ForkJoinPool,
1243     * unschedules and returns, without executing, a task externally
1244     * submitted to the pool, if one is available. Availability may be
1245     * transient, so a {@code null} result does not necessarily imply
1246     * quiescence of the pool. This method is designed primarily to
1247     * support extensions, and is unlikely to be useful otherwise.
1248     *
1249     * @return a task, or {@code null} if none are available
1250 jsr166 1.107 * @since 9
1251 dl 1.94 */
1252     protected static ForkJoinTask<?> pollSubmission() {
1253 dl 1.96 Thread t;
1254 dl 1.124 return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1255     ((ForkJoinWorkerThread)t).pool.pollSubmission() : null);
1256 dl 1.94 }
1257    
1258 dl 1.60 // tag operations
1259 dl 1.54
1260     /**
1261 dl 1.60 * Returns the tag for this task.
1262 dl 1.54 *
1263 dl 1.60 * @return the tag for this task
1264 dl 1.54 * @since 1.8
1265     */
1266 dl 1.60 public final short getForkJoinTaskTag() {
1267     return (short)status;
1268 dl 1.54 }
1269    
1270     /**
1271 jsr166 1.102 * Atomically sets the tag value for this task and returns the old value.
1272 dl 1.54 *
1273 jsr166 1.102 * @param newValue the new tag value
1274 dl 1.60 * @return the previous value of the tag
1275 dl 1.54 * @since 1.8
1276     */
1277 jsr166 1.102 public final short setForkJoinTaskTag(short newValue) {
1278 dl 1.54 for (int s;;) {
1279 dl 1.124 if (casStatus(s = status, (s & ~SMASK) | (newValue & SMASK)))
1280 dl 1.60 return (short)s;
1281 dl 1.54 }
1282     }
1283    
1284     /**
1285 dl 1.60 * Atomically conditionally sets the tag value for this task.
1286     * Among other applications, tags can be used as visit markers
1287 dl 1.61 * in tasks operating on graphs, as in methods that check: {@code
1288 dl 1.60 * if (task.compareAndSetForkJoinTaskTag((short)0, (short)1))}
1289     * before processing, otherwise exiting because the node has
1290     * already been visited.
1291 dl 1.54 *
1292 jsr166 1.102 * @param expect the expected tag value
1293     * @param update the new tag value
1294 jsr166 1.76 * @return {@code true} if successful; i.e., the current value was
1295 jsr166 1.102 * equal to {@code expect} and was changed to {@code update}.
1296 dl 1.54 * @since 1.8
1297     */
1298 jsr166 1.102 public final boolean compareAndSetForkJoinTaskTag(short expect, short update) {
1299 dl 1.54 for (int s;;) {
1300 jsr166 1.102 if ((short)(s = status) != expect)
1301 dl 1.54 return false;
1302 dl 1.124 if (casStatus(s, (s & ~SMASK) | (update & SMASK)))
1303 dl 1.54 return true;
1304     }
1305 jsr166 1.1 }
1306    
1307 jsr166 1.5 /**
1308 jsr166 1.95 * Adapter for Runnables. This implements RunnableFuture
1309 jsr166 1.5 * to be compliant with AbstractExecutorService constraints
1310     * when used in ForkJoinPool.
1311     */
1312     static final class AdaptedRunnable<T> extends ForkJoinTask<T>
1313     implements RunnableFuture<T> {
1314 jsr166 1.123 @SuppressWarnings("serial") // Conditionally serializable
1315 jsr166 1.5 final Runnable runnable;
1316 jsr166 1.123 @SuppressWarnings("serial") // Conditionally serializable
1317 jsr166 1.5 T result;
1318     AdaptedRunnable(Runnable runnable, T result) {
1319     if (runnable == null) throw new NullPointerException();
1320     this.runnable = runnable;
1321 dl 1.59 this.result = result; // OK to set this even before completion
1322 jsr166 1.5 }
1323 dl 1.59 public final T getRawResult() { return result; }
1324     public final void setRawResult(T v) { result = v; }
1325     public final boolean exec() { runnable.run(); return true; }
1326     public final void run() { invoke(); }
1327 jsr166 1.116 public String toString() {
1328     return super.toString() + "[Wrapped task = " + runnable + "]";
1329     }
1330 dl 1.59 private static final long serialVersionUID = 5232453952276885070L;
1331     }
1332    
1333     /**
1334 jsr166 1.99 * Adapter for Runnables without results.
1335 dl 1.59 */
1336     static final class AdaptedRunnableAction extends ForkJoinTask<Void>
1337     implements RunnableFuture<Void> {
1338 jsr166 1.123 @SuppressWarnings("serial") // Conditionally serializable
1339 dl 1.59 final Runnable runnable;
1340     AdaptedRunnableAction(Runnable runnable) {
1341     if (runnable == null) throw new NullPointerException();
1342     this.runnable = runnable;
1343 jsr166 1.5 }
1344 dl 1.59 public final Void getRawResult() { return null; }
1345     public final void setRawResult(Void v) { }
1346     public final boolean exec() { runnable.run(); return true; }
1347     public final void run() { invoke(); }
1348 jsr166 1.116 public String toString() {
1349     return super.toString() + "[Wrapped task = " + runnable + "]";
1350     }
1351 jsr166 1.5 private static final long serialVersionUID = 5232453952276885070L;
1352     }
1353    
1354     /**
1355 jsr166 1.99 * Adapter for Runnables in which failure forces worker exception.
1356 dl 1.73 */
1357     static final class RunnableExecuteAction extends ForkJoinTask<Void> {
1358 jsr166 1.123 @SuppressWarnings("serial") // Conditionally serializable
1359 dl 1.73 final Runnable runnable;
1360     RunnableExecuteAction(Runnable runnable) {
1361     if (runnable == null) throw new NullPointerException();
1362     this.runnable = runnable;
1363     }
1364     public final Void getRawResult() { return null; }
1365     public final void setRawResult(Void v) { }
1366     public final boolean exec() { runnable.run(); return true; }
1367 dl 1.145 int trySetException(Throwable ex) { // if a handler, invoke it
1368     int s; Thread t; java.lang.Thread.UncaughtExceptionHandler h;
1369 dl 1.126 if (isExceptionalStatus(s = trySetThrown(ex)) &&
1370 jsr166 1.146 (h = ((t = Thread.currentThread()).
1371     getUncaughtExceptionHandler())) != null) {
1372     try {
1373     h.uncaughtException(t, ex);
1374     } catch (Throwable ignore) {
1375     }
1376     }
1377 dl 1.124 return s;
1378 dl 1.73 }
1379     private static final long serialVersionUID = 5232453952276885070L;
1380     }
1381    
1382     /**
1383 jsr166 1.99 * Adapter for Callables.
1384 jsr166 1.5 */
1385     static final class AdaptedCallable<T> extends ForkJoinTask<T>
1386     implements RunnableFuture<T> {
1387 jsr166 1.123 @SuppressWarnings("serial") // Conditionally serializable
1388 jsr166 1.6 final Callable<? extends T> callable;
1389 jsr166 1.123 @SuppressWarnings("serial") // Conditionally serializable
1390 jsr166 1.5 T result;
1391 jsr166 1.6 AdaptedCallable(Callable<? extends T> callable) {
1392 jsr166 1.5 if (callable == null) throw new NullPointerException();
1393     this.callable = callable;
1394     }
1395 dl 1.59 public final T getRawResult() { return result; }
1396     public final void setRawResult(T v) { result = v; }
1397     public final boolean exec() {
1398 jsr166 1.5 try {
1399     result = callable.call();
1400     return true;
1401     } catch (RuntimeException rex) {
1402     throw rex;
1403     } catch (Exception ex) {
1404     throw new RuntimeException(ex);
1405     }
1406     }
1407 dl 1.59 public final void run() { invoke(); }
1408 jsr166 1.116 public String toString() {
1409     return super.toString() + "[Wrapped task = " + callable + "]";
1410     }
1411 jsr166 1.117 private static final long serialVersionUID = 2838392045355241008L;
1412 jsr166 1.5 }
1413 jsr166 1.2
1414 dl 1.129 static final class AdaptedInterruptibleCallable<T> extends ForkJoinTask<T>
1415     implements RunnableFuture<T> {
1416     @SuppressWarnings("serial") // Conditionally serializable
1417     final Callable<? extends T> callable;
1418     @SuppressWarnings("serial") // Conditionally serializable
1419     transient volatile Thread runner;
1420     T result;
1421     AdaptedInterruptibleCallable(Callable<? extends T> callable) {
1422     if (callable == null) throw new NullPointerException();
1423     this.callable = callable;
1424     }
1425     public final T getRawResult() { return result; }
1426     public final void setRawResult(T v) { result = v; }
1427     public final boolean exec() {
1428     Thread.interrupted();
1429     runner = Thread.currentThread();
1430     try {
1431 dl 1.133 if (!isDone()) // recheck
1432     result = callable.call();
1433 dl 1.129 return true;
1434     } catch (RuntimeException rex) {
1435     throw rex;
1436     } catch (Exception ex) {
1437     throw new RuntimeException(ex);
1438     } finally {
1439     runner = null;
1440     Thread.interrupted();
1441     }
1442     }
1443     public final void run() { invoke(); }
1444     public final boolean cancel(boolean mayInterruptIfRunning) {
1445     Thread t;
1446     boolean stat = super.cancel(false);
1447     if (mayInterruptIfRunning && (t = runner) != null) {
1448     try {
1449     t.interrupt();
1450     } catch (Throwable ignore) {
1451     }
1452     }
1453     return stat;
1454     }
1455     public String toString() {
1456     return super.toString() + "[Wrapped task = " + callable + "]";
1457     }
1458     private static final long serialVersionUID = 2838392045355241008L;
1459     }
1460    
1461 jsr166 1.2 /**
1462 jsr166 1.6 * Returns a new {@code ForkJoinTask} that performs the {@code run}
1463     * method of the given {@code Runnable} as its action, and returns
1464     * a null result upon {@link #join}.
1465 jsr166 1.2 *
1466     * @param runnable the runnable action
1467     * @return the task
1468     */
1469 jsr166 1.6 public static ForkJoinTask<?> adapt(Runnable runnable) {
1470 dl 1.59 return new AdaptedRunnableAction(runnable);
1471 jsr166 1.2 }
1472    
1473     /**
1474 jsr166 1.6 * Returns a new {@code ForkJoinTask} that performs the {@code run}
1475     * method of the given {@code Runnable} as its action, and returns
1476     * the given result upon {@link #join}.
1477 jsr166 1.2 *
1478     * @param runnable the runnable action
1479     * @param result the result upon completion
1480 jsr166 1.82 * @param <T> the type of the result
1481 jsr166 1.2 * @return the task
1482     */
1483     public static <T> ForkJoinTask<T> adapt(Runnable runnable, T result) {
1484 jsr166 1.5 return new AdaptedRunnable<T>(runnable, result);
1485 jsr166 1.2 }
1486    
1487     /**
1488 jsr166 1.6 * Returns a new {@code ForkJoinTask} that performs the {@code call}
1489     * method of the given {@code Callable} as its action, and returns
1490     * its result upon {@link #join}, translating any checked exceptions
1491     * encountered into {@code RuntimeException}.
1492 jsr166 1.2 *
1493     * @param callable the callable action
1494 jsr166 1.82 * @param <T> the type of the callable's result
1495 jsr166 1.2 * @return the task
1496     */
1497 jsr166 1.6 public static <T> ForkJoinTask<T> adapt(Callable<? extends T> callable) {
1498 jsr166 1.5 return new AdaptedCallable<T>(callable);
1499 jsr166 1.2 }
1500    
1501 dl 1.129 /**
1502 jsr166 1.138 * Returns a new {@code ForkJoinTask} that performs the {@code call}
1503     * method of the given {@code Callable} as its action, and returns
1504     * its result upon {@link #join}, translating any checked exceptions
1505     * encountered into {@code RuntimeException}. Additionally,
1506     * invocations of {@code cancel} with {@code mayInterruptIfRunning
1507     * true} will attempt to interrupt the thread performing the task.
1508 dl 1.129 *
1509     * @param callable the callable action
1510     * @param <T> the type of the callable's result
1511     * @return the task
1512     *
1513 jsr166 1.147 * @since 17
1514 dl 1.129 */
1515 jsr166 1.147 // adaptInterruptible deferred to its own independent change
1516     // https://bugs.openjdk.java.net/browse/JDK-8246587
1517 jsr166 1.148 /* TODO: public */ private static <T> ForkJoinTask<T> adaptInterruptible(Callable<? extends T> callable) {
1518 dl 1.129 return new AdaptedInterruptibleCallable<T>(callable);
1519     }
1520    
1521 jsr166 1.1 // Serialization support
1522    
1523     private static final long serialVersionUID = -7721805057305804111L;
1524    
1525     /**
1526 jsr166 1.53 * Saves this task to a stream (that is, serializes it).
1527 jsr166 1.1 *
1528 jsr166 1.83 * @param s the stream
1529 jsr166 1.84 * @throws java.io.IOException if an I/O error occurs
1530 jsr166 1.1 * @serialData the current run status and the exception thrown
1531 jsr166 1.4 * during execution, or {@code null} if none
1532 jsr166 1.1 */
1533     private void writeObject(java.io.ObjectOutputStream s)
1534     throws java.io.IOException {
1535 dl 1.124 Aux a;
1536 jsr166 1.1 s.defaultWriteObject();
1537 dl 1.124 s.writeObject((a = aux) == null ? null : a.ex);
1538 jsr166 1.1 }
1539    
1540     /**
1541 jsr166 1.53 * Reconstitutes this task from a stream (that is, deserializes it).
1542 jsr166 1.83 * @param s the stream
1543 jsr166 1.84 * @throws ClassNotFoundException if the class of a serialized object
1544     * could not be found
1545     * @throws java.io.IOException if an I/O error occurs
1546 jsr166 1.1 */
1547     private void readObject(java.io.ObjectInputStream s)
1548     throws java.io.IOException, ClassNotFoundException {
1549     s.defaultReadObject();
1550     Object ex = s.readObject();
1551     if (ex != null)
1552 dl 1.124 trySetThrown((Throwable)ex);
1553 jsr166 1.1 }
1554    
1555 dl 1.45 static {
1556 jsr166 1.1 try {
1557 dl 1.109 MethodHandles.Lookup l = MethodHandles.lookup();
1558     STATUS = l.findVarHandle(ForkJoinTask.class, "status", int.class);
1559 dl 1.124 AUX = l.findVarHandle(ForkJoinTask.class, "aux", Aux.class);
1560 jsr166 1.91 } catch (ReflectiveOperationException e) {
1561 jsr166 1.120 throw new ExceptionInInitializerError(e);
1562 jsr166 1.1 }
1563     }
1564 dl 1.45
1565 jsr166 1.1 }