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.45 by jsr166, Thu Oct 22 08:19:44 2009 UTC vs.
Revision 1.46 by dl, Mon Apr 5 15:52:26 2010 UTC

# Line 139 | Line 139 | import java.util.WeakHashMap;
139   */
140   public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
141  
142 +    /*
143 +     * See the internal documentation of class ForkJoinPool for a
144 +     * general implementation overview.  ForkJoinTasks are mainly
145 +     * responsible for maintaining their "status" field amidst relays
146 +     * to methods in ForkJoinWorkerThread and ForkJoinPool. The
147 +     * methods of this class are more-or-less layered into (1) basic
148 +     * status maintenance (2) execution and awaiting completion (3)
149 +     * user-level methods that additionally report results. This is
150 +     * sometimes hard to see because this file orders exported methods
151 +     * in a way that flows well in javadocs.
152 +     */
153 +
154      /**
155       * Run control status bits packed into a single int to minimize
156       * footprint and to ensure atomicity (via CAS).  Status is
# Line 148 | Line 160 | public abstract class ForkJoinTask<V> im
160       * blocking waits by other threads have SIGNAL_MASK bits set --
161       * bit 15 for external (nonFJ) waits, and the rest a count of
162       * waiting FJ threads.  (This representation relies on
163 <     * ForkJoinPool max thread limits). Completion of a stolen task
164 <     * with SIGNAL_MASK bits set awakens waiter via notifyAll. Even
165 <     * though suboptimal for some purposes, we use basic builtin
166 <     * wait/notify to take advantage of "monitor inflation" in JVMs
167 <     * that we would otherwise need to emulate to avoid adding further
168 <     * per-task bookkeeping overhead. Note that bits 16-28 are
169 <     * currently unused. Also value 0x80000000 is available as spare
170 <     * completion value.
163 >     * ForkJoinPool max thread limits). Signal counts are not directly
164 >     * incremented by ForkJoinTask methods, but instead via a call to
165 >     * requestSignal within ForkJoinPool.preJoin, once their need is
166 >     * established.
167 >     *
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.
179       */
180      volatile int status; // accessed directly by pool and workers
181  
182 <    static final int COMPLETION_MASK      = 0xe0000000;
183 <    static final int NORMAL               = 0xe0000000; // == mask
184 <    static final int CANCELLED            = 0xc0000000;
185 <    static final int EXCEPTIONAL          = 0xa0000000;
186 <    static final int SIGNAL_MASK          = 0x0000ffff;
187 <    static final int INTERNAL_SIGNAL_MASK = 0x00007fff;
188 <    static final int EXTERNAL_SIGNAL      = 0x00008000; // top bit of low word
182 >    private static final int COMPLETION_MASK      = 0xe0000000;
183 >    private static final int NORMAL               = 0xe0000000; // == mask
184 >    private static final int CANCELLED            = 0xc0000000;
185 >    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;
189  
190      /**
191       * Table of exceptions thrown by tasks, to enable reporting by
# Line 179 | Line 199 | public abstract class ForkJoinTask<V> im
199          Collections.synchronizedMap
200          (new WeakHashMap<ForkJoinTask<?>, Throwable>());
201  
202 <    // within-package utilities
202 >    // Maintaining completion status
203  
204      /**
205 <     * Gets current worker thread, or null if not a worker thread.
205 >     * Marks completion and wakes up threads waiting to join this task,
206 >     * also clearing signal request bits.
207 >     *
208 >     * @param completion one of NORMAL, CANCELLED, EXCEPTIONAL
209       */
210 <    static ForkJoinWorkerThread getWorker() {
211 <        Thread t = Thread.currentThread();
212 <        return ((t instanceof ForkJoinWorkerThread) ?
213 <                (ForkJoinWorkerThread) t : null);
214 <    }
215 <
216 <    final boolean casStatus(int cmp, int val) {
217 <        return UNSAFE.compareAndSwapInt(this, statusOffset, cmp, val);
210 >    private void setCompletion(int completion) {
211 >        int s;
212 >        while ((s = status) >= 0) {
213 >            if (UNSAFE.compareAndSwapInt(this, statusOffset, s, completion)) {
214 >                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);
219 >                    synchronized (this) { notifyAll(); }
220 >                }
221 >                return;
222 >            }
223 >        }
224      }
225  
226      /**
227 <     * Workaround for not being able to rethrow unchecked exceptions.
227 >     * Record exception and set exceptional completion
228       */
229 <    static void rethrowException(Throwable ex) {
230 <        if (ex != null)
231 <            UNSAFE.throwException(ex);
229 >    private void setDoneExceptionally(Throwable rex) {
230 >        exceptionMap.put(this, rex);
231 >        setCompletion(EXCEPTIONAL);
232      }
233  
205    // Setting completion status
206
234      /**
235 <     * Marks completion and wakes up threads waiting to join this task.
235 >     * Main internal execution method: Unless done, calls exec and
236 >     * records completion.
237       *
238 <     * @param completion one of NORMAL, CANCELLED, EXCEPTIONAL
238 >     * @return true if ran and completed normally
239       */
240 <    final void setCompletion(int completion) {
241 <        ForkJoinPool pool = getPool();
242 <        if (pool != null) {
243 <            int s; // Clear signal bits while setting completion status
244 <            do {} while ((s = status) >= 0 && !casStatus(s, completion));
245 <
246 <            if ((s & SIGNAL_MASK) != 0) {
219 <                if ((s &= INTERNAL_SIGNAL_MASK) != 0)
220 <                    pool.updateRunningCount(s);
221 <                synchronized (this) { notifyAll(); }
222 <            }
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 <        else
249 <            externallySetCompletion(completion);
248 >        setCompletion(NORMAL); // must be outside try block
249 >        return true;
250      }
251  
252      /**
253 <     * Version of setCompletion for non-FJ threads.  Leaves signal
254 <     * bits for unblocked threads to adjust, and always notifies.
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 <    private void externallySetCompletion(int completion) {
259 >    final int requestSignal() {
260          int s;
261 <        do {} while ((s = status) >= 0 &&
262 <                     !casStatus(s, (s & SIGNAL_MASK) | completion));
263 <        synchronized (this) { notifyAll(); }
261 >        do {} while ((s = status) >= 0 &&
262 >                     !UNSAFE.compareAndSwapInt(this, statusOffset, s, s + 1));
263 >        return s;
264      }
265 <
265 >    
266      /**
267 <     * Sets status to indicate normal completion.
267 >     * Sets external signal request unless already done.
268 >     *
269 >     * @return status
270       */
271 <    final void setNormalCompletion() {
272 <        // Try typical fast case -- single CAS, no signal, not already done.
273 <        // Manually expand casStatus to improve chances of inlining it
274 <        if (!UNSAFE.compareAndSwapInt(this, statusOffset, 0, NORMAL))
275 <            setCompletion(NORMAL);
271 >    private int requestExternalSignal() {
272 >        int s;
273 >        do {} while ((s = status) >= 0 &&
274 >                     !UNSAFE.compareAndSwapInt(this, statusOffset,
275 >                                               s, s | EXTERNAL_SIGNAL));
276 >        return s;
277      }
278  
279 <    // internal waiting and notification
280 <
281 <    /**
282 <     * Performs the actual monitor wait for awaitDone.
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       */
254    private void doAwaitDone() {
255        // Minimize lock bias and in/de-flation effects by maximizing
256        // chances of waiting inside sync
257        try {
258            while (status >= 0)
259                synchronized (this) { if (status >= 0) wait(); }
260        } catch (InterruptedException ie) {
261            onInterruptedWait();
262        }
263    }
284  
285      /**
286 <     * Performs the actual timed monitor wait for awaitDone.
286 >     * Blocks a worker until this task is done, also maintaining pool
287 >     * and signal counts
288       */
289 <    private void doAwaitDone(long startTime, long nanos) {
290 <        synchronized (this) {
291 <            try {
292 <                while (status >= 0) {
293 <                    long nt = nanos - (System.nanoTime() - startTime);
294 <                    if (nt <= 0)
295 <                        break;
296 <                    wait(nt / 1000000, (int) (nt % 1000000));
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)
296 >                            wait();
297 >                        else { // help release; also helps avoid lock-biasing
298 >                            notifyAll();
299 >                            break;
300 >                        }
301 >                    }
302 >                } catch (InterruptedException ie) {
303 >                    cancelIfTerminating();
304                  }
277            } catch (InterruptedException ie) {
278                onInterruptedWait();
305              }
306          }
307      }
308  
283    // Awaiting completion
284
309      /**
310 <     * Sets status to indicate there is joiner, then waits for join,
287 <     * surrounded with pool notifications.
288 <     *
289 <     * @return status upon exit
310 >     * Blocks a non-ForkJoin thread until this task is done.
311       */
312 <    private int awaitDone(ForkJoinWorkerThread w,
313 <                          boolean maintainParallelism) {
314 <        ForkJoinPool pool = (w == null) ? null : w.pool;
315 <        int s;
316 <        while ((s = status) >= 0) {
317 <            if (casStatus(s, (pool == null) ? s|EXTERNAL_SIGNAL : s+1)) {
318 <                if (pool == null || !pool.preJoin(this, maintainParallelism))
319 <                    doAwaitDone();
320 <                if (((s = status) & INTERNAL_SIGNAL_MASK) != 0)
321 <                    adjustPoolCountsOnUnblock(pool);
322 <                break;
312 >    private void externalAwaitDone() {
313 >        if (requestExternalSignal() >= 0) {
314 >            boolean interrupted = false;
315 >            while (status >= 0) {
316 >                try {
317 >                    synchronized(this) {
318 >                        if (status >= 0)
319 >                            wait();
320 >                        else {
321 >                            notifyAll();
322 >                            break;
323 >                        }
324 >                    }
325 >                } catch (InterruptedException ie) {
326 >                    interrupted = true;
327 >                }
328              }
329 +            if (interrupted)
330 +                Thread.currentThread().interrupt();
331          }
304        return s;
332      }
333  
334      /**
335 <     * Timed version of awaitDone
309 <     *
310 <     * @return status upon exit
335 >     * Blocks a worker until this task is done or timeout elapses
336       */
337 <    private int awaitDone(ForkJoinWorkerThread w, long nanos) {
338 <        ForkJoinPool pool = (w == null) ? null : w.pool;
339 <        int s;
340 <        while ((s = status) >= 0) {
341 <            if (casStatus(s, (pool == null) ? s|EXTERNAL_SIGNAL : s+1)) {
342 <                long startTime = System.nanoTime();
343 <                if (pool == null || !pool.preJoin(this, false))
344 <                    doAwaitDone(startTime, nanos);
345 <                if ((s = status) >= 0) {
346 <                    adjustPoolCountsOnCancelledWait(pool);
347 <                    s = status;
337 >    private void timedAwaitDone(ForkJoinWorkerThread w, long nanos) {
338 >        if (status >= 0) {
339 >            long startTime = System.nanoTime();
340 >            ForkJoinPool pool = w.pool;
341 >            pool.preJoin(this);
342 >            while (status >= 0) {
343 >                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                  }
324                if (s < 0 && (s & INTERNAL_SIGNAL_MASK) != 0)
325                    adjustPoolCountsOnUnblock(pool);
326                break;
365              }
366          }
329        return s;
367      }
368  
369      /**
370 <     * Notifies pool that thread is unblocked. Called by signalled
334 <     * threads when woken by non-FJ threads (which is atypical).
370 >     * Blocks a non-ForkJoin thread until this task is done or timeout elapses
371       */
372 <    private void adjustPoolCountsOnUnblock(ForkJoinPool pool) {
373 <        int s;
374 <        do {} while ((s = status) < 0 && !casStatus(s, s & COMPLETION_MASK));
375 <        if (pool != null && (s &= INTERNAL_SIGNAL_MASK) != 0)
376 <            pool.updateRunningCount(s);
377 <    }
378 <
343 <    /**
344 <     * Notifies pool to adjust counts on cancelled or timed out wait.
345 <     */
346 <    private void adjustPoolCountsOnCancelledWait(ForkJoinPool pool) {
347 <        if (pool != null) {
348 <            int s;
349 <            while ((s = status) >= 0 && (s & INTERNAL_SIGNAL_MASK) != 0) {
350 <                if (casStatus(s, s - 1)) {
351 <                    pool.updateRunningCount(1);
372 >    private void externalTimedAwaitDone(long nanos) {
373 >        if (requestExternalSignal() >= 0) {
374 >            long startTime = System.nanoTime();
375 >            boolean interrupted = false;
376 >            while (status >= 0) {
377 >                long nt = nanos - (System.nanoTime() - startTime);
378 >                if (nt <= 0)
379                      break;
380 +                long ms = nt / 1000000;
381 +                int ns = (int) (nt % 1000000);
382 +                try {
383 +                    synchronized(this) { if (status >= 0) wait(ms, ns); }
384 +                } catch (InterruptedException ie) {
385 +                    interrupted = true;
386                  }
387              }
388 +            if (interrupted)
389 +                Thread.currentThread().interrupt();
390          }
391      }
392  
393 <    /**
359 <     * Handles interruptions during waits.
360 <     */
361 <    private void onInterruptedWait() {
362 <        ForkJoinWorkerThread w = getWorker();
363 <        if (w == null)
364 <            Thread.currentThread().interrupt(); // re-interrupt
365 <        else if (w.isTerminating())
366 <            cancelIgnoringExceptions();
367 <        // else if FJworker, ignore interrupt
368 <    }
369 <
370 <    // Recording and reporting exceptions
371 <
372 <    private void setDoneExceptionally(Throwable rex) {
373 <        exceptionMap.put(this, rex);
374 <        setCompletion(EXCEPTIONAL);
375 <    }
393 >    // reporting results
394  
395      /**
396 <     * Throws the exception associated with status s.
397 <     *
398 <     * @throws the exception
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 void reportException(int s) {
401 <        if ((s &= COMPLETION_MASK) < NORMAL) {
402 <            if (s == CANCELLED)
403 <                throw new CancellationException();
404 <            else
387 <                rethrowException(exceptionMap.get(this));
400 >    private V reportResult() {
401 >        if ((status & COMPLETION_MASK) < NORMAL) {
402 >            Throwable ex = getException();
403 >            if (ex != null)
404 >                UNSAFE.throwException(ex);
405          }
406 +        return getRawResult();
407      }
408  
409      /**
# Line 428 | Line 446 | public abstract class ForkJoinTask<V> im
446              throw new TimeoutException();
447      }
448  
431    // internal execution methods
432
433    /**
434     * Calls exec, recording completion, and rethrowing exception if
435     * encountered. Caller should normally check status before calling.
436     *
437     * @return true if completed normally
438     */
439    private boolean tryExec() {
440        try { // try block must contain only call to exec
441            if (!exec())
442                return false;
443        } catch (Throwable rex) {
444            setDoneExceptionally(rex);
445            rethrowException(rex);
446            return false; // not reached
447        }
448        setNormalCompletion();
449        return true;
450    }
451
452    /**
453     * Main execution method used by worker threads. Invokes
454     * base computation unless already complete.
455     */
456    final void quietlyExec() {
457        if (status >= 0) {
458            try {
459                if (!exec())
460                    return;
461            } catch (Throwable rex) {
462                setDoneExceptionally(rex);
463                return;
464            }
465            setNormalCompletion();
466        }
467    }
468
469    /**
470     * Calls exec(), recording but not rethrowing exception.
471     * Caller should normally check status before calling.
472     *
473     * @return true if completed normally
474     */
475    private boolean tryQuietlyInvoke() {
476        try {
477            if (!exec())
478                return false;
479        } catch (Throwable rex) {
480            setDoneExceptionally(rex);
481            return false;
482        }
483        setNormalCompletion();
484        return true;
485    }
486
487    /**
488     * Cancels, ignoring any exceptions it throws.
489     */
490    final void cancelIgnoringExceptions() {
491        try {
492            cancel(false);
493        } catch (Throwable ignore) {
494        }
495    }
496
497    /**
498     * Main implementation of helpJoin
499     */
500    private int busyJoin(ForkJoinWorkerThread w) {
501        int s;
502        ForkJoinTask<?> t;
503        while ((s = status) >= 0 && (t = w.scanWhileJoining(this)) != null)
504            t.quietlyExec();
505        return (s >= 0) ? awaitDone(w, false) : s; // block if no work
506    }
507
449      // public methods
450  
451      /**
# Line 540 | Line 481 | public abstract class ForkJoinTask<V> im
481       * @return the computed result
482       */
483      public final V join() {
484 <        ForkJoinWorkerThread w = getWorker();
485 <        if (w == null || status < 0 || !w.unpushTask(this) || !tryExec())
545 <            reportException(awaitDone(w, true));
546 <        return getRawResult();
484 >        quietlyJoin();
485 >        return reportResult();
486      }
487  
488      /**
# Line 554 | Line 493 | public abstract class ForkJoinTask<V> im
493       * @return the computed result
494       */
495      public final V invoke() {
496 <        if (status >= 0 && tryExec())
497 <            return getRawResult();
498 <        else
560 <            return join();
496 >        if (!tryExec())
497 >            quietlyJoin();
498 >        return reportResult();
499      }
500  
501      /**
# Line 634 | Line 572 | public abstract class ForkJoinTask<V> im
572              }
573          }
574          if (ex != null)
575 <            rethrowException(ex);
575 >            UNSAFE.throwException(ex);
576      }
577  
578      /**
# Line 696 | Line 634 | public abstract class ForkJoinTask<V> im
634              }
635          }
636          if (ex != null)
637 <            rethrowException(ex);
637 >            UNSAFE.throwException(ex);
638          return tasks;
639      }
640  
# Line 729 | Line 667 | public abstract class ForkJoinTask<V> im
667          return (status & COMPLETION_MASK) == CANCELLED;
668      }
669  
670 +    /**
671 +     * Cancels, ignoring any exceptions it throws. Used during worker
672 +     * and pool shutdown.
673 +     */
674 +    final void cancelIgnoringExceptions() {
675 +        try {
676 +            cancel(false);
677 +        } catch (Throwable ignore) {
678 +        }
679 +    }
680 +
681 +    /**
682 +     * Cancels ignoring exceptions if worker is terminating
683 +     */
684 +    private void cancelIfTerminating() {
685 +        Thread t = Thread.currentThread();
686 +        if ((t instanceof ForkJoinWorkerThread) &&
687 +            ((ForkJoinWorkerThread) t).isTerminating()) {
688 +            try {
689 +                cancel(false);
690 +            } catch (Throwable ignore) {
691 +            }
692 +        }
693 +    }
694 +
695      public final boolean isDone() {
696          return status < 0;
697      }
# Line 810 | Line 773 | public abstract class ForkJoinTask<V> im
773              setDoneExceptionally(rex);
774              return;
775          }
776 <        setNormalCompletion();
776 >        setCompletion(NORMAL);
777      }
778  
779      public final V get() throws InterruptedException, ExecutionException {
780 <        ForkJoinWorkerThread w = getWorker();
818 <        if (w == null || status < 0 || !w.unpushTask(this) || !tryQuietlyInvoke())
819 <            awaitDone(w, true);
780 >        quietlyJoin();
781          return reportFutureResult();
782      }
783 <
783 >    
784      public final V get(long timeout, TimeUnit unit)
785          throws InterruptedException, ExecutionException, TimeoutException {
786          long nanos = unit.toNanos(timeout);
787 <        ForkJoinWorkerThread w = getWorker();
788 <        if (w == null || status < 0 || !w.unpushTask(this) || !tryQuietlyInvoke())
789 <            awaitDone(w, nanos);
787 >        Thread t = Thread.currentThread();
788 >        if (t instanceof ForkJoinWorkerThread) {
789 >            ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
790 >            if (!w.unpushTask(this) || !tryExec())
791 >                timedAwaitDone(w, nanos);
792 >        }
793 >        else
794 >            externalTimedAwaitDone(nanos);
795          return reportTimedFutureResult();
796      }
797  
# Line 847 | Line 813 | public abstract class ForkJoinTask<V> im
813       * @return the computed result
814       */
815      public final V helpJoin() {
816 <        ForkJoinWorkerThread w = (ForkJoinWorkerThread) Thread.currentThread();
817 <        if (status < 0 || !w.unpushTask(this) || !tryExec())
852 <            reportException(busyJoin(w));
853 <        return getRawResult();
816 >        quietlyHelpJoin();
817 >        return reportResult();
818      }
819  
820      /**
# Line 866 | Line 830 | public abstract class ForkJoinTask<V> im
830       * ClassCastException}.
831       */
832      public final void quietlyHelpJoin() {
833 <        if (status >= 0) {
834 <            ForkJoinWorkerThread w =
835 <                (ForkJoinWorkerThread) Thread.currentThread();
836 <            if (!w.unpushTask(this) || !tryQuietlyInvoke())
837 <                busyJoin(w);
833 >        ForkJoinWorkerThread w = (ForkJoinWorkerThread) Thread.currentThread();
834 >        if (!w.unpushTask(this) || !tryExec()) {
835 >            while (status >= 0) {
836 >                ForkJoinTask<?> t = w.scanWhileJoining(this);
837 >                if (t == null) {
838 >                    if (status >= 0)
839 >                        awaitDone(w);
840 >                    break;
841 >                }
842 >                t.tryExec();
843 >            }
844          }
845      }
846  
# Line 881 | Line 851 | public abstract class ForkJoinTask<V> im
851       * known to have aborted.
852       */
853      public final void quietlyJoin() {
854 <        if (status >= 0) {
855 <            ForkJoinWorkerThread w = getWorker();
856 <            if (w == null || !w.unpushTask(this) || !tryQuietlyInvoke())
857 <                awaitDone(w, true);
854 >        Thread t = Thread.currentThread();
855 >        if (t instanceof ForkJoinWorkerThread) {
856 >            ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
857 >            if (!w.unpushTask(this) || !tryExec())
858 >                awaitDone(w);
859          }
860 +        else
861 +            externalAwaitDone();
862      }
863  
864      /**
# Line 896 | Line 869 | public abstract class ForkJoinTask<V> im
869       * known to have aborted.
870       */
871      public final void quietlyInvoke() {
872 <        if (status >= 0 && !tryQuietlyInvoke())
872 >        if (!tryExec())
873              quietlyJoin();
874      }
875  

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines