--- jsr166/src/jsr166y/ForkJoinTask.java 2010/07/24 20:28:18 1.52 +++ jsr166/src/jsr166y/ForkJoinTask.java 2010/09/07 07:51:13 1.59 @@ -100,7 +100,7 @@ import java.util.WeakHashMap; * ForkJoinTasks (as may be determined using method {@link * #inForkJoinPool}). Attempts to invoke them in other contexts * result in exceptions or errors, possibly including - * ClassCastException. + * {@code ClassCastException}. * *

Most base support methods are {@code final}, to prevent * overriding of implementations that are intrinsically tied to the @@ -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. */ /* @@ -152,7 +153,7 @@ public abstract class ForkJoinTask im * single int to minimize footprint and to ensure atomicity (via * CAS). Status is initially zero, and takes on nonnegative * values until completed, upon which status holds value - * COMPLETED. CANCELLED, or EXCEPTIONAL. Tasks undergoing blocking + * NORMAL, CANCELLED, or EXCEPTIONAL. Tasks undergoing blocking * waits by other threads have the SIGNAL bit set. Completion of * a stolen task with SIGNAL set awakens any waiters via * notifyAll. Even though suboptimal for some purposes, we use @@ -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; - Thread t; - ForkJoinWorkerThread w; - if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) { - if ((stat = status) < 0) - return stat; - 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,28 +330,41 @@ 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) - * exception if the underlying computation did so. + * necessary, and returns its result, or throws an (unchecked) + * {@code RuntimeException} or {@code Error} 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(); } /** * Forks the given tasks, returning when {@code isDone} holds for * each task or an (unchecked) exception is encountered, in which - * case the exception is rethrown. If either task encounters an - * exception, the other one may be, but is not guaranteed to be, - * cancelled. If both tasks throw an exception, then this method - * throws one of them. The individual status of each task may be - * checked using {@link #getException()} and related methods. + * case the exception is rethrown. If more than one task + * encounters an exception, then this method throws any one of + * these exceptions. If any task encounters an exception, the + * other may be cancelled. However, the execution status of + * individual tasks is not guaranteed upon exceptional return. The + * status of each task may be obtained using {@link + * #getException()} and related methods to check if they have been + * cancelled, completed normally or exceptionally, or left + * unprocessed. * *

This method may be invoked only from within {@code * ForkJoinTask} computations (as may be determined using method @@ -407,12 +385,14 @@ public abstract class ForkJoinTask im /** * Forks the given tasks, returning when {@code isDone} holds for * each task or an (unchecked) exception is encountered, in which - * case the exception is rethrown. If any task encounters an - * exception, others may be, but are not guaranteed to be, - * cancelled. If more than one task encounters an exception, then - * this method throws any one of these exceptions. The individual - * status of each task may be checked using {@link #getException()} - * and related methods. + * case the exception is rethrown. If more than one task + * encounters an exception, then this method throws any one of + * these exceptions. If any task encounters an exception, others + * may be cancelled. However, the execution status of individual + * tasks is not guaranteed upon exceptional return. The status of + * each task may be obtained using {@link #getException()} and + * related methods to check if they have been cancelled, completed + * normally or exceptionally, or left unprocessed. * *

This method may be invoked only from within {@code * ForkJoinTask} computations (as may be determined using method @@ -434,16 +414,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) @@ -453,14 +439,15 @@ public abstract class ForkJoinTask im /** * Forks all tasks in the specified collection, returning when * {@code isDone} holds for each task or an (unchecked) exception - * is encountered. If any task encounters an exception, others - * may be, but are not guaranteed to be, cancelled. If more than - * one task encounters an exception, then this method throws any - * one of these exceptions. The individual status of each task - * may be checked using {@link #getException()} and related - * methods. The behavior of this operation is undefined if the - * specified collection is modified while the operation is in - * progress. + * is encountered, in which case the exception is rethrown. If + * more than one task encounters an exception, then this method + * throws any one of these exceptions. If any task encounters an + * exception, others may be cancelled. However, the execution + * status of individual tasks is not guaranteed upon exceptional + * return. The status of each task may be obtained using {@link + * #getException()} and related methods to check if they have been + * cancelled, completed normally or exceptionally, or left + * unprocessed. * *

This method may be invoked only from within {@code * ForkJoinTask} computations (as may be determined using method @@ -490,16 +477,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,7 +525,8 @@ 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; } /** @@ -549,7 +543,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(); @@ -626,13 +621,14 @@ public abstract class ForkJoinTask im /** * Completes this task, and if not already aborted or cancelled, - * returning a {@code null} result upon {@code join} and related - * operations. This method may be used to provide results for - * asynchronous tasks, or to provide alternative handling for - * tasks that would not otherwise complete normally. Its use in - * other situations is discouraged. This method is - * overridable, but overridden versions must invoke {@code super} - * implementation to maintain guarantees. + * returning the given value as the result of subsequent + * invocations of {@code join} and related operations. This method + * may be used to provide results for asynchronous tasks, or to + * provide alternative handling for tasks that would not otherwise + * complete normally. Its use in other situations is + * discouraged. This method is overridable, but overridden + * versions must invoke {@code super} implementation to maintain + * guarantees. * * @param value the result value for this task */ @@ -647,9 +643,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) @@ -667,7 +664,7 @@ 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 @@ -681,8 +678,9 @@ 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) { + if (pool == null && Thread.interrupted()) { interrupted = true; break; } @@ -691,7 +689,6 @@ public abstract class ForkJoinTask im break; 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) { @@ -735,24 +732,55 @@ 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 - * exception. This method may be useful when processing - * collections of tasks when some have been cancelled or otherwise - * known to have aborted. + * necessary, without returning its result or throwing its + * exception. */ 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(); + } } /** @@ -1074,7 +1102,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 @@ -1087,7 +1115,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 */