--- jsr166/src/jsr166y/ForkJoinTask.java 2010/07/23 16:49:11 1.51 +++ jsr166/src/jsr166y/ForkJoinTask.java 2010/09/04 00:21:31 1.56 @@ -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,31 +192,32 @@ 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 + * Records exception and sets 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); } /** - * Blocks a worker thread until completion. Called only by pool. + * Blocks a worker thread until completion. Called only by + * pool. Currently unused -- pool-based waits use timeout + * version below. */ final void internalAwaitDone() { int s; // the odd construction reduces lock bias effects @@ -232,16 +234,37 @@ public abstract class ForkJoinTask im } /** - * Blocks a non-worker-thread until completion. + * Blocks a worker thread until completed or timed out. Called + * only by pool. + * * @return status on exit */ - private int externalAwaitDone() { + final int internalAwaitDone(long millis) { + int s; + if ((s = status) >= 0) { + try { + synchronized(this) { + if (UNSAFE.compareAndSwapInt(this, statusOffset, s,SIGNAL)) + wait(millis, 0); + } + } catch (InterruptedException ie) { + cancelIfTerminating(); + } + s = status; + } + return s; + } + + /** + * Blocks a non-worker-thread until completion. + */ + 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 +277,6 @@ public abstract class ForkJoinTask im } } } - return s; } /** @@ -262,7 +284,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 +295,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,18 +330,26 @@ 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(); } /** * Commences performing this task, awaits its completion if - * necessary, and return its result, or throws an (unchecked) + * necessary, and returns its result, or throws an (unchecked) * exception if the underlying computation did so. * * @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 +407,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 +469,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 +517,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 { @@ -547,7 +535,8 @@ public abstract class ForkJoinTask im } /** - * Cancels ignoring exceptions if worker is terminating + * Cancels if current thread is a terminating worker thread, + * ignoring any exceptions thrown by cancel. */ final void cancelIfTerminating() { Thread t = Thread.currentThread(); @@ -645,9 +634,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 +655,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 +669,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 +678,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 +723,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(); + } } /** @@ -1073,7 +1095,7 @@ public abstract class ForkJoinTask im private static final long serialVersionUID = -7721805057305804111L; /** - * Saves the state to a stream. + * Saves the state to a stream (that is, serializes it). * * @serialData the current run status and the exception thrown * during execution, or {@code null} if none @@ -1086,7 +1108,7 @@ public abstract class ForkJoinTask im } /** - * Reconstitutes the instance from a stream. + * Reconstitutes the instance from a stream (that is, deserializes it). * * @param s the stream */