--- jsr166/src/jsr166y/ForkJoinTask.java 2010/09/18 12:10:21 1.63 +++ jsr166/src/jsr166y/ForkJoinTask.java 2010/11/22 12:24:34 1.69 @@ -6,7 +6,6 @@ package jsr166y; -import java.util.concurrent.*; import java.io.Serializable; import java.util.Collection; import java.util.Collections; @@ -14,6 +13,16 @@ import java.util.List; import java.util.RandomAccess; import java.util.Map; import java.util.WeakHashMap; +import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.RunnableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; /** * Abstract base class for tasks that run within a {@link ForkJoinPool}. @@ -101,6 +110,16 @@ import java.util.WeakHashMap; * 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 @@ -238,13 +257,13 @@ public abstract class ForkJoinTask im * * @return status on exit */ - final int internalAwaitDone(long millis) { + final int internalAwaitDone(long millis, int nanos) { int s; if ((s = status) >= 0) { try { synchronized (this) { if (UNSAFE.compareAndSwapInt(this, statusOffset, s,SIGNAL)) - wait(millis, 0); + wait(millis, nanos); } } catch (InterruptedException ie) { cancelIfTerminating(); @@ -261,7 +280,7 @@ public abstract class ForkJoinTask im int s; while ((s = status) >= 0) { synchronized (this) { - if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)){ + if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)) { boolean interrupted = false; while (status >= 0) { try { @@ -321,10 +340,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 */ @@ -366,7 +388,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}. @@ -394,7 +416,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}. @@ -449,7 +471,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}. @@ -501,25 +523,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 */ @@ -661,7 +686,7 @@ public abstract class ForkJoinTask im else { while ((s = status) >= 0) { synchronized (this) { // interruptible form of awaitDone - if (UNSAFE.compareAndSwapInt(this, statusOffset, + if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)) { while (status >= 0) wait(); @@ -695,66 +720,43 @@ public abstract class ForkJoinTask im */ public final V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { - Thread t = Thread.currentThread(); - ForkJoinPool pool; - if (t instanceof ForkJoinWorkerThread) { - ForkJoinWorkerThread w = (ForkJoinWorkerThread) t; - if (status >= 0 && w.unpushTask(this)) - quietlyExec(); - pool = w.pool; - } - else - pool = null; - /* - * 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 - * than is splitting the FJ and nonFJ cases. - */ - boolean interrupted = false; - boolean dec = false; // true if pool count decremented long nanos = unit.toNanos(timeout); - for (;;) { - if (pool == null && Thread.interrupted()) { - interrupted = true; - break; + 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); } - int s = status; - if (s < 0) - break; - if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)) { + else if (Thread.interrupted()) + throw new InterruptedException(); + else { long startTime = System.nanoTime(); - long nt; // wait time - while (status >= 0 && + int s; long nt; + while ((s = status) >= 0 && (nt = nanos - (System.nanoTime() - startTime)) > 0) { - if (pool != null && !dec) - dec = pool.tryDecrementRunningCount(); - else { + if (UNSAFE.compareAndSwapInt(this, statusOffset, s, + SIGNAL)) { long ms = nt / 1000000; int ns = (int) (nt % 1000000); - try { - synchronized (this) { - if (status >= 0) - wait(ms, ns); - } - } catch (InterruptedException ie) { - if (pool != null) - cancelIfTerminating(); - else { - interrupted = true; - break; - } + synchronized (this) { + if (status >= 0) + wait(ms, ns); // exit on IE throw } } } - break; } } - if (pool != null && dec) - pool.incrementRunningCount(); - if (interrupted) - throw new InterruptedException(); int es = status; if (es != NORMAL) { Throwable ex; @@ -791,7 +793,7 @@ public abstract class ForkJoinTask im return; } } - w.joinTask(this); + w.joinTask(this, false, 0L); } } else @@ -827,7 +829,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}. @@ -846,6 +848,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) @@ -886,7 +894,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}. @@ -905,7 +913,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}. @@ -928,7 +936,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}. @@ -986,7 +994,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}. @@ -1005,7 +1013,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}. @@ -1028,7 +1036,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}.