ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/jsr166y/ForkJoinPool.java
(Generate patch)

Comparing jsr166/src/jsr166y/ForkJoinPool.java (file contents):
Revision 1.66 by dl, Sun Aug 29 23:34:46 2010 UTC vs.
Revision 1.70 by dl, Sat Sep 4 11:33:53 2010 UTC

# Line 7 | Line 7
7   package jsr166y;
8  
9   import java.util.concurrent.*;
10 –
10   import java.util.ArrayList;
11   import java.util.Arrays;
12   import java.util.Collection;
# Line 69 | Line 68 | import java.util.concurrent.CountDownLat
68   *    <td ALIGN=CENTER> <b>Call from within fork/join computations</b></td>
69   *  </tr>
70   *  <tr>
71 < *    <td> <b>Arange async execution</td>
71 > *    <td> <b>Arrange async execution</td>
72   *    <td> {@link #execute(ForkJoinTask)}</td>
73   *    <td> {@link ForkJoinTask#fork}</td>
74   *  </tr>
# Line 140 | Line 139 | public class ForkJoinPool extends Abstra
139       * Beyond work-stealing support and essential bookkeeping, the
140       * main responsibility of this framework is to take actions when
141       * one worker is waiting to join a task stolen (or always held by)
142 <     * another.  Becauae we are multiplexing many tasks on to a pool
142 >     * another.  Because we are multiplexing many tasks on to a pool
143       * of workers, we can't just let them block (as in Thread.join).
144       * We also cannot just reassign the joiner's run-time stack with
145       * another and replace it later, which would be a form of
# Line 157 | Line 156 | public class ForkJoinPool extends Abstra
156       *      links to try to find such a task.
157       *
158       *   Compensating: Unless there are already enough live threads,
159 <     *      method helpMaintainParallelism() may create or or
159 >     *      method helpMaintainParallelism() may create or
160       *      re-activate a spare thread to compensate for blocked
161       *      joiners until they unblock.
162       *
# Line 226 | Line 225 | public class ForkJoinPool extends Abstra
225       * ManagedBlocker), we may create or resume others to take their
226       * place until they unblock (see below). Implementing this
227       * requires counts of the number of "running" threads (i.e., those
228 <     * that are neither blocked nor artifically suspended) as well as
228 >     * that are neither blocked nor artificially suspended) as well as
229       * the total number.  These two values are packed into one field,
230       * "workerCounts" because we need accurate snapshots when deciding
231       * to create, resume or suspend.  Note however that the
232 <     * correspondance of these counts to reality is not guaranteed. In
232 >     * correspondence of these counts to reality is not guaranteed. In
233       * particular updates for unblocked threads may lag until they
234       * actually wake up.
235       *
# Line 315 | Line 314 | public class ForkJoinPool extends Abstra
314       * 7. Deciding when to create new workers. The main dynamic
315       * control in this class is deciding when to create extra threads
316       * in method helpMaintainParallelism. We would like to keep
317 <     * exactly #parallelism threads running, which is an impossble
317 >     * exactly #parallelism threads running, which is an impossible
318       * task. We always need to create one when the number of running
319       * threads would become zero and all workers are busy. Beyond
320 <     * this, we must rely on heuristics that work well in the the
321 <     * presence of transients phenomena such as GC stalls, dynamic
320 >     * this, we must rely on heuristics that work well in the
321 >     * presence of transient phenomena such as GC stalls, dynamic
322       * compilation, and wake-up lags. These transients are extremely
323       * common -- we are normally trying to fully saturate the CPUs on
324       * a machine, so almost any activity other than running tasks
# Line 346 | Line 345 | public class ForkJoinPool extends Abstra
345       * "while ((local = field) != 0)") which are usually the simplest
346       * way to ensure the required read orderings (which are sometimes
347       * critical). Also several occurrences of the unusual "do {}
348 <     * while(!cas...)" which is the simplest way to force an update of
348 >     * while (!cas...)" which is the simplest way to force an update of
349       * a CAS'ed variable. There are also other coding oddities that
350       * help some methods perform reasonably even when interpreted (not
351       * compiled), at the expense of some messy constructions that
# Line 420 | Line 419 | public class ForkJoinPool extends Abstra
419      /**
420       * The time to block in a join (see awaitJoin) before checking if
421       * a new worker should be (re)started to maintain parallelism
422 <     * level. The value should be short enough to maintain gloabal
422 >     * level. The value should be short enough to maintain global
423       * responsiveness and progress but long enough to avoid
424       * counterproductive firings during GC stalls or unrelated system
425       * activity, and to not bog down systems with continual re-firings
# Line 483 | Line 482 | public class ForkJoinPool extends Abstra
482      private volatile long stealCount;
483  
484      /**
485 <     * Encoded record of top of treiber stack of threads waiting for
485 >     * Encoded record of top of Treiber stack of threads waiting for
486       * events. The top 32 bits contain the count being waited for. The
487       * bottom 16 bits contains one plus the pool index of waiting
488       * worker thread. (Bits 16-31 are unused.)
# Line 502 | Line 501 | public class ForkJoinPool extends Abstra
501      private volatile int eventCount;
502  
503      /**
504 <     * Encoded record of top of treiber stack of spare threads waiting
504 >     * Encoded record of top of Treiber stack of spare threads waiting
505       * for resumption. The top 16 bits contain an arbitrary count to
506       * avoid ABA effects. The bottom 16bits contains one plus the pool
507       * index of waiting worker thread.
# Line 678 | Line 677 | public class ForkJoinPool extends Abstra
677       */
678      private void forgetWorker(ForkJoinWorkerThread w) {
679          int idx = w.poolIndex;
680 <        // Locking helps method recordWorker avoid unecessary expansion
680 >        // Locking helps method recordWorker avoid unnecessary expansion
681          final ReentrantLock lock = this.workerLock;
682          lock.lock();
683          try {
# Line 693 | Line 692 | public class ForkJoinPool extends Abstra
692      /**
693       * Final callback from terminating worker.  Removes record of
694       * worker from array, and adjusts counts. If pool is shutting
695 <     * down, tries to complete terminatation.
695 >     * down, tries to complete termination.
696       *
697       * @param w the worker
698       */
# Line 832 | Line 831 | public class ForkJoinPool extends Abstra
831              UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
832                                       sw, w.nextSpare)) {
833              int c; // increment running count before resume
834 <            do {} while(!UNSAFE.compareAndSwapInt
835 <                        (this, workerCountsOffset,
836 <                         c = workerCounts, c + ONE_RUNNING));
834 >            do {} while (!UNSAFE.compareAndSwapInt
835 >                         (this, workerCountsOffset,
836 >                          c = workerCounts, c + ONE_RUNNING));
837              if (w.tryUnsuspend())
838                  LockSupport.unpark(w);
839              else   // back out if w was shutdown
# Line 846 | Line 845 | public class ForkJoinPool extends Abstra
845       * Tries to increase the number of running workers if below target
846       * parallelism: If a spare exists tries to resume it via
847       * tryResumeSpare.  Otherwise, if not enough total workers or all
848 <     * existing workers are busy, adds a new worker. In all casses also
848 >     * existing workers are busy, adds a new worker. In all cases also
849       * helps wake up releasable workers waiting for work.
850       */
851      private void helpMaintainParallelism() {
# Line 985 | Line 984 | public class ForkJoinPool extends Abstra
984                      w.lastEventCount = ec;     // no need to wait
985                      break;
986                  }
987 <                else if (!(inactivate |= active))  
987 >                else if (!(inactivate |= active))
988                      eventSync(w, wec);         // must inactivate before sync
989              }
990              else
# Line 1174 | Line 1173 | public class ForkJoinPool extends Abstra
1173       */
1174      final int idlePerActive() {
1175          int pc = parallelism; // use parallelism, not rc
1176 <        int ac = runState;    // no mask -- artifically boosts during shutdown
1176 >        int ac = runState;    // no mask -- artificially boosts during shutdown
1177          // Use exact results for small values, saturate past 4
1178          return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
1179      }
# Line 1721 | Line 1720 | public class ForkJoinPool extends Abstra
1720          throws InterruptedException {
1721          try {
1722              return termination.awaitAdvanceInterruptibly(0, timeout, unit) > 0;
1723 <        } catch(TimeoutException ex) {
1723 >        } catch (TimeoutException ex) {
1724              return false;
1725          }
1726      }

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines