--- jsr166/src/jsr166y/ForkJoinPool.java 2010/09/01 20:12:39 1.69
+++ jsr166/src/jsr166y/ForkJoinPool.java 2010/11/23 01:06:00 1.88
@@ -6,17 +6,22 @@
package jsr166y;
-import java.util.concurrent.*;
-
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.Callable;
+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;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;
import java.util.concurrent.locks.ReentrantLock;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.CountDownLatch;
/**
* An {@link ExecutorService} for running {@link ForkJoinTask}s.
@@ -301,7 +306,7 @@ public class ForkJoinPool extends Abstra
* about the same time as another is needlessly being created. We
* counteract this and related slop in part by requiring resumed
* spares to immediately recheck (in preStep) to see whether they
- * they should re-suspend.
+ * should re-suspend.
*
* 6. Killing off unneeded workers. A timeout mechanism is used to
* shed unused workers: The oldest (first) event queue waiter uses
@@ -430,10 +435,11 @@ public class ForkJoinPool extends Abstra
/**
* The wakeup interval (in nanoseconds) for the oldest worker
- * worker waiting for an event invokes tryShutdownUnusedWorker to shrink
- * the number of workers. The exact value does not matter too
- * much, but should be long enough to slowly release resources
- * during long periods without use without disrupting normal use.
+ * waiting for an event to invoke tryShutdownUnusedWorker to
+ * shrink the number of workers. The exact value does not matter
+ * too much. It must be short enough to release resources during
+ * sustained periods of idleness, but not so short that threads
+ * are continually re-created.
*/
private static final long SHRINK_RATE_NANOS =
30L * 1000L * 1000L * 1000L; // 2 per minute
@@ -490,8 +496,8 @@ public class ForkJoinPool extends Abstra
*/
private volatile long eventWaiters;
- private static final int EVENT_COUNT_SHIFT = 32;
- private static final long WAITER_ID_MASK = (1L << 16) - 1L;
+ private static final int EVENT_COUNT_SHIFT = 32;
+ private static final int WAITER_ID_MASK = (1 << 16) - 1;
/**
* A counter for events that may wake up worker threads:
@@ -516,7 +522,7 @@ public class ForkJoinPool extends Abstra
* Lifecycle control. The low word contains the number of workers
* that are (probably) executing tasks. This value is atomically
* incremented before a worker gets a task to run, and decremented
- * when worker has no tasks and cannot find any. Bits 16-18
+ * when a worker has no tasks and cannot find any. Bits 16-18
* contain runLevel value. When all are zero, the pool is
* running. Level transitions are monotonic (running -> shutdown
* -> terminating -> terminated) so each transition adds a bit.
@@ -580,7 +586,7 @@ public class ForkJoinPool extends Abstra
// are usually manually inlined by callers
/**
- * Increments running count part of workerCounts
+ * Increments running count part of workerCounts.
*/
final void incrementRunningCount() {
int c;
@@ -590,7 +596,17 @@ public class ForkJoinPool extends Abstra
}
/**
- * Tries to decrement running count unless already zero
+ * Tries to increment running count part of workerCounts.
+ */
+ final boolean tryIncrementRunningCount() {
+ int c;
+ return UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ c = workerCounts,
+ c + ONE_RUNNING);
+ }
+
+ /**
+ * Tries to decrement running count unless already zero.
*/
final boolean tryDecrementRunningCount() {
int wc = workerCounts;
@@ -605,7 +621,7 @@ public class ForkJoinPool extends Abstra
* (rarely) necessary when other count updates lag.
*
* @param dr -- either zero or ONE_RUNNING
- * @param dt == either zero or ONE_TOTAL
+ * @param dt -- either zero or ONE_TOTAL
*/
private void decrementWorkerCounts(int dr, int dt) {
for (;;) {
@@ -663,10 +679,11 @@ public class ForkJoinPool extends Abstra
for (k = 0; k < n && ws[k] != null; ++k)
;
if (k == n)
- ws = Arrays.copyOf(ws, n << 1);
+ ws = workers = Arrays.copyOf(ws, n << 1);
}
ws[k] = w;
- workers = ws; // volatile array write ensures slot visibility
+ int c = eventCount; // advance event count to ensure visibility
+ UNSAFE.compareAndSwapInt(this, eventCountOffset, c, c+1);
} finally {
lock.unlock();
}
@@ -674,7 +691,7 @@ public class ForkJoinPool extends Abstra
}
/**
- * Nulls out record of worker in workers array
+ * Nulls out record of worker in workers array.
*/
private void forgetWorker(ForkJoinWorkerThread w) {
int idx = w.poolIndex;
@@ -699,7 +716,7 @@ public class ForkJoinPool extends Abstra
*/
final void workerTerminated(ForkJoinWorkerThread w) {
forgetWorker(w);
- decrementWorkerCounts(w.isTrimmed()? 0 : ONE_RUNNING, ONE_TOTAL);
+ decrementWorkerCounts(w.isTrimmed() ? 0 : ONE_RUNNING, ONE_TOTAL);
while (w.stealCount != 0) // collect final count
tryAccumulateStealCount(w);
tryTerminate(false);
@@ -720,7 +737,7 @@ public class ForkJoinPool extends Abstra
int ec = eventCount;
boolean releasedOne = false;
ForkJoinWorkerThread w; int id;
- while ((id = ((int)(h & WAITER_ID_MASK)) - 1) >= 0 &&
+ while ((id = (((int)h) & WAITER_ID_MASK) - 1) >= 0 &&
(int)(h >>> EVENT_COUNT_SHIFT) != ec &&
id < n && (w = ws[id]) != null) {
if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
@@ -758,7 +775,7 @@ public class ForkJoinPool extends Abstra
long nh = (((long)ec) << EVENT_COUNT_SHIFT) | ((long)(w.poolIndex+1));
long h;
while ((runState < SHUTDOWN || !tryTerminate(false)) &&
- (((int)((h = eventWaiters) & WAITER_ID_MASK)) == 0 ||
+ (((int)(h = eventWaiters) & WAITER_ID_MASK) == 0 ||
(int)(h >>> EVENT_COUNT_SHIFT) == ec) &&
eventCount == ec) {
if (UNSAFE.compareAndSwapLong(this, eventWaitersOffset,
@@ -785,17 +802,15 @@ public class ForkJoinPool extends Abstra
if (tryAccumulateStealCount(w)) { // transfer while idle
boolean untimed = (w.nextWaiter != 0L ||
(workerCounts & RUNNING_COUNT_MASK) <= 1);
- long startTime = untimed? 0 : System.nanoTime();
+ long startTime = untimed ? 0 : System.nanoTime();
Thread.interrupted(); // clear/ignore interrupt
- if (eventCount != ec || w.runState != 0 ||
- runState >= TERMINATING) // recheck after clear
- break;
+ if (w.isTerminating() || eventCount != ec)
+ break; // recheck after clear
if (untimed)
LockSupport.park(w);
else {
LockSupport.parkNanos(w, SHRINK_RATE_NANOS);
- if (eventCount != ec || w.runState != 0 ||
- runState >= TERMINATING)
+ if (eventCount != ec || w.isTerminating())
break;
if (System.nanoTime() - startTime >= SHRINK_RATE_NANOS)
tryShutdownUnusedWorker(ec);
@@ -807,7 +822,7 @@ public class ForkJoinPool extends Abstra
// Maintaining parallelism
/**
- * Pushes worker onto the spare stack
+ * Pushes worker onto the spare stack.
*/
final void pushSpare(ForkJoinWorkerThread w) {
int ns = (++w.spareCount << SPARE_COUNT_SHIFT) | (w.poolIndex + 1);
@@ -827,7 +842,8 @@ public class ForkJoinPool extends Abstra
if ((sw = spareWaiters) != 0 &&
(id = (sw & SPARE_ID_MASK) - 1) >= 0 &&
id < n && (w = ws[id]) != null &&
- (workerCounts & RUNNING_COUNT_MASK) < parallelism &&
+ (runState >= TERMINATING ||
+ (workerCounts & RUNNING_COUNT_MASK) < parallelism) &&
spareWaiters == sw &&
UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
sw, w.nextSpare)) {
@@ -863,23 +879,26 @@ public class ForkJoinPool extends Abstra
UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
wc + (ONE_RUNNING|ONE_TOTAL))) {
ForkJoinWorkerThread w = null;
+ Throwable fail = null;
try {
w = factory.newThread(this);
- } finally { // adjust on null or exceptional factory return
- if (w == null) {
- decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
- tryTerminate(false); // handle failure during shutdown
- }
+ } catch (Throwable ex) {
+ fail = ex;
}
- if (w == null)
+ if (w == null) { // null or exceptional factory return
+ decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
+ tryTerminate(false); // handle failure during shutdown
+ // If originating from an external caller,
+ // propagate exception, else ignore
+ if (fail != null && runState < TERMINATING &&
+ !(Thread.currentThread() instanceof
+ ForkJoinWorkerThread))
+ UNSAFE.throwException(fail);
break;
+ }
w.start(recordWorker(w), ueh);
- if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc) {
- int c; // advance event count
- UNSAFE.compareAndSwapInt(this, eventCountOffset,
- c = eventCount, c+1);
+ if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc)
break; // add at most one unless total below target
- }
}
}
if (eventWaiters != 0L)
@@ -915,7 +934,7 @@ public class ForkJoinPool extends Abstra
}
else if ((h = eventWaiters) != 0L) {
long nh;
- int id = ((int)(h & WAITER_ID_MASK)) - 1;
+ int id = (((int)h) & WAITER_ID_MASK) - 1;
if (id >= 0 && id < n && (w = ws[id]) != null &&
(nh = w.nextWaiter) != 0L && // keep at least one worker
UNSAFE.compareAndSwapLong(this, eventWaitersOffset, h, nh))
@@ -961,22 +980,33 @@ public class ForkJoinPool extends Abstra
boolean active = w.active;
boolean inactivate = false;
int pc = parallelism;
- int rs;
- while (w.runState == 0 && (rs = runState) < TERMINATING) {
+ while (w.runState == 0) {
+ int rs = runState;
+ if (rs >= TERMINATING) { // propagate shutdown
+ w.shutdown();
+ break;
+ }
if ((inactivate || (active && (rs & ACTIVE_COUNT_MASK) >= pc)) &&
- UNSAFE.compareAndSwapInt(this, runStateOffset, rs, rs - 1))
+ UNSAFE.compareAndSwapInt(this, runStateOffset, rs, --rs)) {
inactivate = active = w.active = false;
- int wc = workerCounts;
+ if (rs == SHUTDOWN) { // all inactive and shut down
+ tryTerminate(false);
+ continue;
+ }
+ }
+ int wc = workerCounts; // try to suspend as spare
if ((wc & RUNNING_COUNT_MASK) > pc) {
if (!(inactivate |= active) && // must inactivate to suspend
- workerCounts == wc && // try to suspend as spare
+ workerCounts == wc &&
UNSAFE.compareAndSwapInt(this, workerCountsOffset,
wc, wc - ONE_RUNNING))
w.suspendAsSpare();
}
else if ((wc >>> TOTAL_COUNT_SHIFT) < pc)
helpMaintainParallelism(); // not enough workers
- else if (!ran) {
+ else if (ran)
+ break;
+ else {
long h = eventWaiters;
int ec = eventCount;
if (h != 0L && (int)(h >>> EVENT_COUNT_SHIFT) != ec)
@@ -988,8 +1018,6 @@ public class ForkJoinPool extends Abstra
else if (!(inactivate |= active))
eventSync(w, wec); // must inactivate before sync
}
- else
- break;
}
}
@@ -999,35 +1027,67 @@ public class ForkJoinPool extends Abstra
*
* @param joinMe the task to join
* @param worker the current worker thread
+ * @param timed true if wait should time out
+ * @param nanos timeout value if timed
*/
- final void awaitJoin(ForkJoinTask joinMe, ForkJoinWorkerThread worker) {
+ final void awaitJoin(ForkJoinTask joinMe, ForkJoinWorkerThread worker,
+ boolean timed, long nanos) {
+ long startTime = timed? System.nanoTime() : 0L;
int retries = 2 + (parallelism >> 2); // #helpJoins before blocking
+ boolean running = true; // false when count decremented
while (joinMe.status >= 0) {
- int wc;
- worker.helpJoinTask(joinMe);
+ if (runState >= TERMINATING) {
+ joinMe.cancelIgnoringExceptions();
+ break;
+ }
+ running = worker.helpJoinTask(joinMe, running);
if (joinMe.status < 0)
break;
- else if (retries > 0)
+ if (retries > 0) {
--retries;
- else if (((wc = workerCounts) & RUNNING_COUNT_MASK) != 0 &&
- UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- wc, wc - ONE_RUNNING)) {
- int stat, c; long h;
- while ((stat = joinMe.status) >= 0 &&
- (h = eventWaiters) != 0L && // help release others
- (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
+ continue;
+ }
+ int wc = workerCounts;
+ if ((wc & RUNNING_COUNT_MASK) != 0) {
+ if (running) {
+ if (!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ wc, wc - ONE_RUNNING))
+ continue;
+ running = false;
+ }
+ long h = eventWaiters;
+ if (h != 0L && (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
releaseEventWaiters();
- if (stat >= 0 &&
- ((workerCounts & RUNNING_COUNT_MASK) == 0 ||
- (stat =
- joinMe.internalAwaitDone(JOIN_TIMEOUT_MILLIS)) >= 0))
- helpMaintainParallelism(); // timeout or no running workers
- do {} while (!UNSAFE.compareAndSwapInt
- (this, workerCountsOffset,
- c = workerCounts, c + ONE_RUNNING));
- if (stat < 0)
- break; // else restart
+ if ((workerCounts & RUNNING_COUNT_MASK) != 0) {
+ long ms; int ns;
+ if (!timed) {
+ ms = JOIN_TIMEOUT_MILLIS;
+ ns = 0;
+ }
+ else { // at most JOIN_TIMEOUT_MILLIS per wait
+ long nt = nanos - (System.nanoTime() - startTime);
+ if (nt <= 0L)
+ break;
+ ms = nt / 1000000;
+ if (ms > JOIN_TIMEOUT_MILLIS) {
+ ms = JOIN_TIMEOUT_MILLIS;
+ ns = 0;
+ }
+ else
+ ns = (int) (nt % 1000000);
+ }
+ joinMe.internalAwaitDone(ms, ns);
+ }
+ if (joinMe.status < 0)
+ break;
}
+ helpMaintainParallelism();
+ }
+ if (!running) {
+ int c;
+ do {} while (!UNSAFE.compareAndSwapInt
+ (this, workerCountsOffset,
+ c = workerCounts, c + ONE_RUNNING));
}
}
@@ -1038,9 +1098,10 @@ public class ForkJoinPool extends Abstra
throws InterruptedException {
while (!blocker.isReleasable()) {
int wc = workerCounts;
- if ((wc & RUNNING_COUNT_MASK) != 0 &&
- UNSAFE.compareAndSwapInt(this, workerCountsOffset,
- wc, wc - ONE_RUNNING)) {
+ if ((wc & RUNNING_COUNT_MASK) == 0)
+ helpMaintainParallelism();
+ else if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
+ wc, wc - ONE_RUNNING)) {
try {
while (!blocker.isReleasable()) {
long h = eventWaiters;
@@ -1085,7 +1146,7 @@ public class ForkJoinPool extends Abstra
// Finish now if all threads terminated; else in some subsequent call
if ((workerCounts >>> TOTAL_COUNT_SHIFT) == 0) {
advanceRunLevel(TERMINATED);
- termination.arrive();
+ termination.forceTermination();
}
return true;
}
@@ -1107,16 +1168,13 @@ public class ForkJoinPool extends Abstra
c = eventCount, c+1);
eventWaiters = 0L; // clobber lists
spareWaiters = 0;
- ForkJoinWorkerThread[] ws = workers;
- int n = ws.length;
- for (int i = 0; i < n; ++i) {
- ForkJoinWorkerThread w = ws[i];
+ for (ForkJoinWorkerThread w : workers) {
if (w != null) {
w.shutdown();
if (passes > 0 && !w.isTerminated()) {
w.cancelTasks();
LockSupport.unpark(w);
- if (passes > 1) {
+ if (passes > 1 && !w.isInterrupted()) {
try {
w.interrupt();
} catch (SecurityException ignore) {
@@ -1129,7 +1187,7 @@ public class ForkJoinPool extends Abstra
}
/**
- * Clear out and cancel submissions, ignoring exceptions
+ * Clears out and cancels submissions, ignoring exceptions.
*/
private void cancelSubmissions() {
ForkJoinTask task;
@@ -1144,15 +1202,15 @@ public class ForkJoinPool extends Abstra
// misc support for ForkJoinWorkerThread
/**
- * Returns pool number
+ * Returns pool number.
*/
final int getPoolNumber() {
return poolNumber;
}
/**
- * Tries to accumulates steal count from a worker, clearing
- * the worker's value.
+ * Tries to accumulate steal count from a worker, clearing
+ * the worker's value if successful.
*
* @return true if worker steal count now zero
*/
@@ -1176,7 +1234,10 @@ public class ForkJoinPool extends Abstra
int pc = parallelism; // use parallelism, not rc
int ac = runState; // no mask -- artificially boosts during shutdown
// Use exact results for small values, saturate past 4
- return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
+ return ((pc <= ac) ? 0 :
+ (pc >>> 1 <= ac) ? 1 :
+ (pc >>> 2 <= ac) ? 3 :
+ pc >>> 3);
}
// Public and protected methods
@@ -1226,13 +1287,13 @@ public class ForkJoinPool extends Abstra
* use {@link #defaultForkJoinWorkerThreadFactory}.
* @param handler the handler for internal worker threads that
* terminate due to unrecoverable errors encountered while executing
- * tasks. For default value, use null.
+ * tasks. For default value, use {@code null}.
* @param asyncMode if true,
* establishes local first-in-first-out scheduling mode for forked
* tasks that are never joined. This mode may be more appropriate
* than default locally stack-based mode in applications in which
* worker threads only process event-style asynchronous tasks.
- * For default value, use false.
+ * For default value, use {@code false}.
* @throws IllegalArgumentException if parallelism less than or
* equal to zero, or greater than implementation limit
* @throws NullPointerException if the factory is null
@@ -1280,17 +1341,13 @@ public class ForkJoinPool extends Abstra
// Execution methods
/**
- * Common code for execute, invoke and submit
+ * Submits task and creates, starts, or resumes some workers if necessary
*/
private void doSubmit(ForkJoinTask task) {
- if (task == null)
- throw new NullPointerException();
- if (runState >= SHUTDOWN)
- throw new RejectedExecutionException();
submissionQueue.offer(task);
int c; // try to increment event count -- CAS failure OK
UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
- helpMaintainParallelism(); // create, start, or resume some workers
+ helpMaintainParallelism();
}
/**
@@ -1303,8 +1360,33 @@ public class ForkJoinPool extends Abstra
* scheduled for execution
*/
public T invoke(ForkJoinTask task) {
- doSubmit(task);
- return task.join();
+ if (task == null)
+ throw new NullPointerException();
+ if (runState >= SHUTDOWN)
+ throw new RejectedExecutionException();
+ Thread t = Thread.currentThread();
+ if ((t instanceof ForkJoinWorkerThread) &&
+ ((ForkJoinWorkerThread)t).pool == this)
+ return task.invoke(); // bypass submit if in same pool
+ else {
+ doSubmit(task);
+ return task.join();
+ }
+ }
+
+ /**
+ * Unless terminating, forks task if within an ongoing FJ
+ * computation in the current pool, else submits as external task.
+ */
+ private void forkOrSubmit(ForkJoinTask task) {
+ if (runState >= SHUTDOWN)
+ throw new RejectedExecutionException();
+ Thread t = Thread.currentThread();
+ if ((t instanceof ForkJoinWorkerThread) &&
+ ((ForkJoinWorkerThread)t).pool == this)
+ task.fork();
+ else
+ doSubmit(task);
}
/**
@@ -1316,7 +1398,9 @@ public class ForkJoinPool extends Abstra
* scheduled for execution
*/
public void execute(ForkJoinTask task) {
- doSubmit(task);
+ if (task == null)
+ throw new NullPointerException();
+ forkOrSubmit(task);
}
// AbstractExecutorService methods
@@ -1327,12 +1411,14 @@ public class ForkJoinPool extends Abstra
* scheduled for execution
*/
public void execute(Runnable task) {
+ if (task == null)
+ throw new NullPointerException();
ForkJoinTask job;
if (task instanceof ForkJoinTask) // avoid re-wrap
job = (ForkJoinTask) task;
else
job = ForkJoinTask.adapt(task, null);
- doSubmit(job);
+ forkOrSubmit(job);
}
/**
@@ -1345,7 +1431,9 @@ public class ForkJoinPool extends Abstra
* scheduled for execution
*/
public ForkJoinTask submit(ForkJoinTask task) {
- doSubmit(task);
+ if (task == null)
+ throw new NullPointerException();
+ forkOrSubmit(task);
return task;
}
@@ -1355,8 +1443,10 @@ public class ForkJoinPool extends Abstra
* scheduled for execution
*/
public ForkJoinTask submit(Callable task) {
+ if (task == null)
+ throw new NullPointerException();
ForkJoinTask job = ForkJoinTask.adapt(task);
- doSubmit(job);
+ forkOrSubmit(job);
return job;
}
@@ -1366,8 +1456,10 @@ public class ForkJoinPool extends Abstra
* scheduled for execution
*/
public ForkJoinTask submit(Runnable task, T result) {
+ if (task == null)
+ throw new NullPointerException();
ForkJoinTask job = ForkJoinTask.adapt(task, result);
- doSubmit(job);
+ forkOrSubmit(job);
return job;
}
@@ -1377,12 +1469,14 @@ public class ForkJoinPool extends Abstra
* scheduled for execution
*/
public ForkJoinTask submit(Runnable task) {
+ if (task == null)
+ throw new NullPointerException();
ForkJoinTask job;
if (task instanceof ForkJoinTask) // avoid re-wrap
job = (ForkJoinTask) task;
else
job = ForkJoinTask.adapt(task, null);
- doSubmit(job);
+ forkOrSubmit(job);
return job;
}
@@ -1442,7 +1536,7 @@ public class ForkJoinPool extends Abstra
/**
* Returns the number of worker threads that have started but not
- * yet terminated. This result returned by this method may differ
+ * yet terminated. The result returned by this method may differ
* from {@link #getParallelism} when threads are created to
* maintain parallelism when others are cooperatively blocked.
*
@@ -1527,13 +1621,9 @@ public class ForkJoinPool extends Abstra
*/
public long getQueuedTaskCount() {
long count = 0;
- ForkJoinWorkerThread[] ws = workers;
- int n = ws.length;
- for (int i = 0; i < n; ++i) {
- ForkJoinWorkerThread w = ws[i];
+ for (ForkJoinWorkerThread w : workers)
if (w != null)
count += w.getQueueSize();
- }
return count;
}
@@ -1588,13 +1678,9 @@ public class ForkJoinPool extends Abstra
*/
protected int drainTasksTo(Collection> c) {
int count = submissionQueue.drainTo(c);
- ForkJoinWorkerThread[] ws = workers;
- int n = ws.length;
- for (int i = 0; i < n; ++i) {
- ForkJoinWorkerThread w = ws[i];
+ for (ForkJoinWorkerThread w : workers)
if (w != null)
count += w.drainTasksTo(c);
- }
return count;
}
@@ -1688,8 +1774,11 @@ public class ForkJoinPool extends Abstra
* commenced but not yet completed. This method may be useful for
* debugging. A return of {@code true} reported a sufficient
* period after shutdown may indicate that submitted tasks have
- * ignored or suppressed interruption, causing this executor not
- * to properly terminate.
+ * ignored or suppressed interruption, or are waiting for IO,
+ * causing this executor not to properly terminate. (See the
+ * advisory notes for class {@link ForkJoinTask} stating that
+ * tasks should not normally entail blocking operations. But if
+ * they do, they must abort them on interrupt.)
*
* @return {@code true} if terminating but not yet terminated
*/
@@ -1698,6 +1787,13 @@ public class ForkJoinPool extends Abstra
}
/**
+ * Returns true if terminating or terminated. Used by ForkJoinWorkerThread.
+ */
+ final boolean isAtLeastTerminating() {
+ return runState >= TERMINATING;
+ }
+
+ /**
* Returns {@code true} if this pool has been shut down.
*
* @return {@code true} if this pool has been shut down
@@ -1720,10 +1816,11 @@ public class ForkJoinPool extends Abstra
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
try {
- return termination.awaitAdvanceInterruptibly(0, timeout, unit) > 0;
+ termination.awaitAdvanceInterruptibly(0, timeout, unit);
} catch (TimeoutException ex) {
return false;
}
+ return true;
}
/**
@@ -1851,11 +1948,11 @@ public class ForkJoinPool extends Abstra
private static final long eventCountOffset =
objectFieldOffset("eventCount", ForkJoinPool.class);
private static final long eventWaitersOffset =
- objectFieldOffset("eventWaiters",ForkJoinPool.class);
+ objectFieldOffset("eventWaiters", ForkJoinPool.class);
private static final long stealCountOffset =
- objectFieldOffset("stealCount",ForkJoinPool.class);
+ objectFieldOffset("stealCount", ForkJoinPool.class);
private static final long spareWaitersOffset =
- objectFieldOffset("spareWaiters",ForkJoinPool.class);
+ objectFieldOffset("spareWaiters", ForkJoinPool.class);
private static long objectFieldOffset(String field, Class klazz) {
try {