/* * Written by Doug Lea with assistance from members of JCP JSR-166 * Expert Group and released to the public domain, as explained at * http://creativecommons.org/licenses/publicdomain */ package jsr166y; import java.io.Serializable; import java.util.*; import java.util.concurrent.*; import java.util.concurrent.atomic.*; import sun.misc.Unsafe; import java.lang.reflect.*; /** * Abstract base class for tasks that run within a ForkJoinPool. A * ForkJoinTask is a thread-like entity that is much lighter weight * than a normal thread. Huge numbers of tasks and subtasks may be * hosted by a small number of actual threads in a ForkJoinPool, * at the price of some usage limitations. * *
ForkJoinTasks are forms of Futures supporting a * limited range of use. The "lightness" of ForkJoinTasks is due to a * set of restrictions (that are only partially statically * enforceable) reflecting their intended use as computational tasks * calculating pure functions or operating on purely isolated objects. * The primary coordination mechanisms supported for ForkJoinTasks are * fork, that arranges asynchronous execution, and * join, that doesn't proceed until the task's result has * been computed. (Cancellation is also supported). The computation * defined in the compute method should avoid * synchronized methods or blocks, and should minimize * blocking synchronization apart from joining other tasks or using * synchronizers such as Phasers that are advertised to cooperate with * fork/join scheduling. Tasks should also not perform blocking IO, * and should ideally access variables that are completely independent * of those accessed by other running tasks. Minor breaches of these * restrictions, for example using shared output streams, may be * tolerable in practice, but frequent use may result in poor * performance, and the potential to indefinitely stall if the number * of threads not waiting for external synchronization becomes * exhausted. This usage restriction is in part enforced by not * permitting checked exceptions such as IOExceptions to be * thrown. However, computations may still encounter unchecked * exceptions, that are rethrown to callers attempting join * them. These exceptions may additionally include * RejectedExecutionExceptions stemming from internal resource * exhaustion such as failure to allocate internal task queues. * *
The ForkJoinTask class is not usually directly * subclassed. Instead, you subclass one of the abstract classes that * support different styles of fork/join processing. Normally, a * concrete ForkJoinTask subclass declares fields comprising its * parameters, established in a constructor, and then defines a * compute method that somehow uses the control methods * supplied by this base class. While these methods have * public access, some of them may only be called from within * other ForkJoinTasks. Attempts to invoke them in other contexts * result in exceptions or errors including ClassCastException. The * only way to invoke a "main" driver task is to submit it to a * ForkJoinPool. Once started, this will usually in turn start other * subtasks. * *
Most base support methods are final because their * implementations are intrinsically tied to the underlying * lightweight task scheduling framework, and so cannot be overridden. * Developers creating new basic styles of fork/join processing should * minimally implement protected methods exec, * setRawResult, and getRawResult, while also * introducing an abstract computational method that can be * implemented in its subclasses. To support such extensions, * instances of ForkJoinTasks maintain an atomically updated * short representing user-defined control state. Control * state is guaranteed initially to be zero, and to be negative upon * completion, but may otherwise be used for any other control * purposes, such as maintaining join counts. The {@link * ForkJoinWorkerThread} class supports additional inspection and * tuning methods that can be useful when developing extensions. * *
ForkJoinTasks should perform relatively small amounts of * computations, othewise splitting into smaller tasks. As a very * rough rule of thumb, a task should perform more than 100 and less * than 10000 basic computational steps. If tasks are too big, then * parellelism cannot improve throughput. If too small, then memory * and internal task maintenance overhead may overwhelm processing. * *
ForkJoinTasks are Serializable, which enables them to
* be used in extensions such as remote execution frameworks. However,
* it is in general safe to serialize tasks only before or after, but
* not during execution. Serialization is not relied on during
* execution itself.
*/
public abstract class ForkJoinTask This method may be overridden in subclasses, but if so, must
* still ensure that these minimal properties hold. In particular,
* the cancel method itself must not throw exceptions.
*
* This method is designed to be invoked by other
* tasks. To terminate the current task, you can just return or
* throw an unchecked exception from its computation method, or
* invoke completeExceptionally(someException).
*
* @param mayInterruptIfRunning this value is ignored in the
* default implementation because tasks are not in general
* cancelled via interruption.
*
* @return true if this task is now cancelled
*/
public boolean cancel(boolean mayInterruptIfRunning) {
setCompletion(CANCELLED);
return (status & COMPLETION_MASK) == CANCELLED;
}
/**
* Completes this task abnormally, and if not already aborted or
* cancelled, causes it to throw the given exception upon
* join and related operations. This method may be used
* to induce exceptions in asynchronous tasks, or to force
* completion of tasks that would not otherwise complete. This
* method is overridable, but overridden versions must invoke
* super implementation to maintain guarantees.
* @param ex the exception to throw. If this exception is
* not a RuntimeException or Error, the actual exception thrown
* will be a RuntimeException with cause ex.
*/
public void completeExceptionally(Throwable ex) {
setDoneExceptionally((ex instanceof RuntimeException) ||
(ex instanceof Error)? ex :
new RuntimeException(ex));
}
/**
* Completes this task, and if not already aborted or cancelled,
* returning a null result upon 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.
*
* @param value the result value for this task.
*/
public void complete(V value) {
try {
setRawResult(value);
} catch(Throwable rex) {
setDoneExceptionally(rex);
return;
}
setNormalCompletion();
}
/**
* Resets the internal bookkeeping state of this task, allowing a
* subsequent fork. This method allows repeated reuse of
* this task, but only if reuse occurs when this task has either
* never been forked, or has been forked, then completed and all
* outstanding joins of this task have also completed. Effects
* under any other usage conditions are not guaranteed, and are
* almost surely wrong. This method may be useful when executing
* pre-constructed trees of subtasks in loops.
*/
public void reinitialize() {
if ((status & COMPLETION_MASK) == EXCEPTIONAL)
exceptionMap.remove(this);
status = 0;
}
/**
* Tries to unschedule this task for execution. This method will
* typically succeed if this task is the next task that would be
* executed by the current thread, and will typically fail (return
* false) otherwise. This method may be useful when arranging
* faster local processing of tasks that could have been, but were
* not, stolen.
* @return true if unforked
*/
public boolean tryUnfork() {
return ((ForkJoinWorkerThread)(Thread.currentThread())).unpushTask(this);
}
/**
* Forks both tasks, returning when isDone holds for both
* of them or an exception is encountered. This method may be
* invoked only from within other ForkJoinTask
* computations. Attempts to invoke in other contexts result in
* exceptions or errors including ClassCastException.
* @param t1 one task
* @param t2 the other task
* @throws NullPointerException if t1 or t2 are null
* @throws RuntimeException or Error if either task did so.
*/
public static void invokeAll(ForkJoinTask>t1, ForkJoinTask> t2) {
t2.fork();
t1.invoke();
t2.join();
}
/**
* Forks the given tasks, returning when isDone holds for
* all of them. If any task encounters an exception, others may be
* cancelled. This method may be invoked only from within other
* ForkJoinTask computations. Attempts to invoke in other contexts
* result in exceptions or errors including ClassCastException.
* @param tasks the array of tasks
* @throws NullPointerException if tasks or any element are null.
* @throws RuntimeException or Error if any task did so.
*/
public static void invokeAll(ForkJoinTask>... tasks) {
Throwable ex = null;
int last = tasks.length - 1;
for (int i = last; i >= 0; --i) {
ForkJoinTask> t = tasks[i];
if (t == null) {
if (ex == null)
ex = new NullPointerException();
}
else if (i != 0)
t.fork();
else {
t.quietlyInvoke();
if (ex == null)
ex = t.getException();
}
}
for (int i = 1; i <= last; ++i) {
ForkJoinTask> t = tasks[i];
if (t != null) {
if (ex != null)
t.cancel(false);
else {
t.quietlyJoin();
if (ex == null)
ex = t.getException();
}
}
}
if (ex != null)
rethrowException(ex);
}
/**
* Forks all tasks in the collection, returning when
* isDone holds for all of them. If any task encounters
* an exception, others may be cancelled. This method may be
* invoked only from within other ForkJoinTask
* computations. Attempts to invoke in other contexts result in
* exceptions or errors including ClassCastException.
* @param tasks the collection of tasks
* @throws NullPointerException if tasks or any element are null.
* @throws RuntimeException or Error if any task did so.
*/
public static void invokeAll(Collection extends ForkJoinTask>> tasks) {
if (!(tasks instanceof List)) {
invokeAll(tasks.toArray(new ForkJoinTask[tasks.size()]));
return;
}
List extends ForkJoinTask>> ts =
(List extends ForkJoinTask>>)tasks;
Throwable ex = null;
int last = ts.size() - 1;
for (int i = last; i >= 0; --i) {
ForkJoinTask> t = ts.get(i);
if (t == null) {
if (ex == null)
ex = new NullPointerException();
}
else if (i != 0)
t.fork();
else {
t.quietlyInvoke();
if (ex == null)
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 {
t.quietlyJoin();
if (ex == null)
ex = t.getException();
}
}
}
if (ex != null)
rethrowException(ex);
}
/**
* Possibly executes tasks until the pool hosting the current task
* {@link ForkJoinPool#isQuiescent}. This method may be of use in
* designs in which many tasks are forked, but none are explicitly
* joined, instead executing them until all are processed.
*/
public static void helpQuiesce() {
((ForkJoinWorkerThread)(Thread.currentThread())).
helpQuiescePool();
}
/**
* Returns a estimate of how many more locally queued tasks are
* held by the current worker thread than there are other worker
* threads that might want to steal them. 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.
* @return the surplus number of tasks, which may be negative
*/
public static int surplus() {
return ((ForkJoinWorkerThread)(Thread.currentThread()))
.getEstimatedSurplusTaskCount();
}
// Extension kit
/**
* Returns the result that would be returned by join, or
* null if this task is not known to have been completed. This
* method is designed to aid debugging, as well as to support
* extensions. Its use in any other context is discouraged.
*
* @return the result, or null if not completed.
*/
public abstract V getRawResult();
/**
* Forces the given value to be returned as a result. This method
* is designed to support extensions, and should not in general be
* called otherwise.
*
* @param value the value
*/
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
* asynchronous actions that require explicit invocations of
* complete to become joinable. It may throw exceptions
* to indicate abnormal exit.
* @return true if completed normally
* @throws Error or RuntimeException if encountered during computation
*/
protected abstract boolean exec();
// Serialization support
private static final long serialVersionUID = -7721805057305804111L;
/**
* Save the state to a stream.
*
* @serialData the current run status and the exception thrown
* during execution, or null if none.
* @param s the stream
*/
private void writeObject(java.io.ObjectOutputStream s)
throws java.io.IOException {
s.defaultWriteObject();
s.writeObject(getException());
}
/**
* Reconstitute the instance from a stream.
* @param s the stream
*/
private void readObject(java.io.ObjectInputStream s)
throws java.io.IOException, ClassNotFoundException {
s.defaultReadObject();
// status &= ~INTERNAL_SIGNAL_MASK; // todo: define policy
Object ex = s.readObject();
if (ex != null)
setDoneExceptionally((Throwable)ex);
}
// Temporary Unsafe mechanics for preliminary release
static final Unsafe _unsafe;
static final long statusOffset;
static {
try {
if (ForkJoinTask.class.getClassLoader() != null) {
Field f = Unsafe.class.getDeclaredField("theUnsafe");
f.setAccessible(true);
_unsafe = (Unsafe)f.get(null);
}
else
_unsafe = Unsafe.getUnsafe();
statusOffset = _unsafe.objectFieldOffset
(ForkJoinTask.class.getDeclaredField("status"));
} catch (Exception ex) { throw new Error(ex); }
}
}