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.20 by jsr166, Sun Jul 26 05:55:34 2009 UTC vs.
Revision 1.56 by jsr166, Sat Sep 4 00:21:31 2010 UTC

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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines