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.47 by dl, Sun Apr 18 12:51:18 2010 UTC vs.
Revision 1.49 by dl, Wed Jul 7 19:52:31 2010 UTC

# Line 64 | Line 64 | import java.util.WeakHashMap;
64   * results of a task is {@link #join}, but there are several variants:
65   * The {@link Future#get} methods support interruptible and/or timed
66   * waits for completion and report results using {@code Future}
67 < * conventions. Method {@link #helpJoin} enables callers to actively
68 < * execute other tasks while awaiting joins, which is sometimes more
69 < * efficient but only applies when all subtasks are known to be
70 < * strictly tree-structured. Method {@link #invoke} is semantically
67 > * conventions. Method {@link #invoke} is semantically
68   * equivalent to {@code fork(); join()} but always attempts to begin
69   * execution in the current thread. The "<em>quiet</em>" forms of
70   * these methods do not extract results or report exceptions. These
# Line 125 | Line 122 | import java.util.WeakHashMap;
122   *
123   * <p>This class provides {@code adapt} methods for {@link Runnable}
124   * and {@link Callable}, that may be of use when mixing execution of
125 < * {@code ForkJoinTasks} with other kinds of tasks. When all tasks
126 < * are of this form, consider using a pool in
130 < * {@linkplain ForkJoinPool#setAsyncMode async mode}.
125 > * {@code ForkJoinTasks} with other kinds of tasks. When all tasks are
126 > * of this form, consider using a pool constructed in <em>asyncMode</em>.
127   *
128   * <p>ForkJoinTasks are {@code Serializable}, which enables them to be
129   * used in extensions such as remote execution frameworks. It is
# Line 155 | Line 151 | public abstract class ForkJoinTask<V> im
151       * Run control status bits packed into a single int to minimize
152       * footprint and to ensure atomicity (via CAS).  Status is
153       * initially zero, and takes on nonnegative values until
154 <     * completed, upon which status holds COMPLETED. CANCELLED, or
155 <     * EXCEPTIONAL, which use the top 3 bits.  Tasks undergoing
156 <     * blocking waits by other threads have SIGNAL_MASK bits set --
157 <     * bit 15 for external (nonFJ) waits, and the rest a count of
158 <     * waiting FJ threads.  (This representation relies on
159 <     * ForkJoinPool max thread limits). Signal counts are not directly
160 <     * incremented by ForkJoinTask methods, but instead via a call to
161 <     * requestSignal within ForkJoinPool.preJoin, once their need is
162 <     * established.
163 <     *
168 <     * Completion of a stolen task with SIGNAL_MASK bits set awakens
169 <     * any waiters via notifyAll. Even though suboptimal for some
170 <     * purposes, we use basic builtin wait/notify to take advantage of
171 <     * "monitor inflation" in JVMs that we would otherwise need to
172 <     * emulate to avoid adding further per-task bookkeeping overhead.
173 <     * We want these monitors to be "fat", i.e., not use biasing or
174 <     * thin-lock techniques, so use some odd coding idioms that tend
175 <     * to avoid them.
176 <     *
177 <     * Note that bits 16-28 are currently unused. Also value
178 <     * 0x80000000 is available as spare completion value.
154 >     * completed, upon which status holds value COMPLETED. CANCELLED,
155 >     * or EXCEPTIONAL. Tasks undergoing blocking waits by other
156 >     * threads have the SIGNAL bit set.  Completion of a stolen task
157 >     * with SIGNAL set awakens any waiters via notifyAll. Even though
158 >     * suboptimal for some purposes, we use basic builtin wait/notify
159 >     * to take advantage of "monitor inflation" in JVMs that we would
160 >     * otherwise need to emulate to avoid adding further per-task
161 >     * bookkeeping overhead.  We want these monitors to be "fat",
162 >     * i.e., not use biasing or thin-lock techniques, so use some odd
163 >     * coding idioms that tend to avoid them.
164       */
165      volatile int status; // accessed directly by pool and workers
166  
167 <    private static final int COMPLETION_MASK      = 0xe0000000;
168 <    private static final int NORMAL               = 0xe0000000; // == mask
169 <    private static final int CANCELLED            = 0xc0000000;
170 <    private static final int EXCEPTIONAL          = 0xa0000000;
186 <    private static final int SIGNAL_MASK          = 0x0000ffff;
187 <    private static final int INTERNAL_SIGNAL_MASK = 0x00007fff;
188 <    private static final int EXTERNAL_SIGNAL      = 0x00008000;
167 >    private static final int NORMAL      = -1;
168 >    private static final int CANCELLED   = -2;
169 >    private static final int EXCEPTIONAL = -3;
170 >    private static final int SIGNAL      =  1;
171  
172      /**
173       * Table of exceptions thrown by tasks, to enable reporting by
# Line 206 | Line 188 | public abstract class ForkJoinTask<V> im
188       * also clearing signal request bits.
189       *
190       * @param completion one of NORMAL, CANCELLED, EXCEPTIONAL
191 +     * @return status on exit
192       */
193 <    private void setCompletion(int completion) {
193 >    private int setCompletion(int completion) {
194          int s;
195          while ((s = status) >= 0) {
196              if (UNSAFE.compareAndSwapInt(this, statusOffset, s, completion)) {
197 <                if ((s & SIGNAL_MASK) != 0) {
215 <                    Thread t = Thread.currentThread();
216 <                    if (t instanceof ForkJoinWorkerThread)
217 <                        ((ForkJoinWorkerThread) t).pool.updateRunningCount
218 <                            (s & INTERNAL_SIGNAL_MASK);
197 >                if (s == SIGNAL)
198                      synchronized (this) { notifyAll(); }
199 <                }
221 <                return;
199 >                return completion;
200              }
201          }
202 +        return s;
203      }
204  
205      /**
206       * Record exception and set exceptional completion
207 +     * @return status on exit
208       */
209 <    private void setDoneExceptionally(Throwable rex) {
209 >    private int setExceptionalCompletion(Throwable rex) {
210          exceptionMap.put(this, rex);
211 <        setCompletion(EXCEPTIONAL);
232 <    }
233 <
234 <    /**
235 <     * Main internal execution method: Unless done, calls exec and
236 <     * records completion.
237 <     *
238 <     * @return true if ran and completed normally
239 <     */
240 <    final boolean tryExec() {
241 <        try {
242 <            if (status < 0 || !exec())
243 <                return false;
244 <        } catch (Throwable rex) {
245 <            setDoneExceptionally(rex);
246 <            return false;
247 <        }
248 <        setCompletion(NORMAL); // must be outside try block
249 <        return true;
250 <    }
251 <
252 <    /**
253 <     * Increments internal signal count (thus requesting signal upon
254 <     * completion) unless already done.  Call only once per join.
255 <     * Used by ForkJoinPool.preJoin.
256 <     *
257 <     * @return status
258 <     */
259 <    final int requestSignal() {
260 <        int s;
261 <        do {} while ((s = status) >= 0 &&
262 <                     !UNSAFE.compareAndSwapInt(this, statusOffset, s, s + 1));
263 <        return s;
211 >        return setCompletion(EXCEPTIONAL);
212      }
213  
214      /**
215 <     * Sets external signal request unless already done.
268 <     *
269 <     * @return status
215 >     * Blocks a worker thread until completion. Called only by pool.
216       */
217 <    private int requestExternalSignal() {
217 >    final int internalAwaitDone() {
218          int s;
219 <        do {} while ((s = status) >= 0 &&
220 <                     !UNSAFE.compareAndSwapInt(this, statusOffset,
221 <                                               s, s | EXTERNAL_SIGNAL));
222 <        return s;
223 <    }
278 <
279 <    /*
280 <     * Awaiting completion. The four versions, internal vs external X
281 <     * untimed vs timed, have the same overall structure but differ
282 <     * from each other enough to defy simple integration.
283 <     */
284 <
285 <    /**
286 <     * Blocks a worker until this task is done, also maintaining pool
287 <     * and signal counts
288 <     */
289 <    private void awaitDone(ForkJoinWorkerThread w) {
290 <        if (status >= 0) {
291 <            w.pool.preJoin(this);
292 <            while (status >= 0) {
293 <                try { // minimize lock scope
294 <                    synchronized(this) {
295 <                        if (status >= 0)
219 >        while ((s = status) >= 0) {
220 >            synchronized(this) {
221 >                if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)) {
222 >                    do {
223 >                        try {
224                              wait();
225 <                        else { // help release; also helps avoid lock-biasing
226 <                            notifyAll();
299 <                            break;
225 >                        } catch (InterruptedException ie) {
226 >                            cancelIfTerminating();
227                          }
228 <                    }
229 <                } catch (InterruptedException ie) {
303 <                    cancelIfTerminating();
228 >                    } while ((s = status) >= 0);
229 >                    break;
230                  }
231              }
232          }
233 +        return s;
234      }
235  
236      /**
237 <     * Blocks a non-ForkJoin thread until this task is done.
237 >     * Blocks a non-worker-thread until completion.
238 >     * @return status on exit
239       */
240 <    private void externalAwaitDone() {
241 <        if (requestExternalSignal() >= 0) {
242 <            boolean interrupted = false;
243 <            while (status >= 0) {
244 <                try {
245 <                    synchronized(this) {
246 <                        if (status >= 0)
240 >    private int externalAwaitDone() {
241 >        int s;
242 >        while ((s = status) >= 0) {
243 >            synchronized(this) {
244 >                if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)){
245 >                    boolean interrupted = false;
246 >                    do {
247 >                        try {
248                              wait();
249 <                        else {
250 <                            notifyAll();
322 <                            break;
249 >                        } catch (InterruptedException ie) {
250 >                            interrupted = true;
251                          }
252 <                    }
253 <                } catch (InterruptedException ie) {
254 <                    interrupted = true;
252 >                    } while ((s = status) >= 0);
253 >                    if (interrupted)
254 >                        Thread.currentThread().interrupt();
255 >                    break;
256                  }
257              }
329            if (interrupted)
330                Thread.currentThread().interrupt();
258          }
259 +        return s;
260      }
261  
262      /**
263 <     * Blocks a worker until this task is done or timeout elapses
263 >     * Unless done, calls exec and records status if completed, but
264 >     * doesn't wait for completion otherwise. Primary execution method
265 >     * for ForkJoinWorkerThread.
266       */
267 <    private void timedAwaitDone(ForkJoinWorkerThread w, long nanos) {
268 <        if (status >= 0) {
269 <            long startTime = System.nanoTime();
270 <            ForkJoinPool pool = w.pool;
271 <            pool.preJoin(this);
272 <            while (status >= 0) {
273 <                long nt = nanos - (System.nanoTime() - startTime);
344 <                if (nt > 0) {
345 <                    long ms = nt / 1000000;
346 <                    int ns = (int) (nt % 1000000);
347 <                    try {
348 <                        synchronized(this) { if (status >= 0) wait(ms, ns); }
349 <                    } catch (InterruptedException ie) {
350 <                        cancelIfTerminating();
351 <                    }
352 <                }
353 <                else {
354 <                    int s; // adjust running count on timeout
355 <                    while ((s = status) >= 0 &&
356 <                           (s & INTERNAL_SIGNAL_MASK) != 0) {
357 <                        if (UNSAFE.compareAndSwapInt(this, statusOffset,
358 <                                                     s, s - 1)) {
359 <                            pool.updateRunningCount(1);
360 <                            break;
361 <                        }
362 <                    }
363 <                    break;
364 <                }
365 <            }
267 >    final void tryExec() {
268 >        try {
269 >            if (status < 0 || !exec())
270 >                return;
271 >        } catch (Throwable rex) {
272 >            setExceptionalCompletion(rex);
273 >            return;
274          }
275 +        setCompletion(NORMAL); // must be outside try block
276      }
277  
278      /**
279 <     * Blocks a non-ForkJoin thread until this task is done or timeout elapses
280 <     */
281 <    private void externalTimedAwaitDone(long nanos) {
282 <        if (requestExternalSignal() >= 0) {
283 <            long startTime = System.nanoTime();
284 <            boolean interrupted = false;
285 <            while (status >= 0) {
286 <                long nt = nanos - (System.nanoTime() - startTime);
287 <                if (nt <= 0)
288 <                    break;
289 <                long ms = nt / 1000000;
290 <                int ns = (int) (nt % 1000000);
279 >     * If not done and this task is next in worker queue, runs it,
280 >     * else waits for it.
281 >     * @return status on exit
282 >     */
283 >    private int doJoin() {
284 >        int stat;
285 >        if ((stat = status) < 0)
286 >            return stat;
287 >        Thread t = Thread.currentThread();
288 >        ForkJoinWorkerThread w;
289 >        if (t instanceof ForkJoinWorkerThread) {
290 >            if ((w = (ForkJoinWorkerThread) t).unpushTask(this)) {
291 >                boolean completed;
292                  try {
293 <                    synchronized(this) { if (status >= 0) wait(ms, ns); }
294 <                } catch (InterruptedException ie) {
295 <                    interrupted = true;
293 >                    completed = exec();
294 >                } catch (Throwable rex) {
295 >                    return setExceptionalCompletion(rex);
296                  }
297 +                if (completed)
298 +                    return setCompletion(NORMAL);
299              }
300 <            if (interrupted)
301 <                Thread.currentThread().interrupt();
390 <        }
391 <    }
392 <
393 <    // reporting results
394 <
395 <    /**
396 <     * Returns result or throws the exception associated with status.
397 <     * Uses Unsafe as a workaround for javac not allowing rethrow of
398 <     * unchecked exceptions.
399 <     */
400 <    private V reportResult() {
401 <        if ((status & COMPLETION_MASK) < NORMAL) {
402 <            Throwable ex = getException();
403 <            if (ex != null)
404 <                UNSAFE.throwException(ex);
300 >            w.joinTask(this);
301 >            return status;
302          }
303 <        return getRawResult();
303 >        return externalAwaitDone();
304      }
305  
306      /**
307 <     * Returns result or throws exception using j.u.c.Future conventions.
308 <     * Only call when {@code isDone} known to be true or thread known
309 <     * to be interrupted.
307 >     * Unless done, calls exec and records status if completed, or
308 >     * waits for completion otherwise.
309 >     * @return status on exit
310       */
311 <    private V reportFutureResult()
312 <        throws InterruptedException, ExecutionException {
313 <        if (Thread.interrupted())
314 <            throw new InterruptedException();
315 <        int s = status & COMPLETION_MASK;
316 <        if (s < NORMAL) {
317 <            Throwable ex;
318 <            if (s == CANCELLED)
319 <                throw new CancellationException();
320 <            if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
321 <                throw new ExecutionException(ex);
311 >    private int doInvoke() {
312 >        int stat;
313 >        if ((stat = status) >= 0) {
314 >            boolean completed;
315 >            try {
316 >                completed = exec();
317 >            } catch (Throwable rex) {
318 >                return setExceptionalCompletion(rex);
319 >            }
320 >            if (completed)
321 >                stat = setCompletion(NORMAL);
322 >            else
323 >                stat = doJoin();
324          }
325 <        return getRawResult();
325 >        return stat;
326      }
327  
328      /**
329 <     * Returns result or throws exception using j.u.c.Future conventions
330 <     * with timeouts.
329 >     * Returns result or throws exception associated with given status.
330 >     * @param s the status
331       */
332 <    private V reportTimedFutureResult()
434 <        throws InterruptedException, ExecutionException, TimeoutException {
435 <        if (Thread.interrupted())
436 <            throw new InterruptedException();
332 >    private V reportResult(int s) {
333          Throwable ex;
334 <        int s = status & COMPLETION_MASK;
335 <        if (s == NORMAL)
336 <            return getRawResult();
441 <        else if (s == CANCELLED)
442 <            throw new CancellationException();
443 <        else if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
444 <            throw new ExecutionException(ex);
445 <        else
446 <            throw new TimeoutException();
334 >        if (s < NORMAL && (ex = getException()) != null)
335 >            UNSAFE.throwException(ex);
336 >        return getRawResult();
337      }
338  
339      // public methods
# Line 481 | Line 371 | public abstract class ForkJoinTask<V> im
371       * @return the computed result
372       */
373      public final V join() {
374 <        quietlyJoin();
485 <        return reportResult();
374 >        return reportResult(doJoin());
375      }
376  
377      /**
# Line 493 | Line 382 | public abstract class ForkJoinTask<V> im
382       * @return the computed result
383       */
384      public final V invoke() {
385 <        if (!tryExec())
497 <            quietlyJoin();
498 <        return reportResult();
385 >        return reportResult(doInvoke());
386      }
387  
388      /**
# Line 553 | Line 440 | public abstract class ForkJoinTask<V> im
440              }
441              else if (i != 0)
442                  t.fork();
443 <            else {
444 <                t.quietlyInvoke();
558 <                if (ex == null)
559 <                    ex = t.getException();
560 <            }
443 >            else if (t.doInvoke() < NORMAL && ex == null)
444 >                ex = t.getException();
445          }
446          for (int i = 1; i <= last; ++i) {
447              ForkJoinTask<?> t = tasks[i];
448              if (t != null) {
449                  if (ex != null)
450                      t.cancel(false);
451 <                else {
452 <                    t.quietlyJoin();
569 <                    if (ex == null)
570 <                        ex = t.getException();
571 <                }
451 >                else if (t.doJoin() < NORMAL && ex == null)
452 >                    ex = t.getException();
453              }
454          }
455          if (ex != null)
# Line 615 | Line 496 | public abstract class ForkJoinTask<V> im
496              }
497              else if (i != 0)
498                  t.fork();
499 <            else {
500 <                t.quietlyInvoke();
620 <                if (ex == null)
621 <                    ex = t.getException();
622 <            }
499 >            else if (t.doInvoke() < NORMAL && ex == null)
500 >                ex = t.getException();
501          }
502          for (int i = 1; i <= last; ++i) {
503              ForkJoinTask<?> t = ts.get(i);
504              if (t != null) {
505                  if (ex != null)
506                      t.cancel(false);
507 <                else {
508 <                    t.quietlyJoin();
631 <                    if (ex == null)
632 <                        ex = t.getException();
633 <                }
507 >                else if (t.doJoin() < NORMAL && ex == null)
508 >                    ex = t.getException();
509              }
510          }
511          if (ex != null)
# Line 664 | Line 539 | public abstract class ForkJoinTask<V> im
539       */
540      public boolean cancel(boolean mayInterruptIfRunning) {
541          setCompletion(CANCELLED);
542 <        return (status & COMPLETION_MASK) == CANCELLED;
542 >        return status == CANCELLED;
543      }
544  
545      /**
# Line 697 | Line 572 | public abstract class ForkJoinTask<V> im
572      }
573  
574      public final boolean isCancelled() {
575 <        return (status & COMPLETION_MASK) == CANCELLED;
575 >        return status == CANCELLED;
576      }
577  
578      /**
# Line 706 | Line 581 | public abstract class ForkJoinTask<V> im
581       * @return {@code true} if this task threw an exception or was cancelled
582       */
583      public final boolean isCompletedAbnormally() {
584 <        return (status & COMPLETION_MASK) < NORMAL;
584 >        return status < NORMAL;
585      }
586  
587      /**
# Line 717 | Line 592 | public abstract class ForkJoinTask<V> im
592       * exception and was not cancelled
593       */
594      public final boolean isCompletedNormally() {
595 <        return (status & COMPLETION_MASK) == NORMAL;
595 >        return status == NORMAL;
596      }
597  
598      /**
# Line 728 | Line 603 | public abstract class ForkJoinTask<V> im
603       * @return the exception, or {@code null} if none
604       */
605      public final Throwable getException() {
606 <        int s = status & COMPLETION_MASK;
606 >        int s = status;
607          return ((s >= NORMAL)    ? null :
608                  (s == CANCELLED) ? new CancellationException() :
609                  exceptionMap.get(this));
# Line 749 | Line 624 | public abstract class ForkJoinTask<V> im
624       * thrown will be a {@code RuntimeException} with cause {@code ex}.
625       */
626      public void completeExceptionally(Throwable ex) {
627 <        setDoneExceptionally((ex instanceof RuntimeException) ||
628 <                             (ex instanceof Error) ? ex :
629 <                             new RuntimeException(ex));
627 >        setExceptionalCompletion((ex instanceof RuntimeException) ||
628 >                                 (ex instanceof Error) ? ex :
629 >                                 new RuntimeException(ex));
630      }
631  
632      /**
# Line 770 | Line 645 | public abstract class ForkJoinTask<V> im
645          try {
646              setRawResult(value);
647          } catch (Throwable rex) {
648 <            setDoneExceptionally(rex);
648 >            setExceptionalCompletion(rex);
649              return;
650          }
651          setCompletion(NORMAL);
652      }
653  
654      public final V get() throws InterruptedException, ExecutionException {
655 <        quietlyJoin();
656 <        return reportFutureResult();
655 >        int s = doJoin();
656 >        if (Thread.interrupted())
657 >            throw new InterruptedException();
658 >        if (s < NORMAL) {
659 >            Throwable ex;
660 >            if (s == CANCELLED)
661 >                throw new CancellationException();
662 >            if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
663 >                throw new ExecutionException(ex);
664 >        }
665 >        return getRawResult();
666      }
667  
668      public final V get(long timeout, TimeUnit unit)
669          throws InterruptedException, ExecutionException, TimeoutException {
786        long nanos = unit.toNanos(timeout);
670          Thread t = Thread.currentThread();
671 +        ForkJoinPool pool;
672          if (t instanceof ForkJoinWorkerThread) {
673              ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
674 <            if (!w.unpushTask(this) || !tryExec())
675 <                timedAwaitDone(w, nanos);
674 >            if (status >= 0 && w.unpushTask(this))
675 >                tryExec();
676 >            pool = w.pool;
677          }
678          else
679 <            externalTimedAwaitDone(nanos);
680 <        return reportTimedFutureResult();
681 <    }
682 <
683 <    /**
684 <     * Possibly executes other tasks until this task {@link #isDone is
685 <     * done}, then returns the result of the computation.  This method
686 <     * may be more efficient than {@code join}, but is only applicable
687 <     * when there are no potential dependencies between continuation
688 <     * of the current task and that of any other task that might be
689 <     * executed while helping. (This usually holds for pure
690 <     * divide-and-conquer tasks).
691 <     *
692 <     * <p>This method may be invoked only from within {@code
693 <     * ForkJoinTask} computations (as may be determined using method
694 <     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
695 <     * result in exceptions or errors, possibly including {@code
696 <     * ClassCastException}.
697 <     *
698 <     * @return the computed result
699 <     */
700 <    public final V helpJoin() {
701 <        quietlyHelpJoin();
702 <        return reportResult();
703 <    }
704 <
705 <    /**
706 <     * Possibly executes other tasks until this task {@link #isDone is
707 <     * done}.  This method may be useful when processing collections
708 <     * of tasks when some have been cancelled or otherwise known to
824 <     * have aborted.
825 <     *
826 <     * <p>This method may be invoked only from within {@code
827 <     * ForkJoinTask} computations (as may be determined using method
828 <     * {@link #inForkJoinPool}).  Attempts to invoke in other contexts
829 <     * result in exceptions or errors, possibly including {@code
830 <     * ClassCastException}.
831 <     */
832 <    public final void quietlyHelpJoin() {
833 <        ForkJoinWorkerThread w = (ForkJoinWorkerThread) Thread.currentThread();
834 <        if (!w.unpushTask(this) || !tryExec()) {
835 <            for (;;) {
836 <                ForkJoinTask<?> t;
837 <                if (status < 0)
838 <                    return;
839 <                else if ((t = w.scanWhileJoining(this)) != null)
840 <                    t.tryExec();
841 <                else if (status < 0)
842 <                    return;
843 <                else if (w.pool.preBlockHelpingJoin(this)) {
844 <                    while (status >= 0) { // variant of awaitDone
679 >            pool = null;
680 >        /*
681 >         * Timed wait loop intermixes cases for fj (pool != null) and
682 >         * non FJ threads. For FJ, decrement pool count but don't try
683 >         * for replacement; increment count on completion. For non-FJ,
684 >         * deal with interrupts. This is messy, but a little less so
685 >         * than is splitting the FJ and nonFJ cases.
686 >         */
687 >        boolean interrupted = false;
688 >        boolean dec = false; // true if pool count decremented
689 >        for (;;) {
690 >            if (Thread.interrupted() && pool == null) {
691 >                interrupted = true;
692 >                break;
693 >            }
694 >            int s = status;
695 >            if (s < 0)
696 >                break;
697 >            if (UNSAFE.compareAndSwapInt(this, statusOffset,
698 >                                         s, s | SIGNAL)) {
699 >                long startTime = System.nanoTime();
700 >                long nanos = unit.toNanos(timeout);
701 >                long nt; // wait time
702 >                while (status >= 0 &&
703 >                       (nt = nanos - (System.nanoTime() - startTime)) > 0) {
704 >                    if (pool != null && !dec)
705 >                        dec = pool.tryDecrementRunningCount();
706 >                    else {
707 >                        long ms = nt / 1000000;
708 >                        int ns = (int) (nt % 1000000);
709                          try {
710                              synchronized(this) {
711                                  if (status >= 0)
712 <                                    wait();
849 <                                else {
850 <                                    notifyAll();
851 <                                    break;
852 <                                }
712 >                                    wait(ms, ns);
713                              }
714                          } catch (InterruptedException ie) {
715 <                            cancelIfTerminating();
715 >                            if (pool != null)
716 >                                cancelIfTerminating();
717 >                            else {
718 >                                interrupted = true;
719 >                                break;
720 >                            }
721                          }
722                      }
858                    return;
723                  }
724 +                break;
725              }
726          }
727 +        if (pool != null && dec)
728 +            pool.incrementRunningCount();
729 +        if (interrupted)
730 +            throw new InterruptedException();
731 +        int es = status;
732 +        if (es != NORMAL) {
733 +            Throwable ex;
734 +            if (es == CANCELLED)
735 +                throw new CancellationException();
736 +            if (es == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
737 +                throw new ExecutionException(ex);
738 +            throw new TimeoutException();
739 +        }
740 +        return getRawResult();
741      }
742  
743      /**
# Line 868 | Line 747 | public abstract class ForkJoinTask<V> im
747       * known to have aborted.
748       */
749      public final void quietlyJoin() {
750 <        Thread t = Thread.currentThread();
872 <        if (t instanceof ForkJoinWorkerThread) {
873 <            ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
874 <            if (!w.unpushTask(this) || !tryExec())
875 <                awaitDone(w);
876 <        }
877 <        else
878 <            externalAwaitDone();
750 >        doJoin();
751      }
752  
753      /**
# Line 886 | Line 758 | public abstract class ForkJoinTask<V> im
758       * known to have aborted.
759       */
760      public final void quietlyInvoke() {
761 <        if (!tryExec())
890 <            quietlyJoin();
761 >        doInvoke();
762      }
763  
764      /**
# Line 919 | Line 790 | public abstract class ForkJoinTask<V> im
790       * pre-constructed trees of subtasks in loops.
791       */
792      public void reinitialize() {
793 <        if ((status & COMPLETION_MASK) == EXCEPTIONAL)
793 >        if (status == EXCEPTIONAL)
794              exceptionMap.remove(this);
795          status = 0;
796      }
# Line 1229 | Line 1100 | public abstract class ForkJoinTask<V> im
1100      private void readObject(java.io.ObjectInputStream s)
1101          throws java.io.IOException, ClassNotFoundException {
1102          s.defaultReadObject();
1232        status &= ~INTERNAL_SIGNAL_MASK; // clear internal signal counts
1233        status |= EXTERNAL_SIGNAL; // conservatively set external signal
1103          Object ex = s.readObject();
1104          if (ex != null)
1105 <            setDoneExceptionally((Throwable) ex);
1105 >            setExceptionalCompletion((Throwable) ex);
1106 >        if (status < 0)
1107 >            synchronized (this) { notifyAll(); }
1108      }
1109  
1110      // Unsafe mechanics

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines