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.70 by dl, Sat Sep 4 11:33:53 2010 UTC vs.
Revision 1.79 by jsr166, Tue Sep 7 23:49:30 2010 UTC

# Line 6 | Line 6
6  
7   package jsr166y;
8  
9 – import java.util.concurrent.*;
9   import java.util.ArrayList;
10   import java.util.Arrays;
11   import java.util.Collection;
12   import java.util.Collections;
13   import java.util.List;
14 + import java.util.concurrent.AbstractExecutorService;
15 + import java.util.concurrent.Callable;
16 + import java.util.concurrent.CountDownLatch;
17 + import java.util.concurrent.ExecutorService;
18 + import java.util.concurrent.Future;
19 + import java.util.concurrent.RejectedExecutionException;
20 + import java.util.concurrent.RunnableFuture;
21 + import java.util.concurrent.TimeUnit;
22 + import java.util.concurrent.TimeoutException;
23 + import java.util.concurrent.atomic.AtomicInteger;
24   import java.util.concurrent.locks.LockSupport;
25   import java.util.concurrent.locks.ReentrantLock;
17 – import java.util.concurrent.atomic.AtomicInteger;
18 – import java.util.concurrent.CountDownLatch;
26  
27   /**
28   * An {@link ExecutorService} for running {@link ForkJoinTask}s.
# Line 300 | Line 307 | public class ForkJoinPool extends Abstra
307       * about the same time as another is needlessly being created. We
308       * counteract this and related slop in part by requiring resumed
309       * spares to immediately recheck (in preStep) to see whether they
310 <     * they should re-suspend.
310 >     * should re-suspend.
311       *
312       * 6. Killing off unneeded workers. A timeout mechanism is used to
313       * shed unused workers: The oldest (first) event queue waiter uses
# Line 429 | Line 436 | public class ForkJoinPool extends Abstra
436  
437      /**
438       * The wakeup interval (in nanoseconds) for the oldest worker
439 <     * worker waiting for an event invokes tryShutdownUnusedWorker to shrink
440 <     * the number of workers.  The exact value does not matter too
441 <     * much, but should be long enough to slowly release resources
442 <     * during long periods without use without disrupting normal use.
439 >     * waiting for an event to invoke tryShutdownUnusedWorker to
440 >     * shrink the number of workers.  The exact value does not matter
441 >     * too much. It must be short enough to release resources during
442 >     * sustained periods of idleness, but not so short that threads
443 >     * are continually re-created.
444       */
445      private static final long SHRINK_RATE_NANOS =
446          30L * 1000L * 1000L * 1000L; // 2 per minute
# Line 515 | Line 523 | public class ForkJoinPool extends Abstra
523       * Lifecycle control. The low word contains the number of workers
524       * that are (probably) executing tasks. This value is atomically
525       * incremented before a worker gets a task to run, and decremented
526 <     * when worker has no tasks and cannot find any.  Bits 16-18
526 >     * when a worker has no tasks and cannot find any.  Bits 16-18
527       * contain runLevel value. When all are zero, the pool is
528       * running. Level transitions are monotonic (running -> shutdown
529       * -> terminating -> terminated) so each transition adds a bit.
# Line 604 | Line 612 | public class ForkJoinPool extends Abstra
612       * (rarely) necessary when other count updates lag.
613       *
614       * @param dr -- either zero or ONE_RUNNING
615 <     * @param dt == either zero or ONE_TOTAL
615 >     * @param dt -- either zero or ONE_TOTAL
616       */
617      private void decrementWorkerCounts(int dr, int dt) {
618          for (;;) {
# Line 673 | Line 681 | public class ForkJoinPool extends Abstra
681      }
682  
683      /**
684 <     * Nulls out record of worker in workers array
684 >     * Nulls out record of worker in workers array.
685       */
686      private void forgetWorker(ForkJoinWorkerThread w) {
687          int idx = w.poolIndex;
# Line 806 | Line 814 | public class ForkJoinPool extends Abstra
814      // Maintaining parallelism
815  
816      /**
817 <     * Pushes worker onto the spare stack
817 >     * Pushes worker onto the spare stack.
818       */
819      final void pushSpare(ForkJoinWorkerThread w) {
820          int ns = (++w.spareCount << SPARE_COUNT_SHIFT) | (w.poolIndex + 1);
# Line 1106 | Line 1114 | public class ForkJoinPool extends Abstra
1114                                       c = eventCount, c+1);
1115              eventWaiters = 0L; // clobber lists
1116              spareWaiters = 0;
1117 <            ForkJoinWorkerThread[] ws = workers;
1110 <            int n = ws.length;
1111 <            for (int i = 0; i < n; ++i) {
1112 <                ForkJoinWorkerThread w = ws[i];
1117 >            for (ForkJoinWorkerThread w : workers) {
1118                  if (w != null) {
1119                      w.shutdown();
1120                      if (passes > 0 && !w.isTerminated()) {
# Line 1128 | Line 1133 | public class ForkJoinPool extends Abstra
1133      }
1134  
1135      /**
1136 <     * Clear out and cancel submissions, ignoring exceptions
1136 >     * Clears out and cancels submissions, ignoring exceptions.
1137       */
1138      private void cancelSubmissions() {
1139          ForkJoinTask<?> task;
# Line 1143 | Line 1148 | public class ForkJoinPool extends Abstra
1148      // misc support for ForkJoinWorkerThread
1149  
1150      /**
1151 <     * Returns pool number
1151 >     * Returns pool number.
1152       */
1153      final int getPoolNumber() {
1154          return poolNumber;
1155      }
1156  
1157      /**
1158 <     * Tries to accumulates steal count from a worker, clearing
1159 <     * the worker's value.
1158 >     * Tries to accumulate steal count from a worker, clearing
1159 >     * the worker's value if successful.
1160       *
1161       * @return true if worker steal count now zero
1162       */
# Line 1175 | Line 1180 | public class ForkJoinPool extends Abstra
1180          int pc = parallelism; // use parallelism, not rc
1181          int ac = runState;    // no mask -- artificially boosts during shutdown
1182          // Use exact results for small values, saturate past 4
1183 <        return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
1183 >        return ((pc <= ac) ? 0 :
1184 >                (pc >>> 1 <= ac) ? 1 :
1185 >                (pc >>> 2 <= ac) ? 3 :
1186 >                pc >>> 3);
1187      }
1188  
1189      // Public and protected methods
# Line 1225 | Line 1233 | public class ForkJoinPool extends Abstra
1233       * use {@link #defaultForkJoinWorkerThreadFactory}.
1234       * @param handler the handler for internal worker threads that
1235       * terminate due to unrecoverable errors encountered while executing
1236 <     * tasks. For default value, use <code>null</code>.
1236 >     * tasks. For default value, use {@code null}.
1237       * @param asyncMode if true,
1238       * establishes local first-in-first-out scheduling mode for forked
1239       * tasks that are never joined. This mode may be more appropriate
1240       * than default locally stack-based mode in applications in which
1241       * worker threads only process event-style asynchronous tasks.
1242 <     * For default value, use <code>false</code>.
1242 >     * For default value, use {@code false}.
1243       * @throws IllegalArgumentException if parallelism less than or
1244       *         equal to zero, or greater than implementation limit
1245       * @throws NullPointerException if the factory is null
# Line 1441 | Line 1449 | public class ForkJoinPool extends Abstra
1449  
1450      /**
1451       * Returns the number of worker threads that have started but not
1452 <     * yet terminated.  This result returned by this method may differ
1452 >     * yet terminated.  The result returned by this method may differ
1453       * from {@link #getParallelism} when threads are created to
1454       * maintain parallelism when others are cooperatively blocked.
1455       *
# Line 1526 | Line 1534 | public class ForkJoinPool extends Abstra
1534       */
1535      public long getQueuedTaskCount() {
1536          long count = 0;
1537 <        ForkJoinWorkerThread[] ws = workers;
1530 <        int n = ws.length;
1531 <        for (int i = 0; i < n; ++i) {
1532 <            ForkJoinWorkerThread w = ws[i];
1537 >        for (ForkJoinWorkerThread w : workers)
1538              if (w != null)
1539                  count += w.getQueueSize();
1535 –        }
1540          return count;
1541      }
1542  
# Line 1587 | Line 1591 | public class ForkJoinPool extends Abstra
1591       */
1592      protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
1593          int count = submissionQueue.drainTo(c);
1594 <        ForkJoinWorkerThread[] ws = workers;
1591 <        int n = ws.length;
1592 <        for (int i = 0; i < n; ++i) {
1593 <            ForkJoinWorkerThread w = ws[i];
1594 >        for (ForkJoinWorkerThread w : workers)
1595              if (w != null)
1596                  count += w.drainTasksTo(c);
1596 –        }
1597          return count;
1598      }
1599  
# Line 1850 | Line 1850 | public class ForkJoinPool extends Abstra
1850      private static final long eventCountOffset =
1851          objectFieldOffset("eventCount", ForkJoinPool.class);
1852      private static final long eventWaitersOffset =
1853 <        objectFieldOffset("eventWaiters",ForkJoinPool.class);
1853 >        objectFieldOffset("eventWaiters", ForkJoinPool.class);
1854      private static final long stealCountOffset =
1855 <        objectFieldOffset("stealCount",ForkJoinPool.class);
1855 >        objectFieldOffset("stealCount", ForkJoinPool.class);
1856      private static final long spareWaitersOffset =
1857 <        objectFieldOffset("spareWaiters",ForkJoinPool.class);
1857 >        objectFieldOffset("spareWaiters", ForkJoinPool.class);
1858  
1859      private static long objectFieldOffset(String field, Class<?> klazz) {
1860          try {

Diff Legend

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