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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines