--- jsr166/src/jsr166y/ForkJoinTask.java 2012/03/04 19:47:08 1.88 +++ jsr166/src/jsr166y/ForkJoinTask.java 2012/10/28 22:36:01 1.91 @@ -5,6 +5,7 @@ */ package jsr166y; + import java.io.Serializable; import java.util.Collection; import java.util.List; @@ -29,15 +30,18 @@ import java.lang.reflect.Constructor; * subtasks may be hosted by a small number of actual threads in a * ForkJoinPool, at the price of some usage limitations. * - *

A "main" {@code ForkJoinTask} begins execution when submitted - * to a {@link ForkJoinPool}. Once started, it will usually in turn - * start other subtasks. As indicated by the name of this class, - * many programs using {@code ForkJoinTask} employ only methods - * {@link #fork} and {@link #join}, or derivatives such as {@link + *

A "main" {@code ForkJoinTask} begins execution when it is + * explicitly submitted to a {@link ForkJoinPool}, or, if not already + * engaged in a ForkJoin computation, commenced in the {@link + * ForkJoinPool#commonPool} via {@link #fork}, {@link #invoke}, or + * related methods. Once started, it will usually in turn start other + * subtasks. As indicated by the name of this class, many programs + * using {@code ForkJoinTask} employ only methods {@link #fork} and + * {@link #join}, or derivatives such as {@link * #invokeAll(ForkJoinTask...) invokeAll}. However, this class also * provides a number of other methods that can come into play in - * advanced usages, as well as extension mechanics that allow - * support of new forms of fork/join processing. + * advanced usages, as well as extension mechanics that allow support + * of new forms of fork/join processing. * *

A {@code ForkJoinTask} is a lightweight form of {@link Future}. * The efficiency of {@code ForkJoinTask}s stems from a set of @@ -70,9 +74,10 @@ import java.lang.reflect.Constructor; * but doing do requires three further considerations: (1) Completion * of few if any other tasks should be dependent on a task * that blocks on external synchronization or IO. Event-style async - * tasks that are never joined often fall into this category. (2) To - * minimize resource impact, tasks should be small; ideally performing - * only the (possibly) blocking action. (3) Unless the {@link + * tasks that are never joined (for example, those subclassing {@link + * CountedCompleter}) often fall into this category. (2) To minimize + * resource impact, tasks should be small; ideally performing only the + * (possibly) blocking action. (3) Unless the {@link * ForkJoinPool.ManagedBlocker} API is used, or the number of possibly * blocked tasks is known to be less than the pool's {@link * ForkJoinPool#getParallelism} level, the pool cannot guarantee that @@ -115,18 +120,13 @@ import java.lang.reflect.Constructor; *

The ForkJoinTask class is not usually directly subclassed. * Instead, you subclass one of the abstract classes that support a * particular style of fork/join processing, typically {@link - * RecursiveAction} for computations that do not return results, or - * {@link RecursiveTask} for those that do. Normally, a concrete - * ForkJoinTask subclass declares fields comprising its parameters, - * established in a constructor, and then defines a {@code compute} - * method that somehow uses the control methods supplied by this base - * class. While these methods have {@code public} access (to allow - * instances of different task subclasses to call each other's - * methods), some of them may only be called from within other - * ForkJoinTasks (as may be determined using method {@link - * #inForkJoinPool}). Attempts to invoke them in other contexts - * result in exceptions or errors, possibly including - * {@code ClassCastException}. + * RecursiveAction} for most computations that do not return results, + * {@link RecursiveTask} for those that do, and {@link + * CountedCompleter} for those in which completed actions trigger + * other actions. Normally, a concrete ForkJoinTask subclass declares + * fields comprising its parameters, established in a constructor, and + * then defines a {@code compute} method that somehow uses the control + * methods supplied by this base class. * *

Method {@link #join} and its variants are appropriate for use * only when completion dependencies are acyclic; that is, the @@ -137,17 +137,16 @@ import java.lang.reflect.Constructor; * {@link Phaser}, {@link #helpQuiesce}, and {@link #complete}) that * may be of use in constructing custom subclasses for problems that * are not statically structured as DAGs. To support such usages a - * ForkJoinTask may be atomically tagged with a {@code - * short} value using {@link #setForkJoinTaskTag} or {@link + * ForkJoinTask may be atomically tagged with a {@code short} + * value using {@link #setForkJoinTaskTag} or {@link * #compareAndSetForkJoinTaskTag} and checked using {@link - * #getForkJoinTaskTag}. The ForkJoinTask implementation does not - * use these {@code protected} methods or tags for any purpose, but - * they may be of use in the construction of specialized subclasses. - * For example, parallel graph traversals can use the supplied methods - * to avoid revisiting nodes/tasks that have already been processed. - * Also, completion based designs can use them to record that subtasks - * have completed. (Method names for tagging are bulky in part to - * encourage definition of methods that reflect their usage patterns.) + * #getForkJoinTaskTag}. The ForkJoinTask implementation does not use + * these {@code protected} methods or tags for any purpose, but they + * may be of use in the construction of specialized subclasses. For + * example, parallel graph traversals can use the supplied methods to + * avoid revisiting nodes/tasks that have already been processed. + * (Method names for tagging are bulky in part to encourage definition + * of methods that reflect their usage patterns.) * *

Most base support methods are {@code final}, to prevent * overriding of implementations that are intrinsically tied to the @@ -269,14 +268,15 @@ public abstract class ForkJoinTask im } /** - * Tries to set SIGNAL status. Used by ForkJoinPool. Other - * variants are directly incorporated into externalAwaitDone etc. + * Tries to set SIGNAL status unless already completed. Used by + * ForkJoinPool. Other variants are directly incorporated into + * externalAwaitDone etc. * * @return true if successful */ final boolean trySetSignal() { - int s; - return U.compareAndSwapInt(this, STATUS, s = status, s | SIGNAL); + int s = status; + return s >= 0 && U.compareAndSwapInt(this, STATUS, s, s | SIGNAL); } /** @@ -284,9 +284,11 @@ public abstract class ForkJoinTask im * @return status upon completion */ private int externalAwaitDone() { - boolean interrupted = false; int s; - while ((s = status) >= 0) { + boolean interrupted = false; + if ((s = status) >= 0 && ForkJoinPool.tryUnsubmitFromCommonPool(this)) + s = doExec(); + while (s >= 0) { if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { synchronized (this) { if (status >= 0) { @@ -300,6 +302,7 @@ public abstract class ForkJoinTask im notifyAll(); } } + s = status; } if (interrupted) Thread.currentThread().interrupt(); @@ -310,10 +313,12 @@ public abstract class ForkJoinTask im * Blocks a non-worker-thread until completion or interruption. */ private int externalInterruptibleAwaitDone() throws InterruptedException { - int s; if (Thread.interrupted()) throw new InterruptedException(); - while ((s = status) >= 0) { + int s; + if ((s = status) >= 0 && ForkJoinPool.tryUnsubmitFromCommonPool(this)) + s = doExec(); + while (s >= 0) { if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { synchronized (this) { if (status >= 0) @@ -322,6 +327,7 @@ public abstract class ForkJoinTask im notifyAll(); } } + s = status; } return s; } @@ -335,16 +341,12 @@ public abstract class ForkJoinTask im */ private int doJoin() { int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w; - if ((s = status) >= 0) { - if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) { - if (!(w = (wt = (ForkJoinWorkerThread)t).workQueue). - tryUnpush(this) || (s = doExec()) >= 0) - s = wt.pool.awaitJoin(w, this); - } - else - s = externalAwaitDone(); - } - return s; + return (s = status) < 0 ? s : + ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + (w = (wt = (ForkJoinWorkerThread)t).workQueue). + tryUnpush(this) && (s = doExec()) < 0 ? s : + wt.pool.awaitJoin(w, this) : + externalAwaitDone(); } /** @@ -354,14 +356,10 @@ public abstract class ForkJoinTask im */ private int doInvoke() { int s; Thread t; ForkJoinWorkerThread wt; - if ((s = doExec()) >= 0) { - if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) - s = (wt = (ForkJoinWorkerThread)t).pool.awaitJoin(wt.workQueue, - this); - else - s = externalAwaitDone(); - } - return s; + return (s = doExec()) < 0 ? s : + ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + (wt = (ForkJoinWorkerThread)t).pool.awaitJoin(wt.workQueue, this) : + externalAwaitDone(); } // Exception table support @@ -409,30 +407,52 @@ public abstract class ForkJoinTask im } /** - * Records exception and sets exceptional completion. + * Records exception and sets status. * * @return status on exit */ - private int setExceptionalCompletion(Throwable ex) { - int h = System.identityHashCode(this); - final ReentrantLock lock = exceptionTableLock; - lock.lock(); - try { - expungeStaleExceptions(); - ExceptionNode[] t = exceptionTable; - int i = h & (t.length - 1); - for (ExceptionNode e = t[i]; ; e = e.next) { - if (e == null) { - t[i] = new ExceptionNode(this, ex, t[i]); - break; + final int recordExceptionalCompletion(Throwable ex) { + int s; + if ((s = status) >= 0) { + int h = System.identityHashCode(this); + final ReentrantLock lock = exceptionTableLock; + lock.lock(); + try { + expungeStaleExceptions(); + ExceptionNode[] t = exceptionTable; + int i = h & (t.length - 1); + for (ExceptionNode e = t[i]; ; e = e.next) { + if (e == null) { + t[i] = new ExceptionNode(this, ex, t[i]); + break; + } + if (e.get() == this) // already present + break; } - if (e.get() == this) // already present - break; + } finally { + lock.unlock(); } - } finally { - lock.unlock(); + s = setCompletion(EXCEPTIONAL); } - return setCompletion(EXCEPTIONAL); + return s; + } + + /** + * Records exception and possibly propagates + * + * @return status on exit + */ + private int setExceptionalCompletion(Throwable ex) { + int s = recordExceptionalCompletion(ex); + if ((s & DONE_MASK) == EXCEPTIONAL) + internalPropagateException(ex); + return s; + } + + /** + * Hook for exception propagation support for tasks with completers. + */ + void internalPropagateException(Throwable ex) { } /** @@ -514,7 +534,7 @@ public abstract class ForkJoinTask im Throwable ex; if (e == null || (ex = e.ex) == null) return null; - if (e.thrower != Thread.currentThread().getId()) { + if (false && e.thrower != Thread.currentThread().getId()) { Class ec = ex.getClass(); try { Constructor noArgCtor = null; @@ -594,25 +614,26 @@ public abstract class ForkJoinTask im // public methods /** - * Arranges to asynchronously execute this task. While it is not - * necessarily enforced, it is a usage error to fork a task more - * than once unless it has completed and been reinitialized. - * Subsequent modifications to the state of this task or any data - * it operates on are not necessarily consistently observable by - * any thread other than the one executing it unless preceded by a - * call to {@link #join} or related methods, or a call to {@link - * #isDone} returning {@code true}. - * - *

This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. + * Arranges to asynchronously execute this task in the pool the + * current task is running in, if applicable, or using the {@link + * ForkJoinPool#commonPool} if not {@link #inForkJoinPool}. While + * it is not necessarily enforced, it is a usage error to fork a + * task more than once unless it has completed and been + * reinitialized. Subsequent modifications to the state of this + * task or any data it operates on are not necessarily + * consistently observable by any thread other than the one + * executing it unless preceded by a call to {@link #join} or + * related methods, or a call to {@link #isDone} returning {@code + * true}. * * @return {@code this}, to simplify usage */ public final ForkJoinTask fork() { - ((ForkJoinWorkerThread)Thread.currentThread()).workQueue.push(this); + Thread t; + if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) + ((ForkJoinWorkerThread)t).workQueue.push(this); + else + ForkJoinPool.submitToCommonPool(this); return this; } @@ -662,12 +683,6 @@ public abstract class ForkJoinTask im * cancelled, completed normally or exceptionally, or left * unprocessed. * - *

This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @param t1 the first task * @param t2 the second task * @throws NullPointerException if any task is null @@ -693,12 +708,6 @@ public abstract class ForkJoinTask im * 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 - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @param tasks the tasks * @throws NullPointerException if any task is null */ @@ -742,12 +751,6 @@ public abstract class ForkJoinTask im * cancelled, completed normally or exceptionally, or left * unprocessed. * - *

This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @param tasks the collection of tasks * @return the tasks argument, to simplify usage * @throws NullPointerException if tasks or any element are null @@ -904,12 +907,12 @@ public abstract class ForkJoinTask im } /** - * Completes this task. The most recent value established by - * {@link #setRawResult} (or {@code null}) will be returned as the - * result of subsequent invocations of {@code join} and related - * operations. This method may be useful when processing sets of - * tasks when some do not otherwise complete normally. Its use in - * other situations is discouraged. + * Completes this task normally without setting a value. The most + * recent value established by {@link #setRawResult} (or {@code + * null} by default) will be returned as the result of subsequent + * invocations of {@code join} and related operations. + * + * @since 1.8 */ public final void quietlyComplete() { setCompletion(NORMAL); @@ -1043,17 +1046,15 @@ public abstract class ForkJoinTask im * be of use in designs in which many tasks are forked, but none * are explicitly joined, instead executing them until all are * processed. - * - *

This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. */ public static void helpQuiesce() { - ForkJoinWorkerThread wt = - (ForkJoinWorkerThread)Thread.currentThread(); - wt.pool.helpQuiescePool(wt.workQueue); + Thread t; + if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) { + ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t; + wt.pool.helpQuiescePool(wt.workQueue); + } + else + ForkJoinPool.externalHelpQuiescePool(); } /** @@ -1106,23 +1107,19 @@ public abstract class ForkJoinTask im /** * Tries to unschedule this task for execution. This method will - * typically succeed if this task is the most recently forked task - * by the current thread, and has not commenced executing in - * another thread. This method may be useful when arranging - * alternative local processing of tasks that could have been, but - * were not, stolen. - * - *

This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. + * typically (but is not guaranteed to) succeed if this task is + * the most recently forked task by the current thread, and has + * not commenced executing in another thread. This method may be + * useful when arranging alternative local processing of tasks + * that could have been, but were not, stolen. * * @return {@code true} if unforked */ public boolean tryUnfork() { - return ((ForkJoinWorkerThread)Thread.currentThread()) - .workQueue.tryUnpush(this); + Thread t; + return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) : + ForkJoinPool.tryUnsubmitFromCommonPool(this); } /** @@ -1131,35 +1128,26 @@ public abstract class ForkJoinTask im * value may be useful for heuristic decisions about whether to * fork other tasks. * - *

This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @return the number of tasks */ public static int getQueuedTaskCount() { - return ((ForkJoinWorkerThread) Thread.currentThread()) - .workQueue.queueSize(); + Thread t; + return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + ((ForkJoinWorkerThread)t).workQueue.queueSize() : + ForkJoinPool.getEstimatedSubmitterQueueLength(); } /** * Returns an estimate of how many more locally queued tasks are * held by the current worker thread than there are other worker - * threads that might steal them. This value may be useful for + * threads that might steal them, or zero if this thread is not + * operating in a ForkJoinPool. This value may be useful for * heuristic decisions about whether to fork other tasks. In many * usages of ForkJoinTasks, at steady state, each worker should * aim to maintain a small constant surplus (for example, 3) of * tasks, and to process computations locally if this threshold is * exceeded. * - *

This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @return the surplus number of tasks, which may be negative */ public static int getSurplusQueuedTaskCount() { @@ -1206,9 +1194,10 @@ public abstract class ForkJoinTask im * have zero queued tasks, so compensate by a factor of * (#idle/#active) threads. */ - ForkJoinWorkerThread wt = - (ForkJoinWorkerThread)Thread.currentThread(); - return wt.workQueue.queueSize() - wt.pool.idlePerActive(); + Thread t; ForkJoinWorkerThread wt; + return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + (wt = (ForkJoinWorkerThread)t).workQueue.queueSize() - wt.pool.idlePerActive() : + 0; } // Extension methods @@ -1234,81 +1223,75 @@ public abstract class ForkJoinTask im protected abstract void setRawResult(V value); /** - * Immediately performs the base action of this task. This method - * is designed to support extensions, and should not in general be - * called otherwise. The return value controls whether this task - * is considered to be done normally. It may return false in + * Immediately performs the base action of this task and returns + * true if, upon return from this method, this task is guaranteed + * to have completed normally. This method may return false + * otherwise, to indicate that this task is not necessarily + * complete (or is not known to be complete), for example in * asynchronous actions that require explicit invocations of - * {@link #complete} to become joinable. It may also throw an - * (unchecked) exception to indicate abnormal exit. + * completion methods. This method may also throw an (unchecked) + * exception to indicate abnormal exit. This method is designed to + * support extensions, and should not in general be called + * otherwise. * - * @return {@code true} if completed normally + * @return {@code true} if this task is known to have completed normally */ protected abstract boolean exec(); /** * Returns, but does not unschedule or execute, a task queued by * the current thread but not yet executed, if one is immediately - * available. There is no guarantee that this task will actually - * be polled or executed next. Conversely, this method may return - * null even if a task exists but cannot be accessed without - * contention with other threads. This method is designed + * available and the current thread is operating in a + * ForkJoinPool. There is no guarantee that this task will + * actually be polled or executed next. Conversely, this method + * may return null even if a task exists but cannot be accessed + * without contention with other threads. This method is designed * primarily to support extensions, and is unlikely to be useful * otherwise. * - *

This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @return the next task, or {@code null} if none are available */ protected static ForkJoinTask peekNextLocalTask() { - return ((ForkJoinWorkerThread) Thread.currentThread()).workQueue.peek(); + Thread t; + return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + ((ForkJoinWorkerThread)t).workQueue.peek() : + null; } /** * Unschedules and returns, without executing, the next task - * queued by the current thread but not yet executed. This method - * is designed primarily to support extensions, and is unlikely to - * be useful otherwise. - * - *

This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. + * queued by the current thread but not yet executed, if the + * current thread is operating in a ForkJoinPool. This method is + * designed primarily to support extensions, and is unlikely to be + * useful otherwise. * * @return the next task, or {@code null} if none are available */ protected static ForkJoinTask pollNextLocalTask() { - return ((ForkJoinWorkerThread) Thread.currentThread()) - .workQueue.nextLocalTask(); + Thread t; + return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + ((ForkJoinWorkerThread)t).workQueue.nextLocalTask() : + null; } /** - * Unschedules and returns, without executing, the next task + * If the current thread is operating in a ForkJoinPool, + * unschedules and returns, without executing, the next task * queued by the current thread but not yet executed, if one is * available, or if not available, a task that was forked by some * other thread, if available. Availability may be transient, so a - * {@code null} result does not necessarily imply quiescence - * of the pool this task is operating in. This method is designed + * {@code null} result does not necessarily imply quiescence of + * the pool this task is operating in. This method is designed * primarily to support extensions, and is unlikely to be useful * otherwise. * - *

This method may be invoked only from within {@code - * ForkJoinPool} computations (as may be determined using method - * {@link #inForkJoinPool}). Attempts to invoke in other contexts - * result in exceptions or errors, possibly including {@code - * ClassCastException}. - * * @return a task, or {@code null} if none are available */ protected static ForkJoinTask pollTask() { - ForkJoinWorkerThread wt = - (ForkJoinWorkerThread)Thread.currentThread(); - return wt.pool.nextTaskFor(wt.workQueue); + Thread t; ForkJoinWorkerThread wt; + return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? + (wt = (ForkJoinWorkerThread)t).pool.nextTaskFor(wt.workQueue) : + null; } // tag operations