ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jsr166y/ForkJoinTask.java
(Generate patch)

Comparing jsr166/src/jsr166y/ForkJoinTask.java (file contents):
Revision 1.16 by jsr166, Fri Jul 24 23:47:01 2009 UTC vs.
Revision 1.71 by dl, Tue Nov 23 10:51:18 2010 UTC

# Line 5 | Line 5
5   */
6  
7   package jsr166y;
8 +
9   import java.io.Serializable;
10 < import java.util.*;
11 < import java.util.concurrent.*;
12 < import java.util.concurrent.atomic.*;
10 > import java.util.Collection;
11 > import java.util.Collections;
12 > import java.util.List;
13 > import java.util.RandomAccess;
14 > import java.util.Map;
15 > import java.util.WeakHashMap;
16 > import java.util.concurrent.Callable;
17 > import java.util.concurrent.CancellationException;
18 > import java.util.concurrent.ExecutionException;
19 > import java.util.concurrent.Executor;
20 > import java.util.concurrent.ExecutorService;
21 > import java.util.concurrent.Future;
22 > import java.util.concurrent.RejectedExecutionException;
23 > import java.util.concurrent.RunnableFuture;
24 > import java.util.concurrent.TimeUnit;
25 > import java.util.concurrent.TimeoutException;
26  
27   /**
28 < * Abstract base class for tasks that run within a {@link
29 < * ForkJoinPool}.  A ForkJoinTask is a thread-like entity that is much
28 > * Abstract base class for tasks that run within a {@link ForkJoinPool}.
29 > * A {@code ForkJoinTask} is a thread-like entity that is much
30   * lighter weight than a normal thread.  Huge numbers of tasks and
31   * subtasks may be hosted by a small number of actual threads in a
32   * ForkJoinPool, at the price of some usage limitations.
33   *
34 < * <p> A "main" ForkJoinTask begins execution when submitted to a
35 < * {@link ForkJoinPool}. Once started, it will usually in turn start
36 < * other subtasks.  As indicated by the name of this class, many
37 < * programs using ForkJoinTasks employ only methods {@code fork}
38 < * and {@code join}, or derivatives such as
39 < * {@code invokeAll}.  However, this class also provides a number
40 < * of other methods that can come into play in advanced usages, as
41 < * well as extension mechanics that allow support of new forms of
42 < * fork/join processing.
34 > * <p>A "main" {@code ForkJoinTask} begins execution when submitted
35 > * to a {@link ForkJoinPool}.  Once started, it will usually in turn
36 > * start other subtasks.  As indicated by the name of this class,
37 > * many programs using {@code ForkJoinTask} employ only methods
38 > * {@link #fork} and {@link #join}, or derivatives such as {@link
39 > * #invokeAll(ForkJoinTask...) invokeAll}.  However, this class also
40 > * provides a number of other methods that can come into play in
41 > * advanced usages, as well as extension mechanics that allow
42 > * support of new forms of fork/join processing.
43   *
44 < * <p>A ForkJoinTask is a lightweight form of {@link Future}.  The
45 < * efficiency of ForkJoinTasks stems from a set of restrictions (that
46 < * are only partially statically enforceable) reflecting their
47 < * intended use as computational tasks calculating pure functions or
48 < * operating on purely isolated objects.  The primary coordination
49 < * mechanisms are {@link #fork}, that arranges asynchronous execution,
50 < * and {@link #join}, that doesn't proceed until the task's result has
51 < * been computed.  Computations should avoid {@code synchronized}
52 < * methods or blocks, and should minimize other blocking
53 < * synchronization apart from joining other tasks or using
54 < * synchronizers such as Phasers that are advertised to cooperate with
55 < * fork/join scheduling. Tasks should also not perform blocking IO,
56 < * and should ideally access variables that are completely independent
57 < * of those accessed by other running tasks. Minor breaches of these
58 < * restrictions, for example using shared output streams, may be
59 < * tolerable in practice, but frequent use may result in poor
60 < * performance, and the potential to indefinitely stall if the number
61 < * of threads not waiting for IO or other external synchronization
62 < * becomes exhausted. This usage restriction is in part enforced by
63 < * not permitting checked exceptions such as {@code IOExceptions}
64 < * to be thrown. However, computations may still encounter unchecked
65 < * exceptions, that are rethrown to callers attempting join
66 < * them. These exceptions may additionally include
67 < * RejectedExecutionExceptions stemming from internal resource
68 < * exhaustion such as failure to allocate internal task queues.
44 > * <p>A {@code ForkJoinTask} is a lightweight form of {@link Future}.
45 > * The efficiency of {@code ForkJoinTask}s stems from a set of
46 > * restrictions (that are only partially statically enforceable)
47 > * reflecting their intended use as computational tasks calculating
48 > * pure functions or operating on purely isolated objects.  The
49 > * primary coordination mechanisms are {@link #fork}, that arranges
50 > * asynchronous execution, and {@link #join}, that doesn't proceed
51 > * until the task's result has been computed.  Computations should
52 > * avoid {@code synchronized} methods or blocks, and should minimize
53 > * other blocking synchronization apart from joining other tasks or
54 > * using synchronizers such as Phasers that are advertised to
55 > * cooperate with fork/join scheduling. Tasks should also not perform
56 > * blocking IO, and should ideally access variables that are
57 > * completely independent of those accessed by other running
58 > * tasks. Minor breaches of these restrictions, for example using
59 > * shared output streams, may be tolerable in practice, but frequent
60 > * use may result in poor performance, and the potential to
61 > * indefinitely stall if the number of threads not waiting for IO or
62 > * other external synchronization becomes exhausted. This usage
63 > * restriction is in part enforced by not permitting checked
64 > * exceptions such as {@code IOExceptions} to be thrown. However,
65 > * computations may still encounter unchecked exceptions, that are
66 > * rethrown to callers attempting to join them. These exceptions may
67 > * additionally include {@link RejectedExecutionException} stemming
68 > * from internal resource exhaustion, such as failure to allocate
69 > * internal task queues.
70   *
71   * <p>The primary method for awaiting completion and extracting
72   * results of a task is {@link #join}, but there are several variants:
73   * The {@link Future#get} methods support interruptible and/or timed
74   * waits for completion and report results using {@code Future}
75 < * conventions. Method {@link #helpJoin} enables callers to actively
76 < * execute other tasks while awaiting joins, which is sometimes more
77 < * efficient but only applies when all subtasks are known to be
78 < * strictly tree-structured. Method {@link #invoke} is semantically
64 < * equivalent to {@code fork(); join()} but always attempts to
65 < * begin execution in the current thread. The "<em>quiet</em>" forms
66 < * of these methods do not extract results or report exceptions. These
75 > * conventions. Method {@link #invoke} is semantically
76 > * equivalent to {@code fork(); join()} but always attempts to begin
77 > * execution in the current thread. The "<em>quiet</em>" forms of
78 > * these methods do not extract results or report exceptions. These
79   * may be useful when a set of tasks are being executed, and you need
80   * to delay processing of results or exceptions until all complete.
81   * Method {@code invokeAll} (available in multiple versions)
82   * performs the most common form of parallel invocation: forking a set
83   * of tasks and joining them all.
84   *
85 < * <p> The ForkJoinTask class is not usually directly subclassed.
85 > * <p>The execution status of tasks may be queried at several levels
86 > * of detail: {@link #isDone} is true if a task completed in any way
87 > * (including the case where a task was cancelled without executing);
88 > * {@link #isCompletedNormally} is true if a task completed without
89 > * cancellation or encountering an exception; {@link #isCancelled} is
90 > * true if the task was cancelled (in which case {@link #getException}
91 > * returns a {@link java.util.concurrent.CancellationException}); and
92 > * {@link #isCompletedAbnormally} is true if a task was either
93 > * cancelled or encountered an exception, in which case {@link
94 > * #getException} will return either the encountered exception or
95 > * {@link java.util.concurrent.CancellationException}.
96 > *
97 > * <p>The ForkJoinTask class is not usually directly subclassed.
98   * Instead, you subclass one of the abstract classes that support a
99 < * particular style of fork/join processing.  Normally, a concrete
99 > * particular style of fork/join processing, typically {@link
100 > * RecursiveAction} for computations that do not return results, or
101 > * {@link RecursiveTask} for those that do.  Normally, a concrete
102   * ForkJoinTask subclass declares fields comprising its parameters,
103   * established in a constructor, and then defines a {@code compute}
104   * method that somehow uses the control methods supplied by this base
105   * class. While these methods have {@code public} access (to allow
106 < * instances of different task subclasses to call each others
106 > * instances of different task subclasses to call each other's
107   * methods), some of them may only be called from within other
108   * ForkJoinTasks (as may be determined using method {@link
109   * #inForkJoinPool}).  Attempts to invoke them in other contexts
110   * result in exceptions or errors, possibly including
111 < * ClassCastException.
111 > * {@code ClassCastException}.
112 > *
113 > * <p>Method {@link #join} and its variants are appropriate for use
114 > * only when completion dependencies are acyclic; that is, the
115 > * parallel computation can be described as a directed acyclic graph
116 > * (DAG). Otherwise, executions may encounter a form of deadlock as
117 > * tasks cyclically wait for each other.  However, this framework
118 > * supports other methods and techniques (for example the use of
119 > * {@link Phaser}, {@link #helpQuiesce}, and {@link #complete}) that
120 > * may be of use in constructing custom subclasses for problems that
121 > * are not statically structured as DAGs.
122   *
123 < * <p>Most base support methods are {@code final} because their
124 < * implementations are intrinsically tied to the underlying
125 < * lightweight task scheduling framework, and so cannot be overridden.
126 < * Developers creating new basic styles of fork/join processing should
127 < * minimally implement {@code protected} methods
128 < * {@code exec}, {@code setRawResult}, and
129 < * {@code getRawResult}, while also introducing an abstract
130 < * computational method that can be implemented in its subclasses,
131 < * possibly relying on other {@code protected} methods provided
96 < * by this class.
123 > * <p>Most base support methods are {@code final}, to prevent
124 > * overriding of implementations that are intrinsically tied to the
125 > * underlying lightweight task scheduling framework.  Developers
126 > * creating new basic styles of fork/join processing should minimally
127 > * implement {@code protected} methods {@link #exec}, {@link
128 > * #setRawResult}, and {@link #getRawResult}, while also introducing
129 > * an abstract computational method that can be implemented in its
130 > * subclasses, possibly relying on other {@code protected} methods
131 > * provided by this class.
132   *
133   * <p>ForkJoinTasks should perform relatively small amounts of
134 < * computations, otherwise splitting into smaller tasks. As a very
135 < * rough rule of thumb, a task should perform more than 100 and less
136 < * than 10000 basic computational steps. If tasks are too big, then
137 < * parallelism cannot improve throughput. If too small, then memory
138 < * and internal task maintenance overhead may overwhelm processing.
134 > * computation. Large tasks should be split into smaller subtasks,
135 > * usually via recursive decomposition. As a very rough rule of thumb,
136 > * a task should perform more than 100 and less than 10000 basic
137 > * computational steps, and should avoid indefinite looping. If tasks
138 > * are too big, then parallelism cannot improve throughput. If too
139 > * small, then memory and internal task maintenance overhead may
140 > * overwhelm processing.
141   *
142 < * <p>ForkJoinTasks are {@code Serializable}, which enables them
143 < * to be used in extensions such as remote execution frameworks. It is
144 < * in general sensible to serialize tasks only before or after, but
145 < * not during execution. Serialization is not relied on during
146 < * execution itself.
142 > * <p>This class provides {@code adapt} methods for {@link Runnable}
143 > * and {@link Callable}, that may be of use when mixing execution of
144 > * {@code ForkJoinTasks} with other kinds of tasks. When all tasks are
145 > * of this form, consider using a pool constructed in <em>asyncMode</em>.
146 > *
147 > * <p>ForkJoinTasks are {@code Serializable}, which enables them to be
148 > * used in extensions such as remote execution frameworks. It is
149 > * sensible to serialize tasks only before or after, but not during,
150 > * execution. Serialization is not relied on during execution itself.
151   *
152   * @since 1.7
153   * @author Doug Lea
154   */
155   public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
156  
157 <    /**
158 <     * Run control status bits packed into a single int to minimize
159 <     * footprint and to ensure atomicity (via CAS).  Status is
160 <     * initially zero, and takes on nonnegative values until
161 <     * completed, upon which status holds COMPLETED. CANCELLED, or
162 <     * EXCEPTIONAL, which use the top 3 bits.  Tasks undergoing
163 <     * blocking waits by other threads have SIGNAL_MASK bits set --
164 <     * bit 15 for external (nonFJ) waits, and the rest a count of
165 <     * waiting FJ threads.  (This representation relies on
166 <     * ForkJoinPool max thread limits). Completion of a stolen task
167 <     * with SIGNAL_MASK bits set awakens waiter via notifyAll. Even
168 <     * though suboptimal for some purposes, we use basic builtin
169 <     * wait/notify to take advantage of "monitor inflation" in JVMs
170 <     * that we would otherwise need to emulate to avoid adding further
171 <     * per-task bookkeeping overhead. Note that bits 16-28 are
172 <     * currently unused. Also value 0x80000000 is available as spare
173 <     * completion value.
157 >    /*
158 >     * See the internal documentation of class ForkJoinPool for a
159 >     * general implementation overview.  ForkJoinTasks are mainly
160 >     * responsible for maintaining their "status" field amidst relays
161 >     * to methods in ForkJoinWorkerThread and ForkJoinPool. The
162 >     * methods of this class are more-or-less layered into (1) basic
163 >     * status maintenance (2) execution and awaiting completion (3)
164 >     * user-level methods that additionally report results. This is
165 >     * sometimes hard to see because this file orders exported methods
166 >     * in a way that flows well in javadocs. In particular, most
167 >     * join mechanics are in method quietlyJoin, below.
168 >     */
169 >
170 >    /*
171 >     * The status field holds run control status bits packed into a
172 >     * single int to minimize footprint and to ensure atomicity (via
173 >     * CAS).  Status is initially zero, and takes on nonnegative
174 >     * values until completed, upon which status holds value
175 >     * NORMAL, CANCELLED, or EXCEPTIONAL. Tasks undergoing blocking
176 >     * waits by other threads have the SIGNAL bit set.  Completion of
177 >     * a stolen task with SIGNAL set awakens any waiters via
178 >     * notifyAll. Even though suboptimal for some purposes, we use
179 >     * basic builtin wait/notify to take advantage of "monitor
180 >     * inflation" in JVMs that we would otherwise need to emulate to
181 >     * avoid adding further per-task bookkeeping overhead.  We want
182 >     * these monitors to be "fat", i.e., not use biasing or thin-lock
183 >     * techniques, so use some odd coding idioms that tend to avoid
184 >     * them.
185       */
186 +
187 +    /** The run status of this task */
188      volatile int status; // accessed directly by pool and workers
189  
190 <    static final int COMPLETION_MASK      = 0xe0000000;
191 <    static final int NORMAL               = 0xe0000000; // == mask
192 <    static final int CANCELLED            = 0xc0000000;
193 <    static final int EXCEPTIONAL          = 0xa0000000;
140 <    static final int SIGNAL_MASK          = 0x0000ffff;
141 <    static final int INTERNAL_SIGNAL_MASK = 0x00007fff;
142 <    static final int EXTERNAL_SIGNAL      = 0x00008000; // top bit of low word
190 >    private static final int NORMAL      = -1;
191 >    private static final int CANCELLED   = -2;
192 >    private static final int EXCEPTIONAL = -3;
193 >    private static final int SIGNAL      =  1;
194  
195      /**
196       * Table of exceptions thrown by tasks, to enable reporting by
# Line 153 | Line 204 | public abstract class ForkJoinTask<V> im
204          Collections.synchronizedMap
205          (new WeakHashMap<ForkJoinTask<?>, Throwable>());
206  
207 <    // within-package utilities
207 >    // Maintaining completion status
208  
209      /**
210 <     * Gets current worker thread, or null if not a worker thread.
211 <     */
161 <    static ForkJoinWorkerThread getWorker() {
162 <        Thread t = Thread.currentThread();
163 <        return ((t instanceof ForkJoinWorkerThread) ?
164 <                (ForkJoinWorkerThread) t : null);
165 <    }
166 <
167 <    final boolean casStatus(int cmp, int val) {
168 <        return UNSAFE.compareAndSwapInt(this, statusOffset, cmp, val);
169 <    }
170 <
171 <    /**
172 <     * Workaround for not being able to rethrow unchecked exceptions.
173 <     */
174 <    static void rethrowException(Throwable ex) {
175 <        if (ex != null)
176 <            UNSAFE.throwException(ex);
177 <    }
178 <
179 <    // Setting completion status
180 <
181 <    /**
182 <     * Marks completion and wakes up threads waiting to join this task.
210 >     * Marks completion and wakes up threads waiting to join this task,
211 >     * also clearing signal request bits.
212       *
213       * @param completion one of NORMAL, CANCELLED, EXCEPTIONAL
214       */
215 <    final void setCompletion(int completion) {
187 <        ForkJoinPool pool = getPool();
188 <        if (pool != null) {
189 <            int s; // Clear signal bits while setting completion status
190 <            do {} while ((s = status) >= 0 && !casStatus(s, completion));
191 <
192 <            if ((s & SIGNAL_MASK) != 0) {
193 <                if ((s &= INTERNAL_SIGNAL_MASK) != 0)
194 <                    pool.updateRunningCount(s);
195 <                synchronized (this) { notifyAll(); }
196 <            }
197 <        }
198 <        else
199 <            externallySetCompletion(completion);
200 <    }
201 <
202 <    /**
203 <     * Version of setCompletion for non-FJ threads.  Leaves signal
204 <     * bits for unblocked threads to adjust, and always notifies.
205 <     */
206 <    private void externallySetCompletion(int completion) {
207 <        int s;
208 <        do {} while ((s = status) >= 0 &&
209 <                     !casStatus(s, (s & SIGNAL_MASK) | completion));
210 <        synchronized (this) { notifyAll(); }
211 <    }
212 <
213 <    /**
214 <     * Sets status to indicate normal completion.
215 <     */
216 <    final void setNormalCompletion() {
217 <        // Try typical fast case -- single CAS, no signal, not already done.
218 <        // Manually expand casStatus to improve chances of inlining it
219 <        if (!UNSAFE.compareAndSwapInt(this, statusOffset, 0, NORMAL))
220 <            setCompletion(NORMAL);
221 <    }
222 <
223 <    // internal waiting and notification
224 <
225 <    /**
226 <     * Performs the actual monitor wait for awaitDone.
227 <     */
228 <    private void doAwaitDone() {
229 <        // Minimize lock bias and in/de-flation effects by maximizing
230 <        // chances of waiting inside sync
231 <        try {
232 <            while (status >= 0)
233 <                synchronized (this) { if (status >= 0) wait(); }
234 <        } catch (InterruptedException ie) {
235 <            onInterruptedWait();
236 <        }
237 <    }
238 <
239 <    /**
240 <     * Performs the actual timed monitor wait for awaitDone.
241 <     */
242 <    private void doAwaitDone(long startTime, long nanos) {
243 <        synchronized (this) {
244 <            try {
245 <                while (status >= 0) {
246 <                    long nt = nanos - System.nanoTime() - startTime;
247 <                    if (nt <= 0)
248 <                        break;
249 <                    wait(nt / 1000000, (int) (nt % 1000000));
250 <                }
251 <            } catch (InterruptedException ie) {
252 <                onInterruptedWait();
253 <            }
254 <        }
255 <    }
256 <
257 <    // Awaiting completion
258 <
259 <    /**
260 <     * Sets status to indicate there is joiner, then waits for join,
261 <     * surrounded with pool notifications.
262 <     *
263 <     * @return status upon exit
264 <     */
265 <    private int awaitDone(ForkJoinWorkerThread w,
266 <                          boolean maintainParallelism) {
267 <        ForkJoinPool pool = (w == null) ? null : w.pool;
215 >    private void setCompletion(int completion) {
216          int s;
217          while ((s = status) >= 0) {
218 <            if (casStatus(s, (pool == null) ? s|EXTERNAL_SIGNAL : s+1)) {
219 <                if (pool == null || !pool.preJoin(this, maintainParallelism))
220 <                    doAwaitDone();
273 <                if (((s = status) & INTERNAL_SIGNAL_MASK) != 0)
274 <                    adjustPoolCountsOnUnblock(pool);
218 >            if (UNSAFE.compareAndSwapInt(this, statusOffset, s, completion)) {
219 >                if (s != 0)
220 >                    synchronized (this) { notifyAll(); }
221                  break;
222              }
223          }
278        return s;
224      }
225  
226      /**
227 <     * Timed version of awaitDone
227 >     * Records exception and sets exceptional completion.
228       *
229 <     * @return status upon exit
229 >     * @return status on exit
230       */
231 <    private int awaitDone(ForkJoinWorkerThread w, long nanos) {
287 <        ForkJoinPool pool = (w == null) ? null : w.pool;
288 <        int s;
289 <        while ((s = status) >= 0) {
290 <            if (casStatus(s, (pool == null) ? s|EXTERNAL_SIGNAL : s+1)) {
291 <                long startTime = System.nanoTime();
292 <                if (pool == null || !pool.preJoin(this, false))
293 <                    doAwaitDone(startTime, nanos);
294 <                if ((s = status) >= 0) {
295 <                    adjustPoolCountsOnCancelledWait(pool);
296 <                    s = status;
297 <                }
298 <                if (s < 0 && (s & INTERNAL_SIGNAL_MASK) != 0)
299 <                    adjustPoolCountsOnUnblock(pool);
300 <                break;
301 <            }
302 <        }
303 <        return s;
304 <    }
305 <
306 <    /**
307 <     * Notifies pool that thread is unblocked. Called by signalled
308 <     * threads when woken by non-FJ threads (which is atypical).
309 <     */
310 <    private void adjustPoolCountsOnUnblock(ForkJoinPool pool) {
311 <        int s;
312 <        do {} while ((s = status) < 0 && !casStatus(s, s & COMPLETION_MASK));
313 <        if (pool != null && (s &= INTERNAL_SIGNAL_MASK) != 0)
314 <            pool.updateRunningCount(s);
315 <    }
316 <
317 <    /**
318 <     * Notifies pool to adjust counts on cancelled or timed out wait.
319 <     */
320 <    private void adjustPoolCountsOnCancelledWait(ForkJoinPool pool) {
321 <        if (pool != null) {
322 <            int s;
323 <            while ((s = status) >= 0 && (s & INTERNAL_SIGNAL_MASK) != 0) {
324 <                if (casStatus(s, s - 1)) {
325 <                    pool.updateRunningCount(1);
326 <                    break;
327 <                }
328 <            }
329 <        }
330 <    }
331 <
332 <    /**
333 <     * Handles interruptions during waits.
334 <     */
335 <    private void onInterruptedWait() {
336 <        ForkJoinWorkerThread w = getWorker();
337 <        if (w == null)
338 <            Thread.currentThread().interrupt(); // re-interrupt
339 <        else if (w.isTerminating())
340 <            cancelIgnoringExceptions();
341 <        // else if FJworker, ignore interrupt
342 <    }
343 <
344 <    // Recording and reporting exceptions
345 <
346 <    private void setDoneExceptionally(Throwable rex) {
231 >    private void setExceptionalCompletion(Throwable rex) {
232          exceptionMap.put(this, rex);
233          setCompletion(EXCEPTIONAL);
234      }
235  
236      /**
237 <     * Throws the exception associated with status s.
238 <     *
354 <     * @throws the exception
237 >     * Blocks a worker thread until completed or timed out.  Called
238 >     * only by pool.
239       */
240 <    private void reportException(int s) {
241 <        if ((s &= COMPLETION_MASK) < NORMAL) {
242 <            if (s == CANCELLED)
243 <                throw new CancellationException();
244 <            else
245 <                rethrowException(exceptionMap.get(this));
240 >    final void internalAwaitDone(long millis, int nanos) {
241 >        if (status >= 0) {
242 >            try {     // the odd construction reduces lock bias effects
243 >                synchronized (this) {
244 >                    if (status > 0 ||
245 >                        UNSAFE.compareAndSwapInt(this, statusOffset,
246 >                                                 0, SIGNAL))
247 >                        wait(millis, nanos);
248 >                }
249 >            } catch (InterruptedException ie) {
250 >                cancelIfTerminating();
251 >            }
252          }
253      }
254  
255      /**
256 <     * Returns result or throws exception using j.u.c.Future conventions.
367 <     * Only call when {@code isDone} known to be true.
256 >     * Blocks a non-worker-thread until completion.
257       */
258 <    private V reportFutureResult()
259 <        throws ExecutionException, InterruptedException {
260 <        int s = status & COMPLETION_MASK;
261 <        if (s < NORMAL) {
262 <            Throwable ex;
263 <            if (s == CANCELLED)
264 <                throw new CancellationException();
265 <            if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
266 <                throw new ExecutionException(ex);
267 <            if (Thread.interrupted())
268 <                throw new InterruptedException();
258 >    private void externalAwaitDone() {
259 >        if (status >= 0) {
260 >            boolean interrupted = false;
261 >            synchronized(this) {
262 >                int s;
263 >                while ((s = status) >= 0) {
264 >                    if (s == 0 &&
265 >                        !UNSAFE.compareAndSwapInt(this, statusOffset,
266 >                                                  0, SIGNAL))
267 >                        continue;
268 >                    try {
269 >                        wait();
270 >                    } catch (InterruptedException ie) {
271 >                        interrupted = true;
272 >                    }
273 >                }
274 >            }
275 >            if (interrupted)
276 >                Thread.currentThread().interrupt();
277          }
381        return getRawResult();
278      }
279  
280      /**
281 <     * Returns result or throws exception using j.u.c.Future conventions
386 <     * with timeouts.
281 >     * Blocks a non-worker-thread until completion or interruption or timeout.
282       */
283 <    private V reportTimedFutureResult()
284 <        throws InterruptedException, ExecutionException, TimeoutException {
390 <        Throwable ex;
391 <        int s = status & COMPLETION_MASK;
392 <        if (s == NORMAL)
393 <            return getRawResult();
394 <        if (s == CANCELLED)
395 <            throw new CancellationException();
396 <        if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
397 <            throw new ExecutionException(ex);
283 >    private void externalInterruptibleAwaitDone(boolean timed, long nanos)
284 >        throws InterruptedException {
285          if (Thread.interrupted())
286              throw new InterruptedException();
400        throw new TimeoutException();
401    }
402
403    // internal execution methods
404
405    /**
406     * Calls exec, recording completion, and rethrowing exception if
407     * encountered. Caller should normally check status before calling.
408     *
409     * @return true if completed normally
410     */
411    private boolean tryExec() {
412        try { // try block must contain only call to exec
413            if (!exec())
414                return false;
415        } catch (Throwable rex) {
416            setDoneExceptionally(rex);
417            rethrowException(rex);
418            return false; // not reached
419        }
420        setNormalCompletion();
421        return true;
422    }
423
424    /**
425     * Main execution method used by worker threads. Invokes
426     * base computation unless already complete.
427     */
428    final void quietlyExec() {
287          if (status >= 0) {
288 <            try {
289 <                if (!exec())
290 <                    return;
291 <            } catch (Throwable rex) {
292 <                setDoneExceptionally(rex);
293 <                return;
288 >            long startTime = timed ? System.nanoTime() : 0L;
289 >            synchronized(this) {
290 >                int s;
291 >                while ((s = status) >= 0) {
292 >                    long nt;
293 >                    if (s == 0 &&
294 >                        !UNSAFE.compareAndSwapInt(this, statusOffset,
295 >                                                  0, SIGNAL))
296 >                        continue;
297 >                    else if (!timed)
298 >                        wait();
299 >                    else if ((nt = nanos - (System.nanoTime()-startTime)) > 0L)
300 >                        wait(nt / 1000000, (int)(nt % 1000000));
301 >                    else
302 >                        break;
303 >                }
304              }
437            setNormalCompletion();
305          }
306      }
307  
308      /**
309 <     * Calls exec(), recording but not rethrowing exception.
310 <     * Caller should normally check status before calling.
311 <     *
445 <     * @return true if completed normally
309 >     * Unless done, calls exec and records status if completed, but
310 >     * doesn't wait for completion otherwise. Primary execution method
311 >     * for ForkJoinWorkerThread.
312       */
313 <    private boolean tryQuietlyInvoke() {
313 >    final void quietlyExec() {
314          try {
315 <            if (!exec())
316 <                return false;
315 >            if (status < 0 || !exec())
316 >                return;
317          } catch (Throwable rex) {
318 <            setDoneExceptionally(rex);
319 <            return false;
454 <        }
455 <        setNormalCompletion();
456 <        return true;
457 <    }
458 <
459 <    /**
460 <     * Cancels, ignoring any exceptions it throws.
461 <     */
462 <    final void cancelIgnoringExceptions() {
463 <        try {
464 <            cancel(false);
465 <        } catch (Throwable ignore) {
318 >            setExceptionalCompletion(rex);
319 >            return;
320          }
321 <    }
468 <
469 <    /**
470 <     * Main implementation of helpJoin
471 <     */
472 <    private int busyJoin(ForkJoinWorkerThread w) {
473 <        int s;
474 <        ForkJoinTask<?> t;
475 <        while ((s = status) >= 0 && (t = w.scanWhileJoining(this)) != null)
476 <            t.quietlyExec();
477 <        return (s >= 0) ? awaitDone(w, false) : s; // block if no work
321 >        setCompletion(NORMAL); // must be outside try block
322      }
323  
324      // public methods
# Line 482 | Line 326 | public abstract class ForkJoinTask<V> im
326      /**
327       * Arranges to asynchronously execute this task.  While it is not
328       * necessarily enforced, it is a usage error to fork a task more
329 <     * than once unless it has completed and been reinitialized.  This
330 <     * method may be invoked only from within ForkJoinTask
331 <     * computations (as may be determined using method {@link
332 <     * #inForkJoinPool}). Attempts to invoke in other contexts result
333 <     * in exceptions or errors, possibly including ClassCastException.
329 >     * than once unless it has completed and been reinitialized.
330 >     * Subsequent modifications to the state of this task or any data
331 >     * it operates on are not necessarily consistently observable by
332 >     * any thread other than the one executing it unless preceded by a
333 >     * call to {@link #join} or related methods, or a call to {@link
334 >     * #isDone} returning {@code true}.
335 >     *
336 >     * <p>This method may be invoked only from within {@code
337 >     * ForkJoinPool} computations (as may be determined using method
338 >     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
339 >     * result in exceptions or errors, possibly including {@code
340 >     * ClassCastException}.
341 >     *
342 >     * @return {@code this}, to simplify usage
343       */
344 <    public final void fork() {
344 >    public final ForkJoinTask<V> fork() {
345          ((ForkJoinWorkerThread) Thread.currentThread())
346              .pushTask(this);
347 +        return this;
348      }
349  
350      /**
351 <     * Returns the result of the computation when it is ready.
352 <     * This method differs from {@code get} in that abnormal
353 <     * completion results in RuntimeExceptions or Errors, not
354 <     * ExecutionExceptions.
351 >     * Returns the result of the computation when it {@link #isDone is
352 >     * done}.  This method differs from {@link #get()} in that
353 >     * abnormal completion results in {@code RuntimeException} or
354 >     * {@code Error}, not {@code ExecutionException}, and that
355 >     * interrupts of the calling thread do <em>not</em> cause the
356 >     * method to abruptly return by throwing {@code
357 >     * InterruptedException}.
358       *
359       * @return the computed result
360       */
361      public final V join() {
362 <        ForkJoinWorkerThread w = getWorker();
363 <        if (w == null || status < 0 || !w.unpushTask(this) || !tryExec())
364 <            reportException(awaitDone(w, true));
362 >        quietlyJoin();
363 >        Throwable ex;
364 >        if (status < NORMAL && (ex = getException()) != null)
365 >            UNSAFE.throwException(ex);
366          return getRawResult();
367      }
368  
369      /**
370       * Commences performing this task, awaits its completion if
371 <     * necessary, and return its result.
371 >     * necessary, and returns its result, or throws an (unchecked)
372 >     * {@code RuntimeException} or {@code Error} if the underlying
373 >     * computation did so.
374       *
515     * @throws Throwable (a RuntimeException, Error, or unchecked
516     * exception) if the underlying computation did so
375       * @return the computed result
376       */
377      public final V invoke() {
378 <        if (status >= 0 && tryExec())
379 <            return getRawResult();
380 <        else
381 <            return join();
378 >        quietlyInvoke();
379 >        Throwable ex;
380 >        if (status < NORMAL && (ex = getException()) != null)
381 >            UNSAFE.throwException(ex);
382 >        return getRawResult();
383      }
384  
385      /**
386 <     * Forks both tasks, returning when {@code isDone} holds for
387 <     * both of them or an exception is encountered. This method may be
388 <     * invoked only from within ForkJoinTask computations (as may be
389 <     * determined using method {@link #inForkJoinPool}). Attempts to
390 <     * invoke in other contexts result in exceptions or errors,
391 <     * possibly including ClassCastException.
392 <     *
393 <     * @param t1 one task
394 <     * @param t2 the other task
395 <     * @throws NullPointerException if t1 or t2 are null
396 <     * @throws RuntimeException or Error if either task did so
386 >     * Forks the given tasks, returning when {@code isDone} holds for
387 >     * each task or an (unchecked) exception is encountered, in which
388 >     * case the exception is rethrown. If more than one task
389 >     * encounters an exception, then this method throws any one of
390 >     * these exceptions. If any task encounters an exception, the
391 >     * other may be cancelled. However, the execution status of
392 >     * individual tasks is not guaranteed upon exceptional return. The
393 >     * status of each task may be obtained using {@link
394 >     * #getException()} and related methods to check if they have been
395 >     * cancelled, completed normally or exceptionally, or left
396 >     * unprocessed.
397 >     *
398 >     * <p>This method may be invoked only from within {@code
399 >     * ForkJoinPool} computations (as may be determined using method
400 >     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
401 >     * result in exceptions or errors, possibly including {@code
402 >     * ClassCastException}.
403 >     *
404 >     * @param t1 the first task
405 >     * @param t2 the second task
406 >     * @throws NullPointerException if any task is null
407       */
408 <    public static void invokeAll(ForkJoinTask<?>t1, ForkJoinTask<?> t2) {
408 >    public static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) {
409          t2.fork();
410          t1.invoke();
411          t2.join();
412      }
413  
414      /**
415 <     * Forks the given tasks, returning when {@code isDone} holds
416 <     * for all of them. If any task encounters an exception, others
417 <     * may be cancelled.  This method may be invoked only from within
418 <     * ForkJoinTask computations (as may be determined using method
419 <     * {@link #inForkJoinPool}). Attempts to invoke in other contexts
420 <     * result in exceptions or errors, possibly including
421 <     * ClassCastException.
415 >     * Forks the given tasks, returning when {@code isDone} holds for
416 >     * each task or an (unchecked) exception is encountered, in which
417 >     * case the exception is rethrown. If more than one task
418 >     * encounters an exception, then this method throws any one of
419 >     * these exceptions. If any task encounters an exception, others
420 >     * may be cancelled. However, the execution status of individual
421 >     * tasks is not guaranteed upon exceptional return. The status of
422 >     * each task may be obtained using {@link #getException()} and
423 >     * related methods to check if they have been cancelled, completed
424 >     * normally or exceptionally, or left unprocessed.
425 >     *
426 >     * <p>This method may be invoked only from within {@code
427 >     * ForkJoinPool} computations (as may be determined using method
428 >     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
429 >     * result in exceptions or errors, possibly including {@code
430 >     * ClassCastException}.
431       *
432 <     * @param tasks the array of tasks
433 <     * @throws NullPointerException if tasks or any element are null
556 <     * @throws RuntimeException or Error if any task did so
432 >     * @param tasks the tasks
433 >     * @throws NullPointerException if any task is null
434       */
435      public static void invokeAll(ForkJoinTask<?>... tasks) {
436          Throwable ex = null;
# Line 568 | Line 445 | public abstract class ForkJoinTask<V> im
445                  t.fork();
446              else {
447                  t.quietlyInvoke();
448 <                if (ex == null)
448 >                if (ex == null && t.status < NORMAL)
449                      ex = t.getException();
450              }
451          }
# Line 579 | Line 456 | public abstract class ForkJoinTask<V> im
456                      t.cancel(false);
457                  else {
458                      t.quietlyJoin();
459 <                    if (ex == null)
459 >                    if (ex == null && t.status < NORMAL)
460                          ex = t.getException();
461                  }
462              }
463          }
464          if (ex != null)
465 <            rethrowException(ex);
465 >            UNSAFE.throwException(ex);
466      }
467  
468      /**
469 <     * Forks all tasks in the collection, returning when
470 <     * {@code isDone} holds for all of them. If any task
471 <     * encounters an exception, others may be cancelled.  This method
472 <     * may be invoked only from within ForkJoinTask computations (as
473 <     * may be determined using method {@link
474 <     * #inForkJoinPool}). Attempts to invoke in other contexts result
475 <     * in exceptions or errors, possibly including ClassCastException.
469 >     * Forks all tasks in the specified collection, returning when
470 >     * {@code isDone} holds for each task or an (unchecked) exception
471 >     * is encountered, in which case the exception is rethrown. If
472 >     * more than one task encounters an exception, then this method
473 >     * throws any one of these exceptions. If any task encounters an
474 >     * exception, others may be cancelled. However, the execution
475 >     * status of individual tasks is not guaranteed upon exceptional
476 >     * return. The status of each task may be obtained using {@link
477 >     * #getException()} and related methods to check if they have been
478 >     * cancelled, completed normally or exceptionally, or left
479 >     * unprocessed.
480 >     *
481 >     * <p>This method may be invoked only from within {@code
482 >     * ForkJoinPool} computations (as may be determined using method
483 >     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
484 >     * result in exceptions or errors, possibly including {@code
485 >     * ClassCastException}.
486       *
487       * @param tasks the collection of tasks
488 +     * @return the tasks argument, to simplify usage
489       * @throws NullPointerException if tasks or any element are null
602     * @throws RuntimeException or Error if any task did so
490       */
491 <    public static void invokeAll(Collection<? extends ForkJoinTask<?>> tasks) {
492 <        if (!(tasks instanceof List<?>)) {
491 >    public static <T extends ForkJoinTask<?>> Collection<T> invokeAll(Collection<T> tasks) {
492 >        if (!(tasks instanceof RandomAccess) || !(tasks instanceof List<?>)) {
493              invokeAll(tasks.toArray(new ForkJoinTask<?>[tasks.size()]));
494 <            return;
494 >            return tasks;
495          }
496          @SuppressWarnings("unchecked")
497          List<? extends ForkJoinTask<?>> ts =
# Line 621 | Line 508 | public abstract class ForkJoinTask<V> im
508                  t.fork();
509              else {
510                  t.quietlyInvoke();
511 <                if (ex == null)
511 >                if (ex == null && t.status < NORMAL)
512                      ex = t.getException();
513              }
514          }
# Line 632 | Line 519 | public abstract class ForkJoinTask<V> im
519                      t.cancel(false);
520                  else {
521                      t.quietlyJoin();
522 <                    if (ex == null)
522 >                    if (ex == null && t.status < NORMAL)
523                          ex = t.getException();
524                  }
525              }
526          }
527          if (ex != null)
528 <            rethrowException(ex);
528 >            UNSAFE.throwException(ex);
529 >        return tasks;
530      }
531  
532      /**
533 <     * Returns true if the computation performed by this task has
534 <     * completed (or has been cancelled).
533 >     * Attempts to cancel execution of this task. This attempt will
534 >     * fail if the task has already completed or could not be
535 >     * cancelled for some other reason. If successful, and this task
536 >     * has not started when {@code cancel} is called, execution of
537 >     * this task is suppressed. After this method returns
538 >     * successfully, unless there is an intervening call to {@link
539 >     * #reinitialize}, subsequent calls to {@link #isCancelled},
540 >     * {@link #isDone}, and {@code cancel} will return {@code true}
541 >     * and calls to {@link #join} and related methods will result in
542 >     * {@code CancellationException}.
543 >     *
544 >     * <p>This method may be overridden in subclasses, but if so, must
545 >     * still ensure that these properties hold. In particular, the
546 >     * {@code cancel} method itself must not throw exceptions.
547 >     *
548 >     * <p>This method is designed to be invoked by <em>other</em>
549 >     * tasks. To terminate the current task, you can just return or
550 >     * throw an unchecked exception from its computation method, or
551 >     * invoke {@link #completeExceptionally}.
552 >     *
553 >     * @param mayInterruptIfRunning this value has no effect in the
554 >     * default implementation because interrupts are not used to
555 >     * control cancellation.
556       *
557 <     * @return true if this computation has completed
557 >     * @return {@code true} if this task is now cancelled
558       */
559 <    public final boolean isDone() {
560 <        return status < 0;
559 >    public boolean cancel(boolean mayInterruptIfRunning) {
560 >        setCompletion(CANCELLED);
561 >        return status == CANCELLED;
562      }
563  
564      /**
565 <     * Returns true if this task was cancelled.
566 <     *
567 <     * @return true if this task was cancelled
565 >     * Cancels, ignoring any exceptions thrown by cancel. Used during
566 >     * worker and pool shutdown. Cancel is spec'ed not to throw any
567 >     * exceptions, but if it does anyway, we have no recourse during
568 >     * shutdown, so guard against this case.
569       */
570 +    final void cancelIgnoringExceptions() {
571 +        try {
572 +            cancel(false);
573 +        } catch (Throwable ignore) {
574 +        }
575 +    }
576 +
577 +    /**
578 +     * Cancels if current thread is a terminating worker thread,
579 +     * ignoring any exceptions thrown by cancel.
580 +     */
581 +    final void cancelIfTerminating() {
582 +        Thread t = Thread.currentThread();
583 +        if ((t instanceof ForkJoinWorkerThread) &&
584 +            ((ForkJoinWorkerThread) t).isTerminating()) {
585 +            try {
586 +                cancel(false);
587 +            } catch (Throwable ignore) {
588 +            }
589 +        }
590 +    }
591 +
592 +    public final boolean isDone() {
593 +        return status < 0;
594 +    }
595 +
596      public final boolean isCancelled() {
597 <        return (status & COMPLETION_MASK) == CANCELLED;
597 >        return status == CANCELLED;
598      }
599  
600      /**
601 <     * Asserts that the results of this task's computation will not be
665 <     * used. If a cancellation occurs before attempting to execute this
666 <     * task, then execution will be suppressed, {@code isCancelled}
667 <     * will report true, and {@code join} will result in a
668 <     * {@code CancellationException} being thrown. Otherwise, when
669 <     * cancellation races with completion, there are no guarantees
670 <     * about whether {@code isCancelled} will report true, whether
671 <     * {@code join} will return normally or via an exception, or
672 <     * whether these behaviors will remain consistent upon repeated
673 <     * invocation.
674 <     *
675 <     * <p>This method may be overridden in subclasses, but if so, must
676 <     * still ensure that these minimal properties hold. In particular,
677 <     * the cancel method itself must not throw exceptions.
678 <     *
679 <     * <p> This method is designed to be invoked by <em>other</em>
680 <     * tasks. To terminate the current task, you can just return or
681 <     * throw an unchecked exception from its computation method, or
682 <     * invoke {@code completeExceptionally}.
601 >     * Returns {@code true} if this task threw an exception or was cancelled.
602       *
603 <     * @param mayInterruptIfRunning this value is ignored in the
685 <     * default implementation because tasks are not in general
686 <     * cancelled via interruption
687 <     *
688 <     * @return true if this task is now cancelled
603 >     * @return {@code true} if this task threw an exception or was cancelled
604       */
605 <    public boolean cancel(boolean mayInterruptIfRunning) {
606 <        setCompletion(CANCELLED);
692 <        return (status & COMPLETION_MASK) == CANCELLED;
605 >    public final boolean isCompletedAbnormally() {
606 >        return status < NORMAL;
607      }
608  
609      /**
610 <     * Returns true if this task threw an exception or was cancelled.
610 >     * Returns {@code true} if this task completed without throwing an
611 >     * exception and was not cancelled.
612       *
613 <     * @return true if this task threw an exception or was cancelled
613 >     * @return {@code true} if this task completed without throwing an
614 >     * exception and was not cancelled
615       */
616 <    public final boolean isCompletedAbnormally() {
617 <        return (status & COMPLETION_MASK) < NORMAL;
616 >    public final boolean isCompletedNormally() {
617 >        return status == NORMAL;
618      }
619  
620      /**
621       * Returns the exception thrown by the base computation, or a
622 <     * CancellationException if cancelled, or null if none or if the
623 <     * method has not yet completed.
622 >     * {@code CancellationException} if cancelled, or {@code null} if
623 >     * none or if the method has not yet completed.
624       *
625 <     * @return the exception, or null if none
625 >     * @return the exception, or {@code null} if none
626       */
627      public final Throwable getException() {
628 <        int s = status & COMPLETION_MASK;
629 <        if (s >= NORMAL)
630 <            return null;
631 <        if (s == CANCELLED)
716 <            return new CancellationException();
717 <        return exceptionMap.get(this);
628 >        int s = status;
629 >        return ((s >= NORMAL)    ? null :
630 >                (s == CANCELLED) ? new CancellationException() :
631 >                exceptionMap.get(this));
632      }
633  
634      /**
# Line 723 | Line 637 | public abstract class ForkJoinTask<V> im
637       * {@code join} and related operations. This method may be used
638       * to induce exceptions in asynchronous tasks, or to force
639       * completion of tasks that would not otherwise complete.  Its use
640 <     * in other situations is likely to be wrong.  This method is
640 >     * in other situations is discouraged.  This method is
641       * overridable, but overridden versions must invoke {@code super}
642       * implementation to maintain guarantees.
643       *
644 <     * @param ex the exception to throw. If this exception is
645 <     * not a RuntimeException or Error, the actual exception thrown
646 <     * will be a RuntimeException with cause ex.
644 >     * @param ex the exception to throw. If this exception is not a
645 >     * {@code RuntimeException} or {@code Error}, the actual exception
646 >     * thrown will be a {@code RuntimeException} with cause {@code ex}.
647       */
648      public void completeExceptionally(Throwable ex) {
649 <        setDoneExceptionally((ex instanceof RuntimeException) ||
650 <                             (ex instanceof Error) ? ex :
651 <                             new RuntimeException(ex));
649 >        setExceptionalCompletion((ex instanceof RuntimeException) ||
650 >                                 (ex instanceof Error) ? ex :
651 >                                 new RuntimeException(ex));
652      }
653  
654      /**
655       * Completes this task, and if not already aborted or cancelled,
656 <     * returning a {@code null} result upon {@code join} and related
657 <     * operations. This method may be used to provide results for
658 <     * asynchronous tasks, or to provide alternative handling for
659 <     * tasks that would not otherwise complete normally. Its use in
660 <     * other situations is likely to be wrong. This method is
661 <     * overridable, but overridden versions must invoke {@code super}
662 <     * implementation to maintain guarantees.
656 >     * returning the given value as the result of subsequent
657 >     * invocations of {@code join} and related operations. This method
658 >     * may be used to provide results for asynchronous tasks, or to
659 >     * provide alternative handling for tasks that would not otherwise
660 >     * complete normally. Its use in other situations is
661 >     * discouraged. This method is overridable, but overridden
662 >     * versions must invoke {@code super} implementation to maintain
663 >     * guarantees.
664       *
665       * @param value the result value for this task
666       */
# Line 753 | Line 668 | public abstract class ForkJoinTask<V> im
668          try {
669              setRawResult(value);
670          } catch (Throwable rex) {
671 <            setDoneExceptionally(rex);
671 >            setExceptionalCompletion(rex);
672              return;
673          }
674 <        setNormalCompletion();
760 <    }
761 <
762 <    public final V get() throws InterruptedException, ExecutionException {
763 <        ForkJoinWorkerThread w = getWorker();
764 <        if (w == null || status < 0 || !w.unpushTask(this) || !tryQuietlyInvoke())
765 <            awaitDone(w, true);
766 <        return reportFutureResult();
767 <    }
768 <
769 <    public final V get(long timeout, TimeUnit unit)
770 <        throws InterruptedException, ExecutionException, TimeoutException {
771 <        ForkJoinWorkerThread w = getWorker();
772 <        if (w == null || status < 0 || !w.unpushTask(this) || !tryQuietlyInvoke())
773 <            awaitDone(w, unit.toNanos(timeout));
774 <        return reportTimedFutureResult();
674 >        setCompletion(NORMAL);
675      }
676  
677      /**
678 <     * Possibly executes other tasks until this task is ready, then
679 <     * returns the result of the computation.  This method may be more
780 <     * efficient than {@code join}, but is only applicable when
781 <     * there are no potential dependencies between continuation of the
782 <     * current task and that of any other task that might be executed
783 <     * while helping. (This usually holds for pure divide-and-conquer
784 <     * tasks). This method may be invoked only from within
785 <     * ForkJoinTask computations (as may be determined using method
786 <     * {@link #inForkJoinPool}). Attempts to invoke in other contexts
787 <     * result in exceptions or errors, possibly including
788 <     * ClassCastException.
678 >     * Waits if necessary for the computation to complete, and then
679 >     * retrieves its result.
680       *
681       * @return the computed result
682 +     * @throws CancellationException if the computation was cancelled
683 +     * @throws ExecutionException if the computation threw an
684 +     * exception
685 +     * @throws InterruptedException if the current thread is not a
686 +     * member of a ForkJoinPool and was interrupted while waiting
687       */
688 <    public final V helpJoin() {
689 <        ForkJoinWorkerThread w = (ForkJoinWorkerThread) Thread.currentThread();
690 <        if (status < 0 || !w.unpushTask(this) || !tryExec())
691 <            reportException(busyJoin(w));
688 >    public final V get() throws InterruptedException, ExecutionException {
689 >        Thread t = Thread.currentThread();
690 >        if (t instanceof ForkJoinWorkerThread)
691 >            quietlyJoin();
692 >        else
693 >            externalInterruptibleAwaitDone(false, 0L);
694 >        int s = status;
695 >        if (s != NORMAL) {
696 >            Throwable ex;
697 >            if (s == CANCELLED)
698 >                throw new CancellationException();
699 >            if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
700 >                throw new ExecutionException(ex);
701 >        }
702          return getRawResult();
703      }
704  
705      /**
706 <     * Possibly executes other tasks until this task is ready.  This
707 <     * method may be invoked only from within ForkJoinTask
708 <     * computations (as may be determined using method {@link
709 <     * #inForkJoinPool}). Attempts to invoke in other contexts result
710 <     * in exceptions or errors, possibly including ClassCastException.
706 >     * Waits if necessary for at most the given time for the computation
707 >     * to complete, and then retrieves its result, if available.
708 >     *
709 >     * @param timeout the maximum time to wait
710 >     * @param unit the time unit of the timeout argument
711 >     * @return the computed result
712 >     * @throws CancellationException if the computation was cancelled
713 >     * @throws ExecutionException if the computation threw an
714 >     * exception
715 >     * @throws InterruptedException if the current thread is not a
716 >     * member of a ForkJoinPool and was interrupted while waiting
717 >     * @throws TimeoutException if the wait timed out
718       */
719 <    public final void quietlyHelpJoin() {
720 <        if (status >= 0) {
721 <            ForkJoinWorkerThread w =
722 <                (ForkJoinWorkerThread) Thread.currentThread();
723 <            if (!w.unpushTask(this) || !tryQuietlyInvoke())
724 <                busyJoin(w);
719 >    public final V get(long timeout, TimeUnit unit)
720 >        throws InterruptedException, ExecutionException, TimeoutException {
721 >        long nanos = unit.toNanos(timeout);
722 >        Thread t = Thread.currentThread();
723 >        if (t instanceof ForkJoinWorkerThread)
724 >            ((ForkJoinWorkerThread)t).joinTask(this, true, nanos);
725 >        else
726 >            externalInterruptibleAwaitDone(true, nanos);
727 >        int s = status;
728 >        if (s != NORMAL) {
729 >            Throwable ex;
730 >            if (s == CANCELLED)
731 >                throw new CancellationException();
732 >            if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
733 >                throw new ExecutionException(ex);
734 >            throw new TimeoutException();
735          }
736 +        return getRawResult();
737      }
738  
739      /**
740 <     * Joins this task, without returning its result or throwing an
740 >     * Joins this task, without returning its result or throwing its
741       * exception. This method may be useful when processing
742       * collections of tasks when some have been cancelled or otherwise
743       * known to have aborted.
744       */
745      public final void quietlyJoin() {
746 <        if (status >= 0) {
747 <            ForkJoinWorkerThread w = getWorker();
748 <            if (w == null || !w.unpushTask(this) || !tryQuietlyInvoke())
749 <                awaitDone(w, true);
746 >        Thread t;
747 >        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
748 >            ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
749 >            if (status >= 0) {
750 >                if (w.unpushTask(this)) {
751 >                    boolean completed;
752 >                    try {
753 >                        completed = exec();
754 >                    } catch (Throwable rex) {
755 >                        setExceptionalCompletion(rex);
756 >                        return;
757 >                    }
758 >                    if (completed) {
759 >                        setCompletion(NORMAL);
760 >                        return;
761 >                    }
762 >                }
763 >                w.joinTask(this, false, 0L);
764 >            }
765          }
766 +        else
767 +            externalAwaitDone();
768      }
769  
770      /**
771       * Commences performing this task and awaits its completion if
772 <     * necessary, without returning its result or throwing an
773 <     * exception. This method may be useful when processing
833 <     * collections of tasks when some have been cancelled or otherwise
834 <     * known to have aborted.
772 >     * necessary, without returning its result or throwing its
773 >     * exception.
774       */
775      public final void quietlyInvoke() {
776 <        if (status >= 0 && !tryQuietlyInvoke())
777 <            quietlyJoin();
776 >        if (status >= 0) {
777 >            boolean completed;
778 >            try {
779 >                completed = exec();
780 >            } catch (Throwable rex) {
781 >                setExceptionalCompletion(rex);
782 >                return;
783 >            }
784 >            if (completed)
785 >                setCompletion(NORMAL);
786 >            else
787 >                quietlyJoin();
788 >        }
789      }
790  
791      /**
792       * Possibly executes tasks until the pool hosting the current task
793 <     * {@link ForkJoinPool#isQuiescent}. This method may be of use in
794 <     * designs in which many tasks are forked, but none are explicitly
795 <     * joined, instead executing them until all are processed.
793 >     * {@link ForkJoinPool#isQuiescent is quiescent}. This method may
794 >     * be of use in designs in which many tasks are forked, but none
795 >     * are explicitly joined, instead executing them until all are
796 >     * processed.
797 >     *
798 >     * <p>This method may be invoked only from within {@code
799 >     * ForkJoinPool} computations (as may be determined using method
800 >     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
801 >     * result in exceptions or errors, possibly including {@code
802 >     * ClassCastException}.
803       */
804      public static void helpQuiesce() {
805          ((ForkJoinWorkerThread) Thread.currentThread())
# Line 855 | Line 812 | public abstract class ForkJoinTask<V> im
812       * this task, but only if reuse occurs when this task has either
813       * never been forked, or has been forked, then completed and all
814       * outstanding joins of this task have also completed. Effects
815 <     * under any other usage conditions are not guaranteed, and are
816 <     * almost surely wrong. This method may be useful when executing
815 >     * under any other usage conditions are not guaranteed.
816 >     * This method may be useful when executing
817       * pre-constructed trees of subtasks in loops.
818 +     *
819 +     * <p>Upon completion of this method, {@code isDone()} reports
820 +     * {@code false}, and {@code getException()} reports {@code
821 +     * null}. However, the value returned by {@code getRawResult} is
822 +     * unaffected. To clear this value, you can invoke {@code
823 +     * setRawResult(null)}.
824       */
825      public void reinitialize() {
826 <        if ((status & COMPLETION_MASK) == EXCEPTIONAL)
826 >        if (status == EXCEPTIONAL)
827              exceptionMap.remove(this);
828          status = 0;
829      }
# Line 869 | Line 832 | public abstract class ForkJoinTask<V> im
832       * Returns the pool hosting the current task execution, or null
833       * if this task is executing outside of any ForkJoinPool.
834       *
835 <     * @return the pool, or null if none
835 >     * @see #inForkJoinPool
836 >     * @return the pool, or {@code null} if none
837       */
838      public static ForkJoinPool getPool() {
839          Thread t = Thread.currentThread();
# Line 878 | Line 842 | public abstract class ForkJoinTask<V> im
842      }
843  
844      /**
845 <     * Returns {@code true} if the current thread is executing as a
846 <     * ForkJoinPool computation.
845 >     * Returns {@code true} if the current thread is a {@link
846 >     * ForkJoinWorkerThread} executing as a ForkJoinPool computation.
847       *
848 <     * @return {@code true} if the current thread is executing as a
849 <     * ForkJoinPool computation, or false otherwise
848 >     * @return {@code true} if the current thread is a {@link
849 >     * ForkJoinWorkerThread} executing as a ForkJoinPool computation,
850 >     * or {@code false} otherwise
851       */
852      public static boolean inForkJoinPool() {
853          return Thread.currentThread() instanceof ForkJoinWorkerThread;
# Line 894 | Line 859 | public abstract class ForkJoinTask<V> im
859       * by the current thread, and has not commenced executing in
860       * another thread.  This method may be useful when arranging
861       * alternative local processing of tasks that could have been, but
862 <     * were not, stolen. This method may be invoked only from within
898 <     * ForkJoinTask computations (as may be determined using method
899 <     * {@link #inForkJoinPool}). Attempts to invoke in other contexts
900 <     * result in exceptions or errors, possibly including
901 <     * ClassCastException.
862 >     * were not, stolen.
863       *
864 <     * @return true if unforked
864 >     * <p>This method may be invoked only from within {@code
865 >     * ForkJoinPool} computations (as may be determined using method
866 >     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
867 >     * result in exceptions or errors, possibly including {@code
868 >     * ClassCastException}.
869 >     *
870 >     * @return {@code true} if unforked
871       */
872      public boolean tryUnfork() {
873          return ((ForkJoinWorkerThread) Thread.currentThread())
# Line 913 | Line 880 | public abstract class ForkJoinTask<V> im
880       * value may be useful for heuristic decisions about whether to
881       * fork other tasks.
882       *
883 +     * <p>This method may be invoked only from within {@code
884 +     * ForkJoinPool} computations (as may be determined using method
885 +     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
886 +     * result in exceptions or errors, possibly including {@code
887 +     * ClassCastException}.
888 +     *
889       * @return the number of tasks
890       */
891      public static int getQueuedTaskCount() {
# Line 930 | Line 903 | public abstract class ForkJoinTask<V> im
903       * tasks, and to process computations locally if this threshold is
904       * exceeded.
905       *
906 +     * <p>This method may be invoked only from within {@code
907 +     * ForkJoinPool} computations (as may be determined using method
908 +     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
909 +     * result in exceptions or errors, possibly including {@code
910 +     * ClassCastException}.
911 +     *
912       * @return the surplus number of tasks, which may be negative
913       */
914      public static int getSurplusQueuedTaskCount() {
# Line 940 | Line 919 | public abstract class ForkJoinTask<V> im
919      // Extension methods
920  
921      /**
922 <     * Returns the result that would be returned by {@code join},
923 <     * even if this task completed abnormally, or null if this task is
924 <     * not known to have been completed.  This method is designed to
925 <     * aid debugging, as well as to support extensions. Its use in any
926 <     * other context is discouraged.
922 >     * Returns the result that would be returned by {@link #join}, even
923 >     * if this task completed abnormally, or {@code null} if this task
924 >     * is not known to have been completed.  This method is designed
925 >     * to aid debugging, as well as to support extensions. Its use in
926 >     * any other context is discouraged.
927       *
928 <     * @return the result, or null if not completed
928 >     * @return the result, or {@code null} if not completed
929       */
930      public abstract V getRawResult();
931  
# Line 965 | Line 944 | public abstract class ForkJoinTask<V> im
944       * called otherwise. The return value controls whether this task
945       * is considered to be done normally. It may return false in
946       * asynchronous actions that require explicit invocations of
947 <     * {@code complete} to become joinable. It may throw exceptions
948 <     * to indicate abnormal exit.
947 >     * {@link #complete} to become joinable. It may also throw an
948 >     * (unchecked) exception to indicate abnormal exit.
949       *
950 <     * @return true if completed normally
972 <     * @throws Error or RuntimeException if encountered during computation
950 >     * @return {@code true} if completed normally
951       */
952      protected abstract boolean exec();
953  
954      /**
955 <     * Returns, but does not unschedule or execute, the task queued by
956 <     * the current thread but not yet executed, if one is
955 >     * Returns, but does not unschedule or execute, a task queued by
956 >     * the current thread but not yet executed, if one is immediately
957       * available. There is no guarantee that this task will actually
958 <     * be polled or executed next.  This method is designed primarily
959 <     * to support extensions, and is unlikely to be useful otherwise.
960 <     * This method may be invoked only from within ForkJoinTask
961 <     * computations (as may be determined using method {@link
962 <     * #inForkJoinPool}). Attempts to invoke in other contexts result
963 <     * in exceptions or errors, possibly including ClassCastException.
958 >     * be polled or executed next. Conversely, this method may return
959 >     * null even if a task exists but cannot be accessed without
960 >     * contention with other threads.  This method is designed
961 >     * primarily to support extensions, and is unlikely to be useful
962 >     * otherwise.
963 >     *
964 >     * <p>This method may be invoked only from within {@code
965 >     * ForkJoinPool} computations (as may be determined using method
966 >     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
967 >     * result in exceptions or errors, possibly including {@code
968 >     * ClassCastException}.
969       *
970 <     * @return the next task, or null if none are available
970 >     * @return the next task, or {@code null} if none are available
971       */
972      protected static ForkJoinTask<?> peekNextLocalTask() {
973          return ((ForkJoinWorkerThread) Thread.currentThread())
# Line 995 | Line 978 | public abstract class ForkJoinTask<V> im
978       * Unschedules and returns, without executing, the next task
979       * queued by the current thread but not yet executed.  This method
980       * is designed primarily to support extensions, and is unlikely to
981 <     * be useful otherwise.  This method may be invoked only from
999 <     * within ForkJoinTask computations (as may be determined using
1000 <     * method {@link #inForkJoinPool}). Attempts to invoke in other
1001 <     * contexts result in exceptions or errors, possibly including
1002 <     * ClassCastException.
981 >     * be useful otherwise.
982       *
983 <     * @return the next task, or null if none are available
983 >     * <p>This method may be invoked only from within {@code
984 >     * ForkJoinPool} computations (as may be determined using method
985 >     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
986 >     * result in exceptions or errors, possibly including {@code
987 >     * ClassCastException}.
988 >     *
989 >     * @return the next task, or {@code null} if none are available
990       */
991      protected static ForkJoinTask<?> pollNextLocalTask() {
992          return ((ForkJoinWorkerThread) Thread.currentThread())
# Line 1016 | Line 1001 | public abstract class ForkJoinTask<V> im
1001       * {@code null} result does not necessarily imply quiescence
1002       * of the pool this task is operating in.  This method is designed
1003       * primarily to support extensions, and is unlikely to be useful
1004 <     * otherwise.  This method may be invoked only from within
1005 <     * ForkJoinTask computations (as may be determined using method
1006 <     * {@link #inForkJoinPool}). Attempts to invoke in other contexts
1007 <     * result in exceptions or errors, possibly including
1008 <     * ClassCastException.
1004 >     * otherwise.
1005 >     *
1006 >     * <p>This method may be invoked only from within {@code
1007 >     * ForkJoinPool} computations (as may be determined using method
1008 >     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1009 >     * result in exceptions or errors, possibly including {@code
1010 >     * ClassCastException}.
1011       *
1012 <     * @return a task, or null if none are available
1012 >     * @return a task, or {@code null} if none are available
1013       */
1014      protected static ForkJoinTask<?> pollTask() {
1015          return ((ForkJoinWorkerThread) Thread.currentThread())
1016              .pollTask();
1017      }
1018  
1019 +    /**
1020 +     * Adaptor for Runnables. This implements RunnableFuture
1021 +     * to be compliant with AbstractExecutorService constraints
1022 +     * when used in ForkJoinPool.
1023 +     */
1024 +    static final class AdaptedRunnable<T> extends ForkJoinTask<T>
1025 +        implements RunnableFuture<T> {
1026 +        final Runnable runnable;
1027 +        final T resultOnCompletion;
1028 +        T result;
1029 +        AdaptedRunnable(Runnable runnable, T result) {
1030 +            if (runnable == null) throw new NullPointerException();
1031 +            this.runnable = runnable;
1032 +            this.resultOnCompletion = result;
1033 +        }
1034 +        public T getRawResult() { return result; }
1035 +        public void setRawResult(T v) { result = v; }
1036 +        public boolean exec() {
1037 +            runnable.run();
1038 +            result = resultOnCompletion;
1039 +            return true;
1040 +        }
1041 +        public void run() { invoke(); }
1042 +        private static final long serialVersionUID = 5232453952276885070L;
1043 +    }
1044 +
1045 +    /**
1046 +     * Adaptor for Callables
1047 +     */
1048 +    static final class AdaptedCallable<T> extends ForkJoinTask<T>
1049 +        implements RunnableFuture<T> {
1050 +        final Callable<? extends T> callable;
1051 +        T result;
1052 +        AdaptedCallable(Callable<? extends T> callable) {
1053 +            if (callable == null) throw new NullPointerException();
1054 +            this.callable = callable;
1055 +        }
1056 +        public T getRawResult() { return result; }
1057 +        public void setRawResult(T v) { result = v; }
1058 +        public boolean exec() {
1059 +            try {
1060 +                result = callable.call();
1061 +                return true;
1062 +            } catch (Error err) {
1063 +                throw err;
1064 +            } catch (RuntimeException rex) {
1065 +                throw rex;
1066 +            } catch (Exception ex) {
1067 +                throw new RuntimeException(ex);
1068 +            }
1069 +        }
1070 +        public void run() { invoke(); }
1071 +        private static final long serialVersionUID = 2838392045355241008L;
1072 +    }
1073 +
1074 +    /**
1075 +     * Returns a new {@code ForkJoinTask} that performs the {@code run}
1076 +     * method of the given {@code Runnable} as its action, and returns
1077 +     * a null result upon {@link #join}.
1078 +     *
1079 +     * @param runnable the runnable action
1080 +     * @return the task
1081 +     */
1082 +    public static ForkJoinTask<?> adapt(Runnable runnable) {
1083 +        return new AdaptedRunnable<Void>(runnable, null);
1084 +    }
1085 +
1086 +    /**
1087 +     * Returns a new {@code ForkJoinTask} that performs the {@code run}
1088 +     * method of the given {@code Runnable} as its action, and returns
1089 +     * the given result upon {@link #join}.
1090 +     *
1091 +     * @param runnable the runnable action
1092 +     * @param result the result upon completion
1093 +     * @return the task
1094 +     */
1095 +    public static <T> ForkJoinTask<T> adapt(Runnable runnable, T result) {
1096 +        return new AdaptedRunnable<T>(runnable, result);
1097 +    }
1098 +
1099 +    /**
1100 +     * Returns a new {@code ForkJoinTask} that performs the {@code call}
1101 +     * method of the given {@code Callable} as its action, and returns
1102 +     * its result upon {@link #join}, translating any checked exceptions
1103 +     * encountered into {@code RuntimeException}.
1104 +     *
1105 +     * @param callable the callable action
1106 +     * @return the task
1107 +     */
1108 +    public static <T> ForkJoinTask<T> adapt(Callable<? extends T> callable) {
1109 +        return new AdaptedCallable<T>(callable);
1110 +    }
1111 +
1112      // Serialization support
1113  
1114      private static final long serialVersionUID = -7721805057305804111L;
1115  
1116      /**
1117 <     * Save the state to a stream.
1117 >     * Saves the state to a stream (that is, serializes it).
1118       *
1119       * @serialData the current run status and the exception thrown
1120 <     * during execution, or null if none
1120 >     * during execution, or {@code null} if none
1121       * @param s the stream
1122       */
1123      private void writeObject(java.io.ObjectOutputStream s)
# Line 1047 | Line 1127 | public abstract class ForkJoinTask<V> im
1127      }
1128  
1129      /**
1130 <     * Reconstitute the instance from a stream.
1130 >     * Reconstitutes the instance from a stream (that is, deserializes it).
1131       *
1132       * @param s the stream
1133       */
1134      private void readObject(java.io.ObjectInputStream s)
1135          throws java.io.IOException, ClassNotFoundException {
1136          s.defaultReadObject();
1057        status &= ~INTERNAL_SIGNAL_MASK; // clear internal signal counts
1058        status |= EXTERNAL_SIGNAL; // conservatively set external signal
1137          Object ex = s.readObject();
1138          if (ex != null)
1139 <            setDoneExceptionally((Throwable) ex);
1139 >            setExceptionalCompletion((Throwable) ex);
1140 >    }
1141 >
1142 >    // Unsafe mechanics
1143 >
1144 >    private static final sun.misc.Unsafe UNSAFE = getUnsafe();
1145 >    private static final long statusOffset =
1146 >        objectFieldOffset("status", ForkJoinTask.class);
1147 >
1148 >    private static long objectFieldOffset(String field, Class<?> klazz) {
1149 >        try {
1150 >            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
1151 >        } catch (NoSuchFieldException e) {
1152 >            // Convert Exception to corresponding Error
1153 >            NoSuchFieldError error = new NoSuchFieldError(field);
1154 >            error.initCause(e);
1155 >            throw error;
1156 >        }
1157      }
1158  
1159 <    // Unsafe mechanics for jsr166y 3rd party package.
1159 >    /**
1160 >     * Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
1161 >     * Replace with a simple call to Unsafe.getUnsafe when integrating
1162 >     * into a jdk.
1163 >     *
1164 >     * @return a sun.misc.Unsafe
1165 >     */
1166      private static sun.misc.Unsafe getUnsafe() {
1167          try {
1168              return sun.misc.Unsafe.getUnsafe();
1169          } catch (SecurityException se) {
1170              try {
1171                  return java.security.AccessController.doPrivileged
1172 <                    (new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1172 >                    (new java.security
1173 >                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
1174                          public sun.misc.Unsafe run() throws Exception {
1175 <                            return getUnsafeByReflection();
1175 >                            java.lang.reflect.Field f = sun.misc
1176 >                                .Unsafe.class.getDeclaredField("theUnsafe");
1177 >                            f.setAccessible(true);
1178 >                            return (sun.misc.Unsafe) f.get(null);
1179                          }});
1180              } catch (java.security.PrivilegedActionException e) {
1181                  throw new RuntimeException("Could not initialize intrinsics",
# Line 1078 | Line 1183 | public abstract class ForkJoinTask<V> im
1183              }
1184          }
1185      }
1081
1082    private static sun.misc.Unsafe getUnsafeByReflection()
1083            throws NoSuchFieldException, IllegalAccessException {
1084        java.lang.reflect.Field f =
1085            sun.misc.Unsafe.class.getDeclaredField("theUnsafe");
1086        f.setAccessible(true);
1087        return (sun.misc.Unsafe) f.get(null);
1088    }
1089
1090    private static long fieldOffset(String fieldName, Class<?> klazz) {
1091        try {
1092            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(fieldName));
1093        } catch (NoSuchFieldException e) {
1094            // Convert Exception to Error
1095            NoSuchFieldError error = new NoSuchFieldError(fieldName);
1096            error.initCause(e);
1097            throw error;
1098        }
1099    }
1100
1101    private static final sun.misc.Unsafe UNSAFE = getUnsafe();
1102    static final long statusOffset =
1103        fieldOffset("status", ForkJoinTask.class);
1104
1186   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines