--- jsr166/src/jsr166e/ForkJoinTask.java 2012/11/14 17:20:29 1.4
+++ jsr166/src/jsr166e/ForkJoinTask.java 2013/09/20 10:52:13 1.16
@@ -33,7 +33,7 @@ import java.lang.reflect.Constructor;
*
A "main" {@code ForkJoinTask} begins execution when it is
* explicitly submitted to a {@link ForkJoinPool}, or, if not already
* engaged in a ForkJoin computation, commenced in the {@link
- * ForkJoinPool#commonPool} via {@link #fork}, {@link #invoke}, or
+ * ForkJoinPool#commonPool()} via {@link #fork}, {@link #invoke}, or
* related methods. Once started, it will usually in turn start other
* subtasks. As indicated by the name of this class, many programs
* using {@code ForkJoinTask} employ only methods {@link #fork} and
@@ -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) {
@@ -395,11 +411,13 @@ public abstract class ForkJoinTask im
final Throwable ex;
ExceptionNode next;
final long thrower; // use id not ref to avoid weak cycles
+ final int hashCode; // store task hashCode before weak ref disappears
ExceptionNode(ForkJoinTask> task, Throwable ex, ExceptionNode next) {
super(task, exceptionTableRefQueue);
this.ex = ex;
this.next = next;
this.thrower = Thread.currentThread().getId();
+ this.hashCode = System.identityHashCode(task);
}
}
@@ -435,7 +453,7 @@ public abstract class ForkJoinTask im
}
/**
- * Records exception and possibly propagates
+ * Records exception and possibly propagates.
*
* @return status on exit
*/
@@ -468,7 +486,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);
@@ -558,12 +576,15 @@ public abstract class ForkJoinTask im
/**
* Poll stale refs and remove them. Call only while holding lock.
*/
+ /**
+ * Poll stale refs and remove them. Call only while holding lock.
+ */
private static void expungeStaleExceptions() {
for (Object x; (x = exceptionTableRefQueue.poll()) != null;) {
if (x instanceof ExceptionNode) {
- ForkJoinTask> key = ((ExceptionNode)x).get();
+ int hashCode = ((ExceptionNode)x).hashCode;
ExceptionNode[] t = exceptionTable;
- int i = System.identityHashCode(key) & (t.length - 1);
+ int i = hashCode & (t.length - 1);
ExceptionNode e = t[i];
ExceptionNode pred = null;
while (e != null) {
@@ -600,14 +621,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 +632,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
}
/**
@@ -635,7 +651,7 @@ public abstract class ForkJoinTask im
/**
* Arranges to asynchronously execute this task in the pool the
* current task is running in, if applicable, or using the {@link
- * ForkJoinPool#commonPool} if not {@link #inForkJoinPool}. While
+ * ForkJoinPool#commonPool()} if not {@link #inForkJoinPool}. While
* it is not necessarily enforced, it is a usage error to fork a
* task more than once unless it has completed and been
* reinitialized. Subsequent modifications to the state of this
@@ -652,7 +668,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 +787,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 +845,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 +995,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 +1009,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 +1022,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 +1098,7 @@ public abstract class ForkJoinTask im
wt.pool.helpQuiescePool(wt.workQueue);
}
else
- ForkJoinPool.externalHelpQuiescePool();
+ ForkJoinPool.quiesceCommonPool();
}
/**
@@ -1140,7 +1163,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 +1332,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 +1385,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 +1450,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 +1464,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 extends T> callable) {
@@ -1435,6 +1478,8 @@ public abstract class ForkJoinTask im
/**
* Saves this task to a stream (that is, serializes it).
*
+ * @param s the stream
+ * @throws java.io.IOException if an I/O error occurs
* @serialData the current run status and the exception thrown
* during execution, or {@code null} if none
*/
@@ -1446,6 +1491,10 @@ public abstract class ForkJoinTask im
/**
* Reconstitutes this task from a stream (that is, deserializes it).
+ * @param s the stream
+ * @throws ClassNotFoundException if the class of a serialized object
+ * could not be found
+ * @throws java.io.IOException if an I/O error occurs
*/
private void readObject(java.io.ObjectInputStream s)
throws java.io.IOException, ClassNotFoundException {
@@ -1483,21 +1532,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());
}
}
}