--- jsr166/src/jsr166y/ForkJoinPool.java 2010/09/07 06:32:45 1.74 +++ jsr166/src/jsr166y/ForkJoinPool.java 2010/10/10 11:56:11 1.82 @@ -6,16 +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. @@ -429,10 +435,11 @@ public class ForkJoinPool extends Abstra /** * The wakeup interval (in nanoseconds) for the oldest 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 @@ -515,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. @@ -604,7 +611,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 (;;) { @@ -786,15 +793,13 @@ public class ForkJoinPool extends Abstra (workerCounts & RUNNING_COUNT_MASK) <= 1); long startTime = untimed? 0 : System.nanoTime(); Thread.interrupted(); // clear/ignore interrupt - if (eventCount != ec || w.runState != 0 || - runState >= TERMINATING) // recheck after clear - break; + if (eventCount != ec || w.isTerminating()) + 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); @@ -862,16 +867,23 @@ 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 @@ -960,8 +972,12 @@ 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)) inactivate = active = w.active = false; @@ -1003,6 +1019,10 @@ public class ForkJoinPool extends Abstra int retries = 2 + (parallelism >> 2); // #helpJoins before blocking while (joinMe.status >= 0) { int wc; + if (runState >= TERMINATING) { + joinMe.cancelIgnoringExceptions(); + break; + } worker.helpJoinTask(joinMe); if (joinMe.status < 0) break; @@ -1089,6 +1109,7 @@ public class ForkJoinPool extends Abstra return true; } + /** * Actions on transition to TERMINATING * @@ -1112,7 +1133,7 @@ public class ForkJoinPool extends Abstra if (passes > 0 && !w.isTerminated()) { w.cancelTasks(); LockSupport.unpark(w); - if (passes > 1) { + if (passes > 1 && !w.isInterrupted()) { try { w.interrupt(); } catch (SecurityException ignore) { @@ -1279,17 +1300,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(); } /** @@ -1302,8 +1319,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); } /** @@ -1315,7 +1357,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 @@ -1326,12 +1370,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); } /** @@ -1344,7 +1390,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; } @@ -1354,8 +1402,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; } @@ -1365,8 +1415,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; } @@ -1376,12 +1428,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; } @@ -1441,7 +1495,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. * @@ -1689,6 +1743,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 @@ -1842,11 +1903,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 {