--- jsr166/src/jsr166y/ForkJoinTask.java 2010/10/24 19:37:26 1.66 +++ jsr166/src/jsr166y/ForkJoinTask.java 2010/11/28 21:21:03 1.73 @@ -110,6 +110,16 @@ import java.util.concurrent.TimeoutExcep * result in exceptions or errors, possibly including * {@code ClassCastException}. * + *

Method {@link #join} and its variants are appropriate for use + * only when completion dependencies are acyclic; that is, the + * parallel computation can be described as a directed acyclic graph + * (DAG). Otherwise, executions may encounter a form of deadlock as + * tasks cyclically wait for each other. However, this framework + * supports other methods and techniques (for example the use of + * {@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. + * *

Most base support methods are {@code final}, to prevent * overriding of implementations that are intrinsically tied to the * underlying lightweight task scheduling framework. Developers @@ -124,9 +134,10 @@ import java.util.concurrent.TimeoutExcep * computation. Large tasks should be split into smaller subtasks, * usually via recursive decomposition. 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 parallelism cannot - * improve throughput. If too small, then memory and internal task - * maintenance overhead may overwhelm processing. + * computational steps, and should avoid indefinite looping. If tasks + * are too big, then parallelism cannot improve throughput. If too + * small, then memory and internal task maintenance overhead may + * overwhelm processing. * *

This class provides {@code adapt} methods for {@link Runnable} * and {@link Callable}, that may be of use when mixing execution of @@ -223,65 +234,83 @@ public abstract class ForkJoinTask im } /** - * 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 - while ((s = status) >= 0) { - 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 nanos) { - int s; - if ((s = status) >= 0) { - try { + final void internalAwaitDone(long millis, int nanos) { + int s = status; + if ((s == 0 && + UNSAFE.compareAndSwapInt(this, statusOffset, 0, SIGNAL)) || + s > 0) { + try { // the odd construction reduces lock bias effects synchronized (this) { - if (UNSAFE.compareAndSwapInt(this, statusOffset, s,SIGNAL)) + if (status > 0) wait(millis, nanos); + else + notifyAll(); } } catch (InterruptedException ie) { cancelIfTerminating(); } - s = status; } - return s; } /** * Blocks a non-worker-thread until completion. */ private void externalAwaitDone() { - int s; - while ((s = status) >= 0) { + if (status >= 0) { + boolean interrupted = false; synchronized (this) { - if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)) { - boolean interrupted = false; - while (status >= 0) { + for (;;) { + int s = status; + if (s == 0) + UNSAFE.compareAndSwapInt(this, statusOffset, + 0, SIGNAL); + else if (s < 0) { + notifyAll(); + break; + } + else { try { wait(); } catch (InterruptedException ie) { interrupted = true; } } - if (interrupted) - Thread.currentThread().interrupt(); - break; + } + } + if (interrupted) + Thread.currentThread().interrupt(); + } + } + + /** + * Blocks a non-worker-thread until completion or interruption or timeout. + */ + private void externalInterruptibleAwaitDone(boolean timed, long nanos) + throws InterruptedException { + if (Thread.interrupted()) + throw new InterruptedException(); + if (status >= 0) { + long startTime = timed ? System.nanoTime() : 0L; + synchronized (this) { + for (;;) { + long nt; + int s = status; + if (s == 0) + UNSAFE.compareAndSwapInt(this, statusOffset, + 0, SIGNAL); + else if (s < 0) { + notifyAll(); + break; + } + else if (!timed) + wait(); + else if ((nt = nanos - (System.nanoTime()-startTime)) > 0L) + wait(nt / 1000000, (int)(nt % 1000000)); + else + break; } } } @@ -316,7 +345,7 @@ public abstract class ForkJoinTask im * #isDone} returning {@code true}. * *

This method may be invoked only from within {@code - * ForkJoinTask} computations (as may be determined using method + * 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}. @@ -330,10 +359,13 @@ public abstract class ForkJoinTask im } /** - * Returns the result of the computation when it {@link #isDone is done}. - * This method differs from {@link #get()} in that + * Returns the result of the computation when it {@link #isDone is + * done}. This method differs from {@link #get()} in that * abnormal completion results in {@code RuntimeException} or - * {@code Error}, not {@code ExecutionException}. + * {@code Error}, not {@code ExecutionException}, and that + * interrupts of the calling thread do not cause the + * method to abruptly return by throwing {@code + * InterruptedException}. * * @return the computed result */ @@ -375,7 +407,7 @@ public abstract class ForkJoinTask im * unprocessed. * *

This method may be invoked only from within {@code - * ForkJoinTask} computations (as may be determined using method + * 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}. @@ -403,7 +435,7 @@ public abstract class ForkJoinTask im * normally or exceptionally, or left unprocessed. * *

This method may be invoked only from within {@code - * ForkJoinTask} computations (as may be determined using method + * 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}. @@ -458,7 +490,7 @@ public abstract class ForkJoinTask im * unprocessed. * *

This method may be invoked only from within {@code - * ForkJoinTask} computations (as may be determined using method + * 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}. @@ -510,25 +542,28 @@ public abstract class ForkJoinTask im /** * Attempts to cancel execution of this task. This attempt will - * fail if the task has already completed, has already been - * cancelled, or could not be cancelled for some other reason. If - * successful, and this task has not started when cancel is - * called, execution of this task is suppressed, {@link - * #isCancelled} will report true, and {@link #join} will result - * in a {@code CancellationException} being thrown. + * fail if the task has already completed or could not be + * cancelled for some other reason. If successful, and this task + * has not started when {@code cancel} is called, execution of + * this task is suppressed. After this method returns + * successfully, unless there is an intervening call to {@link + * #reinitialize}, subsequent calls to {@link #isCancelled}, + * {@link #isDone}, and {@code cancel} will return {@code true} + * and calls to {@link #join} and related methods will result in + * {@code CancellationException}. * *

This method may be overridden in subclasses, but if so, must - * still ensure that these minimal properties hold. In particular, - * the {@code cancel} method itself must not throw exceptions. + * still ensure that these properties hold. In particular, the + * {@code 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 {@link #completeExceptionally}. * - * @param mayInterruptIfRunning this value is ignored in the - * default implementation because tasks are not - * cancelled via interruption + * @param mayInterruptIfRunning this value has no effect in the + * default implementation because interrupts are not used to + * control cancellation. * * @return {@code true} if this task is now cancelled */ @@ -662,23 +697,13 @@ public abstract class ForkJoinTask im * member of a ForkJoinPool and was interrupted while waiting */ public final V get() throws InterruptedException, ExecutionException { - int s; - if (Thread.currentThread() instanceof ForkJoinWorkerThread) { + Thread t = Thread.currentThread(); + if (t instanceof ForkJoinWorkerThread) quietlyJoin(); - s = status; - } - else { - while ((s = status) >= 0) { - synchronized (this) { // interruptible form of awaitDone - if (UNSAFE.compareAndSwapInt(this, statusOffset, - s, SIGNAL)) { - while (status >= 0) - wait(); - } - } - } - } - if (s < NORMAL) { + else + externalInterruptibleAwaitDone(false, 0L); + int s = status; + if (s != NORMAL) { Throwable ex; if (s == CANCELLED) throw new CancellationException(); @@ -705,48 +730,17 @@ public abstract class ForkJoinTask im public final V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { long nanos = unit.toNanos(timeout); - if (status >= 0) { - Thread t = Thread.currentThread(); - if (t instanceof ForkJoinWorkerThread) { - ForkJoinWorkerThread w = (ForkJoinWorkerThread) t; - boolean completed = false; // timed variant of quietlyJoin - if (w.unpushTask(this)) { - try { - completed = exec(); - } catch (Throwable rex) { - setExceptionalCompletion(rex); - } - } - if (completed) - setCompletion(NORMAL); - else if (status >= 0) - w.joinTask(this, true, nanos); - } - else if (Thread.interrupted()) - throw new InterruptedException(); - else { - long startTime = System.nanoTime(); - int s; long nt; - while ((s = status) >= 0 && - (nt = nanos - (System.nanoTime() - startTime)) > 0) { - if (UNSAFE.compareAndSwapInt(this, statusOffset, s, - SIGNAL)) { - long ms = nt / 1000000; - int ns = (int) (nt % 1000000); - synchronized (this) { - if (status >= 0) - wait(ms, ns); // exit on IE throw - } - } - } - } - } - int es = status; - if (es != NORMAL) { + Thread t = Thread.currentThread(); + if (t instanceof ForkJoinWorkerThread) + ((ForkJoinWorkerThread)t).joinTask(this, true, nanos); + else + externalInterruptibleAwaitDone(true, nanos); + int s = status; + if (s != NORMAL) { Throwable ex; - if (es == CANCELLED) + if (s == CANCELLED) throw new CancellationException(); - if (es == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null) + if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null) throw new ExecutionException(ex); throw new TimeoutException(); } @@ -813,7 +807,7 @@ public abstract class ForkJoinTask im * processed. * *

This method may be invoked only from within {@code - * ForkJoinTask} computations (as may be determined using method + * 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}. @@ -832,6 +826,12 @@ public abstract class ForkJoinTask im * under any other usage conditions are not guaranteed. * This method may be useful when executing * pre-constructed trees of subtasks in loops. + * + *

Upon completion of this method, {@code isDone()} reports + * {@code false}, and {@code getException()} reports {@code + * null}. However, the value returned by {@code getRawResult} is + * unaffected. To clear this value, you can invoke {@code + * setRawResult(null)}. */ public void reinitialize() { if (status == EXCEPTIONAL) @@ -853,11 +853,12 @@ public abstract class ForkJoinTask im } /** - * Returns {@code true} if the current thread is executing as a - * ForkJoinPool computation. + * Returns {@code true} if the current thread is a {@link + * ForkJoinWorkerThread} executing as a ForkJoinPool computation. * - * @return {@code true} if the current thread is executing as a - * ForkJoinPool computation, or false otherwise + * @return {@code true} if the current thread is a {@link + * ForkJoinWorkerThread} executing as a ForkJoinPool computation, + * or {@code false} otherwise */ public static boolean inForkJoinPool() { return Thread.currentThread() instanceof ForkJoinWorkerThread; @@ -872,7 +873,7 @@ public abstract class ForkJoinTask im * were not, stolen. * *

This method may be invoked only from within {@code - * ForkJoinTask} computations (as may be determined using method + * 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}. @@ -891,7 +892,7 @@ public abstract class ForkJoinTask im * fork other tasks. * *

This method may be invoked only from within {@code - * ForkJoinTask} computations (as may be determined using method + * 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}. @@ -914,7 +915,7 @@ public abstract class ForkJoinTask im * exceeded. * *

This method may be invoked only from within {@code - * ForkJoinTask} computations (as may be determined using method + * 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}. @@ -972,7 +973,7 @@ public abstract class ForkJoinTask im * otherwise. * *

This method may be invoked only from within {@code - * ForkJoinTask} computations (as may be determined using method + * 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}. @@ -991,7 +992,7 @@ public abstract class ForkJoinTask im * be useful otherwise. * *

This method may be invoked only from within {@code - * ForkJoinTask} computations (as may be determined using method + * 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}. @@ -1014,7 +1015,7 @@ public abstract class ForkJoinTask im * otherwise. * *

This method may be invoked only from within {@code - * ForkJoinTask} computations (as may be determined using method + * 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}.