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

Comparing jsr166/src/jsr166e/ForkJoinPool.java (file contents):
Revision 1.28 by dl, Thu Dec 13 20:47:34 2012 UTC vs.
Revision 1.48 by jsr166, Tue Jan 22 23:00:06 2013 UTC

# Line 440 | Line 440 | public class ForkJoinPool extends Abstra
440       * Common Pool
441       * ===========
442       *
443 <     * The static commonPool always exists after static
443 >     * The static common Pool always exists after static
444       * initialization.  Since it (or any other created pool) need
445       * never be used, we minimize initial construction overhead and
446       * footprint to the setup of about a dozen fields, with no nested
# Line 987 | Line 987 | public class ForkJoinPool extends Abstra
987              if (t != null) {
988                  (currentSteal = t).doExec();
989                  currentSteal = null;
990 +                ++nsteals;
991                  if (base - top < 0) {       // process remaining local tasks
992                      if (mode == 0)
993                          popAndExecAll();
994                      else
995                          pollAndExecAll();
996                  }
996                ++nsteals;
997                hint = -1;
997              }
998          }
999  
# Line 1021 | Line 1020 | public class ForkJoinPool extends Abstra
1020                      s != Thread.State.TIMED_WAITING);
1021          }
1022  
1024        /**
1025         * If this owned and is not already interrupted, try to
1026         * interrupt and/or unpark, ignoring exceptions.
1027         */
1028        final void interruptOwner() {
1029            Thread wt, p;
1030            if ((wt = owner) != null && !wt.isInterrupted()) {
1031                try {
1032                    wt.interrupt();
1033                } catch (SecurityException ignore) {
1034                }
1035            }
1036            if ((p = parker) != null)
1037                U.unpark(p);
1038        }
1039
1023          // Unsafe mechanics
1024          private static final sun.misc.Unsafe U;
1025          private static final long QLOCK;
1026          private static final int ABASE;
1027          private static final int ASHIFT;
1028          static {
1046            int s;
1029              try {
1030                  U = getUnsafe();
1031                  Class<?> k = WorkQueue.class;
# Line 1051 | Line 1033 | public class ForkJoinPool extends Abstra
1033                  QLOCK = U.objectFieldOffset
1034                      (k.getDeclaredField("qlock"));
1035                  ABASE = U.arrayBaseOffset(ak);
1036 <                s = U.arrayIndexScale(ak);
1036 >                int scale = U.arrayIndexScale(ak);
1037 >                if ((scale & (scale - 1)) != 0)
1038 >                    throw new Error("data type scale not a power of two");
1039 >                ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
1040              } catch (Exception e) {
1041                  throw new Error(e);
1042              }
1058            if ((s & (s-1)) != 0)
1059                throw new Error("data type scale not a power of two");
1060            ASHIFT = 31 - Integer.numberOfLeadingZeros(s);
1043          }
1044      }
1045  
# Line 1091 | Line 1073 | public class ForkJoinPool extends Abstra
1073       * to paranoically avoid potential initialization circularities
1074       * as well as to simplify generated code.
1075       */
1076 <    static final ForkJoinPool commonPool;
1076 >    static final ForkJoinPool common;
1077  
1078      /**
1079 <     * Common pool parallelism. Must equal commonPool.parallelism.
1079 >     * Common pool parallelism. Must equal common.parallelism.
1080       */
1081 <    static final int commonPoolParallelism;
1081 >    static final int commonParallelism;
1082  
1083      /**
1084       * Sequence number for creating workerNamePrefix.
# Line 1104 | Line 1086 | public class ForkJoinPool extends Abstra
1086      private static int poolNumberSequence;
1087  
1088      /**
1089 <     * Return the next sequence number. We don't expect this to
1090 <     * ever contend so use simple builtin sync.
1089 >     * Returns the next sequence number. We don't expect this to
1090 >     * ever contend, so use simple builtin sync.
1091       */
1092      private static final synchronized int nextPoolId() {
1093          return ++poolNumberSequence;
# Line 1131 | Line 1113 | public class ForkJoinPool extends Abstra
1113      /**
1114       * Tolerance for idle timeouts, to cope with timer undershoots
1115       */
1116 <    private static final long TIMEOUT_SLOP = 2000000L; // 20ms
1116 >    private static final long TIMEOUT_SLOP = 2000000L;
1117  
1118      /**
1119       * The maximum stolen->joining link depth allowed in method
# Line 1265 | Line 1247 | public class ForkJoinPool extends Abstra
1247      /*
1248       * Acquires the plock lock to protect worker array and related
1249       * updates. This method is called only if an initial CAS on plock
1250 <     * fails. This acts as a spinLock for normal cases, but falls back
1250 >     * fails. This acts as a spinlock for normal cases, but falls back
1251       * to builtin monitor to block when (rarely) needed. This would be
1252       * a terrible idea for a highly contended lock, but works fine as
1253 <     * a more conservative alternative to a pure spinlock.  See
1272 <     * internal ConcurrentHashMap documentation for further
1273 <     * explanation of nearly the same construction.
1253 >     * a more conservative alternative to a pure spinlock.
1254       */
1255      private int acquirePlock() {
1256          int spins = PL_SPINS, r = 0, ps, nps;
# Line 1322 | Line 1302 | public class ForkJoinPool extends Abstra
1302      }
1303  
1304      /**
1325     * Performs secondary initialization, called when plock is zero.
1326     * Creates workQueue array and sets plock to a valid value.  The
1327     * lock body must be exception-free (so no try/finally) so we
1328     * optimistically allocate new array outside the lock and throw
1329     * away if (very rarely) not needed. (A similar tactic is used in
1330     * fullExternalPush.)  Because the plock seq value can eventually
1331     * wrap around zero, this method harmlessly fails to reinitialize
1332     * if workQueues exists, while still advancing plock.
1333     *
1334     * Additionally tries to create the first worker.
1335     */
1336    private void initWorkers() {
1337        WorkQueue[] ws, nws; int ps;
1338        int p = config & SMASK;        // find power of two table size
1339        int n = (p > 1) ? p - 1 : 1;   // ensure at least 2 slots
1340        n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; n |= n >>> 8; n |= n >>> 16;
1341        n = (n + 1) << 1;
1342        if ((ws = workQueues) == null || ws.length == 0)
1343            nws = new WorkQueue[n];
1344        else
1345            nws = null;
1346        if (((ps = plock) & PL_LOCK) != 0 ||
1347            !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
1348            ps = acquirePlock();
1349        if (((ws = workQueues) == null || ws.length == 0) && nws != null)
1350            workQueues = nws;
1351        int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
1352        if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
1353            releasePlock(nps);
1354        tryAddWorker();
1355    }
1356
1357    /**
1305       * Tries to create and start one worker if fewer than target
1306       * parallelism level exist. Adjusts counts etc on failure.
1307       */
# Line 1466 | Line 1413 | public class ForkJoinPool extends Abstra
1413              }
1414          }
1415  
1416 <        long c;                             // adjust ctl counts
1416 >        long c;                          // adjust ctl counts
1417          do {} while (!U.compareAndSwapLong
1418                       (this, CTL, c = ctl, (((c - AC_UNIT) & AC_MASK) |
1419                                             ((c - TC_UNIT) & TC_MASK) |
1420                                             (c & ~(AC_MASK|TC_MASK)))));
1421  
1422          if (!tryTerminate(false, false) && w != null && w.array != null) {
1423 <            w.cancelAll();                  // cancel remaining tasks
1424 <            int e, u, i, n; WorkQueue[] ws; WorkQueue v; Thread p;
1425 <            while ((u = (int)((c = ctl) >>> 32)) < 0) {
1426 <                if ((e = (int)c) > 0) {     // activate or create replacement
1427 <                    if ((ws = workQueues) != null &&
1428 <                        ws.length > (i = e & SMASK) &&
1429 <                        (v = ws[i]) != null && v.eventCount == (e | INT_SIGN)) {
1430 <                        long nc = (((long)(v.nextWait & E_MASK)) |
1431 <                                   ((long)(u + UAC_UNIT) << 32));
1432 <                        if (U.compareAndSwapLong(this, CTL, c, nc)) {
1433 <                            v.eventCount = (e + E_SEQ) & E_MASK;
1434 <                            if ((p = v.parker) != null)
1435 <                                U.unpark(p);
1436 <                            break;
1437 <                        }
1438 <                    }
1492 <                    else
1423 >            w.cancelAll();               // cancel remaining tasks
1424 >            WorkQueue[] ws; WorkQueue v; Thread p; int u, i, e;
1425 >            while ((u = (int)((c = ctl) >>> 32)) < 0 && (e = (int)c) >= 0) {
1426 >                if (e > 0) {             // activate or create replacement
1427 >                    if ((ws = workQueues) == null ||
1428 >                        (i = e & SMASK) >= ws.length ||
1429 >                        (v = ws[i]) == null)
1430 >                        break;
1431 >                    long nc = (((long)(v.nextWait & E_MASK)) |
1432 >                               ((long)(u + UAC_UNIT) << 32));
1433 >                    if (v.eventCount != (e | INT_SIGN))
1434 >                        break;
1435 >                    if (U.compareAndSwapLong(this, CTL, c, nc)) {
1436 >                        v.eventCount = (e + E_SEQ) & E_MASK;
1437 >                        if ((p = v.parker) != null)
1438 >                            U.unpark(p);
1439                          break;
1440 +                    }
1441                  }
1442                  else {
1443                      if ((short)u < 0)
# Line 1539 | Line 1486 | public class ForkJoinPool extends Abstra
1486      /**
1487       * Full version of externalPush. This method is called, among
1488       * other times, upon the first submission of the first task to the
1489 <     * pool, so must perform secondary initialization (via
1490 <     * initWorkers). It also detects first submission by an external
1491 <     * thread by looking up its ThreadLocal, and creates a new shared
1492 <     * queue if the one at index if empty or contended. The plock lock
1493 <     * body must be exception-free (so no try/finally) so we
1494 <     * optimistically allocate new queues outside the lock and throw
1495 <     * them away if (very rarely) not needed.
1489 >     * pool, so must perform secondary initialization.  It also
1490 >     * detects first submission by an external thread by looking up
1491 >     * its ThreadLocal, and creates a new shared queue if the one at
1492 >     * index if empty or contended. The plock lock body must be
1493 >     * exception-free (so no try/finally) so we optimistically
1494 >     * allocate new queues outside the lock and throw them away if
1495 >     * (very rarely) not needed.
1496 >     *
1497 >     * Secondary initialization occurs when plock is zero, to create
1498 >     * workQueue array and set plock to a valid value.  This lock body
1499 >     * must also be exception-free. Because the plock seq value can
1500 >     * eventually wrap around zero, this method harmlessly fails to
1501 >     * reinitialize if workQueues exists, while still advancing plock.
1502       */
1503      private void fullExternalPush(ForkJoinTask<?> task) {
1504          int r = 0; // random index seed
# Line 1556 | Line 1509 | public class ForkJoinPool extends Abstra
1509                                          r += SEED_INCREMENT) && r != 0)
1510                      submitters.set(z = new Submitter(r));
1511              }
1512 <            else if (r == 0) {               // move to a different index
1512 >            else if (r == 0) {                  // move to a different index
1513                  r = z.seed;
1514 <                r ^= r << 13;                // same xorshift as WorkQueues
1514 >                r ^= r << 13;                   // same xorshift as WorkQueues
1515                  r ^= r >>> 17;
1516                  z.seed = r ^ (r << 5);
1517              }
1518              else if ((ps = plock) < 0)
1519                  throw new RejectedExecutionException();
1520              else if (ps == 0 || (ws = workQueues) == null ||
1521 <                     (m = ws.length - 1) < 0)
1522 <                initWorkers();
1521 >                     (m = ws.length - 1) < 0) { // initialize workQueues
1522 >                int p = config & SMASK;         // find power of two table size
1523 >                int n = (p > 1) ? p - 1 : 1;    // ensure at least 2 slots
1524 >                n |= n >>> 1; n |= n >>> 2;  n |= n >>> 4;
1525 >                n |= n >>> 8; n |= n >>> 16; n = (n + 1) << 1;
1526 >                WorkQueue[] nws = ((ws = workQueues) == null || ws.length == 0 ?
1527 >                                   new WorkQueue[n] : null);
1528 >                if (((ps = plock) & PL_LOCK) != 0 ||
1529 >                    !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
1530 >                    ps = acquirePlock();
1531 >                if (((ws = workQueues) == null || ws.length == 0) && nws != null)
1532 >                    workQueues = nws;
1533 >                int nps = (ps & SHUTDOWN) | ((ps + PL_LOCK) & ~SHUTDOWN);
1534 >                if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
1535 >                    releasePlock(nps);
1536 >            }
1537              else if ((q = ws[k = r & m & SQMASK]) != null) {
1538                  if (q.qlock == 0 && U.compareAndSwapInt(q, QLOCK, 0, 1)) {
1539                      ForkJoinTask<?>[] a = q.array;
# Line 1686 | Line 1653 | public class ForkJoinPool extends Abstra
1653       *
1654       * * If not already enqueued, try to inactivate and enqueue the
1655       * worker on wait queue. Or, if inactivating has caused the pool
1656 <     * to be quiescent, relay to idleAwaitWork to check for
1657 <     * termination and possibly shrink pool.
1656 >     * to be quiescent, relay to idleAwaitWork to possibly shrink
1657 >     * pool.
1658       *
1659       * * If already enqueued and none of the above apply, possibly
1660 <     * (with 1/2 probability) park awaiting signal, else lingering to
1661 <     * help scan and signal.
1660 >     * park awaiting signal, else lingering to help scan and signal.
1661 >     *
1662 >     * * If a non-empty queue discovered or left as a hint,
1663 >     * help wake up other workers before return.
1664       *
1665       * @param w the worker (via its WorkQueue)
1666       * @return a task or null if none found
# Line 1702 | Line 1671 | public class ForkJoinPool extends Abstra
1671          if (w != null && (ws = workQueues) != null && (m = ws.length - 1) >= 0) {
1672              int ec = w.eventCount;               // ec is negative if inactive
1673              int r = w.seed; r ^= r << 13; r ^= r >>> 17; w.seed = r ^= r << 5;
1674 +            w.hint = -1;                         // update seed and clear hint
1675              int j = ((m + m + 1) | MIN_SCAN) & MAX_SCAN;
1676              do {
1677                  WorkQueue q; ForkJoinTask<?>[] a; int b;
# Line 1723 | Line 1693 | public class ForkJoinPool extends Abstra
1693                  }
1694              } while (--j >= 0);
1695  
1696 <            long c, sc; int e, ns, h;
1697 <            if ((h = w.hint) < 0) {
1698 <                if ((ns = w.nsteals) != 0) {
1699 <                    if (U.compareAndSwapLong(this, STEALCOUNT,
1700 <                                             sc = stealCount, sc + ns))
1701 <                        w.nsteals = 0;           // collect steals
1702 <                }
1703 <                else if (plock != ps)            // consistency check
1704 <                    ;                            // skip
1705 <                else if ((e = (int)(c = ctl)) < 0)
1706 <                    w.qlock = -1;                // pool is terminating
1707 <                else if (ec >= 0) {              // try to enqueue/inactivate
1708 <                    long nc = ((long)ec | ((c - AC_UNIT) & (AC_MASK|TC_MASK)));
1709 <                    w.nextWait = e;              // link and mark inactive
1710 <                    w.eventCount = ec | INT_SIGN;
1711 <                    if (ctl != c || !U.compareAndSwapLong(this, CTL, c, nc))
1712 <                        w.eventCount = ec;       // unmark on CAS failure
1713 <                    else if ((int)(c >> AC_SHIFT) == 1 - (config & SMASK))
1714 <                        idleAwaitWork(w, nc, c);
1715 <                }
1716 <                else if (w.eventCount < 0) {     // block
1717 <                    Thread wt = Thread.currentThread();
1718 <                    Thread.interrupted();        // clear status
1719 <                    U.putObject(wt, PARKBLOCKER, this);
1720 <                    w.parker = wt;               // emulate LockSupport.park
1721 <                    if (w.eventCount < 0)        // recheck
1722 <                        U.park(false, 0L);
1723 <                    w.parker = null;
1724 <                    U.putObject(wt, PARKBLOCKER, null);
1696 >            int h, e, ns; long c, sc; WorkQueue q;
1697 >            if ((ns = w.nsteals) != 0) {
1698 >                if (U.compareAndSwapLong(this, STEALCOUNT,
1699 >                                         sc = stealCount, sc + ns))
1700 >                    w.nsteals = 0;               // collect steals and rescan
1701 >            }
1702 >            else if (plock != ps)                // consistency check
1703 >                ;                                // skip
1704 >            else if ((e = (int)(c = ctl)) < 0)
1705 >                w.qlock = -1;                    // pool is terminating
1706 >            else {
1707 >                if ((h = w.hint) < 0) {
1708 >                    if (ec >= 0) {               // try to enqueue/inactivate
1709 >                        long nc = (((long)ec |
1710 >                                    ((c - AC_UNIT) & (AC_MASK|TC_MASK))));
1711 >                        w.nextWait = e;          // link and mark inactive
1712 >                        w.eventCount = ec | INT_SIGN;
1713 >                        if (ctl != c || !U.compareAndSwapLong(this, CTL, c, nc))
1714 >                            w.eventCount = ec;   // unmark on CAS failure
1715 >                        else if ((int)(c >> AC_SHIFT) == 1 - (config & SMASK))
1716 >                            idleAwaitWork(w, nc, c);
1717 >                    }
1718 >                    else if (w.eventCount < 0 && ctl == c) {
1719 >                        Thread wt = Thread.currentThread();
1720 >                        Thread.interrupted();    // clear status
1721 >                        U.putObject(wt, PARKBLOCKER, this);
1722 >                        w.parker = wt;           // emulate LockSupport.park
1723 >                        if (w.eventCount < 0)    // recheck
1724 >                            U.park(false, 0L);   // block
1725 >                        w.parker = null;
1726 >                        U.putObject(wt, PARKBLOCKER, null);
1727 >                    }
1728 >                }
1729 >                if ((h >= 0 || (h = w.hint) >= 0) &&
1730 >                    (ws = workQueues) != null && h < ws.length &&
1731 >                    (q = ws[h]) != null) {      // signal others before retry
1732 >                    WorkQueue v; Thread p; int u, i, s;
1733 >                    for (int n = (config & SMASK) - 1;;) {
1734 >                        int idleCount = (w.eventCount < 0) ? 0 : -1;
1735 >                        if (((s = idleCount - q.base + q.top) <= n &&
1736 >                             (n = s) <= 0) ||
1737 >                            (u = (int)((c = ctl) >>> 32)) >= 0 ||
1738 >                            (e = (int)c) <= 0 || m < (i = e & SMASK) ||
1739 >                            (v = ws[i]) == null)
1740 >                            break;
1741 >                        long nc = (((long)(v.nextWait & E_MASK)) |
1742 >                                   ((long)(u + UAC_UNIT) << 32));
1743 >                        if (v.eventCount != (e | INT_SIGN) ||
1744 >                            !U.compareAndSwapLong(this, CTL, c, nc))
1745 >                            break;
1746 >                        v.hint = h;
1747 >                        v.eventCount = (e + E_SEQ) & E_MASK;
1748 >                        if ((p = v.parker) != null)
1749 >                            U.unpark(p);
1750 >                        if (--n <= 0)
1751 >                            break;
1752 >                    }
1753                  }
1754              }
1757            if (h >= 0 || w.hint >= 0)           // signal others before retry
1758                helpSignalHint(w);
1755          }
1756          return null;
1757      }
# Line 1774 | Line 1770 | public class ForkJoinPool extends Abstra
1770       */
1771      private void idleAwaitWork(WorkQueue w, long currentCtl, long prevCtl) {
1772          if (w != null && w.eventCount < 0 &&
1773 <            !tryTerminate(false, false) && (int)prevCtl != 0) {
1773 >            !tryTerminate(false, false) && (int)prevCtl != 0 &&
1774 >            ctl == currentCtl) {
1775              int dc = -(short)(currentCtl >>> TC_SHIFT);
1776              long parkTime = dc < 0 ? FAST_IDLE_TIMEOUT: (dc + 1) * IDLE_TIMEOUT;
1777              long deadline = System.nanoTime() + parkTime - TIMEOUT_SLOP;
# Line 1792 | Line 1789 | public class ForkJoinPool extends Abstra
1789                  if (deadline - System.nanoTime() <= 0L &&
1790                      U.compareAndSwapLong(this, CTL, currentCtl, prevCtl)) {
1791                      w.eventCount = (w.eventCount + E_SEQ) | E_MASK;
1792 +                    w.hint = -1;
1793                      w.qlock = -1;   // shrink
1794                      break;
1795                  }
# Line 1826 | Line 1824 | public class ForkJoinPool extends Abstra
1824                          break outer;
1825                      long nc = (((long)(w.nextWait & E_MASK)) |
1826                                 ((long)(u + UAC_UNIT) << 32));
1827 <                    if (w.eventCount == (e | INT_SIGN) &&
1828 <                        U.compareAndSwapLong(this, CTL, c, nc)) {
1829 <                        w.eventCount = (e + E_SEQ) & E_MASK;
1832 <                        if ((p = w.parker) != null)
1833 <                            U.unpark(p);
1834 <                        if (--n <= 0)
1835 <                            break;
1836 <                    }
1837 <                }
1838 <            }
1839 <        }
1840 <    }
1841 <
1842 <    /**
1843 <     * Signals other workers if tasks are present in hinted queue.
1844 <     *
1845 <     * @param caller the worker with the hint
1846 <     */
1847 <    private void helpSignalHint(WorkQueue caller) {
1848 <        WorkQueue[] ws; WorkQueue q, w; Thread p; long c; int h, m, u, e, i, s;
1849 <        if (caller != null && (h = caller.hint) >= 0) {
1850 <            caller.hint = -1;
1851 <            if ((u = (int)(ctl >>> 32)) < 0 && (u >> UAC_SHIFT) < 0 &&
1852 <                (ws = workQueues) != null && (m = ws.length - 1) >= 0 &&
1853 <                (q = ws[h & m]) != null) {
1854 <                for (int n = 2;;) { // limit to at most 2 signals
1855 <                    int idleCount = (caller.eventCount < 0) ? 0 : -1;
1856 <                    if (((s = idleCount - q.base + q.top) <= n &&
1857 <                         (n = s) <= 0) ||
1858 <                        (u = (int)((c = ctl) >>> 32)) >= 0 ||
1859 <                        (e = (int)c) <= 0 || m < (i = e & SMASK) ||
1860 <                        (w = ws[i]) == null)
1861 <                        break;
1862 <                    long nc = (((long)(w.nextWait & E_MASK)) |
1863 <                               ((long)(u + UAC_UNIT) << 32));
1864 <                    if (w.eventCount == (e | INT_SIGN) &&
1865 <                        U.compareAndSwapLong(this, CTL, c, nc)) {
1866 <                        w.hint = h;
1827 >                    if (w.eventCount != (e | INT_SIGN))
1828 >                        break outer;
1829 >                    if (U.compareAndSwapLong(this, CTL, c, nc)) {
1830                          w.eventCount = (e + E_SEQ) & E_MASK;
1831                          if ((p = w.parker) != null)
1832                              U.unpark(p);
# Line 1968 | Line 1931 | public class ForkJoinPool extends Abstra
1931       * @param task the task to join
1932       * @param mode if shared, exit upon completing any task
1933       * if all workers are active
1971     *
1934       */
1935      private int helpComplete(ForkJoinTask<?> task, int mode) {
1936          WorkQueue[] ws; WorkQueue q; int m, n, s, u;
# Line 2120 | Line 2082 | public class ForkJoinPool extends Abstra
2082  
2083      /**
2084       * Returns a (probably) non-empty steal queue, if one is found
2085 <     * during a random, then cyclic scan, else null.  This method must
2086 <     * be retried by caller if, by the time it tries to use the queue,
2125 <     * it is empty.
2085 >     * during a scan, else null.  This method must be retried by
2086 >     * caller if, by the time it tries to use the queue, it is empty.
2087       * @param r a (random) seed for scanning
2088       */
2089      private WorkQueue findNonEmptyStealQueue(int r) {
2090 <        for (WorkQueue[] ws;;) {
2091 <            int ps = plock, m, n;
2092 <            if ((ws = workQueues) == null || (m = ws.length - 1) < 1)
2093 <                return null;
2094 <            for (int j = (m + 1) << 2; ;) {
2095 <                WorkQueue q = ws[(((r + j) << 1) | 1) & m];
2096 <                if (q != null && (n = q.base - q.top) < 0) {
2136 <                    if (n < -1)
2137 <                        signalWork(q);
2138 <                    return q;
2139 <                }
2140 <                else if (--j < 0) {
2141 <                    if (plock == ps)
2142 <                        return null;
2143 <                    break;
2090 >        for (;;) {
2091 >            int ps = plock, m; WorkQueue[] ws; WorkQueue q;
2092 >            if ((ws = workQueues) != null && (m = ws.length - 1) >= 0) {
2093 >                for (int j = (m + 1) << 2; j >= 0; --j) {
2094 >                    if ((q = ws[(((r + j) << 1) | 1) & m]) != null &&
2095 >                        q.base - q.top < 0)
2096 >                        return q;
2097                  }
2098              }
2099 +            if (plock == ps)
2100 +                return null;
2101          }
2102      }
2103  
# Line 2154 | Line 2109 | public class ForkJoinPool extends Abstra
2109       */
2110      final void helpQuiescePool(WorkQueue w) {
2111          for (boolean active = true;;) {
2112 <            ForkJoinTask<?> localTask; // exhaust local queue
2113 <            while ((localTask = w.nextLocalTask()) != null)
2114 <                localTask.doExec();
2115 <            // Similar to loop in scan(), but ignoring submissions
2116 <            WorkQueue q = findNonEmptyStealQueue(w.nextSeed());
2117 <            if (q != null) {
2118 <                ForkJoinTask<?> t; int b;
2112 >            long c; WorkQueue q; ForkJoinTask<?> t; int b;
2113 >            while ((t = w.nextLocalTask()) != null) {
2114 >                if (w.base - w.top < 0)
2115 >                    signalWork(w);
2116 >                t.doExec();
2117 >            }
2118 >            if ((q = findNonEmptyStealQueue(w.nextSeed())) != null) {
2119                  if (!active) {      // re-establish active count
2165                    long c;
2120                      active = true;
2121                      do {} while (!U.compareAndSwapLong
2122                                   (this, CTL, c = ctl, c + AC_UNIT));
2123                  }
2124 <                if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null)
2124 >                if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null) {
2125 >                    if (q.base - q.top < 0)
2126 >                        signalWork(q);
2127                      w.runSubtask(t);
2128 +                }
2129              }
2130 <            else {
2131 <                long c;
2132 <                if (active) {       // decrement active count without queuing
2130 >            else if (active) {       // decrement active count without queuing
2131 >                long nc = (c = ctl) - AC_UNIT;
2132 >                if ((int)(nc >> AC_SHIFT) + (config & SMASK) == 0)
2133 >                    return;          // bypass decrement-then-increment
2134 >                if (U.compareAndSwapLong(this, CTL, c, nc))
2135                      active = false;
2177                    do {} while (!U.compareAndSwapLong
2178                                 (this, CTL, c = ctl, c -= AC_UNIT));
2179                }
2180                else
2181                    c = ctl;        // re-increment on exit
2182                if ((int)(c >> AC_SHIFT) + (config & SMASK) == 0) {
2183                    do {} while (!U.compareAndSwapLong
2184                                 (this, CTL, c = ctl, c + AC_UNIT));
2185                    break;
2186                }
2136              }
2137 +            else if ((int)((c = ctl) >> AC_SHIFT) + (config & SMASK) == 0 &&
2138 +                     U.compareAndSwapLong(this, CTL, c, c + AC_UNIT))
2139 +                return;
2140          }
2141      }
2142  
# Line 2200 | Line 2152 | public class ForkJoinPool extends Abstra
2152                  return t;
2153              if ((q = findNonEmptyStealQueue(w.nextSeed())) == null)
2154                  return null;
2155 <            if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null)
2155 >            if ((b = q.base) - q.top < 0 && (t = q.pollAt(b)) != null) {
2156 >                if (q.base - q.top < 0)
2157 >                    signalWork(q);
2158                  return t;
2159 +            }
2160          }
2161      }
2162  
# Line 2283 | Line 2238 | public class ForkJoinPool extends Abstra
2238       * @return true if now terminating or terminated
2239       */
2240      private boolean tryTerminate(boolean now, boolean enable) {
2241 <        if (this == commonPool)                     // cannot shut down
2241 >        int ps;
2242 >        if (this == common)                    // cannot shut down
2243              return false;
2244 +        if ((ps = plock) >= 0) {                   // enable by setting plock
2245 +            if (!enable)
2246 +                return false;
2247 +            if ((ps & PL_LOCK) != 0 ||
2248 +                !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
2249 +                ps = acquirePlock();
2250 +            int nps = ((ps + PL_LOCK) & ~SHUTDOWN) | SHUTDOWN;
2251 +            if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
2252 +                releasePlock(nps);
2253 +        }
2254          for (long c;;) {
2255 <            if (((c = ctl) & STOP_BIT) != 0) {      // already terminating
2255 >            if (((c = ctl) & STOP_BIT) != 0) {     // already terminating
2256                  if ((short)(c >>> TC_SHIFT) == -(config & SMASK)) {
2257                      synchronized (this) {
2258 <                        notifyAll();                // signal when 0 workers
2258 >                        notifyAll();               // signal when 0 workers
2259                      }
2260                  }
2261                  return true;
2262              }
2263 <            if (plock >= 0) {                       // not yet enabled
2264 <                int ps;
2265 <                if (!enable)
2263 >            if (!now) {                            // check if idle & no tasks
2264 >                WorkQueue[] ws; WorkQueue w;
2265 >                if ((int)(c >> AC_SHIFT) != -(config & SMASK))
2266                      return false;
2267 <                if (((ps = plock) & PL_LOCK) != 0 ||
2268 <                    !U.compareAndSwapInt(this, PLOCK, ps, ps += PL_LOCK))
2269 <                    ps = acquirePlock();
2270 <                int nps = SHUTDOWN;
2271 <                if (!U.compareAndSwapInt(this, PLOCK, ps, nps))
2272 <                    releasePlock(nps);
2273 <            }
2274 <            if (!now) {                             // check if idle & no tasks
2275 <                if ((int)(c >> AC_SHIFT) != -(config & SMASK) ||
2276 <                    hasQueuedSubmissions())
2311 <                    return false;
2312 <                // Check for unqueued inactive workers. One pass suffices.
2313 <                WorkQueue[] ws = workQueues; WorkQueue w;
2314 <                if (ws != null) {
2315 <                    for (int i = 1; i < ws.length; i += 2) {
2316 <                        if ((w = ws[i]) != null && w.eventCount >= 0)
2317 <                            return false;
2267 >                if ((ws = workQueues) != null) {
2268 >                    for (int i = 0; i < ws.length; ++i) {
2269 >                        if ((w = ws[i]) != null) {
2270 >                            if (!w.isEmpty()) {    // signal unprocessed tasks
2271 >                                signalWork(w);
2272 >                                return false;
2273 >                            }
2274 >                            if ((i & 1) != 0 && w.eventCount >= 0)
2275 >                                return false;      // unqueued inactive worker
2276 >                        }
2277                      }
2278                  }
2279              }
2280              if (U.compareAndSwapLong(this, CTL, c, c | STOP_BIT)) {
2281                  for (int pass = 0; pass < 3; ++pass) {
2282 <                    WorkQueue[] ws = workQueues;
2283 <                    if (ws != null) {
2325 <                        WorkQueue w;
2282 >                    WorkQueue[] ws; WorkQueue w; Thread wt;
2283 >                    if ((ws = workQueues) != null) {
2284                          int n = ws.length;
2285                          for (int i = 0; i < n; ++i) {
2286                              if ((w = ws[i]) != null) {
2287                                  w.qlock = -1;
2288                                  if (pass > 0) {
2289                                      w.cancelAll();
2290 <                                    if (pass > 1)
2291 <                                        w.interruptOwner();
2290 >                                    if (pass > 1 && (wt = w.owner) != null) {
2291 >                                        if (!wt.isInterrupted()) {
2292 >                                            try {
2293 >                                                wt.interrupt();
2294 >                                            } catch (Throwable ignore) {
2295 >                                            }
2296 >                                        }
2297 >                                        U.unpark(wt);
2298 >                                    }
2299                                  }
2300                              }
2301                          }
2302                          // Wake up workers parked on event queue
2303                          int i, e; long cc; Thread p;
2304                          while ((e = (int)(cc = ctl) & E_MASK) != 0 &&
2305 <                               (i = e & SMASK) < n &&
2305 >                               (i = e & SMASK) < n && i >= 0 &&
2306                                 (w = ws[i]) != null) {
2307                              long nc = ((long)(w.nextWait & E_MASK) |
2308                                         ((cc + AC_UNIT) & AC_MASK) |
# Line 2365 | Line 2330 | public class ForkJoinPool extends Abstra
2330      static WorkQueue commonSubmitterQueue() {
2331          ForkJoinPool p; WorkQueue[] ws; int m; Submitter z;
2332          return ((z = submitters.get()) != null &&
2333 <                (p = commonPool) != null &&
2333 >                (p = common) != null &&
2334                  (ws = p.workQueues) != null &&
2335                  (m = ws.length - 1) >= 0) ?
2336              ws[m & z.seed & SQMASK] : null;
# Line 2379 | Line 2344 | public class ForkJoinPool extends Abstra
2344          ForkJoinTask<?>[] a;  int m, s;
2345          if (t != null &&
2346              (z = submitters.get()) != null &&
2347 <            (p = commonPool) != null &&
2347 >            (p = common) != null &&
2348              (ws = p.workQueues) != null &&
2349              (m = ws.length - 1) >= 0 &&
2350              (q = ws[m & z.seed & SQMASK]) != null &&
# Line 2456 | Line 2421 | public class ForkJoinPool extends Abstra
2421          ForkJoinTask<?>[] a;  int m, s, n;
2422          if (t != null &&
2423              (z = submitters.get()) != null &&
2424 <            (p = commonPool) != null &&
2424 >            (p = common) != null &&
2425              (ws = p.workQueues) != null &&
2426              (m = ws.length - 1) >= 0 &&
2427              (q = ws[m & z.seed & SQMASK]) != null &&
# Line 2485 | Line 2450 | public class ForkJoinPool extends Abstra
2450          }
2451      }
2452  
2488    /**
2489     * Restricted version of helpQuiescePool for external callers
2490     */
2491    static void externalHelpQuiescePool() {
2492        ForkJoinPool p; ForkJoinTask<?> t; WorkQueue q; int b;
2493        if ((p = commonPool) != null &&
2494            (q = p.findNonEmptyStealQueue(1)) != null &&
2495            (b = q.base) - q.top < 0 &&
2496            (t = q.pollAt(b)) != null)
2497            t.doExec();
2498    }
2499
2453      // Exported methods
2454  
2455      // Constructors
# Line 2595 | Line 2548 | public class ForkJoinPool extends Abstra
2548      }
2549  
2550      /**
2551 <     * Returns the common pool instance.
2551 >     * Returns the common pool instance. This pool is statically
2552 >     * constructed; its run state is unaffected by attempts to {@link
2553 >     * #shutdown} or {@link #shutdownNow}. However this pool and any
2554 >     * ongoing processing are automatically terminated upon program
2555 >     * {@link System#exit}.  Any program that relies on asynchronous
2556 >     * task processing to complete before program termination should
2557 >     * invoke {@code commonPool().}{@link #awaitQuiescence}, before
2558 >     * exit.
2559       *
2560       * @return the common pool instance
2561 +     * @since 1.8
2562       */
2563      public static ForkJoinPool commonPool() {
2564 <        // assert commonPool != null : "static init error";
2565 <        return commonPool;
2564 >        // assert common != null : "static init error";
2565 >        return common;
2566      }
2567  
2568      // Execution methods
# Line 2724 | Line 2685 | public class ForkJoinPool extends Abstra
2685          // In previous versions of this class, this method constructed
2686          // a task to run ForkJoinTask.invokeAll, but now external
2687          // invocation of multiple tasks is at least as efficient.
2688 <        List<ForkJoinTask<T>> fs = new ArrayList<ForkJoinTask<T>>(tasks.size());
2728 <        // Workaround needed because method wasn't declared with
2729 <        // wildcards in return type but should have been.
2730 <        @SuppressWarnings({"unchecked", "rawtypes"})
2731 <            List<Future<T>> futures = (List<Future<T>>) (List) fs;
2688 >        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
2689  
2690          boolean done = false;
2691          try {
2692              for (Callable<T> t : tasks) {
2693                  ForkJoinTask<T> f = new ForkJoinTask.AdaptedCallable<T>(t);
2694                  externalPush(f);
2695 <                fs.add(f);
2695 >                futures.add(f);
2696              }
2697 <            for (ForkJoinTask<T> f : fs)
2698 <                f.quietlyJoin();
2697 >            for (int i = 0, size = futures.size(); i < size; i++)
2698 >                ((ForkJoinTask<?>)futures.get(i)).quietlyJoin();
2699              done = true;
2700              return futures;
2701          } finally {
2702              if (!done)
2703 <                for (ForkJoinTask<T> f : fs)
2704 <                    f.cancel(false);
2703 >                for (int i = 0, size = futures.size(); i < size; i++)
2704 >                    futures.get(i).cancel(false);
2705          }
2706      }
2707  
# Line 2780 | Line 2737 | public class ForkJoinPool extends Abstra
2737       * Returns the targeted parallelism level of the common pool.
2738       *
2739       * @return the targeted parallelism level of the common pool
2740 +     * @since 1.8
2741       */
2742      public static int getCommonPoolParallelism() {
2743 <        return commonPoolParallelism;
2743 >        return commonParallelism;
2744      }
2745  
2746      /**
# Line 3040 | Line 2998 | public class ForkJoinPool extends Abstra
2998       * Possibly initiates an orderly shutdown in which previously
2999       * submitted tasks are executed, but no new tasks will be
3000       * accepted. Invocation has no effect on execution state if this
3001 <     * is the {@link #commonPool}, and no additional effect if
3001 >     * is the {@link #commonPool()}, and no additional effect if
3002       * already shut down.  Tasks that are in the process of being
3003       * submitted concurrently during the course of this method may or
3004       * may not be rejected.
# Line 3058 | Line 3016 | public class ForkJoinPool extends Abstra
3016      /**
3017       * Possibly attempts to cancel and/or stop all tasks, and reject
3018       * all subsequently submitted tasks.  Invocation has no effect on
3019 <     * execution state if this is the {@link #commonPool}, and no
3019 >     * execution state if this is the {@link #commonPool()}, and no
3020       * additional effect if already shut down. Otherwise, tasks that
3021       * are in the process of being submitted or executed concurrently
3022       * during the course of this method may or may not be
# Line 3121 | Line 3079 | public class ForkJoinPool extends Abstra
3079      /**
3080       * Blocks until all tasks have completed execution after a
3081       * shutdown request, or the timeout occurs, or the current thread
3082 <     * is interrupted, whichever happens first. Note that the {@link
3083 <     * #commonPool()} never terminates until program shutdown so
3084 <     * this method will always time out.
3082 >     * is interrupted, whichever happens first. Because the {@link
3083 >     * #commonPool()} never terminates until program shutdown, when
3084 >     * applied to the common pool, this method is equivalent to {@link
3085 >     * #awaitQuiescence} but always returns {@code false}.
3086       *
3087       * @param timeout the maximum time to wait
3088       * @param unit the time unit of the timeout argument
# Line 3133 | Line 3092 | public class ForkJoinPool extends Abstra
3092       */
3093      public boolean awaitTermination(long timeout, TimeUnit unit)
3094          throws InterruptedException {
3095 +        if (Thread.interrupted())
3096 +            throw new InterruptedException();
3097 +        if (this == common) {
3098 +            awaitQuiescence(timeout, unit);
3099 +            return false;
3100 +        }
3101          long nanos = unit.toNanos(timeout);
3102          if (isTerminated())
3103              return true;
# Line 3152 | Line 3117 | public class ForkJoinPool extends Abstra
3117      }
3118  
3119      /**
3120 +     * If called by a ForkJoinTask operating in this pool, equivalent
3121 +     * in effect to {@link ForkJoinTask#helpQuiesce}. Otherwise,
3122 +     * waits and/or attempts to assist performing tasks until this
3123 +     * pool {@link #isQuiescent} or the indicated timeout elapses.
3124 +     *
3125 +     * @param timeout the maximum time to wait
3126 +     * @param unit the time unit of the timeout argument
3127 +     * @return {@code true} if quiescent; {@code false} if the
3128 +     * timeout elapsed.
3129 +     */
3130 +    public boolean awaitQuiescence(long timeout, TimeUnit unit) {
3131 +        long nanos = unit.toNanos(timeout);
3132 +        ForkJoinWorkerThread wt;
3133 +        Thread thread = Thread.currentThread();
3134 +        if ((thread instanceof ForkJoinWorkerThread) &&
3135 +            (wt = (ForkJoinWorkerThread)thread).pool == this) {
3136 +            helpQuiescePool(wt.workQueue);
3137 +            return true;
3138 +        }
3139 +        long startTime = System.nanoTime();
3140 +        WorkQueue[] ws;
3141 +        int r = 0, m;
3142 +        boolean found = true;
3143 +        while (!isQuiescent() && (ws = workQueues) != null &&
3144 +               (m = ws.length - 1) >= 0) {
3145 +            if (!found) {
3146 +                if ((System.nanoTime() - startTime) > nanos)
3147 +                    return false;
3148 +                Thread.yield(); // cannot block
3149 +            }
3150 +            found = false;
3151 +            for (int j = (m + 1) << 2; j >= 0; --j) {
3152 +                ForkJoinTask<?> t; WorkQueue q; int b;
3153 +                if ((q = ws[r++ & m]) != null && (b = q.base) - q.top < 0) {
3154 +                    found = true;
3155 +                    if ((t = q.pollAt(b)) != null) {
3156 +                        if (q.base - q.top < 0)
3157 +                            signalWork(q);
3158 +                        t.doExec();
3159 +                    }
3160 +                    break;
3161 +                }
3162 +            }
3163 +        }
3164 +        return true;
3165 +    }
3166 +
3167 +    /**
3168 +     * Waits and/or attempts to assist performing tasks indefinitely
3169 +     * until the {@link #commonPool()} {@link #isQuiescent}
3170 +     */
3171 +    static void quiesceCommonPool() {
3172 +        common.awaitQuiescence(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
3173 +    }
3174 +
3175 +    /**
3176       * Interface for extending managed parallelism for tasks running
3177       * in {@link ForkJoinPool}s.
3178       *
# Line 3304 | Line 3325 | public class ForkJoinPool extends Abstra
3325      private static final long QLOCK;
3326  
3327      static {
3328 <        int s; // initialize field offsets for CAS etc
3328 >        // initialize field offsets for CAS etc
3329          try {
3330              U = getUnsafe();
3331              Class<?> k = ForkJoinPool.class;
# Line 3324 | Line 3345 | public class ForkJoinPool extends Abstra
3345                  (wk.getDeclaredField("qlock"));
3346              Class<?> ak = ForkJoinTask[].class;
3347              ABASE = U.arrayBaseOffset(ak);
3348 <            s = U.arrayIndexScale(ak);
3349 <            ASHIFT = 31 - Integer.numberOfLeadingZeros(s);
3348 >            int scale = U.arrayIndexScale(ak);
3349 >            if ((scale & (scale - 1)) != 0)
3350 >                throw new Error("data type scale not a power of two");
3351 >            ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
3352          } catch (Exception e) {
3353              throw new Error(e);
3354          }
3332        if ((s & (s-1)) != 0)
3333            throw new Error("data type scale not a power of two");
3355  
3356          submitters = new ThreadLocal<Submitter>();
3357          ForkJoinWorkerThreadFactory fac = defaultForkJoinWorkerThreadFactory =
# Line 3367 | Line 3388 | public class ForkJoinPool extends Abstra
3388              par = Runtime.getRuntime().availableProcessors();
3389          if (par > MAX_CAP)
3390              par = MAX_CAP;
3391 <        commonPoolParallelism = par;
3391 >        commonParallelism = par;
3392          long np = (long)(-par); // precompute initial ctl value
3393          long ct = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK);
3394  
3395 <        commonPool = new ForkJoinPool(par, ct, fac, handler);
3395 >        common = new ForkJoinPool(par, ct, fac, handler);
3396      }
3397  
3398      /**
# Line 3384 | Line 3405 | public class ForkJoinPool extends Abstra
3405      private static sun.misc.Unsafe getUnsafe() {
3406          try {
3407              return sun.misc.Unsafe.getUnsafe();
3408 <        } catch (SecurityException se) {
3409 <            try {
3410 <                return java.security.AccessController.doPrivileged
3411 <                    (new java.security
3412 <                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
3413 <                        public sun.misc.Unsafe run() throws Exception {
3414 <                            java.lang.reflect.Field f = sun.misc
3415 <                                .Unsafe.class.getDeclaredField("theUnsafe");
3416 <                            f.setAccessible(true);
3417 <                            return (sun.misc.Unsafe) f.get(null);
3418 <                        }});
3419 <            } catch (java.security.PrivilegedActionException e) {
3420 <                throw new RuntimeException("Could not initialize intrinsics",
3421 <                                           e.getCause());
3422 <            }
3408 >        } catch (SecurityException tryReflectionInstead) {}
3409 >        try {
3410 >            return java.security.AccessController.doPrivileged
3411 >            (new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
3412 >                public sun.misc.Unsafe run() throws Exception {
3413 >                    Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class;
3414 >                    for (java.lang.reflect.Field f : k.getDeclaredFields()) {
3415 >                        f.setAccessible(true);
3416 >                        Object x = f.get(null);
3417 >                        if (k.isInstance(x))
3418 >                            return k.cast(x);
3419 >                    }
3420 >                    throw new NoSuchFieldError("the Unsafe");
3421 >                }});
3422 >        } catch (java.security.PrivilegedActionException e) {
3423 >            throw new RuntimeException("Could not initialize intrinsics",
3424 >                                       e.getCause());
3425          }
3426      }
3404
3427   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines