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.74 by dl, Tue Feb 22 00:39:31 2011 UTC

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

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines