--- jsr166/src/jsr166e/ForkJoinTask.java 2012/11/19 18:12:28 1.5 +++ jsr166/src/jsr166e/ForkJoinTask.java 2013/07/21 06:32:28 1.14 @@ -55,7 +55,7 @@ import java.lang.reflect.Constructor; * minimize other blocking synchronization apart from joining other * tasks or using synchronizers such as Phasers that are advertised to * cooperate with fork/join scheduling. Subdividable tasks should also - * not perform blocking IO, and should ideally access variables that + * not perform blocking I/O, and should ideally access variables that * are completely independent of those accessed by other running * tasks. These guidelines are loosely enforced by not permitting * checked exceptions such as {@code IOExceptions} to be @@ -73,7 +73,7 @@ import java.lang.reflect.Constructor; *

It is possible to define and use ForkJoinTasks that may block, * 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 + * that blocks on external synchronization or I/O. Event-style async * 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 @@ -134,9 +134,9 @@ import java.lang.reflect.Constructor; * (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 + * {@link java.util.concurrent.Phaser 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 + * 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 * #compareAndSetForkJoinTaskTag} and checked using {@link @@ -285,25 +285,35 @@ public abstract class ForkJoinTask im */ private int externalAwaitDone() { int s; - ForkJoinPool.externalHelpJoin(this); - boolean interrupted = false; - while ((s = status) >= 0) { - if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { - synchronized (this) { - if (status >= 0) { - try { - wait(); - } catch (InterruptedException ie) { - interrupted = true; + ForkJoinPool cp = ForkJoinPool.common; + if ((s = status) >= 0) { + if (cp != null) { + if (this instanceof CountedCompleter) + s = cp.externalHelpComplete((CountedCompleter)this); + else if (cp.tryExternalUnpush(this)) + s = doExec(); + } + if (s >= 0 && (s = status) >= 0) { + boolean interrupted = false; + do { + if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { + synchronized (this) { + if (status >= 0) { + try { + wait(); + } catch (InterruptedException ie) { + interrupted = true; + } + } + else + notifyAll(); } } - else - notifyAll(); - } + } while ((s = status) >= 0); + if (interrupted) + Thread.currentThread().interrupt(); } } - if (interrupted) - Thread.currentThread().interrupt(); return s; } @@ -312,9 +322,15 @@ public abstract class ForkJoinTask im */ private int externalInterruptibleAwaitDone() throws InterruptedException { int s; + ForkJoinPool cp = ForkJoinPool.common; if (Thread.interrupted()) throw new InterruptedException(); - ForkJoinPool.externalHelpJoin(this); + if ((s = status) >= 0 && cp != null) { + if (this instanceof CountedCompleter) + cp.externalHelpComplete((CountedCompleter)this); + else if (cp.tryExternalUnpush(this)) + doExec(); + } while ((s = status) >= 0) { if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) { synchronized (this) { @@ -435,7 +451,7 @@ public abstract class ForkJoinTask im } /** - * Records exception and possibly propagates + * Records exception and possibly propagates. * * @return status on exit */ @@ -468,7 +484,7 @@ public abstract class ForkJoinTask im } /** - * Removes exception node and clears status + * Removes exception node and clears status. */ private void clearExceptionalCompletion() { int h = System.identityHashCode(this); @@ -600,14 +616,9 @@ public abstract class ForkJoinTask im /** * A version of "sneaky throw" to relay exceptions */ - static void rethrow(final Throwable ex) { - if (ex != null) { - if (ex instanceof Error) - throw (Error)ex; - if (ex instanceof RuntimeException) - throw (RuntimeException)ex; - throw uncheckedThrowable(ex, RuntimeException.class); - } + static void rethrow(Throwable ex) { + if (ex != null) + ForkJoinTask.uncheckedThrow(ex); } /** @@ -616,8 +627,8 @@ public abstract class ForkJoinTask im * unchecked exceptions */ @SuppressWarnings("unchecked") static - T uncheckedThrowable(final Throwable t, final Class c) { - return (T)t; // rely on vacuous cast + void uncheckedThrow(Throwable t) throws T { + throw (T)t; // rely on vacuous cast } /** @@ -652,7 +663,7 @@ public abstract class ForkJoinTask im if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ((ForkJoinWorkerThread)t).workQueue.push(this); else - ForkJoinPool.commonPool.externalPush(this); + ForkJoinPool.common.externalPush(this); return this; } @@ -771,6 +782,7 @@ public abstract class ForkJoinTask im * unprocessed. * * @param tasks the collection of tasks + * @param the type of the values returned from the tasks * @return the tasks argument, to simplify usage * @throws NullPointerException if tasks or any element are null */ @@ -828,7 +840,7 @@ public abstract class ForkJoinTask im *

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}. + * invoke {@link #completeExceptionally(Throwable)}. * * @param mayInterruptIfRunning this value has no effect in the * default implementation because interrupts are not used to @@ -978,8 +990,10 @@ public abstract class ForkJoinTask im if (Thread.interrupted()) throw new InterruptedException(); // Messy in part because we measure in nanosecs, but wait in millisecs - int s; long ns, ms; - if ((s = status) >= 0 && (ns = unit.toNanos(timeout)) > 0L) { + int s; long ms; + long ns = unit.toNanos(timeout); + ForkJoinPool cp; + if ((s = status) >= 0 && ns > 0L) { long deadline = System.nanoTime() + ns; ForkJoinPool p = null; ForkJoinPool.WorkQueue w = null; @@ -990,8 +1004,12 @@ public abstract class ForkJoinTask im w = wt.workQueue; p.helpJoinOnce(w, this); // no retries on failure } - else - ForkJoinPool.externalHelpJoin(this); + else if ((cp = ForkJoinPool.common) != null) { + if (this instanceof CountedCompleter) + cp.externalHelpComplete((CountedCompleter)this); + else if (cp.tryExternalUnpush(this)) + doExec(); + } boolean canBlock = false; boolean interrupted = false; try { @@ -999,7 +1017,7 @@ public abstract class ForkJoinTask im if (w != null && w.qlock < 0) cancelIgnoringExceptions(this); else if (!canBlock) { - if (p == null || p.tryCompensate()) + if (p == null || p.tryCompensate(p.ctl)) canBlock = true; } else { @@ -1075,7 +1093,7 @@ public abstract class ForkJoinTask im wt.pool.helpQuiescePool(wt.workQueue); } else - ForkJoinPool.externalHelpQuiescePool(); + ForkJoinPool.quiesceCommonPool(); } /** @@ -1140,7 +1158,7 @@ public abstract class ForkJoinTask im Thread t; return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ? ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) : - ForkJoinPool.tryExternalUnpush(this)); + ForkJoinPool.common.tryExternalUnpush(this)); } /** @@ -1309,7 +1327,7 @@ public abstract class ForkJoinTask im * * @param e the expected tag value * @param tag the new tag value - * @return true if successful; i.e., the current value was + * @return {@code true} if successful; i.e., the current value was * equal to e and is now tag. * @since 1.8 */ @@ -1362,6 +1380,24 @@ public abstract class ForkJoinTask im } /** + * Adaptor for Runnables in which failure forces worker exception + */ + static final class RunnableExecuteAction extends ForkJoinTask { + final Runnable runnable; + RunnableExecuteAction(Runnable runnable) { + if (runnable == null) throw new NullPointerException(); + this.runnable = runnable; + } + public final Void getRawResult() { return null; } + public final void setRawResult(Void v) { } + public final boolean exec() { runnable.run(); return true; } + void internalPropagateException(Throwable ex) { + rethrow(ex); // rethrow outside exec() catches. + } + private static final long serialVersionUID = 5232453952276885070L; + } + + /** * Adaptor for Callables */ static final class AdaptedCallable extends ForkJoinTask @@ -1409,6 +1445,7 @@ public abstract class ForkJoinTask im * * @param runnable the runnable action * @param result the result upon completion + * @param the type of the result * @return the task */ public static ForkJoinTask adapt(Runnable runnable, T result) { @@ -1422,6 +1459,7 @@ public abstract class ForkJoinTask im * encountered into {@code RuntimeException}. * * @param callable the callable action + * @param the type of the callable's result * @return the task */ public static ForkJoinTask adapt(Callable callable) { @@ -1483,21 +1521,23 @@ public abstract class ForkJoinTask im private static sun.misc.Unsafe getUnsafe() { try { return sun.misc.Unsafe.getUnsafe(); - } catch (SecurityException se) { - try { - return java.security.AccessController.doPrivileged - (new java.security - .PrivilegedExceptionAction() { - public sun.misc.Unsafe run() throws Exception { - java.lang.reflect.Field f = sun.misc - .Unsafe.class.getDeclaredField("theUnsafe"); - f.setAccessible(true); - return (sun.misc.Unsafe) f.get(null); - }}); - } catch (java.security.PrivilegedActionException e) { - throw new RuntimeException("Could not initialize intrinsics", - e.getCause()); - } + } catch (SecurityException tryReflectionInstead) {} + try { + return java.security.AccessController.doPrivileged + (new java.security.PrivilegedExceptionAction() { + public sun.misc.Unsafe run() throws Exception { + Class k = sun.misc.Unsafe.class; + for (java.lang.reflect.Field f : k.getDeclaredFields()) { + f.setAccessible(true); + Object x = f.get(null); + if (k.isInstance(x)) + return k.cast(x); + } + throw new NoSuchFieldError("the Unsafe"); + }}); + } catch (java.security.PrivilegedActionException e) { + throw new RuntimeException("Could not initialize intrinsics", + e.getCause()); } } }