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
@@ -285,10 +285,9 @@ public abstract class ForkJoinTask im
*/
private int externalAwaitDone() {
int s;
+ ForkJoinPool.externalHelpJoin(this);
boolean interrupted = false;
- if ((s = status) >= 0 && ForkJoinPool.tryUnsubmitFromCommonPool(this))
- s = doExec();
- while (s >= 0) {
+ while ((s = status) >= 0) {
if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
synchronized (this) {
if (status >= 0) {
@@ -302,7 +301,6 @@ public abstract class ForkJoinTask im
notifyAll();
}
}
- s = status;
}
if (interrupted)
Thread.currentThread().interrupt();
@@ -313,12 +311,11 @@ public abstract class ForkJoinTask im
* Blocks a non-worker-thread until completion or interruption.
*/
private int externalInterruptibleAwaitDone() throws InterruptedException {
+ int s;
if (Thread.interrupted())
throw new InterruptedException();
- int s;
- if ((s = status) >= 0 && ForkJoinPool.tryUnsubmitFromCommonPool(this))
- s = doExec();
- while (s >= 0) {
+ ForkJoinPool.externalHelpJoin(this);
+ while ((s = status) >= 0) {
if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
synchronized (this) {
if (status >= 0)
@@ -327,11 +324,11 @@ public abstract class ForkJoinTask im
notifyAll();
}
}
- s = status;
}
return s;
}
+
/**
* Implementation for join, get, quietlyJoin. Directly handles
* only cases of already-completed, external wait, and
@@ -601,14 +598,36 @@ 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);
+ }
+ }
+
+ /**
+ * The sneaky part of sneaky throw, relying on generics
+ * limitations to evade compiler complaints about rethrowing
+ * unchecked exceptions
+ */
+ @SuppressWarnings("unchecked") static
+ T uncheckedThrowable(final Throwable t, final Class c) {
+ return (T)t; // rely on vacuous cast
+ }
+
+ /**
* Throws exception, if any, associated with the given status.
*/
private void reportException(int s) {
- Throwable ex = ((s == CANCELLED) ? new CancellationException() :
- (s == EXCEPTIONAL) ? getThrowableException() :
- null);
- if (ex != null)
- U.throwException(ex);
+ if (s == CANCELLED)
+ throw new CancellationException();
+ if (s == EXCEPTIONAL)
+ rethrow(getThrowableException());
}
// public methods
@@ -616,7 +635,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
@@ -633,7 +652,7 @@ public abstract class ForkJoinTask im
if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
((ForkJoinWorkerThread)t).workQueue.push(this);
else
- ForkJoinPool.submitToCommonPool(this);
+ ForkJoinPool.commonPool.externalPush(this);
return this;
}
@@ -735,7 +754,7 @@ public abstract class ForkJoinTask im
}
}
if (ex != null)
- U.throwException(ex);
+ rethrow(ex);
}
/**
@@ -786,7 +805,7 @@ public abstract class ForkJoinTask im
}
}
if (ex != null)
- U.throwException(ex);
+ rethrow(ex);
return tasks;
}
@@ -959,8 +978,9 @@ 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);
+ if ((s = status) >= 0 && ns > 0L) {
long deadline = System.nanoTime() + ns;
ForkJoinPool p = null;
ForkJoinPool.WorkQueue w = null;
@@ -969,16 +989,18 @@ public abstract class ForkJoinTask im
ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
p = wt.pool;
w = wt.workQueue;
- s = p.helpJoinOnce(w, this); // no retries on failure
+ p.helpJoinOnce(w, this); // no retries on failure
}
+ else
+ ForkJoinPool.externalHelpJoin(this);
boolean canBlock = false;
boolean interrupted = false;
try {
while ((s = status) >= 0) {
- if (w != null && w.runState < 0)
+ if (w != null && w.qlock < 0)
cancelIgnoringExceptions(this);
else if (!canBlock) {
- if (p == null || p.tryCompensate(this, null))
+ if (p == null || p.tryCompensate())
canBlock = true;
}
else {
@@ -1117,9 +1139,9 @@ public abstract class ForkJoinTask im
*/
public boolean tryUnfork() {
Thread t;
- return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
- ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) :
- ForkJoinPool.tryUnsubmitFromCommonPool(this);
+ return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
+ ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) :
+ ForkJoinPool.tryExternalUnpush(this));
}
/**
@@ -1131,10 +1153,12 @@ public abstract class ForkJoinTask im
* @return the number of tasks
*/
public static int getQueuedTaskCount() {
- Thread t;
- return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
- ((ForkJoinWorkerThread)t).workQueue.queueSize() :
- ForkJoinPool.getEstimatedSubmitterQueueLength();
+ Thread t; ForkJoinPool.WorkQueue q;
+ if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
+ q = ((ForkJoinWorkerThread)t).workQueue;
+ else
+ q = ForkJoinPool.commonSubmitterQueue();
+ return (q == null) ? 0 : q.queueSize();
}
/**
@@ -1151,53 +1175,7 @@ public abstract class ForkJoinTask im
* @return the surplus number of tasks, which may be negative
*/
public static int getSurplusQueuedTaskCount() {
- /*
- * The aim of this method is to return a cheap heuristic guide
- * for task partitioning when programmers, frameworks, tools,
- * or languages have little or no idea about task granularity.
- * In essence by offering this method, we ask users only about
- * tradeoffs in overhead vs expected throughput and its
- * variance, rather than how finely to partition tasks.
- *
- * In a steady state strict (tree-structured) computation,
- * each thread makes available for stealing enough tasks for
- * other threads to remain active. Inductively, if all threads
- * play by the same rules, each thread should make available
- * only a constant number of tasks.
- *
- * The minimum useful constant is just 1. But using a value of
- * 1 would require immediate replenishment upon each steal to
- * maintain enough tasks, which is infeasible. Further,
- * partitionings/granularities of offered tasks should
- * minimize steal rates, which in general means that threads
- * nearer the top of computation tree should generate more
- * than those nearer the bottom. In perfect steady state, each
- * thread is at approximately the same level of computation
- * tree. However, producing extra tasks amortizes the
- * uncertainty of progress and diffusion assumptions.
- *
- * So, users will want to use values larger, but not much
- * larger than 1 to both smooth over transient shortages and
- * hedge against uneven progress; as traded off against the
- * cost of extra task overhead. We leave the user to pick a
- * threshold value to compare with the results of this call to
- * guide decisions, but recommend values such as 3.
- *
- * When all threads are active, it is on average OK to
- * estimate surplus strictly locally. In steady-state, if one
- * thread is maintaining say 2 surplus tasks, then so are
- * others. So we can just use estimated queue length.
- * However, this strategy alone leads to serious mis-estimates
- * in some non-steady-state conditions (ramp-up, ramp-down,
- * other stalls). We can detect many of these by further
- * considering the number of "idle" threads, that are known to
- * have zero queued tasks, so compensate by a factor of
- * (#idle/#active) threads.
- */
- Thread t; ForkJoinWorkerThread wt;
- return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
- (wt = (ForkJoinWorkerThread)t).workQueue.queueSize() - wt.pool.idlePerActive() :
- 0;
+ return ForkJoinPool.getSurplusQueuedTaskCount();
}
// Extension methods
@@ -1241,21 +1219,22 @@ public abstract class ForkJoinTask im
/**
* Returns, but does not unschedule or execute, a task queued by
* the current thread but not yet executed, if one is immediately
- * available and the current thread is operating in a
- * ForkJoinPool. There is no guarantee that this task will
- * actually be polled or executed next. Conversely, this method
- * may return null even if a task exists but cannot be accessed
- * without contention with other threads. This method is designed
+ * available. There is no guarantee that this task will actually
+ * be polled or executed next. Conversely, this method may return
+ * null even if a task exists but cannot be accessed without
+ * contention with other threads. This method is designed
* primarily to support extensions, and is unlikely to be useful
* otherwise.
*
* @return the next task, or {@code null} if none are available
*/
protected static ForkJoinTask> peekNextLocalTask() {
- Thread t;
- return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
- ((ForkJoinWorkerThread)t).workQueue.peek() :
- null;
+ Thread t; ForkJoinPool.WorkQueue q;
+ if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
+ q = ((ForkJoinWorkerThread)t).workQueue;
+ else
+ q = ForkJoinPool.commonSubmitterQueue();
+ return (q == null) ? null : q.peek();
}
/**
@@ -1480,14 +1459,16 @@ public abstract class ForkJoinTask im
// Unsafe mechanics
private static final sun.misc.Unsafe U;
private static final long STATUS;
+
static {
exceptionTableLock = new ReentrantLock();
exceptionTableRefQueue = new ReferenceQueue