--- jsr166/src/jsr166y/ForkJoinPool.java 2009/11/16 04:57:09 1.49 +++ jsr166/src/jsr166y/ForkJoinPool.java 2009/12/04 12:09:46 1.50 @@ -526,14 +526,16 @@ public class ForkJoinPool extends Abstra ws = workers; if (ws == null) { int ps = parallelism; + updateWorkerCount(ps); ws = ensureWorkerArrayCapacity(ps); for (int i = 0; i < ps; ++i) { ForkJoinWorkerThread w = createWorker(i); if (w != null) { ws[i] = w; w.start(); - updateWorkerCount(1); } + else + updateWorkerCount(-1); } } } finally { @@ -799,10 +801,12 @@ public class ForkJoinPool extends Abstra if (isProcessingTasks()) { int p = this.parallelism; this.parallelism = parallelism; - if (parallelism > p) - createAndStartAddedWorkers(); - else - trimSpares(); + if (workers != null) { + if (parallelism > p) + createAndStartAddedWorkers(); + else + trimSpares(); + } } } finally { lock.unlock(); @@ -1376,7 +1380,6 @@ public class ForkJoinPool extends Abstra } } - /* * Nodes for event barrier to manage idle threads. Queue nodes * are basic Treiber stack nodes, also used for spare stack. @@ -1400,15 +1403,14 @@ public class ForkJoinPool extends Abstra * handling: Method signalWork returns without advancing count if * the queue appears to be empty. This would ordinarily result in * races causing some queued waiters not to be woken up. To avoid - * this, the first worker enqueued in method sync (see - * syncIsReleasable) rescans for tasks after being enqueued, and - * helps signal if any are found. This works well because the - * worker has nothing better to do, and so might as well help - * alleviate the overhead and contention on the threads actually - * doing work. Also, since event counts increments on task - * availability exist to maintain liveness (rather than to force - * refreshes etc), it is OK for callers to exit early if - * contending with another signaller. + * this, the first worker enqueued in method sync rescans for + * tasks after being enqueued, and helps signal if any are + * found. This works well because the worker has nothing better to + * do, and so might as well help alleviate the overhead and + * contention on the threads actually doing work. Also, since + * event counts increments on task availability exist to maintain + * liveness (rather than to force refreshes etc), it is OK for + * callers to exit early if contending with another signaller. */ static final class WaitQueueNode { WaitQueueNode next; // only written before enqueued @@ -1421,7 +1423,7 @@ public class ForkJoinPool extends Abstra } /** - * Wakes up waiter, returning false if known to already + * Wakes up waiter, returning false if known to already be awake */ boolean signal() { ForkJoinWorkerThread t = thread; @@ -1431,34 +1433,14 @@ public class ForkJoinPool extends Abstra LockSupport.unpark(t); return true; } - - /** - * Awaits release on sync. - */ - void awaitSyncRelease(ForkJoinPool p) { - while (thread != null && !p.syncIsReleasable(this)) - LockSupport.park(this); - } - - /** - * Awaits resumption as spare. - */ - void awaitSpareRelease() { - while (thread != null) { - if (!Thread.interrupted()) - LockSupport.park(this); - } - } } /** * Ensures that no thread is waiting for count to advance from the * current value of eventCount read on entry to this method, by * releasing waiting threads if necessary. - * - * @return the count */ - final long ensureSync() { + final void ensureSync() { long c = eventCount; WaitQueueNode q; while ((q = syncStack) != null && q.count < c) { @@ -1469,7 +1451,6 @@ public class ForkJoinPool extends Abstra break; } } - return c; } /** @@ -1484,17 +1465,18 @@ public class ForkJoinPool extends Abstra /** * Signals threads waiting to poll a task. Because method sync * rechecks availability, it is OK to only proceed if queue - * appears to be non-empty, and OK to skip under contention to - * increment count (since some other thread succeeded). + * appears to be non-empty, and OK if CAS to increment count + * fails (since some other thread succeeded). */ final void signalWork() { - long c; - WaitQueueNode q; - if (syncStack != null && - casEventCount(c = eventCount, c+1) && - (((q = syncStack) != null && q.count <= c) && - (!casBarrierStack(q, q.next) || !q.signal()))) - ensureSync(); + if (syncStack != null) { + long c; + casEventCount(c = eventCount, c+1); + WaitQueueNode q = syncStack; + if (q != null && q.count <= c && + (!casBarrierStack(q, q.next) || !q.signal())) + ensureSync(); + } } /** @@ -1507,53 +1489,38 @@ public class ForkJoinPool extends Abstra final void sync(ForkJoinWorkerThread w) { updateStealCount(w); // Transfer w's count while it is idle - while (!w.isShutdown() && isProcessingTasks() && !suspendIfSpare(w)) { + if (!w.isShutdown() && isProcessingTasks() && !suspendIfSpare(w)) { long prev = w.lastEventCount; WaitQueueNode node = null; WaitQueueNode h; + boolean helpSignal = false; while (eventCount == prev && ((h = syncStack) == null || h.count == prev)) { if (node == null) node = new WaitQueueNode(prev, w); if (casBarrierStack(node.next = h, node)) { - node.awaitSyncRelease(this); + if (!Thread.interrupted() && node.thread != null && + eventCount == prev) { + if (h == null && // cover signalWork race + ForkJoinWorkerThread.hasQueuedTasks(workers)) + helpSignal = true; + else + LockSupport.park(this); + } + if (node.thread != null) + node.thread = null; break; } } - long ec = ensureSync(); - if (ec != prev) { + long ec = eventCount; + if (ec != prev) w.lastEventCount = ec; - break; - } + else if (helpSignal) + casEventCount(ec, ec + 1); + ensureSync(); } } - /** - * Returns {@code true} if worker waiting on sync can proceed: - * - on signal (thread == null) - * - on event count advance (winning race to notify vs signaller) - * - on interrupt - * - if the first queued node, we find work available - * If node was not signalled and event count not advanced on exit, - * then we also help advance event count. - * - * @return {@code true} if node can be released - */ - final boolean syncIsReleasable(WaitQueueNode node) { - long prev = node.count; - if (!Thread.interrupted() && node.thread != null && - (node.next != null || - !ForkJoinWorkerThread.hasQueuedTasks(workers)) && - eventCount == prev) - return false; - if (node.thread != null) { - node.thread = null; - long ec = eventCount; - if (prev <= ec) // help signal - casEventCount(ec, ec+1); - } - return true; - } /** * Returns {@code true} if a new sync event occurred since last @@ -1561,11 +1528,11 @@ public class ForkJoinPool extends Abstra */ final boolean hasNewSyncEvent(ForkJoinWorkerThread w) { long lc = w.lastEventCount; - long ec = ensureSync(); - if (ec == lc) - return false; - w.lastEventCount = ec; - return true; + long ec = eventCount; + if (lc != ec) + w.lastEventCount = ec; + ensureSync(); + return lc != ec || lc != eventCount; } // Parallelism maintenance @@ -1731,19 +1698,25 @@ public class ForkJoinPool extends Abstra */ private boolean suspendIfSpare(ForkJoinWorkerThread w) { WaitQueueNode node = null; - int s; - while (parallelism < runningCountOf(s = workerCounts)) { + for (;;) { + int p = parallelism; + int s = workerCounts; + int r = runningCountOf(s); + int t = totalCountOf(s); + // use t as bound if r transiently out of sync + if (t <= p || r <= p) + return false; // not a spare if (node == null) node = new WaitQueueNode(0, w); - if (casWorkerCounts(s, s-1)) { // representation-dependent - // push onto stack - do {} while (!casSpareStack(node.next = spareStack, node)); - // block until released by resumeSpare - node.awaitSpareRelease(); - return true; - } + if (casWorkerCounts(s, workerCountsFor(t, r - 1))) + break; } - return false; + // push onto stack + do {} while (!casSpareStack(node.next = spareStack, node)); + // block until released by resumeSpare + while (!Thread.interrupted() && node.thread != null) + LockSupport.park(this); + return true; } /**