--- jsr166/src/jsr166y/ForkJoinTask.java 2010/07/07 19:52:31 1.49 +++ jsr166/src/jsr166y/ForkJoinTask.java 2010/09/04 11:33:53 1.57 @@ -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,24 +144,28 @@ 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. */ - /** - * Run control status bits packed into a 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 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 basic builtin wait/notify - * to take advantage of "monitor inflation" in JVMs that we would - * otherwise need to emulate to avoid adding further per-task - * bookkeeping overhead. We want these monitors to be "fat", - * i.e., not use biasing or thin-lock techniques, so use some odd - * coding idioms that tend to avoid them. + /* + * The status field holds run control status bits packed into a + * 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 + * 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 + * basic builtin wait/notify to take advantage of "monitor + * inflation" in JVMs that we would otherwise need to emulate to + * avoid adding further per-task bookkeeping overhead. We want + * these monitors to be "fat", i.e., not use biasing or thin-lock + * techniques, so use some odd coding idioms that tend to avoid + * them. */ + + /** The run status of this task */ volatile int status; // accessed directly by pool and workers private static final int NORMAL = -1; @@ -188,75 +192,91 @@ 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 == SIGNAL) + 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 int internalAwaitDone() { - int s; + final void internalAwaitDone() { + int s; // the odd construction reduces lock bias effects while ((s = status) >= 0) { - synchronized(this) { - if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)) { - do { - try { - wait(); - } catch (InterruptedException ie) { - cancelIfTerminating(); - } - } while ((s = status) >= 0); - break; + try { + synchronized(this) { + if (UNSAFE.compareAndSwapInt(this, statusOffset, s,SIGNAL)) + wait(); } + } catch (InterruptedException ie) { + cancelIfTerminating(); } } + } + + /** + * Blocks a worker thread until completed or timed out. Called + * only by pool. + * + * @return status on exit + */ + 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. - * @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; - do { + while (status >= 0) { try { wait(); } catch (InterruptedException ie) { interrupted = true; } - } while ((s = status) >= 0); + } if (interrupted) Thread.currentThread().interrupt(); break; } } } - return s; } /** @@ -264,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; @@ -275,67 +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); - } - w.joinTask(this); - return status; - } - 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); - } - if (completed) - stat = setCompletion(NORMAL); - else - stat = 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 /** @@ -371,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 @@ -413,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 @@ -440,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) @@ -459,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 @@ -496,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) @@ -543,8 +530,10 @@ public abstract class ForkJoinTask im } /** - * 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 { @@ -554,9 +543,10 @@ 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. */ - private void cancelIfTerminating() { + final void cancelIfTerminating() { Thread t = Thread.currentThread(); if ((t instanceof ForkJoinWorkerThread) && ((ForkJoinWorkerThread) t).isTerminating()) { @@ -652,9 +642,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) @@ -672,13 +663,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 @@ -686,6 +677,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; @@ -694,10 +686,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) { @@ -741,24 +731,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(); + } } /** @@ -1080,7 +1103,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 @@ -1093,7 +1116,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 */ @@ -1103,8 +1126,6 @@ public abstract class ForkJoinTask im Object ex = s.readObject(); if (ex != null) setExceptionalCompletion((Throwable) ex); - if (status < 0) - synchronized (this) { notifyAll(); } } // Unsafe mechanics