--- jsr166/src/jsr166y/ForkJoinTask.java 2010/09/17 14:24:56 1.62 +++ jsr166/src/jsr166y/ForkJoinTask.java 2010/11/21 19:04:45 1.68 @@ -6,8 +6,6 @@ package jsr166y; -import java.util.concurrent.*; - import java.io.Serializable; import java.util.Collection; import java.util.Collections; @@ -15,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}. @@ -239,13 +247,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(); @@ -262,7 +270,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 { @@ -643,7 +651,15 @@ public abstract class ForkJoinTask im } /** - * @throws CancellationException {@inheritDoc} + * Waits if necessary for the computation to complete, and then + * retrieves its result. + * + * @return the computed result + * @throws CancellationException if the computation was cancelled + * @throws ExecutionException if the computation threw an + * exception + * @throws InterruptedException if the current thread is not a + * member of a ForkJoinPool and was interrupted while waiting */ public final V get() throws InterruptedException, ExecutionException { int s; @@ -654,7 +670,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(); @@ -673,70 +689,58 @@ public abstract class ForkJoinTask im } /** - * @throws CancellationException {@inheritDoc} + * Waits if necessary for at most the given time for the computation + * to complete, and then retrieves its result, if available. + * + * @param timeout the maximum time to wait + * @param unit the time unit of the timeout argument + * @return the computed result + * @throws CancellationException if the computation was cancelled + * @throws ExecutionException if the computation threw an + * exception + * @throws InterruptedException if the current thread is not a + * member of a ForkJoinPool and was interrupted while waiting + * @throws TimeoutException if the wait timed out */ 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; @@ -773,7 +777,7 @@ public abstract class ForkJoinTask im return; } } - w.joinTask(this); + w.joinTask(this, false, 0L); } } else @@ -828,6 +832,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)