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.47 by dl, Sun Apr 18 12:51:18 2010 UTC vs.
Revision 1.70 by dl, Tue Nov 23 00:10:39 2010 UTC

# Line 6 | Line 6
6  
7   package jsr166y;
8  
9 import java.util.concurrent.*;
10
9   import java.io.Serializable;
10   import java.util.Collection;
11   import java.util.Collections;
# Line 15 | Line 13 | 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 ForkJoinPool}.
# Line 28 | Line 36 | import java.util.WeakHashMap;
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}.  However, this class also provides a number of other
40 < * methods that can come into play in advanced usages, as well as
41 < * extension mechanics that allow support of new forms of fork/join
42 < * processing.
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 {@code ForkJoinTask} is a lightweight form of {@link Future}.
45   * The efficiency of {@code ForkJoinTask}s stems from a set of
# Line 64 | Line 72 | import java.util.WeakHashMap;
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
68 < * execute other tasks while awaiting joins, which is sometimes more
69 < * efficient but only applies when all subtasks are known to be
70 < * strictly tree-structured. Method {@link #invoke} is semantically
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
# Line 103 | Line 108 | import java.util.WeakHashMap;
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}, to prevent
124   * overriding of implementations that are intrinsically tied to the
# Line 119 | Line 134 | import java.util.WeakHashMap;
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. If tasks are too big, then parallelism cannot
138 < * improve throughput. If too small, then memory and internal task
139 < * maintenance overhead may overwhelm processing.
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>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
145 < * are of this form, consider using a pool in
130 < * {@linkplain ForkJoinPool#setAsyncMode async mode}.
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
# Line 148 | Line 163 | public abstract class ForkJoinTask<V> im
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.
166 >     * in a way that flows well in javadocs. In particular, most
167 >     * join mechanics are in method quietlyJoin, below.
168       */
169  
170 <    /**
171 <     * Run control status bits packed into a single int to minimize
172 <     * footprint and to ensure atomicity (via CAS).  Status is
173 <     * initially zero, and takes on nonnegative values until
174 <     * completed, upon which status holds COMPLETED. CANCELLED, or
175 <     * EXCEPTIONAL, which use the top 3 bits.  Tasks undergoing
176 <     * blocking waits by other threads have SIGNAL_MASK bits set --
177 <     * bit 15 for external (nonFJ) waits, and the rest a count of
178 <     * waiting FJ threads.  (This representation relies on
179 <     * ForkJoinPool max thread limits). Signal counts are not directly
180 <     * incremented by ForkJoinTask methods, but instead via a call to
181 <     * requestSignal within ForkJoinPool.preJoin, once their need is
182 <     * established.
183 <     *
184 <     * Completion of a stolen task with SIGNAL_MASK bits set awakens
169 <     * any waiters via notifyAll. Even though suboptimal for some
170 <     * purposes, we use basic builtin wait/notify to take advantage of
171 <     * "monitor inflation" in JVMs that we would otherwise need to
172 <     * emulate to avoid adding further per-task bookkeeping overhead.
173 <     * We want these monitors to be "fat", i.e., not use biasing or
174 <     * thin-lock techniques, so use some odd coding idioms that tend
175 <     * to avoid them.
176 <     *
177 <     * Note that bits 16-28 are currently unused. Also value
178 <     * 0x80000000 is available as spare completion value.
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 <    private static final int COMPLETION_MASK      = 0xe0000000;
191 <    private static final int NORMAL               = 0xe0000000; // == mask
192 <    private static final int CANCELLED            = 0xc0000000;
193 <    private static final int EXCEPTIONAL          = 0xa0000000;
186 <    private static final int SIGNAL_MASK          = 0x0000ffff;
187 <    private static final int INTERNAL_SIGNAL_MASK = 0x00007fff;
188 <    private static final int EXTERNAL_SIGNAL      = 0x00008000;
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 211 | Line 216 | public abstract class ForkJoinTask<V> im
216          int s;
217          while ((s = status) >= 0) {
218              if (UNSAFE.compareAndSwapInt(this, statusOffset, s, completion)) {
219 <                if ((s & SIGNAL_MASK) != 0) {
215 <                    Thread t = Thread.currentThread();
216 <                    if (t instanceof ForkJoinWorkerThread)
217 <                        ((ForkJoinWorkerThread) t).pool.updateRunningCount
218 <                            (s & INTERNAL_SIGNAL_MASK);
219 >                if (s != 0)
220                      synchronized (this) { notifyAll(); }
221 <                }
221 <                return;
221 >                break;
222              }
223          }
224      }
225  
226      /**
227 <     * Record exception and set exceptional completion
227 >     * Records exception and sets exceptional completion.
228 >     *
229 >     * @return status on exit
230       */
231 <    private void setDoneExceptionally(Throwable rex) {
231 >    private void setExceptionalCompletion(Throwable rex) {
232          exceptionMap.put(this, rex);
233          setCompletion(EXCEPTIONAL);
234      }
235  
236      /**
237 <     * Main internal execution method: Unless done, calls exec and
238 <     * records completion.
237 <     *
238 <     * @return true if ran and completed normally
237 >     * Blocks a worker thread until completed or timed out.  Called
238 >     * only by pool.
239       */
240 <    final boolean tryExec() {
241 <        try {
242 <            if (status < 0 || !exec())
243 <                return false;
244 <        } catch (Throwable rex) {
245 <            setDoneExceptionally(rex);
246 <            return false;
247 <        }
248 <        setCompletion(NORMAL); // must be outside try block
249 <        return true;
250 <    }
251 <
252 <    /**
253 <     * Increments internal signal count (thus requesting signal upon
254 <     * completion) unless already done.  Call only once per join.
255 <     * Used by ForkJoinPool.preJoin.
256 <     *
257 <     * @return status
258 <     */
259 <    final int requestSignal() {
260 <        int s;
261 <        do {} while ((s = status) >= 0 &&
262 <                     !UNSAFE.compareAndSwapInt(this, statusOffset, s, s + 1));
263 <        return s;
264 <    }
265 <
266 <    /**
267 <     * Sets external signal request unless already done.
268 <     *
269 <     * @return status
270 <     */
271 <    private int requestExternalSignal() {
272 <        int s;
273 <        do {} while ((s = status) >= 0 &&
274 <                     !UNSAFE.compareAndSwapInt(this, statusOffset,
275 <                                               s, s | EXTERNAL_SIGNAL));
276 <        return s;
277 <    }
278 <
279 <    /*
280 <     * Awaiting completion. The four versions, internal vs external X
281 <     * untimed vs timed, have the same overall structure but differ
282 <     * from each other enough to defy simple integration.
283 <     */
284 <
285 <    /**
286 <     * Blocks a worker until this task is done, also maintaining pool
287 <     * and signal counts
288 <     */
289 <    private void awaitDone(ForkJoinWorkerThread w) {
240 >    final void internalAwaitDone(long millis, int nanos) {
241          if (status >= 0) {
242 <            w.pool.preJoin(this);
243 <            while (status >= 0) {
244 <                try { // minimize lock scope
245 <                    synchronized(this) {
246 <                        if (status >= 0)
247 <                            wait();
297 <                        else { // help release; also helps avoid lock-biasing
298 <                            notifyAll();
299 <                            break;
300 <                        }
301 <                    }
302 <                } catch (InterruptedException ie) {
303 <                    cancelIfTerminating();
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 <     * Blocks a non-ForkJoin thread until this task is done.
256 >     * Blocks a non-worker-thread until completion.
257       */
258      private void externalAwaitDone() {
313        if (requestExternalSignal() >= 0) {
314            boolean interrupted = false;
315            while (status >= 0) {
316                try {
317                    synchronized(this) {
318                        if (status >= 0)
319                            wait();
320                        else {
321                            notifyAll();
322                            break;
323                        }
324                    }
325                } catch (InterruptedException ie) {
326                    interrupted = true;
327                }
328            }
329            if (interrupted)
330                Thread.currentThread().interrupt();
331        }
332    }
333
334    /**
335     * Blocks a worker until this task is done or timeout elapses
336     */
337    private void timedAwaitDone(ForkJoinWorkerThread w, long nanos) {
259          if (status >= 0) {
260 <            long startTime = System.nanoTime();
261 <            ForkJoinPool pool = w.pool;
262 <            pool.preJoin(this);
263 <            while (status >= 0) {
264 <                long nt = nanos - (System.nanoTime() - startTime);
265 <                if (nt > 0) {
266 <                    long ms = nt / 1000000;
267 <                    int ns = (int) (nt % 1000000);
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 <                        synchronized(this) { if (status >= 0) wait(ms, ns); }
269 >                        wait();
270                      } catch (InterruptedException ie) {
271 <                        cancelIfTerminating();
271 >                        interrupted = true;
272                      }
273                  }
353                else {
354                    int s; // adjust running count on timeout
355                    while ((s = status) >= 0 &&
356                           (s & INTERNAL_SIGNAL_MASK) != 0) {
357                        if (UNSAFE.compareAndSwapInt(this, statusOffset,
358                                                     s, s - 1)) {
359                            pool.updateRunningCount(1);
360                            break;
361                        }
362                    }
363                    break;
364                }
365            }
366        }
367    }
368
369    /**
370     * Blocks a non-ForkJoin thread until this task is done or timeout elapses
371     */
372    private void externalTimedAwaitDone(long nanos) {
373        if (requestExternalSignal() >= 0) {
374            long startTime = System.nanoTime();
375            boolean interrupted = false;
376            while (status >= 0) {
377                long nt = nanos - (System.nanoTime() - startTime);
378                if (nt <= 0)
379                    break;
380                long ms = nt / 1000000;
381                int ns = (int) (nt % 1000000);
382                try {
383                    synchronized(this) { if (status >= 0) wait(ms, ns); }
384                } catch (InterruptedException ie) {
385                    interrupted = true;
386                }
274              }
275              if (interrupted)
276                  Thread.currentThread().interrupt();
277          }
278      }
279  
393    // reporting results
394
395    /**
396     * Returns result or throws the exception associated with status.
397     * Uses Unsafe as a workaround for javac not allowing rethrow of
398     * unchecked exceptions.
399     */
400    private V reportResult() {
401        if ((status & COMPLETION_MASK) < NORMAL) {
402            Throwable ex = getException();
403            if (ex != null)
404                UNSAFE.throwException(ex);
405        }
406        return getRawResult();
407    }
408
280      /**
281 <     * Returns result or throws exception using j.u.c.Future conventions.
411 <     * Only call when {@code isDone} known to be true or thread known
412 <     * to be interrupted.
281 >     * Blocks a non-worker-thread until completion or interruption or timeout
282       */
283 <    private V reportFutureResult()
284 <        throws InterruptedException, ExecutionException {
283 >    private void externalInterruptibleAwaitDone(boolean timed, long nanos)
284 >        throws InterruptedException {
285          if (Thread.interrupted())
286              throw new InterruptedException();
287 <        int s = status & COMPLETION_MASK;
288 <        if (s < NORMAL) {
289 <            Throwable ex;
290 <            if (s == CANCELLED)
291 <                throw new CancellationException();
292 <            if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
293 <                throw new ExecutionException(ex);
287 >        if (status >= 0) {
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 >            }
305          }
426        return getRawResult();
306      }
307  
308      /**
309 <     * Returns result or throws exception using j.u.c.Future conventions
310 <     * with timeouts.
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 V reportTimedFutureResult()
314 <        throws InterruptedException, ExecutionException, TimeoutException {
315 <        if (Thread.interrupted())
316 <            throw new InterruptedException();
317 <        Throwable ex;
318 <        int s = status & COMPLETION_MASK;
319 <        if (s == NORMAL)
320 <            return getRawResult();
321 <        else if (s == CANCELLED)
442 <            throw new CancellationException();
443 <        else if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
444 <            throw new ExecutionException(ex);
445 <        else
446 <            throw new TimeoutException();
313 >    final void quietlyExec() {
314 >        try {
315 >            if (status < 0 || !exec())
316 >                return;
317 >        } catch (Throwable rex) {
318 >            setExceptionalCompletion(rex);
319 >            return;
320 >        }
321 >        setCompletion(NORMAL); // must be outside try block
322      }
323  
324      // public methods
# Line 459 | Line 334 | public abstract class ForkJoinTask<V> im
334       * #isDone} returning {@code true}.
335       *
336       * <p>This method may be invoked only from within {@code
337 <     * ForkJoinTask} computations (as may be determined using method
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}.
# Line 473 | Line 348 | public abstract class ForkJoinTask<V> im
348      }
349  
350      /**
351 <     * Returns the result of the computation when it {@link #isDone is done}.
352 <     * This method differs from {@link #get()} in that
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}.
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          quietlyJoin();
363 <        return reportResult();
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, or throws an (unchecked)
372 <     * exception if the underlying computation did so.
371 >     * necessary, and returns its result, or throws an (unchecked)
372 >     * {@code RuntimeException} or {@code Error} if the underlying
373 >     * computation did so.
374       *
375       * @return the computed result
376       */
377      public final V invoke() {
378 <        if (!tryExec())
379 <            quietlyJoin();
380 <        return reportResult();
378 >        quietlyInvoke();
379 >        Throwable ex;
380 >        if (status < NORMAL && (ex = getException()) != null)
381 >            UNSAFE.throwException(ex);
382 >        return getRawResult();
383      }
384  
385      /**
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 either task encounters an
389 <     * exception, the other one may be, but is not guaranteed to be,
390 <     * cancelled.  If both tasks throw an exception, then this method
391 <     * throws one of them.  The individual status of each task may be
392 <     * checked using {@link #getException()} and related methods.
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 <     * ForkJoinTask} computations (as may be determined using method
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}.
# Line 526 | Line 414 | public abstract class ForkJoinTask<V> im
414      /**
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 any task encounters an
418 <     * exception, others may be, but are not guaranteed to be,
419 <     * cancelled.  If more than one task encounters an exception, then
420 <     * this method throws any one of these exceptions.  The individual
421 <     * status of each task may be checked using {@link #getException()}
422 <     * and related methods.
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 <     * ForkJoinTask} computations (as may be determined using method
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}.
# Line 555 | 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 566 | 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              }
# Line 578 | Line 468 | public abstract class ForkJoinTask<V> im
468      /**
469       * Forks all tasks in the specified collection, returning when
470       * {@code isDone} holds for each task or an (unchecked) exception
471 <     * is encountered.  If any task encounters an exception, others
472 <     * may be, but are not guaranteed to be, cancelled.  If more than
473 <     * one task encounters an exception, then this method throws any
474 <     * one of these exceptions.  The individual status of each task
475 <     * may be checked using {@link #getException()} and related
476 <     * methods.  The behavior of this operation is undefined if the
477 <     * specified collection is modified while the operation is in
478 <     * progress.
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 <     * ForkJoinTask} computations (as may be determined using method
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}.
# Line 617 | 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 628 | 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              }
# Line 640 | Line 531 | public abstract class ForkJoinTask<V> im
531  
532      /**
533       * Attempts to cancel execution of this task. This attempt will
534 <     * fail if the task has already completed, has already been
535 <     * cancelled, or could not be cancelled for some other reason. If
536 <     * successful, and this task has not started when cancel is
537 <     * called, execution of this task is suppressed, {@link
538 <     * #isCancelled} will report true, and {@link #join} will result
539 <     * in a {@code CancellationException} being thrown.
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 minimal properties hold. In particular,
546 <     * the {@code cancel} method itself must not throw exceptions.
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 is ignored in the
554 <     * default implementation because tasks are not
555 <     * cancelled via interruption
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 {@code true} if this task is now cancelled
558       */
559      public boolean cancel(boolean mayInterruptIfRunning) {
560          setCompletion(CANCELLED);
561 <        return (status & COMPLETION_MASK) == CANCELLED;
561 >        return status == CANCELLED;
562      }
563  
564      /**
565 <     * Cancels, ignoring any exceptions it throws. Used during worker
566 <     * and pool shutdown.
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 {
# Line 679 | Line 575 | public abstract class ForkJoinTask<V> im
575      }
576  
577      /**
578 <     * Cancels ignoring exceptions if worker is terminating
578 >     * Cancels if current thread is a terminating worker thread,
579 >     * ignoring any exceptions thrown by cancel.
580       */
581 <    private void cancelIfTerminating() {
581 >    final void cancelIfTerminating() {
582          Thread t = Thread.currentThread();
583          if ((t instanceof ForkJoinWorkerThread) &&
584              ((ForkJoinWorkerThread) t).isTerminating()) {
# Line 697 | Line 594 | public abstract class ForkJoinTask<V> im
594      }
595  
596      public final boolean isCancelled() {
597 <        return (status & COMPLETION_MASK) == CANCELLED;
597 >        return status == CANCELLED;
598      }
599  
600      /**
# Line 706 | Line 603 | public abstract class ForkJoinTask<V> im
603       * @return {@code true} if this task threw an exception or was cancelled
604       */
605      public final boolean isCompletedAbnormally() {
606 <        return (status & COMPLETION_MASK) < NORMAL;
606 >        return status < NORMAL;
607      }
608  
609      /**
# Line 717 | Line 614 | public abstract class ForkJoinTask<V> im
614       * exception and was not cancelled
615       */
616      public final boolean isCompletedNormally() {
617 <        return (status & COMPLETION_MASK) == NORMAL;
617 >        return status == NORMAL;
618      }
619  
620      /**
# Line 728 | Line 625 | public abstract class ForkJoinTask<V> im
625       * @return the exception, or {@code null} if none
626       */
627      public final Throwable getException() {
628 <        int s = status & COMPLETION_MASK;
628 >        int s = status;
629          return ((s >= NORMAL)    ? null :
630                  (s == CANCELLED) ? new CancellationException() :
631                  exceptionMap.get(this));
# Line 749 | Line 646 | public abstract class ForkJoinTask<V> im
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 discouraged. 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 770 | 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          setCompletion(NORMAL);
675      }
676  
779    public final V get() throws InterruptedException, ExecutionException {
780        quietlyJoin();
781        return reportFutureResult();
782    }
783
784    public final V get(long timeout, TimeUnit unit)
785        throws InterruptedException, ExecutionException, TimeoutException {
786        long nanos = unit.toNanos(timeout);
787        Thread t = Thread.currentThread();
788        if (t instanceof ForkJoinWorkerThread) {
789            ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
790            if (!w.unpushTask(this) || !tryExec())
791                timedAwaitDone(w, nanos);
792        }
793        else
794            externalTimedAwaitDone(nanos);
795        return reportTimedFutureResult();
796    }
797
677      /**
678 <     * Possibly executes other tasks until this task {@link #isDone is
679 <     * done}, then returns the result of the computation.  This method
801 <     * may be more efficient than {@code join}, but is only applicable
802 <     * when there are no potential dependencies between continuation
803 <     * of the current task and that of any other task that might be
804 <     * executed while helping. (This usually holds for pure
805 <     * divide-and-conquer tasks).
806 <     *
807 <     * <p>This method may be invoked only from within {@code
808 <     * ForkJoinTask} computations (as may be determined using method
809 <     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
810 <     * result in exceptions or errors, possibly including {@code
811 <     * 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 <        quietlyHelpJoin();
690 <        return reportResult();
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 {@link #isDone is
707 <     * done}.  This method may be useful when processing collections
823 <     * of tasks when some have been cancelled or otherwise known to
824 <     * have aborted.
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 <     * <p>This method may be invoked only from within {@code
710 <     * ForkJoinTask} computations (as may be determined using method
711 <     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
712 <     * result in exceptions or errors, possibly including {@code
713 <     * ClassCastException}.
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 <        ForkJoinWorkerThread w = (ForkJoinWorkerThread) Thread.currentThread();
721 <        if (!w.unpushTask(this) || !tryExec()) {
722 <            for (;;) {
723 <                ForkJoinTask<?> t;
724 <                if (status < 0)
725 <                    return;
726 <                else if ((t = w.scanWhileJoining(this)) != null)
727 <                    t.tryExec();
728 <                else if (status < 0)
729 <                    return;
730 <                else if (w.pool.preBlockHelpingJoin(this)) {
731 <                    while (status >= 0) { // variant of awaitDone
732 <                        try {
733 <                            synchronized(this) {
734 <                                if (status >= 0)
848 <                                    wait();
849 <                                else {
850 <                                    notifyAll();
851 <                                    break;
852 <                                }
853 <                            }
854 <                        } catch (InterruptedException ie) {
855 <                            cancelIfTerminating();
856 <                        }
857 <                    }
858 <                    return;
859 <                }
860 <            }
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 <        Thread t = Thread.currentThread();
747 <        if (t instanceof ForkJoinWorkerThread) {
746 >        Thread t;
747 >        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
748              ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
749 <            if (!w.unpushTask(this) || !tryExec())
750 <                awaitDone(w);
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();
# Line 880 | Line 769 | public abstract class ForkJoinTask<V> im
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
885 <     * collections of tasks when some have been cancelled or otherwise
886 <     * known to have aborted.
772 >     * necessary, without returning its result or throwing its
773 >     * exception.
774       */
775      public final void quietlyInvoke() {
776 <        if (!tryExec())
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      /**
# Line 898 | Line 796 | public abstract class ForkJoinTask<V> im
796       * processed.
797       *
798       * <p>This method may be invoked only from within {@code
799 <     * ForkJoinTask} computations (as may be determined using method
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}.
# Line 917 | Line 815 | public abstract class ForkJoinTask<V> im
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 957 | Line 861 | public abstract class ForkJoinTask<V> im
861       * were not, stolen.
862       *
863       * <p>This method may be invoked only from within {@code
864 <     * ForkJoinTask} computations (as may be determined using method
864 >     * ForkJoinPool} computations (as may be determined using method
865       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
866       * result in exceptions or errors, possibly including {@code
867       * ClassCastException}.
# Line 976 | Line 880 | public abstract class ForkJoinTask<V> im
880       * fork other tasks.
881       *
882       * <p>This method may be invoked only from within {@code
883 <     * ForkJoinTask} computations (as may be determined using method
883 >     * ForkJoinPool} computations (as may be determined using method
884       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
885       * result in exceptions or errors, possibly including {@code
886       * ClassCastException}.
# Line 999 | Line 903 | public abstract class ForkJoinTask<V> im
903       * exceeded.
904       *
905       * <p>This method may be invoked only from within {@code
906 <     * ForkJoinTask} computations (as may be determined using method
906 >     * ForkJoinPool} computations (as may be determined using method
907       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
908       * result in exceptions or errors, possibly including {@code
909       * ClassCastException}.
# Line 1057 | Line 961 | public abstract class ForkJoinTask<V> im
961       * otherwise.
962       *
963       * <p>This method may be invoked only from within {@code
964 <     * ForkJoinTask} computations (as may be determined using method
964 >     * ForkJoinPool} computations (as may be determined using method
965       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
966       * result in exceptions or errors, possibly including {@code
967       * ClassCastException}.
# Line 1076 | Line 980 | public abstract class ForkJoinTask<V> im
980       * be useful otherwise.
981       *
982       * <p>This method may be invoked only from within {@code
983 <     * ForkJoinTask} computations (as may be determined using method
983 >     * ForkJoinPool} computations (as may be determined using method
984       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
985       * result in exceptions or errors, possibly including {@code
986       * ClassCastException}.
# Line 1099 | Line 1003 | public abstract class ForkJoinTask<V> im
1003       * otherwise.
1004       *
1005       * <p>This method may be invoked only from within {@code
1006 <     * ForkJoinTask} computations (as may be determined using method
1006 >     * ForkJoinPool} computations (as may be determined using method
1007       * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1008       * result in exceptions or errors, possibly including {@code
1009       * ClassCastException}.
# Line 1209 | Line 1113 | public abstract class ForkJoinTask<V> im
1113      private static final long serialVersionUID = -7721805057305804111L;
1114  
1115      /**
1116 <     * Saves the state to a stream.
1116 >     * Saves the state to a stream (that is, serializes it).
1117       *
1118       * @serialData the current run status and the exception thrown
1119       * during execution, or {@code null} if none
# Line 1222 | Line 1126 | public abstract class ForkJoinTask<V> im
1126      }
1127  
1128      /**
1129 <     * Reconstitutes the instance from a stream.
1129 >     * Reconstitutes the instance from a stream (that is, deserializes it).
1130       *
1131       * @param s the stream
1132       */
1133      private void readObject(java.io.ObjectInputStream s)
1134          throws java.io.IOException, ClassNotFoundException {
1135          s.defaultReadObject();
1232        status &= ~INTERNAL_SIGNAL_MASK; // clear internal signal counts
1233        status |= EXTERNAL_SIGNAL; // conservatively set external signal
1136          Object ex = s.readObject();
1137          if (ex != null)
1138 <            setDoneExceptionally((Throwable) ex);
1138 >            setExceptionalCompletion((Throwable) ex);
1139      }
1140  
1141      // Unsafe mechanics

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines