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.1 by dl, Tue Jan 6 14:30:31 2009 UTC vs.
Revision 1.24 by jsr166, Mon Jul 27 21:41:53 2009 UTC

# Line 5 | Line 5
5   */
6  
7   package jsr166y;
8 < import java.io.Serializable;
9 < import java.util.*;
8 >
9   import java.util.concurrent.*;
10 < import java.util.concurrent.atomic.*;
11 < import sun.misc.Unsafe;
12 < import java.lang.reflect.*;
10 >
11 > import java.io.Serializable;
12 > import java.util.Collection;
13 > import java.util.Collections;
14 > import java.util.List;
15 > import java.util.Map;
16 > import java.util.WeakHashMap;
17  
18   /**
19 < * Abstract base class for tasks that run within a ForkJoinPool.  A
20 < * ForkJoinTask is a thread-like entity that is much lighter weight
21 < * than a normal thread.  Huge numbers of tasks and subtasks may be
22 < * hosted by a small number of actual threads in a ForkJoinPool,
23 < * at the price of some usage limitations.
19 > * Abstract base class for tasks that run within a {@link
20 > * ForkJoinPool}.  A ForkJoinTask is a thread-like entity that is much
21 > * lighter weight than a normal thread.  Huge numbers of tasks and
22 > * subtasks may be hosted by a small number of actual threads in a
23 > * ForkJoinPool, at the price of some usage limitations.
24   *
25 < * <p> ForkJoinTasks are forms of <tt>Futures</tt> supporting a
26 < * limited range of use.  The "lightness" of ForkJoinTasks is due to a
27 < * set of restrictions (that are only partially statically
28 < * enforceable) reflecting their intended use as computational tasks
29 < * calculating pure functions or operating on purely isolated objects.
30 < * The primary coordination mechanisms supported for ForkJoinTasks are
31 < * <tt>fork</tt>, that arranges asynchronous execution, and
32 < * <tt>join</tt>, that doesn't proceed until the task's result has
33 < * been computed. (Cancellation is also supported).  The computation
34 < * defined in the <tt>compute</tt> method should avoid
35 < * <tt>synchronized</tt> methods or blocks, and should minimize
36 < * blocking synchronization apart from joining other tasks or using
25 > * <p> A "main" ForkJoinTask begins execution when submitted to a
26 > * {@link ForkJoinPool}. Once started, it will usually in turn start
27 > * other subtasks.  As indicated by the name of this class, many
28 > * programs using ForkJoinTasks employ only methods {@code fork} and
29 > * {@code join}, or derivatives such as {@code invokeAll}.  However,
30 > * this class also provides a number of other methods that can come
31 > * into play in advanced usages, as well as extension mechanics that
32 > * allow support of new forms of fork/join processing.
33 > *
34 > * <p>A ForkJoinTask is a lightweight form of {@link Future}.  The
35 > * efficiency of ForkJoinTasks stems from a set of restrictions (that
36 > * are only partially statically enforceable) reflecting their
37 > * intended use as computational tasks calculating pure functions or
38 > * operating on purely isolated objects.  The primary coordination
39 > * mechanisms are {@link #fork}, that arranges asynchronous execution,
40 > * and {@link #join}, that doesn't proceed until the task's result has
41 > * been computed.  Computations should avoid {@code synchronized}
42 > * methods or blocks, and should minimize other blocking
43 > * synchronization apart from joining other tasks or using
44   * synchronizers such as Phasers that are advertised to cooperate with
45   * fork/join scheduling. Tasks should also not perform blocking IO,
46   * and should ideally access variables that are completely independent
# Line 38 | Line 48 | import java.lang.reflect.*;
48   * restrictions, for example using shared output streams, may be
49   * tolerable in practice, but frequent use may result in poor
50   * performance, and the potential to indefinitely stall if the number
51 < * of threads not waiting for external synchronization becomes
52 < * exhausted. This usage restriction is in part enforced by not
53 < * permitting checked exceptions such as IOExceptions to be
54 < * thrown. However, computations may still encounter unchecked
51 > * of threads not waiting for IO or other external synchronization
52 > * becomes exhausted. This usage restriction is in part enforced by
53 > * not permitting checked exceptions such as {@code IOExceptions}
54 > * to be thrown. However, computations may still encounter unchecked
55   * exceptions, that are rethrown to callers attempting join
56   * them. These exceptions may additionally include
57   * RejectedExecutionExceptions stemming from internal resource
58   * exhaustion such as failure to allocate internal task queues.
59   *
60 < * <p> The <tt>ForkJoinTask</tt> class is not usually directly
61 < * subclassed.  Instead, you subclass one of the abstract classes that
62 < * support different styles of fork/join processing.  Normally, a
63 < * concrete ForkJoinTask subclass declares fields comprising its
64 < * parameters, established in a constructor, and then defines a
65 < * <tt>compute</tt> method that somehow uses the control methods
66 < * supplied by this base class. While these methods have
67 < * <tt>public</tt> access, some of them may only be called from within
68 < * other ForkJoinTasks. Attempts to invoke them in other contexts
69 < * result in exceptions or errors including ClassCastException.  The
70 < * only way to invoke a "main" driver task is to submit it to a
71 < * ForkJoinPool. Once started, this will usually in turn start other
72 < * subtasks.
60 > * <p>The primary method for awaiting completion and extracting
61 > * results of a task is {@link #join}, but there are several variants:
62 > * The {@link Future#get} methods support interruptible and/or timed
63 > * waits for completion and report results using {@code Future}
64 > * conventions. Method {@link #helpJoin} enables callers to actively
65 > * execute other tasks while awaiting joins, which is sometimes more
66 > * efficient but only applies when all subtasks are known to be
67 > * strictly tree-structured. Method {@link #invoke} is semantically
68 > * equivalent to {@code fork(); join()} but always attempts to
69 > * begin execution in the current thread. The "<em>quiet</em>" forms
70 > * of these methods do not extract results or report exceptions. These
71 > * may be useful when a set of tasks are being executed, and you need
72 > * to delay processing of results or exceptions until all complete.
73 > * Method {@code invokeAll} (available in multiple versions)
74 > * performs the most common form of parallel invocation: forking a set
75 > * of tasks and joining them all.
76 > *
77 > * <p> The ForkJoinTask class is not usually directly subclassed.
78 > * Instead, you subclass one of the abstract classes that support a
79 > * particular style of fork/join processing.  Normally, a concrete
80 > * ForkJoinTask subclass declares fields comprising its parameters,
81 > * established in a constructor, and then defines a {@code compute}
82 > * method that somehow uses the control methods supplied by this base
83 > * class. While these methods have {@code public} access (to allow
84 > * instances of different task subclasses to call each others
85 > * methods), some of them may only be called from within other
86 > * ForkJoinTasks (as may be determined using method {@link
87 > * #inForkJoinPool}).  Attempts to invoke them in other contexts
88 > * result in exceptions or errors, possibly including
89 > * ClassCastException.
90   *
91 < * <p>Most base support methods are <tt>final</tt> because their
91 > * <p>Most base support methods are {@code final} because their
92   * implementations are intrinsically tied to the underlying
93   * lightweight task scheduling framework, and so cannot be overridden.
94   * Developers creating new basic styles of fork/join processing should
95 < * minimally implement protected methods <tt>exec</tt>,
96 < * <tt>setRawResult</tt>, and <tt>getRawResult</tt>, while also
97 < * introducing an abstract computational method that can be
98 < * implemented in its subclasses. To support such extensions,
99 < * instances of ForkJoinTasks maintain an atomically updated
100 < * <tt>short</tt> representing user-defined control state.  Control
74 < * state is guaranteed initially to be zero, and to be negative upon
75 < * completion, but may otherwise be used for any other control
76 < * purposes, such as maintaining join counts.  The {@link
77 < * ForkJoinWorkerThread} class supports additional inspection and
78 < * tuning methods that can be useful when developing extensions.
95 > * minimally implement {@code protected} methods
96 > * {@link #exec}, {@link #setRawResult}, and
97 > * {@link #getRawResult}, while also introducing an abstract
98 > * computational method that can be implemented in its subclasses,
99 > * possibly relying on other {@code protected} methods provided
100 > * by this class.
101   *
102   * <p>ForkJoinTasks should perform relatively small amounts of
103 < * computations, othewise splitting into smaller tasks. As a very
103 > * computations, otherwise splitting into smaller tasks. As a very
104   * rough rule of thumb, a task should perform more than 100 and less
105   * than 10000 basic computational steps. If tasks are too big, then
106 < * parellelism cannot improve throughput. If too small, then memory
106 > * parallelism cannot improve throughput. If too small, then memory
107   * and internal task maintenance overhead may overwhelm processing.
108   *
109 < * <p>ForkJoinTasks are <tt>Serializable</tt>, which enables them to
110 < * be used in extensions such as remote execution frameworks. However,
111 < * it is in general safe to serialize tasks only before or after, but
109 > * <p>ForkJoinTasks are {@code Serializable}, which enables them
110 > * to be used in extensions such as remote execution frameworks. It is
111 > * in general sensible to serialize tasks only before or after, but
112   * not during execution. Serialization is not relied on during
113   * execution itself.
114 + *
115 + * @since 1.7
116 + * @author Doug Lea
117   */
118   public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
119 +
120      /**
121 <     * Status field holding all run status. We pack this into a single
122 <     * int both to minimize footprint overhead and to ensure atomicity
123 <     * (updates are via CAS).
98 <     *
99 <     * Status is initially zero, and takes on nonnegative values until
121 >     * Run control status bits packed into a single int to minimize
122 >     * footprint and to ensure atomicity (via CAS).  Status is
123 >     * initially zero, and takes on nonnegative values until
124       * completed, upon which status holds COMPLETED. CANCELLED, or
125       * EXCEPTIONAL, which use the top 3 bits.  Tasks undergoing
126       * blocking waits by other threads have SIGNAL_MASK bits set --
# Line 111 | Line 135 | public abstract class ForkJoinTask<V> im
135       * currently unused. Also value 0x80000000 is available as spare
136       * completion value.
137       */
138 <    volatile int status; // accessed directy by pool and workers
138 >    volatile int status; // accessed directly by pool and workers
139  
140      static final int COMPLETION_MASK      = 0xe0000000;
141      static final int NORMAL               = 0xe0000000; // == mask
# Line 124 | Line 148 | public abstract class ForkJoinTask<V> im
148      /**
149       * Table of exceptions thrown by tasks, to enable reporting by
150       * callers. Because exceptions are rare, we don't directly keep
151 <     * them with task objects, but instead us a weak ref table.  Note
151 >     * them with task objects, but instead use a weak ref table.  Note
152       * that cancellation exceptions don't appear in the table, but are
153       * instead recorded as status values.
154 <     * Todo: Use ConcurrentReferenceHashMap
154 >     * TODO: Use ConcurrentReferenceHashMap
155       */
156      static final Map<ForkJoinTask<?>, Throwable> exceptionMap =
157          Collections.synchronizedMap
# Line 136 | Line 160 | public abstract class ForkJoinTask<V> im
160      // within-package utilities
161  
162      /**
163 <     * Get current worker thread, or null if not a worker thread
163 >     * Gets current worker thread, or null if not a worker thread.
164       */
165      static ForkJoinWorkerThread getWorker() {
166          Thread t = Thread.currentThread();
167 <        return ((t instanceof ForkJoinWorkerThread)?
168 <                (ForkJoinWorkerThread)t : null);
145 <    }
146 <
147 <    /**
148 <     * Get pool of current worker thread, or null if not a worker thread
149 <     */
150 <    static ForkJoinPool getWorkerPool() {
151 <        Thread t = Thread.currentThread();
152 <        return ((t instanceof ForkJoinWorkerThread)?
153 <                ((ForkJoinWorkerThread)t).pool : null);
167 >        return ((t instanceof ForkJoinWorkerThread) ?
168 >                (ForkJoinWorkerThread) t : null);
169      }
170  
171      final boolean casStatus(int cmp, int val) {
172 <        return _unsafe.compareAndSwapInt(this, statusOffset, cmp, val);
172 >        return UNSAFE.compareAndSwapInt(this, statusOffset, cmp, val);
173      }
174  
175      /**
# Line 162 | Line 177 | public abstract class ForkJoinTask<V> im
177       */
178      static void rethrowException(Throwable ex) {
179          if (ex != null)
180 <            _unsafe.throwException(ex);
180 >            UNSAFE.throwException(ex);
181      }
182  
183      // Setting completion status
184  
185      /**
186 <     * Mark completion and wake up threads waiting to join this task.
186 >     * Marks completion and wakes up threads waiting to join this task.
187 >     *
188       * @param completion one of NORMAL, CANCELLED, EXCEPTIONAL
189       */
190      final void setCompletion(int completion) {
191 <        ForkJoinPool pool = getWorkerPool();
191 >        ForkJoinPool pool = getPool();
192          if (pool != null) {
193              int s; // Clear signal bits while setting completion status
194 <            do;while ((s = status) >= 0 && !casStatus(s, completion));
194 >            do {} while ((s = status) >= 0 && !casStatus(s, completion));
195  
196              if ((s & SIGNAL_MASK) != 0) {
197                  if ((s &= INTERNAL_SIGNAL_MASK) != 0)
198                      pool.updateRunningCount(s);
199 <                synchronized(this) { notifyAll(); }
199 >                synchronized (this) { notifyAll(); }
200              }
201          }
202          else
# Line 193 | Line 209 | public abstract class ForkJoinTask<V> im
209       */
210      private void externallySetCompletion(int completion) {
211          int s;
212 <        do;while ((s = status) >= 0 &&
213 <                  !casStatus(s, (s & SIGNAL_MASK) | completion));
214 <        synchronized(this) { notifyAll(); }
212 >        do {} while ((s = status) >= 0 &&
213 >                     !casStatus(s, (s & SIGNAL_MASK) | completion));
214 >        synchronized (this) { notifyAll(); }
215      }
216  
217      /**
218 <     * Sets status to indicate normal completion
218 >     * Sets status to indicate normal completion.
219       */
220      final void setNormalCompletion() {
221          // Try typical fast case -- single CAS, no signal, not already done.
222          // Manually expand casStatus to improve chances of inlining it
223 <        if (!_unsafe.compareAndSwapInt(this, statusOffset, 0, NORMAL))
223 >        if (!UNSAFE.compareAndSwapInt(this, statusOffset, 0, NORMAL))
224              setCompletion(NORMAL);
225      }
226  
227      // internal waiting and notification
228  
229      /**
230 <     * Performs the actual monitor wait for awaitDone
230 >     * Performs the actual monitor wait for awaitDone.
231       */
232      private void doAwaitDone() {
233          // Minimize lock bias and in/de-flation effects by maximizing
234          // chances of waiting inside sync
235          try {
236              while (status >= 0)
237 <                synchronized(this) { if (status >= 0) wait(); }
237 >                synchronized (this) { if (status >= 0) wait(); }
238          } catch (InterruptedException ie) {
239              onInterruptedWait();
240          }
241      }
242  
243      /**
244 <     * Performs the actual monitor wait for awaitDone
244 >     * Performs the actual timed monitor wait for awaitDone.
245       */
246      private void doAwaitDone(long startTime, long nanos) {
247 <        synchronized(this) {
247 >        synchronized (this) {
248              try {
249                  while (status >= 0) {
250                      long nt = nanos - System.nanoTime() - startTime;
251                      if (nt <= 0)
252                          break;
253 <                    wait(nt / 1000000, (int)(nt % 1000000));
253 >                    wait(nt / 1000000, (int) (nt % 1000000));
254                  }
255              } catch (InterruptedException ie) {
256                  onInterruptedWait();
# Line 247 | Line 263 | public abstract class ForkJoinTask<V> im
263      /**
264       * Sets status to indicate there is joiner, then waits for join,
265       * surrounded with pool notifications.
266 +     *
267       * @return status upon exit
268       */
269 <    final int awaitDone(ForkJoinWorkerThread w, boolean maintainParallelism) {
270 <        ForkJoinPool pool = w == null? null : w.pool;
269 >    private int awaitDone(ForkJoinWorkerThread w,
270 >                          boolean maintainParallelism) {
271 >        ForkJoinPool pool = (w == null) ? null : w.pool;
272          int s;
273          while ((s = status) >= 0) {
274 <            if (casStatus(s, pool == null? s|EXTERNAL_SIGNAL : s+1)) {
274 >            if (casStatus(s, (pool == null) ? s|EXTERNAL_SIGNAL : s+1)) {
275                  if (pool == null || !pool.preJoin(this, maintainParallelism))
276                      doAwaitDone();
277                  if (((s = status) & INTERNAL_SIGNAL_MASK) != 0)
# Line 266 | Line 284 | public abstract class ForkJoinTask<V> im
284  
285      /**
286       * Timed version of awaitDone
287 +     *
288       * @return status upon exit
289       */
290 <    final int awaitDone(ForkJoinWorkerThread w, long nanos) {
291 <        ForkJoinPool pool = w == null? null : w.pool;
290 >    private int awaitDone(ForkJoinWorkerThread w, long nanos) {
291 >        ForkJoinPool pool = (w == null) ? null : w.pool;
292          int s;
293          while ((s = status) >= 0) {
294 <            if (casStatus(s, pool == null? s|EXTERNAL_SIGNAL : s+1)) {
294 >            if (casStatus(s, (pool == null) ? s|EXTERNAL_SIGNAL : s+1)) {
295                  long startTime = System.nanoTime();
296                  if (pool == null || !pool.preJoin(this, false))
297                      doAwaitDone(startTime, nanos);
# Line 289 | Line 308 | public abstract class ForkJoinTask<V> im
308      }
309  
310      /**
311 <     * Notify pool that thread is unblocked. Called by signalled
311 >     * Notifies pool that thread is unblocked. Called by signalled
312       * threads when woken by non-FJ threads (which is atypical).
313       */
314      private void adjustPoolCountsOnUnblock(ForkJoinPool pool) {
315          int s;
316 <        do;while ((s = status) < 0 && !casStatus(s, s & COMPLETION_MASK));
316 >        do {} while ((s = status) < 0 && !casStatus(s, s & COMPLETION_MASK));
317          if (pool != null && (s &= INTERNAL_SIGNAL_MASK) != 0)
318              pool.updateRunningCount(s);
319      }
320  
321      /**
322 <     * Notify pool to adjust counts on cancelled or timed out wait
322 >     * Notifies pool to adjust counts on cancelled or timed out wait.
323       */
324      private void adjustPoolCountsOnCancelledWait(ForkJoinPool pool) {
325          if (pool != null) {
# Line 314 | Line 333 | public abstract class ForkJoinTask<V> im
333          }
334      }
335  
336 +    /**
337 +     * Handles interruptions during waits.
338 +     */
339      private void onInterruptedWait() {
340 <        Thread t = Thread.currentThread();
341 <        if (t instanceof ForkJoinWorkerThread) {
342 <            ForkJoinWorkerThread w = (ForkJoinWorkerThread)t;
343 <            if (w.isTerminating())
344 <                cancelIgnoreExceptions();
345 <        }
324 <        else { // re-interrupt
325 <            try {
326 <                t.interrupt();
327 <            } catch (SecurityException ignore) {
328 <            }
329 <        }
340 >        ForkJoinWorkerThread w = getWorker();
341 >        if (w == null)
342 >            Thread.currentThread().interrupt(); // re-interrupt
343 >        else if (w.isTerminating())
344 >            cancelIgnoringExceptions();
345 >        // else if FJworker, ignore interrupt
346      }
347  
348      // Recording and reporting exceptions
# Line 337 | Line 353 | public abstract class ForkJoinTask<V> im
353      }
354  
355      /**
356 <     * Throws the exception associated with status s;
356 >     * Throws the exception associated with status s.
357 >     *
358       * @throws the exception
359       */
360      private void reportException(int s) {
# Line 350 | Line 367 | public abstract class ForkJoinTask<V> im
367      }
368  
369      /**
370 <     * Returns result or throws exception using j.u.c.Future conventions
371 <     * Only call when isDone known to be true.
370 >     * Returns result or throws exception using j.u.c.Future conventions.
371 >     * Only call when {@code isDone} known to be true.
372       */
373      private V reportFutureResult()
374          throws ExecutionException, InterruptedException {
# Line 370 | Line 387 | public abstract class ForkJoinTask<V> im
387  
388      /**
389       * Returns result or throws exception using j.u.c.Future conventions
390 <     * with timeouts
390 >     * with timeouts.
391       */
392      private V reportTimedFutureResult()
393          throws InterruptedException, ExecutionException, TimeoutException {
# Line 391 | Line 408 | public abstract class ForkJoinTask<V> im
408  
409      /**
410       * Calls exec, recording completion, and rethrowing exception if
411 <     * encountered. Caller should normally check status before calling
411 >     * encountered. Caller should normally check status before calling.
412 >     *
413       * @return true if completed normally
414       */
415      private boolean tryExec() {
# Line 409 | Line 427 | public abstract class ForkJoinTask<V> im
427  
428      /**
429       * Main execution method used by worker threads. Invokes
430 <     * base computation unless already complete
430 >     * base computation unless already complete.
431       */
432      final void quietlyExec() {
433          if (status >= 0) {
434              try {
435                  if (!exec())
436                      return;
437 <            } catch(Throwable rex) {
437 >            } catch (Throwable rex) {
438                  setDoneExceptionally(rex);
439                  return;
440              }
# Line 425 | Line 443 | public abstract class ForkJoinTask<V> im
443      }
444  
445      /**
446 <     * Calls exec, recording but not rethrowing exception
447 <     * Caller should normally check status before calling
446 >     * Calls exec(), recording but not rethrowing exception.
447 >     * Caller should normally check status before calling.
448 >     *
449       * @return true if completed normally
450       */
451      private boolean tryQuietlyInvoke() {
# Line 442 | Line 461 | public abstract class ForkJoinTask<V> im
461      }
462  
463      /**
464 <     * Cancel, ignoring any exceptions it throws
464 >     * Cancels, ignoring any exceptions it throws.
465       */
466 <    final void cancelIgnoreExceptions() {
466 >    final void cancelIgnoringExceptions() {
467          try {
468              cancel(false);
469 <        } catch(Throwable ignore) {
469 >        } catch (Throwable ignore) {
470          }
471      }
472  
473 +    /**
474 +     * Main implementation of helpJoin
475 +     */
476 +    private int busyJoin(ForkJoinWorkerThread w) {
477 +        int s;
478 +        ForkJoinTask<?> t;
479 +        while ((s = status) >= 0 && (t = w.scanWhileJoining(this)) != null)
480 +            t.quietlyExec();
481 +        return (s >= 0) ? awaitDone(w, false) : s; // block if no work
482 +    }
483 +
484      // public methods
485  
486      /**
487       * Arranges to asynchronously execute this task.  While it is not
488       * necessarily enforced, it is a usage error to fork a task more
489       * than once unless it has completed and been reinitialized.  This
490 <     * method may be invoked only from within other ForkJoinTask
491 <     * computations. Attempts to invoke in other contexts result in
492 <     * exceptions or errors including ClassCastException.
490 >     * method may be invoked only from within ForkJoinTask
491 >     * computations (as may be determined using method {@link
492 >     * #inForkJoinPool}). Attempts to invoke in other contexts result
493 >     * in exceptions or errors, possibly including ClassCastException.
494 >     *
495 >     * @return {@code this}, to simplify usage.
496       */
497 <    public final void fork() {
498 <        ((ForkJoinWorkerThread)(Thread.currentThread())).pushTask(this);
497 >    public final ForkJoinTask<V> fork() {
498 >        ((ForkJoinWorkerThread) Thread.currentThread())
499 >            .pushTask(this);
500 >        return this;
501      }
502  
503      /**
504       * Returns the result of the computation when it is ready.
505 <     * This method differs from <tt>get</tt> in that abnormal
505 >     * This method differs from {@code get} in that abnormal
506       * completion results in RuntimeExceptions or Errors, not
507       * ExecutionExceptions.
508       *
# Line 480 | Line 515 | public abstract class ForkJoinTask<V> im
515          return getRawResult();
516      }
517  
483    public final V get() throws InterruptedException, ExecutionException {
484        ForkJoinWorkerThread w = getWorker();
485        if (w == null || status < 0 || !w.unpushTask(this) || !tryQuietlyInvoke())
486            awaitDone(w, true);
487        return reportFutureResult();
488    }
489
490    public final V get(long timeout, TimeUnit unit)
491        throws InterruptedException, ExecutionException, TimeoutException {
492        ForkJoinWorkerThread w = getWorker();
493        if (w == null || status < 0 || !w.unpushTask(this) || !tryQuietlyInvoke())
494            awaitDone(w, unit.toNanos(timeout));
495        return reportTimedFutureResult();
496    }
497
498    /**
499     * Possibly executes other tasks until this task is ready, then
500     * returns the result of the computation.  This method may be more
501     * efficient than <tt>join</tt>, but is only applicable when there
502     * are no potemtial dependencies between continuation of the
503     * current task and that of any other task that might be executed
504     * while helping. (This usually holds for pure divide-and-conquer
505     * tasks).
506     * @return the computed result
507     */
508    public final V helpJoin() {
509        ForkJoinWorkerThread w = (ForkJoinWorkerThread)(Thread.currentThread());
510        if (status < 0 || !w.unpushTask(this) || !tryExec())
511            reportException(w.helpJoinTask(this));
512        return getRawResult();
513    }
514
518      /**
519 <     * Performs this task, awaits its completion if necessary, and
520 <     * return its result.
519 >     * Commences performing this task, awaits its completion if
520 >     * necessary, and return its result.
521 >     *
522       * @throws Throwable (a RuntimeException, Error, or unchecked
523 <     * exception) if the underlying computation did so.
523 >     * exception) if the underlying computation did so
524       * @return the computed result
525       */
526      public final V invoke() {
# Line 527 | Line 531 | public abstract class ForkJoinTask<V> im
531      }
532  
533      /**
534 <     * Joins this task, without returning its result or throwing an
535 <     * exception. This method may be useful when processing
536 <     * collections of tasks when some have been cancelled or otherwise
537 <     * known to have aborted.
534 >     * Forks both tasks, returning when {@code isDone} holds for
535 >     * both of them or an exception is encountered. This method may be
536 >     * invoked only from within ForkJoinTask computations (as may be
537 >     * determined using method {@link #inForkJoinPool}). Attempts to
538 >     * invoke in other contexts result in exceptions or errors,
539 >     * possibly including ClassCastException.
540 >     *
541 >     * @param t1 one task
542 >     * @param t2 the other task
543 >     * @throws NullPointerException if t1 or t2 are null
544 >     * @throws RuntimeException or Error if either task did so
545       */
546 <    public final void quietlyJoin() {
547 <        if (status >= 0) {
548 <            ForkJoinWorkerThread w = getWorker();
549 <            if (w == null || !w.unpushTask(this) || !tryQuietlyInvoke())
539 <                awaitDone(w, true);
540 <        }
546 >    public static void invokeAll(ForkJoinTask<?>t1, ForkJoinTask<?> t2) {
547 >        t2.fork();
548 >        t1.invoke();
549 >        t2.join();
550      }
551  
552      /**
553 <     * Possibly executes other tasks until this task is ready.
553 >     * Forks the given tasks, returning when {@code isDone} holds
554 >     * for all of them. If any task encounters an exception, others
555 >     * may be cancelled.  This method may be invoked only from within
556 >     * ForkJoinTask computations (as may be determined using method
557 >     * {@link #inForkJoinPool}). Attempts to invoke in other contexts
558 >     * result in exceptions or errors, possibly including
559 >     * ClassCastException.
560 >     *
561 >     * @param tasks the array of tasks
562 >     * @throws NullPointerException if tasks or any element are null
563 >     * @throws RuntimeException or Error if any task did so
564       */
565 <    public final void quietlyHelpJoin() {
566 <        if (status >= 0) {
567 <            ForkJoinWorkerThread w =
568 <                (ForkJoinWorkerThread)(Thread.currentThread());
569 <            if (!w.unpushTask(this) || !tryQuietlyInvoke())
570 <                w.helpJoinTask(this);
565 >    public static void invokeAll(ForkJoinTask<?>... tasks) {
566 >        Throwable ex = null;
567 >        int last = tasks.length - 1;
568 >        for (int i = last; i >= 0; --i) {
569 >            ForkJoinTask<?> t = tasks[i];
570 >            if (t == null) {
571 >                if (ex == null)
572 >                    ex = new NullPointerException();
573 >            }
574 >            else if (i != 0)
575 >                t.fork();
576 >            else {
577 >                t.quietlyInvoke();
578 >                if (ex == null)
579 >                    ex = t.getException();
580 >            }
581 >        }
582 >        for (int i = 1; i <= last; ++i) {
583 >            ForkJoinTask<?> t = tasks[i];
584 >            if (t != null) {
585 >                if (ex != null)
586 >                    t.cancel(false);
587 >                else {
588 >                    t.quietlyJoin();
589 >                    if (ex == null)
590 >                        ex = t.getException();
591 >                }
592 >            }
593          }
594 +        if (ex != null)
595 +            rethrowException(ex);
596      }
597  
598      /**
599 <     * Performs this task and awaits its completion if necessary,
600 <     * without returning its result or throwing an exception. This
601 <     * method may be useful when processing collections of tasks when
602 <     * some have been cancelled or otherwise known to have aborted.
603 <     */
604 <    public final void quietlyInvoke() {
605 <        if (status >= 0 && !tryQuietlyInvoke())
606 <            quietlyJoin();
599 >     * Forks all tasks in the collection, returning when
600 >     * {@code isDone} holds for all of them. If any task
601 >     * encounters an exception, others may be cancelled.  This method
602 >     * may be invoked only from within ForkJoinTask computations (as
603 >     * may be determined using method {@link
604 >     * #inForkJoinPool}). Attempts to invoke in other contexts result
605 >     * in exceptions or errors, possibly including ClassCastException.
606 >     *
607 >     * @param tasks the collection of tasks
608 >     * @return the tasks argument, to simplify usage
609 >     * @throws NullPointerException if tasks or any element are null
610 >     * @throws RuntimeException or Error if any task did so
611 >     */
612 >    public static <T extends ForkJoinTask<?>> Collection<T> invokeAll(Collection<T> tasks) {
613 >        if (!(tasks instanceof List<?>)) {
614 >            invokeAll(tasks.toArray(new ForkJoinTask<?>[tasks.size()]));
615 >            return tasks;
616 >        }
617 >        @SuppressWarnings("unchecked")
618 >        List<? extends ForkJoinTask<?>> ts =
619 >            (List<? extends ForkJoinTask<?>>) tasks;
620 >        Throwable ex = null;
621 >        int last = ts.size() - 1;
622 >        for (int i = last; i >= 0; --i) {
623 >            ForkJoinTask<?> t = ts.get(i);
624 >            if (t == null) {
625 >                if (ex == null)
626 >                    ex = new NullPointerException();
627 >            }
628 >            else if (i != 0)
629 >                t.fork();
630 >            else {
631 >                t.quietlyInvoke();
632 >                if (ex == null)
633 >                    ex = t.getException();
634 >            }
635 >        }
636 >        for (int i = 1; i <= last; ++i) {
637 >            ForkJoinTask<?> t = ts.get(i);
638 >            if (t != null) {
639 >                if (ex != null)
640 >                    t.cancel(false);
641 >                else {
642 >                    t.quietlyJoin();
643 >                    if (ex == null)
644 >                        ex = t.getException();
645 >                }
646 >            }
647 >        }
648 >        if (ex != null)
649 >            rethrowException(ex);
650 >        return tasks;
651      }
652  
653      /**
654 <     * Returns true if the computation performed by this task has
655 <     * completed (or has been cancelled).
656 <     * @return true if this computation has completed
654 >     * Returns {@code true} if the computation performed by this task
655 >     * has completed (or has been cancelled).
656 >     *
657 >     * @return {@code true} if this computation has completed
658       */
659      public final boolean isDone() {
660          return status < 0;
661      }
662  
663      /**
664 <     * Returns true if this task was cancelled.
665 <     * @return true if this task was cancelled
664 >     * Returns {@code true} if this task was cancelled.
665 >     *
666 >     * @return {@code true} if this task was cancelled
667       */
668      public final boolean isCancelled() {
669          return (status & COMPLETION_MASK) == CANCELLED;
670      }
671  
672      /**
584     * Returns true if this task threw an exception or was cancelled
585     * @return true if this task threw an exception or was cancelled
586     */
587    public final boolean completedAbnormally() {
588        return (status & COMPLETION_MASK) < NORMAL;
589    }
590
591    /**
592     * Returns the exception thrown by the base computation, or a
593     * CancellationException if cancelled, or null if none or if the
594     * method has not yet completed.
595     * @return the exception, or null if none
596     */
597    public final Throwable getException() {
598        int s = status & COMPLETION_MASK;
599        if (s >= NORMAL)
600            return null;
601        if (s == CANCELLED)
602            return new CancellationException();
603        return exceptionMap.get(this);
604    }
605
606    /**
673       * Asserts that the results of this task's computation will not be
674 <     * used. If a cancellation occurs before this task is processed,
675 <     * then its <tt>compute</tt> method will not be executed,
676 <     * <tt>isCancelled</tt> will report true, and <tt>join</tt> will
677 <     * result in a CancellationException being thrown. Otherwise, when
674 >     * used. If a cancellation occurs before attempting to execute this
675 >     * task, execution will be suppressed, {@link #isCancelled}
676 >     * will report true, and {@link #join} will result in a
677 >     * {@code CancellationException} being thrown. Otherwise, when
678       * cancellation races with completion, there are no guarantees
679 <     * about whether <tt>isCancelled</tt> will report true, whether
680 <     * <tt>join</tt> will return normally or via an exception, or
681 <     * whether these behaviors will remain consistent upon repeated
679 >     * about whether {@code isCancelled} will report {@code true},
680 >     * whether {@code join} will return normally or via an exception,
681 >     * or whether these behaviors will remain consistent upon repeated
682       * invocation.
683       *
684       * <p>This method may be overridden in subclasses, but if so, must
# Line 622 | Line 688 | public abstract class ForkJoinTask<V> im
688       * <p> This method is designed to be invoked by <em>other</em>
689       * tasks. To terminate the current task, you can just return or
690       * throw an unchecked exception from its computation method, or
691 <     * invoke <tt>completeExceptionally(someException)</tt>.
691 >     * invoke {@link #completeExceptionally}.
692       *
693       * @param mayInterruptIfRunning this value is ignored in the
694       * default implementation because tasks are not in general
695 <     * cancelled via interruption.
695 >     * cancelled via interruption
696       *
697 <     * @return true if this task is now cancelled
697 >     * @return {@code true} if this task is now cancelled
698       */
699      public boolean cancel(boolean mayInterruptIfRunning) {
700          setCompletion(CANCELLED);
# Line 636 | Line 702 | public abstract class ForkJoinTask<V> im
702      }
703  
704      /**
705 +     * Returns {@code true} if this task threw an exception or was cancelled.
706 +     *
707 +     * @return {@code true} if this task threw an exception or was cancelled
708 +     */
709 +    public final boolean isCompletedAbnormally() {
710 +        return (status & COMPLETION_MASK) < NORMAL;
711 +    }
712 +
713 +    /**
714 +     * Returns the exception thrown by the base computation, or a
715 +     * CancellationException if cancelled, or null if none or if the
716 +     * method has not yet completed.
717 +     *
718 +     * @return the exception, or {@code null} if none
719 +     */
720 +    public final Throwable getException() {
721 +        int s = status & COMPLETION_MASK;
722 +        if (s >= NORMAL)
723 +            return null;
724 +        if (s == CANCELLED)
725 +            return new CancellationException();
726 +        return exceptionMap.get(this);
727 +    }
728 +
729 +    /**
730       * Completes this task abnormally, and if not already aborted or
731       * cancelled, causes it to throw the given exception upon
732 <     * <tt>join</tt> and related operations. This method may be used
732 >     * {@code join} and related operations. This method may be used
733       * to induce exceptions in asynchronous tasks, or to force
734 <     * completion of tasks that would not otherwise complete.  This
735 <     * method is overridable, but overridden versions must invoke
736 <     * <tt>super</tt> implementation to maintain guarantees.
734 >     * completion of tasks that would not otherwise complete.  Its use
735 >     * in other situations is likely to be wrong.  This method is
736 >     * overridable, but overridden versions must invoke {@code super}
737 >     * implementation to maintain guarantees.
738 >     *
739       * @param ex the exception to throw. If this exception is
740       * not a RuntimeException or Error, the actual exception thrown
741       * will be a RuntimeException with cause ex.
742       */
743      public void completeExceptionally(Throwable ex) {
744          setDoneExceptionally((ex instanceof RuntimeException) ||
745 <                             (ex instanceof Error)? ex :
745 >                             (ex instanceof Error) ? ex :
746                               new RuntimeException(ex));
747      }
748  
749      /**
750       * Completes this task, and if not already aborted or cancelled,
751 <     * returning a <tt>null</tt> result upon <tt>join</tt> and related
751 >     * returning a {@code null} result upon {@code join} and related
752       * operations. This method may be used to provide results for
753       * asynchronous tasks, or to provide alternative handling for
754 <     * tasks that would not otherwise complete normally.
754 >     * tasks that would not otherwise complete normally. Its use in
755 >     * other situations is likely to be wrong. This method is
756 >     * overridable, but overridden versions must invoke {@code super}
757 >     * implementation to maintain guarantees.
758       *
759 <     * @param value the result value for this task.
759 >     * @param value the result value for this task
760       */
761      public void complete(V value) {
762          try {
763              setRawResult(value);
764 <        } catch(Throwable rex) {
764 >        } catch (Throwable rex) {
765              setDoneExceptionally(rex);
766              return;
767          }
768          setNormalCompletion();
769      }
770  
771 +    public final V get() throws InterruptedException, ExecutionException {
772 +        ForkJoinWorkerThread w = getWorker();
773 +        if (w == null || status < 0 || !w.unpushTask(this) || !tryQuietlyInvoke())
774 +            awaitDone(w, true);
775 +        return reportFutureResult();
776 +    }
777 +
778 +    public final V get(long timeout, TimeUnit unit)
779 +        throws InterruptedException, ExecutionException, TimeoutException {
780 +        ForkJoinWorkerThread w = getWorker();
781 +        if (w == null || status < 0 || !w.unpushTask(this) || !tryQuietlyInvoke())
782 +            awaitDone(w, unit.toNanos(timeout));
783 +        return reportTimedFutureResult();
784 +    }
785 +
786 +    /**
787 +     * Possibly executes other tasks until this task is ready, then
788 +     * returns the result of the computation.  This method may be more
789 +     * efficient than {@code join}, but is only applicable when
790 +     * there are no potential dependencies between continuation of the
791 +     * current task and that of any other task that might be executed
792 +     * while helping. (This usually holds for pure divide-and-conquer
793 +     * tasks). This method may be invoked only from within
794 +     * ForkJoinTask computations (as may be determined using method
795 +     * {@link #inForkJoinPool}). Attempts to invoke in other contexts
796 +     * result in exceptions or errors, possibly including
797 +     * ClassCastException.
798 +     *
799 +     * @return the computed result
800 +     */
801 +    public final V helpJoin() {
802 +        ForkJoinWorkerThread w = (ForkJoinWorkerThread) Thread.currentThread();
803 +        if (status < 0 || !w.unpushTask(this) || !tryExec())
804 +            reportException(busyJoin(w));
805 +        return getRawResult();
806 +    }
807 +
808 +    /**
809 +     * Possibly executes other tasks until this task is ready.  This
810 +     * method may be invoked only from within ForkJoinTask
811 +     * computations (as may be determined using method {@link
812 +     * #inForkJoinPool}). Attempts to invoke in other contexts result
813 +     * in exceptions or errors, possibly including ClassCastException.
814 +     */
815 +    public final void quietlyHelpJoin() {
816 +        if (status >= 0) {
817 +            ForkJoinWorkerThread w =
818 +                (ForkJoinWorkerThread) Thread.currentThread();
819 +            if (!w.unpushTask(this) || !tryQuietlyInvoke())
820 +                busyJoin(w);
821 +        }
822 +    }
823 +
824 +    /**
825 +     * Joins this task, without returning its result or throwing an
826 +     * exception. This method may be useful when processing
827 +     * collections of tasks when some have been cancelled or otherwise
828 +     * known to have aborted.
829 +     */
830 +    public final void quietlyJoin() {
831 +        if (status >= 0) {
832 +            ForkJoinWorkerThread w = getWorker();
833 +            if (w == null || !w.unpushTask(this) || !tryQuietlyInvoke())
834 +                awaitDone(w, true);
835 +        }
836 +    }
837 +
838 +    /**
839 +     * Commences performing this task and awaits its completion if
840 +     * necessary, without returning its result or throwing an
841 +     * exception. This method may be useful when processing
842 +     * collections of tasks when some have been cancelled or otherwise
843 +     * known to have aborted.
844 +     */
845 +    public final void quietlyInvoke() {
846 +        if (status >= 0 && !tryQuietlyInvoke())
847 +            quietlyJoin();
848 +    }
849 +
850 +    /**
851 +     * Possibly executes tasks until the pool hosting the current task
852 +     * {@link ForkJoinPool#isQuiescent}. This method may be of use in
853 +     * designs in which many tasks are forked, but none are explicitly
854 +     * joined, instead executing them until all are processed.
855 +     */
856 +    public static void helpQuiesce() {
857 +        ((ForkJoinWorkerThread) Thread.currentThread())
858 +            .helpQuiescePool();
859 +    }
860 +
861      /**
862       * Resets the internal bookkeeping state of this task, allowing a
863 <     * subsequent <tt>fork</tt>. This method allows repeated reuse of
863 >     * subsequent {@code fork}. This method allows repeated reuse of
864       * this task, but only if reuse occurs when this task has either
865       * never been forked, or has been forked, then completed and all
866       * outstanding joins of this task have also completed. Effects
# Line 689 | Line 875 | public abstract class ForkJoinTask<V> im
875      }
876  
877      /**
878 <     * Tries to unschedule this task for execution. This method will
879 <     * typically succeed if this task is the next task that would be
880 <     * executed by the current thread, and will typically fail (return
881 <     * false) otherwise. This method may be useful when arranging
696 <     * faster local processing of tasks that could have been, but were
697 <     * not, stolen.
698 <     * @return true if unforked
699 <     */
700 <    public boolean tryUnfork() {
701 <        return ((ForkJoinWorkerThread)(Thread.currentThread())).unpushTask(this);
702 <    }
703 <
704 <    /**
705 <     * Forks both tasks, returning when <tt>isDone</tt> holds for both
706 <     * of them or an exception is encountered. This method may be
707 <     * invoked only from within other ForkJoinTask
708 <     * computations. Attempts to invoke in other contexts result in
709 <     * exceptions or errors including ClassCastException.
710 <     * @param t1 one task
711 <     * @param t2 the other task
712 <     * @throws NullPointerException if t1 or t2 are null
713 <     * @throws RuntimeException or Error if either task did so.
878 >     * Returns the pool hosting the current task execution, or null
879 >     * if this task is executing outside of any ForkJoinPool.
880 >     *
881 >     * @return the pool, or {@code null} if none
882       */
883 <    public static void invokeAll(ForkJoinTask<?>t1, ForkJoinTask<?> t2) {
884 <        t2.fork();
885 <        t1.invoke();
886 <        t2.join();
883 >    public static ForkJoinPool getPool() {
884 >        Thread t = Thread.currentThread();
885 >        return (t instanceof ForkJoinWorkerThread) ?
886 >            ((ForkJoinWorkerThread) t).pool : null;
887      }
888  
889      /**
890 <     * Forks the given tasks, returning when <tt>isDone</tt> holds for
891 <     * all of them. If any task encounters an exception, others may be
892 <     * cancelled.  This method may be invoked only from within other
893 <     * ForkJoinTask computations. Attempts to invoke in other contexts
894 <     * result in exceptions or errors including ClassCastException.
727 <     * @param tasks the array of tasks
728 <     * @throws NullPointerException if tasks or any element are null.
729 <     * @throws RuntimeException or Error if any task did so.
890 >     * Returns {@code true} if the current thread is executing as a
891 >     * ForkJoinPool computation.
892 >     *
893 >     * @return {@code true} if the current thread is executing as a
894 >     * ForkJoinPool computation, or false otherwise
895       */
896 <    public static void invokeAll(ForkJoinTask<?>... tasks) {
897 <        Throwable ex = null;
733 <        int last = tasks.length - 1;
734 <        for (int i = last; i >= 0; --i) {
735 <            ForkJoinTask<?> t = tasks[i];
736 <            if (t == null) {
737 <                if (ex == null)
738 <                    ex = new NullPointerException();
739 <            }
740 <            else if (i != 0)
741 <                t.fork();
742 <            else {
743 <                t.quietlyInvoke();
744 <                if (ex == null)
745 <                    ex = t.getException();
746 <            }
747 <        }
748 <        for (int i = 1; i <= last; ++i) {
749 <            ForkJoinTask<?> t = tasks[i];
750 <            if (t != null) {
751 <                if (ex != null)
752 <                    t.cancel(false);
753 <                else {
754 <                    t.quietlyJoin();
755 <                    if (ex == null)
756 <                        ex = t.getException();
757 <                }
758 <            }
759 <        }
760 <        if (ex != null)
761 <            rethrowException(ex);
896 >    public static boolean inForkJoinPool() {
897 >        return Thread.currentThread() instanceof ForkJoinWorkerThread;
898      }
899  
900      /**
901 <     * Forks all tasks in the collection, returning when
902 <     * <tt>isDone</tt> holds for all of them. If any task encounters
903 <     * an exception, others may be cancelled.  This method may be
904 <     * invoked only from within other ForkJoinTask
905 <     * computations. Attempts to invoke in other contexts result in
906 <     * exceptions or errors including ClassCastException.
907 <     * @param tasks the collection of tasks
908 <     * @throws NullPointerException if tasks or any element are null.
909 <     * @throws RuntimeException or Error if any task did so.
901 >     * Tries to unschedule this task for execution. This method will
902 >     * typically succeed if this task is the most recently forked task
903 >     * by the current thread, and has not commenced executing in
904 >     * another thread.  This method may be useful when arranging
905 >     * alternative local processing of tasks that could have been, but
906 >     * were not, stolen. This method may be invoked only from within
907 >     * ForkJoinTask computations (as may be determined using method
908 >     * {@link #inForkJoinPool}). Attempts to invoke in other contexts
909 >     * result in exceptions or errors, possibly including
910 >     * ClassCastException.
911 >     *
912 >     * @return {@code true} if unforked
913       */
914 <    public static void invokeAll(Collection<? extends ForkJoinTask<?>> tasks) {
915 <        if (!(tasks instanceof List)) {
916 <            invokeAll(tasks.toArray(new ForkJoinTask[tasks.size()]));
778 <            return;
779 <        }
780 <        List<? extends ForkJoinTask<?>> ts =
781 <            (List<? extends ForkJoinTask<?>>)tasks;
782 <        Throwable ex = null;
783 <        int last = ts.size() - 1;
784 <        for (int i = last; i >= 0; --i) {
785 <            ForkJoinTask<?> t = ts.get(i);
786 <            if (t == null) {
787 <                if (ex == null)
788 <                    ex = new NullPointerException();
789 <            }
790 <            else if (i != 0)
791 <                t.fork();
792 <            else {
793 <                t.quietlyInvoke();
794 <                if (ex == null)
795 <                    ex = t.getException();
796 <            }
797 <        }
798 <        for (int i = 1; i <= last; ++i) {
799 <            ForkJoinTask<?> t = ts.get(i);
800 <            if (t != null) {
801 <                if (ex != null)
802 <                    t.cancel(false);
803 <                else {
804 <                    t.quietlyJoin();
805 <                    if (ex == null)
806 <                        ex = t.getException();
807 <                }
808 <            }
809 <        }
810 <        if (ex != null)
811 <            rethrowException(ex);
914 >    public boolean tryUnfork() {
915 >        return ((ForkJoinWorkerThread) Thread.currentThread())
916 >            .unpushTask(this);
917      }
918  
919      /**
920 <     * Possibly executes tasks until the pool hosting the current task
921 <     * {@link ForkJoinPool#isQuiescent}. This method may be of use in
922 <     * designs in which many tasks are forked, but none are explicitly
923 <     * joined, instead executing them until all are processed.
920 >     * Returns an estimate of the number of tasks that have been
921 >     * forked by the current worker thread but not yet executed. This
922 >     * value may be useful for heuristic decisions about whether to
923 >     * fork other tasks.
924 >     *
925 >     * @return the number of tasks
926       */
927 <    public static void helpQuiesce() {
928 <        ((ForkJoinWorkerThread)(Thread.currentThread())).
929 <            helpQuiescePool();
927 >    public static int getQueuedTaskCount() {
928 >        return ((ForkJoinWorkerThread) Thread.currentThread())
929 >            .getQueueSize();
930      }
931  
932      /**
933 <     * Returns a estimate of how many more locally queued tasks are
933 >     * Returns an estimate of how many more locally queued tasks are
934       * held by the current worker thread than there are other worker
935 <     * threads that might want to steal them.  This value may be
936 <     * useful for heuristic decisions about whether to fork other
937 <     * tasks. In many usages of ForkJoinTasks, at steady state, each
938 <     * worker should aim to maintain a small constant surplus (for
939 <     * example, 3) of tasks, and to process computations locally if
940 <     * this threshold is exceeded.
935 >     * threads that might steal them.  This value may be useful for
936 >     * heuristic decisions about whether to fork other tasks. In many
937 >     * usages of ForkJoinTasks, at steady state, each worker should
938 >     * aim to maintain a small constant surplus (for example, 3) of
939 >     * tasks, and to process computations locally if this threshold is
940 >     * exceeded.
941 >     *
942       * @return the surplus number of tasks, which may be negative
943       */
944 <    public static int surplus() {
945 <        return ((ForkJoinWorkerThread)(Thread.currentThread()))
944 >    public static int getSurplusQueuedTaskCount() {
945 >        return ((ForkJoinWorkerThread) Thread.currentThread())
946              .getEstimatedSurplusTaskCount();
947      }
948  
949 <    // Extension kit
949 >    // Extension methods
950  
951      /**
952 <     * Returns the result that would be returned by <tt>join</tt>, or
953 <     * null if this task is not known to have been completed.  This
954 <     * method is designed to aid debugging, as well as to support
955 <     * extensions. Its use in any other context is discouraged.
952 >     * Returns the result that would be returned by {@link #join}, even
953 >     * if this task completed abnormally, or {@code null} if this task
954 >     * is not known to have been completed.  This method is designed
955 >     * to aid debugging, as well as to support extensions. Its use in
956 >     * any other context is discouraged.
957       *
958 <     * @return the result, or null if not completed.
958 >     * @return the result, or {@code null} if not completed
959       */
960      public abstract V getRawResult();
961  
# Line 865 | Line 974 | public abstract class ForkJoinTask<V> im
974       * called otherwise. The return value controls whether this task
975       * is considered to be done normally. It may return false in
976       * asynchronous actions that require explicit invocations of
977 <     * <tt>complete</tt> to become joinable. It may throw exceptions
977 >     * {@link #complete} to become joinable. It may throw exceptions
978       * to indicate abnormal exit.
979 <     * @return true if completed normally
979 >     *
980 >     * @return {@code true} if completed normally
981       * @throws Error or RuntimeException if encountered during computation
982       */
983      protected abstract boolean exec();
984  
985 +    /**
986 +     * Returns, but does not unschedule or execute, the task queued by
987 +     * the current thread but not yet executed, if one is
988 +     * available. There is no guarantee that this task will actually
989 +     * be polled or executed next.  This method is designed primarily
990 +     * to support extensions, and is unlikely to be useful otherwise.
991 +     * This method may be invoked only from within ForkJoinTask
992 +     * computations (as may be determined using method {@link
993 +     * #inForkJoinPool}). Attempts to invoke in other contexts result
994 +     * in exceptions or errors, possibly including ClassCastException.
995 +     *
996 +     * @return the next task, or {@code null} if none are available
997 +     */
998 +    protected static ForkJoinTask<?> peekNextLocalTask() {
999 +        return ((ForkJoinWorkerThread) Thread.currentThread())
1000 +            .peekTask();
1001 +    }
1002 +
1003 +    /**
1004 +     * Unschedules and returns, without executing, the next task
1005 +     * queued by the current thread but not yet executed.  This method
1006 +     * is designed primarily to support extensions, and is unlikely to
1007 +     * be useful otherwise.  This method may be invoked only from
1008 +     * within ForkJoinTask computations (as may be determined using
1009 +     * method {@link #inForkJoinPool}). Attempts to invoke in other
1010 +     * contexts result in exceptions or errors, possibly including
1011 +     * ClassCastException.
1012 +     *
1013 +     * @return the next task, or {@code null} if none are available
1014 +     */
1015 +    protected static ForkJoinTask<?> pollNextLocalTask() {
1016 +        return ((ForkJoinWorkerThread) Thread.currentThread())
1017 +            .pollLocalTask();
1018 +    }
1019 +
1020 +    /**
1021 +     * Unschedules and returns, without executing, the next task
1022 +     * queued by the current thread but not yet executed, if one is
1023 +     * available, or if not available, a task that was forked by some
1024 +     * other thread, if available. Availability may be transient, so a
1025 +     * {@code null} result does not necessarily imply quiescence
1026 +     * of the pool this task is operating in.  This method is designed
1027 +     * primarily to support extensions, and is unlikely to be useful
1028 +     * otherwise.  This method may be invoked only from within
1029 +     * ForkJoinTask computations (as may be determined using method
1030 +     * {@link #inForkJoinPool}). Attempts to invoke in other contexts
1031 +     * result in exceptions or errors, possibly including
1032 +     * ClassCastException.
1033 +     *
1034 +     * @return a task, or {@code null} if none are available
1035 +     */
1036 +    protected static ForkJoinTask<?> pollTask() {
1037 +        return ((ForkJoinWorkerThread) Thread.currentThread())
1038 +            .pollTask();
1039 +    }
1040 +
1041 +    // adaptors
1042 +
1043 +    /**
1044 +     * Returns a new ForkJoinTask that performs the {@code run}
1045 +     * method of the given Runnable as its action, and returns a null
1046 +     * result upon {@code join}.
1047 +     *
1048 +     * @param runnable the runnable action
1049 +     * @return the task
1050 +     */
1051 +    public static ForkJoinTask<Void> adapt(Runnable runnable) {
1052 +        return new ForkJoinPool.AdaptedRunnable<Void>(runnable, null);
1053 +    }
1054 +
1055 +    /**
1056 +     * Returns a new ForkJoinTask that performs the {@code run}
1057 +     * method of the given Runnable as its action, and returns the
1058 +     * given result upon {@code join}.
1059 +     *
1060 +     * @param runnable the runnable action
1061 +     * @param result the result upon completion
1062 +     * @return the task
1063 +     */
1064 +    public static <T> ForkJoinTask<T> adapt(Runnable runnable, T result) {
1065 +        return new ForkJoinPool.AdaptedRunnable<T>(runnable, result);
1066 +    }
1067 +
1068 +    /**
1069 +     * Returns a new ForkJoinTask that performs the {@code call}
1070 +     * method of the given Callable as its action, and returns its
1071 +     * result upon {@code join}, translating any checked
1072 +     * exceptions encountered into {@code RuntimeException}.
1073 +     *
1074 +     * @param callable the callable action
1075 +     * @return the task
1076 +     */
1077 +    public static <T> ForkJoinTask<T> adapt(Callable<T> callable) {
1078 +        return new ForkJoinPool.AdaptedCallable<T>(callable);
1079 +    }
1080 +
1081      // Serialization support
1082  
1083      private static final long serialVersionUID = -7721805057305804111L;
# Line 880 | Line 1086 | public abstract class ForkJoinTask<V> im
1086       * Save the state to a stream.
1087       *
1088       * @serialData the current run status and the exception thrown
1089 <     * during execution, or null if none.
1089 >     * during execution, or {@code null} if none
1090       * @param s the stream
1091       */
1092      private void writeObject(java.io.ObjectOutputStream s)
# Line 891 | Line 1097 | public abstract class ForkJoinTask<V> im
1097  
1098      /**
1099       * Reconstitute the instance from a stream.
1100 +     *
1101       * @param s the stream
1102       */
1103      private void readObject(java.io.ObjectInputStream s)
1104          throws java.io.IOException, ClassNotFoundException {
1105          s.defaultReadObject();
1106 <        //        status &= ~INTERNAL_SIGNAL_MASK; //  todo: define policy
1106 >        status &= ~INTERNAL_SIGNAL_MASK; // clear internal signal counts
1107 >        status |= EXTERNAL_SIGNAL; // conservatively set external signal
1108          Object ex = s.readObject();
1109          if (ex != null)
1110 <            setDoneExceptionally((Throwable)ex);
1110 >            setDoneExceptionally((Throwable) ex);
1111      }
1112  
1113 <    // Temporary Unsafe mechanics for preliminary release
1113 >    // Unsafe mechanics
1114  
1115 <    static final Unsafe _unsafe;
1116 <    static final long statusOffset;
1115 >    private static final sun.misc.Unsafe UNSAFE = getUnsafe();
1116 >    private static final long statusOffset =
1117 >        objectFieldOffset("status", ForkJoinTask.class);
1118  
1119 <    static {
1119 >    private static long objectFieldOffset(String field, Class<?> klazz) {
1120          try {
1121 <            if (ForkJoinTask.class.getClassLoader() != null) {
1122 <                Field f = Unsafe.class.getDeclaredField("theUnsafe");
1123 <                f.setAccessible(true);
1124 <                _unsafe = (Unsafe)f.get(null);
1125 <            }
1126 <            else
1127 <                _unsafe = Unsafe.getUnsafe();
919 <            statusOffset = _unsafe.objectFieldOffset
920 <                (ForkJoinTask.class.getDeclaredField("status"));
921 <        } catch (Exception ex) { throw new Error(ex); }
1121 >            return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
1122 >        } catch (NoSuchFieldException e) {
1123 >            // Convert Exception to corresponding Error
1124 >            NoSuchFieldError error = new NoSuchFieldError(field);
1125 >            error.initCause(e);
1126 >            throw error;
1127 >        }
1128      }
1129  
1130 +    /**
1131 +     * Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
1132 +     * Replace with a simple call to Unsafe.getUnsafe when integrating
1133 +     * into a jdk.
1134 +     *
1135 +     * @return a sun.misc.Unsafe
1136 +     */
1137 +    private static sun.misc.Unsafe getUnsafe() {
1138 +        try {
1139 +            return sun.misc.Unsafe.getUnsafe();
1140 +        } catch (SecurityException se) {
1141 +            try {
1142 +                return java.security.AccessController.doPrivileged
1143 +                    (new java.security
1144 +                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
1145 +                        public sun.misc.Unsafe run() throws Exception {
1146 +                            java.lang.reflect.Field f = sun.misc
1147 +                                .Unsafe.class.getDeclaredField("theUnsafe");
1148 +                            f.setAccessible(true);
1149 +                            return (sun.misc.Unsafe) f.get(null);
1150 +                        }});
1151 +            } catch (java.security.PrivilegedActionException e) {
1152 +                throw new RuntimeException("Could not initialize intrinsics",
1153 +                                           e.getCause());
1154 +            }
1155 +        }
1156 +    }
1157   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines