--- jsr166/src/jsr166y/ForkJoinTask.java 2010/07/23 16:49:11 1.51 +++ jsr166/src/jsr166y/ForkJoinTask.java 2010/08/11 19:44:30 1.54 @@ -144,7 +144,8 @@ public abstract class ForkJoinTask im * status maintenance (2) execution and awaiting completion (3) * user-level methods that additionally report results. This is * sometimes hard to see because this file orders exported methods - * in a way that flows well in javadocs. + * in a way that flows well in javadocs. In particular, most + * join mechanics are in method quietlyJoin, below. */ /* @@ -164,7 +165,7 @@ public abstract class ForkJoinTask im * them. */ - /** Run status of this task */ + /** The run status of this task */ volatile int status; // accessed directly by pool and workers private static final int NORMAL = -1; @@ -191,27 +192,25 @@ public abstract class ForkJoinTask im * also clearing signal request bits. * * @param completion one of NORMAL, CANCELLED, EXCEPTIONAL - * @return status on exit */ - private int setCompletion(int completion) { + private void setCompletion(int completion) { int s; while ((s = status) >= 0) { if (UNSAFE.compareAndSwapInt(this, statusOffset, s, completion)) { if (s != 0) synchronized (this) { notifyAll(); } - return completion; + break; } } - return s; } /** * Record exception and set exceptional completion * @return status on exit */ - private int setExceptionalCompletion(Throwable rex) { + private void setExceptionalCompletion(Throwable rex) { exceptionMap.put(this, rex); - return setCompletion(EXCEPTIONAL); + setCompletion(EXCEPTIONAL); } /** @@ -233,15 +232,14 @@ public abstract class ForkJoinTask im /** * Blocks a non-worker-thread until completion. - * @return status on exit */ - private int externalAwaitDone() { + private void externalAwaitDone() { int s; while ((s = status) >= 0) { synchronized(this) { if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)){ boolean interrupted = false; - while ((s = status) >= 0) { + while (status >= 0) { try { wait(); } catch (InterruptedException ie) { @@ -254,7 +252,6 @@ public abstract class ForkJoinTask im } } } - return s; } /** @@ -262,7 +259,7 @@ public abstract class ForkJoinTask im * doesn't wait for completion otherwise. Primary execution method * for ForkJoinWorkerThread. */ - final void tryExec() { + final void quietlyExec() { try { if (status < 0 || !exec()) return; @@ -273,63 +270,6 @@ public abstract class ForkJoinTask im setCompletion(NORMAL); // must be outside try block } - /** - * If not done and this task is next in worker queue, runs it, - * else waits for it. - * @return status on exit - */ - private int doJoin() { - int stat; - if ((stat = status) < 0) - return stat; - Thread t = Thread.currentThread(); - ForkJoinWorkerThread w; - if (t instanceof ForkJoinWorkerThread) { - if ((w = (ForkJoinWorkerThread) t).unpushTask(this)) { - boolean completed; - try { - completed = exec(); - } catch (Throwable rex) { - return setExceptionalCompletion(rex); - } - if (completed) - return setCompletion(NORMAL); - } - return w.joinTask(this); - } - return externalAwaitDone(); - } - - /** - * Unless done, calls exec and records status if completed, or - * waits for completion otherwise. - * @return status on exit - */ - private int doInvoke() { - int stat; - if ((stat = status) >= 0) { - boolean completed; - try { - completed = exec(); - } catch (Throwable rex) { - return setExceptionalCompletion(rex); - } - stat = completed ? setCompletion(NORMAL) : doJoin(); - } - return stat; - } - - /** - * Returns result or throws exception associated with given status. - * @param s the status - */ - private V reportResult(int s) { - Throwable ex; - if (s < NORMAL && (ex = getException()) != null) - UNSAFE.throwException(ex); - return getRawResult(); - } - // public methods /** @@ -365,7 +305,11 @@ public abstract class ForkJoinTask im * @return the computed result */ public final V join() { - return reportResult(doJoin()); + quietlyJoin(); + Throwable ex; + if (status < NORMAL && (ex = getException()) != null) + UNSAFE.throwException(ex); + return getRawResult(); } /** @@ -376,7 +320,11 @@ public abstract class ForkJoinTask im * @return the computed result */ public final V invoke() { - return reportResult(doInvoke()); + quietlyInvoke(); + Throwable ex; + if (status < NORMAL && (ex = getException()) != null) + UNSAFE.throwException(ex); + return getRawResult(); } /** @@ -434,16 +382,22 @@ public abstract class ForkJoinTask im } else if (i != 0) t.fork(); - else if (t.doInvoke() < NORMAL && ex == null) - ex = t.getException(); + else { + t.quietlyInvoke(); + if (ex == null && t.status < NORMAL) + ex = t.getException(); + } } for (int i = 1; i <= last; ++i) { ForkJoinTask t = tasks[i]; if (t != null) { if (ex != null) t.cancel(false); - else if (t.doJoin() < NORMAL && ex == null) - ex = t.getException(); + else { + t.quietlyJoin(); + if (ex == null && t.status < NORMAL) + ex = t.getException(); + } } } if (ex != null) @@ -490,16 +444,22 @@ public abstract class ForkJoinTask im } else if (i != 0) t.fork(); - else if (t.doInvoke() < NORMAL && ex == null) - ex = t.getException(); + else { + t.quietlyInvoke(); + if (ex == null && t.status < NORMAL) + ex = t.getException(); + } } for (int i = 1; i <= last; ++i) { ForkJoinTask t = ts.get(i); if (t != null) { if (ex != null) t.cancel(false); - else if (t.doJoin() < NORMAL && ex == null) - ex = t.getException(); + else { + t.quietlyJoin(); + if (ex == null && t.status < NORMAL) + ex = t.getException(); + } } } if (ex != null) @@ -532,12 +492,15 @@ public abstract class ForkJoinTask im * @return {@code true} if this task is now cancelled */ public boolean cancel(boolean mayInterruptIfRunning) { - return setCompletion(CANCELLED) == CANCELLED; + setCompletion(CANCELLED); + return status == CANCELLED; } /** - * Cancels, ignoring any exceptions it throws. Used during worker - * and pool shutdown. + * Cancels, ignoring any exceptions thrown by cancel. Used during + * worker and pool shutdown. Cancel is spec'ed not to throw any + * exceptions, but if it does anyway, we have no recourse during + * shutdown, so guard against this case. */ final void cancelIgnoringExceptions() { try { @@ -645,9 +608,10 @@ public abstract class ForkJoinTask im } public final V get() throws InterruptedException, ExecutionException { - int s = doJoin(); + quietlyJoin(); if (Thread.interrupted()) throw new InterruptedException(); + int s = status; if (s < NORMAL) { Throwable ex; if (s == CANCELLED) @@ -665,13 +629,13 @@ public abstract class ForkJoinTask im if (t instanceof ForkJoinWorkerThread) { ForkJoinWorkerThread w = (ForkJoinWorkerThread) t; if (status >= 0 && w.unpushTask(this)) - tryExec(); + quietlyExec(); pool = w.pool; } else pool = null; /* - * Timed wait loop intermixes cases for fj (pool != null) and + * Timed wait loop intermixes cases for FJ (pool != null) and * non FJ threads. For FJ, decrement pool count but don't try * for replacement; increment count on completion. For non-FJ, * deal with interrupts. This is messy, but a little less so @@ -679,6 +643,7 @@ public abstract class ForkJoinTask im */ boolean interrupted = false; boolean dec = false; // true if pool count decremented + long nanos = unit.toNanos(timeout); for (;;) { if (Thread.interrupted() && pool == null) { interrupted = true; @@ -687,10 +652,8 @@ public abstract class ForkJoinTask im int s = status; if (s < 0) break; - if (UNSAFE.compareAndSwapInt(this, statusOffset, - s, s | SIGNAL)) { + if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)) { long startTime = System.nanoTime(); - long nanos = unit.toNanos(timeout); long nt; // wait time while (status >= 0 && (nt = nanos - (System.nanoTime() - startTime)) > 0) { @@ -734,24 +697,57 @@ public abstract class ForkJoinTask im } /** - * Joins this task, without returning its result or throwing an + * Joins this task, without returning its result or throwing its * exception. This method may be useful when processing * collections of tasks when some have been cancelled or otherwise * known to have aborted. */ public final void quietlyJoin() { - doJoin(); + Thread t; + if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) { + ForkJoinWorkerThread w = (ForkJoinWorkerThread) t; + if (status >= 0) { + if (w.unpushTask(this)) { + boolean completed; + try { + completed = exec(); + } catch (Throwable rex) { + setExceptionalCompletion(rex); + return; + } + if (completed) { + setCompletion(NORMAL); + return; + } + } + w.joinTask(this); + } + } + else + externalAwaitDone(); } /** * Commences performing this task and awaits its completion if - * necessary, without returning its result or throwing an + * necessary, without returning its result or throwing its * exception. This method may be useful when processing * collections of tasks when some have been cancelled or otherwise * known to have aborted. */ public final void quietlyInvoke() { - doInvoke(); + if (status >= 0) { + boolean completed; + try { + completed = exec(); + } catch (Throwable rex) { + setExceptionalCompletion(rex); + return; + } + if (completed) + setCompletion(NORMAL); + else + quietlyJoin(); + } } /**