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

Comparing jsr166/src/jsr166y/ForkJoinTask.java (file contents):
Revision 1.1 by dl, Tue Jan 6 14:30:31 2009 UTC vs.
Revision 1.79 by jsr166, Fri Jun 10 18:10:53 2011 UTC

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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines