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.100 by dl, Fri Apr 1 20:20:37 2011 UTC vs.
Revision 1.110 by jsr166, Fri Dec 23 00:58:29 2011 UTC

# Line 19 | Line 19 | 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;
22   import java.util.concurrent.atomic.AtomicInteger;
23   import java.util.concurrent.locks.LockSupport;
24   import java.util.concurrent.locks.ReentrantLock;
# Line 102 | Line 101 | import java.util.concurrent.locks.Condit
101   * daemon} mode, there is typically no need to explicitly {@link
102   * #shutdown} such a pool upon program exit.
103   *
104 < * <pre>
104 > *  <pre> {@code
105   * static final ForkJoinPool mainPool = new ForkJoinPool();
106   * ...
107   * public void sort(long[] array) {
108   *   mainPool.invoke(new SortTask(array, 0, array.length));
109 < * }
111 < * </pre>
109 > * }}</pre>
110   *
111   * <p><b>Implementation notes</b>: This implementation restricts the
112   * maximum number of running threads to 32767. Attempts to create
# Line 292 | Line 290 | public class ForkJoinPool extends Abstra
290       * "terminate" status, cancels all unprocessed tasks, and wakes up
291       * all waiting workers.  Detecting whether termination should
292       * commence after a non-abrupt shutdown() call requires more work
293 <     * and bookkeeping. We need consensus about quiesence (i.e., that
293 >     * and bookkeeping. We need consensus about quiescence (i.e., that
294       * there is no more work) which is reflected in active counts so
295       * long as there are no current blockers, as well as possible
296       * re-evaluations during independent changes in blocking or
# Line 467 | Line 465 | public class ForkJoinPool extends Abstra
465      /**
466       * Main pool control -- a long packed with:
467       * AC: Number of active running workers minus target parallelism (16 bits)
468 <     * TC: Number of total workers minus target parallelism (16bits)
468 >     * TC: Number of total workers minus target parallelism (16 bits)
469       * ST: true if pool is terminating (1 bit)
470       * EC: the wait count of top waiting thread (15 bits)
471       * ID: ~poolIndex of top of Treiber stack of waiting threads (16 bits)
# Line 547 | Line 545 | public class ForkJoinPool extends Abstra
545      volatile boolean shutdown;
546  
547      /**
548 <     * True if use local fifo, not default lifo, for local polling
549 <     * Read by, and replicated by ForkJoinWorkerThreads
548 >     * True if use local fifo, not default lifo, for local polling.
549 >     * Read by, and replicated by ForkJoinWorkerThreads.
550       */
551      final boolean locallyFifo;
552  
# Line 786 | Line 784 | public class ForkJoinPool extends Abstra
784          long nc = (long)(v & E_MASK) | ((c - AC_UNIT) & (AC_MASK|TC_MASK));
785          if (ctl != c || !UNSAFE.compareAndSwapLong(this, ctlOffset, c, nc)) {
786              long d = ctl; // return true if lost to a deq, to force scan
787 <            return (int)d != (int)c && ((d - c) & AC_MASK) >= 0L;
787 >            return (int)d != (int)c && (d & AC_MASK) >= (c & AC_MASK);
788          }
789          for (int sc = w.stealCount; sc != 0;) {   // accumulate stealCount
790              long s = stealCount;
# Line 795 | Line 793 | public class ForkJoinPool extends Abstra
793              else if (w.eventCount != v)
794                  return true;                      // update next time
795          }
796 <        if ((int)c != 0 && parallelism + (int)(nc >> AC_SHIFT) == 0 &&
796 >        if ((!shutdown || !tryTerminate(false)) &&
797 >            (int)c != 0 && parallelism + (int)(nc >> AC_SHIFT) == 0 &&
798              blockedCount == 0 && quiescerCount == 0)
799              idleAwaitWork(w, nc, c, v);           // quiescent
800          for (boolean rescanned = false;;) {
# Line 865 | Line 864 | public class ForkJoinPool extends Abstra
864                  w.parked = false;
865                  if (w.eventCount != v)
866                      break;
867 <                else if (System.nanoTime() - startTime <
867 >                else if (System.nanoTime() - startTime <
868                           SHRINK_RATE - (SHRINK_RATE / 10)) // timing slop
869                      Thread.interrupted();          // spurious wakeup
870                  else if (UNSAFE.compareAndSwapLong(this, ctlOffset,
# Line 948 | Line 947 | public class ForkJoinPool extends Abstra
947              int pc = parallelism;
948              do {
949                  ForkJoinWorkerThread[] ws; ForkJoinWorkerThread w;
950 <                int e, ac, tc, rc, i;
950 >                int e, ac, tc, i;
951                  long c = ctl;
952                  int u = (int)(c >>> 32);
953                  if ((e = (int)c) < 0) {
# Line 988 | Line 987 | public class ForkJoinPool extends Abstra
987      }
988  
989      /**
990 <     * Decrements blockedCount and increments active count
990 >     * Decrements blockedCount and increments active count.
991       */
992      private void postBlock() {
993          long c;
994          do {} while (!UNSAFE.compareAndSwapLong(this, ctlOffset,  // no mask
995                                                  c = ctl, c + AC_UNIT));
996          int b;
997 <        do {} while(!UNSAFE.compareAndSwapInt(this, blockedCountOffset,
998 <                                              b = blockedCount, b - 1));
997 >        do {} while (!UNSAFE.compareAndSwapInt(this, blockedCountOffset,
998 >                                               b = blockedCount, b - 1));
999      }
1000  
1001      /**
# Line 1006 | Line 1005 | public class ForkJoinPool extends Abstra
1005       * @param joinMe the task
1006       */
1007      final void tryAwaitJoin(ForkJoinTask<?> joinMe) {
1009 –        int s;
1008          Thread.interrupted(); // clear interrupts before checking termination
1009          if (joinMe.status >= 0) {
1010              if (tryPreBlock()) {
# Line 1020 | Line 1018 | public class ForkJoinPool extends Abstra
1018  
1019      /**
1020       * Possibly blocks the given worker waiting for joinMe to
1021 <     * complete or timeout
1021 >     * complete or timeout.
1022       *
1023       * @param joinMe the task
1024       * @param millis the wait time for underlying Object.wait
# Line 1056 | Line 1054 | public class ForkJoinPool extends Abstra
1054      }
1055  
1056      /**
1057 <     * If necessary, compensates for blocker, and blocks
1057 >     * If necessary, compensates for blocker, and blocks.
1058       */
1059      private void awaitBlocker(ManagedBlocker blocker)
1060          throws InterruptedException {
# Line 1072 | Line 1070 | public class ForkJoinPool extends Abstra
1070          }
1071      }
1072  
1073 <    // Creating, registering and deregistring workers
1073 >    // Creating, registering and deregistering workers
1074  
1075      /**
1076       * Tries to create and start a worker; minimally rolls back counts
# Line 1148 | Line 1146 | public class ForkJoinPool extends Abstra
1146                          ws[k] = w;
1147                          nextWorkerIndex = k + 1;
1148                          int m = g & SMASK;
1149 <                        g = k > m? ((m << 1) + 1) & SMASK : g + (SG_UNIT<<1);
1149 >                        g = (k > m) ? ((m << 1) + 1) & SMASK : g + (SG_UNIT<<1);
1150                      }
1151                  } finally {
1152                      scanGuard = g;
# Line 1323 | Line 1321 | public class ForkJoinPool extends Abstra
1321      // misc ForkJoinWorkerThread support
1322  
1323      /**
1324 <     * Increment or decrement quiescerCount. Needed only to prevent
1324 >     * Increments or decrements quiescerCount. Needed only to prevent
1325       * triggering shutdown if a worker is transiently inactive while
1326       * checking quiescence.
1327       *
# Line 1331 | Line 1329 | public class ForkJoinPool extends Abstra
1329       */
1330      final void addQuiescerCount(int delta) {
1331          int c;
1332 <        do {} while(!UNSAFE.compareAndSwapInt(this, quiescerCountOffset,
1333 <                                              c = quiescerCount, c + delta));
1332 >        do {} while (!UNSAFE.compareAndSwapInt(this, quiescerCountOffset,
1333 >                                               c = quiescerCount, c + delta));
1334      }
1335  
1336      /**
1337 <     * Directly increment or decrement active count without
1338 <     * queuing. This method is used to transiently assert inactivation
1339 <     * while checking quiescence.
1337 >     * Directly increments or decrements active count without queuing.
1338 >     * This method is used to transiently assert inactivation while
1339 >     * checking quiescence.
1340       *
1341       * @param delta 1 for increment, -1 for decrement
1342       */
1343      final void addActiveCount(int delta) {
1344 <        long d = delta < 0 ? -AC_UNIT : AC_UNIT;
1344 >        long d = (long)delta << AC_SHIFT;
1345          long c;
1346 <        do {} while (!UNSAFE.compareAndSwapLong(this, ctlOffset, c = ctl,
1347 <                                                ((c + d) & AC_MASK) |
1350 <                                                (c & ~AC_MASK)));
1346 >        do {} while (!UNSAFE.compareAndSwapLong(this, ctlOffset,
1347 >                                                c = ctl, c + d));
1348      }
1349  
1350      /**
# Line 1357 | Line 1354 | public class ForkJoinPool extends Abstra
1354      final int idlePerActive() {
1355          // Approximate at powers of two for small values, saturate past 4
1356          int p = parallelism;
1357 <        int a = p + (int)(ctl >> AC_SHIFT);
1358 <        return (a > (p >>>= 1) ? 0 :
1359 <                a > (p >>>= 1) ? 1 :
1360 <                a > (p >>>= 1) ? 2 :
1361 <                a > (p >>>= 1) ? 4 :
1362 <                8);
1357 >        int a = p + (int)(ctl >> AC_SHIFT);
1358 >        return (a > (p >>>= 1) ? 0 :
1359 >                a > (p >>>= 1) ? 1 :
1360 >                a > (p >>>= 1) ? 2 :
1361 >                a > (p >>>= 1) ? 4 :
1362 >                8);
1363      }
1364  
1365      // Exported methods
# Line 1685 | Line 1682 | public class ForkJoinPool extends Abstra
1682       */
1683      public int getRunningThreadCount() {
1684          int r = parallelism + (int)(ctl >> AC_SHIFT);
1685 <        return r <= 0? 0 : r; // suppress momentarily negative values
1685 >        return (r <= 0) ? 0 : r; // suppress momentarily negative values
1686      }
1687  
1688      /**
# Line 1697 | Line 1694 | public class ForkJoinPool extends Abstra
1694       */
1695      public int getActiveThreadCount() {
1696          int r = parallelism + (int)(ctl >> AC_SHIFT) + blockedCount;
1697 <        return r <= 0? 0 : r; // suppress momentarily negative values
1697 >        return (r <= 0) ? 0 : r; // suppress momentarily negative values
1698      }
1699  
1700      /**
# Line 1852 | Line 1849 | public class ForkJoinPool extends Abstra
1849          int ac = rc + blockedCount;
1850          String level;
1851          if ((c & STOP_BIT) != 0)
1852 <            level = (tc == 0)? "Terminated" : "Terminating";
1852 >            level = (tc == 0) ? "Terminated" : "Terminating";
1853          else
1854 <            level = shutdown? "Shutting down" : "Running";
1854 >            level = shutdown ? "Shutting down" : "Running";
1855          return super.toString() +
1856              "[" + level +
1857              ", parallelism = " + pc +
# Line 2117 | Line 2114 | public class ForkJoinPool extends Abstra
2114          modifyThreadPermission = new RuntimePermission("modifyThread");
2115          defaultForkJoinWorkerThreadFactory =
2116              new DefaultForkJoinWorkerThreadFactory();
2120 –        int s;
2117          try {
2118              UNSAFE = getUnsafe();
2119 <            Class k = ForkJoinPool.class;
2119 >            Class<?> k = ForkJoinPool.class;
2120              ctlOffset = UNSAFE.objectFieldOffset
2121                  (k.getDeclaredField("ctl"));
2122              stealCountOffset = UNSAFE.objectFieldOffset
# Line 2133 | Line 2129 | public class ForkJoinPool extends Abstra
2129                  (k.getDeclaredField("scanGuard"));
2130              nextWorkerNumberOffset = UNSAFE.objectFieldOffset
2131                  (k.getDeclaredField("nextWorkerNumber"));
2136 –            Class a = ForkJoinTask[].class;
2137 –            ABASE = UNSAFE.arrayBaseOffset(a);
2138 –            s = UNSAFE.arrayIndexScale(a);
2132          } catch (Exception e) {
2133              throw new Error(e);
2134          }
2135 +        Class<?> a = ForkJoinTask[].class;
2136 +        ABASE = UNSAFE.arrayBaseOffset(a);
2137 +        int s = UNSAFE.arrayIndexScale(a);
2138          if ((s & (s-1)) != 0)
2139              throw new Error("data type scale not a power of two");
2140          ASHIFT = 31 - Integer.numberOfLeadingZeros(s);

Diff Legend

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