sync up ForkJoinPool from jsr166 CVS

Author: AlanBateman

src/java.base/share/classes/java/util/concurrent/ForkJoinPool.java

@@ -792,25 +792,25 @@ static final class DefaultForkJoinWorkerThreadFactory
         private static final AccessControlContext ACC = contextWithPermissions(
             new RuntimePermission("getClassLoader"),
             new RuntimePermission("setContextClassLoader"));
-
         public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
             return AccessController.doPrivileged(
                 new PrivilegedAction<>() {
                     public ForkJoinWorkerThread run() {
-                        return new ForkJoinWorkerThread(null, pool, true, true);
+                        return new ForkJoinWorkerThread(null, pool, true, false);
                     }},
                 ACC);
         }
     }
 
     /**
-     * Factory for InnocuousForkJoinWorkerThread. Support requires
-     * that we break quite a lot of encapsulation (some via helper
-     * methods in ThreadLocalRandom) to access and set Thread fields.
+     * Factory for CommonPool unless overridded by System
+     * property. Creates InnocuousForkJoinWorkerThreads if a security
+     * manager is present at time of invocation. Support requires that
+     * we break quite a lot of encapsulation (some via helper methods
+     * in ThreadLocalRandom) to access and set Thread fields.
      */
-    static final class InnocuousForkJoinWorkerThreadFactory
+    static final class DefaultCommonPoolForkJoinWorkerThreadFactory
         implements ForkJoinWorkerThreadFactory {
-        // ACC for access to the factory
         private static final AccessControlContext ACC = contextWithPermissions(
             modifyThreadPermission,
             new RuntimePermission("enableContextClassLoaderOverride"),
@@ -820,11 +820,13 @@ static final class InnocuousForkJoinWorkerThreadFactory
 
         public final ForkJoinWorkerThread newThread(ForkJoinPool pool) {
             return AccessController.doPrivileged(
-                new PrivilegedAction<>() {
-                    public ForkJoinWorkerThread run() {
-                        return new ForkJoinWorkerThread.
-                            InnocuousForkJoinWorkerThread(pool); }},
-                ACC);
+                 new PrivilegedAction<>() {
+                     public ForkJoinWorkerThread run() {
+                         return System.getSecurityManager() == null ?
+                             new ForkJoinWorkerThread(null, pool, true, true):
+                             new ForkJoinWorkerThread.
+                             InnocuousForkJoinWorkerThread(pool); }},
+                 ACC);
         }
     }
 
@@ -1689,15 +1691,18 @@ else if (mode < 0)
                 return -1;
             else if ((int)(ctl >> RC_SHIFT) > ac)
                 Thread.onSpinWait();         // signal in progress
-            else if (!(alt = !alt)) {        // check between park calls
-                if (!Thread.interrupted() && deadline != 0L &&
-                    deadline - System.currentTimeMillis() <= TIMEOUT_SLOP &&
-                    compareAndSetCtl(c, ((UC_MASK & (c - TC_UNIT)) |
-                                         (w.stackPred & SP_MASK)))) {
+            else if (deadline != 0L &&
+                     deadline - System.currentTimeMillis() <= TIMEOUT_SLOP) {
+                if (c != (c = ctl))          // ensure consistent
+                    ac = (int)(c >> RC_SHIFT);
+                else if (compareAndSetCtl(c, ((UC_MASK & (c - TC_UNIT)) |
+                                              (w.stackPred & SP_MASK)))) {
                     w.phase = QUIET;
                     return -1;               // drop on timeout
                 }
             }
+            else if (!(alt = !alt))    // check between park calls
+                Thread.interrupted();
             else if (deadline != 0L)
                 LockSupport.parkUntil(deadline);
             else
@@ -1710,11 +1715,19 @@ else if (deadline != 0L)
     // Utilities used by ForkJoinTask
 
     /**
-     * Returns true if all workers are busy
+     * Returns true if all workers are busy, possibly creating one if allowed
      */
     final boolean isSaturated() {
-        long c;
-        return (int)((c = ctl) >> RC_SHIFT) >= 0 && ((int)c & ~UNSIGNALLED) == 0;
+        int maxTotal = bounds >>> SWIDTH;
+        for (long c;;) {
+            if (((int)(c = ctl) & ~UNSIGNALLED) != 0)
+                return false;
+            if ((short)(c >>> TC_SHIFT) >= maxTotal)
+                return true;
+            long nc = ((c + TC_UNIT) & TC_MASK) | (c & ~TC_MASK);
+            if (compareAndSetCtl(c, nc))
+                return !createWorker();
+        }
     }
 
     /**
@@ -2534,9 +2547,7 @@ private ForkJoinPool(byte forCommonPoolOnly) {
         int p = this.mode = Math.min(Math.max(parallelism, 0), MAX_CAP);
         int size = 1 << (33 - Integer.numberOfLeadingZeros(p > 0 ? p - 1 : 1));
         this.factory = (fac != null) ? fac :
-            (System.getSecurityManager() == null ?
-             defaultForkJoinWorkerThreadFactory :
-             new InnocuousForkJoinWorkerThreadFactory());
+            new DefaultCommonPoolForkJoinWorkerThreadFactory();
         this.ueh = handler;
         this.keepAlive = DEFAULT_KEEPALIVE;
         this.saturate = null;
@@ -2676,7 +2687,10 @@ public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
                 ForkJoinTask<T> f =
                     new ForkJoinTask.AdaptedInterruptibleCallable<T>(t);
                 futures.add(f);
-                externalSubmit(f);
+                if (isSaturated())
+                    f.doExec();
+                else
+                    externalSubmit(f);
             }
             for (int i = futures.size() - 1; i >= 0; --i)
                 ((ForkJoinTask<?>)futures.get(i)).quietlyJoin();
@@ -2699,7 +2713,10 @@ public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                 ForkJoinTask<T> f =
                     new ForkJoinTask.AdaptedInterruptibleCallable<T>(t);
                 futures.add(f);
-                externalSubmit(f);
+                if (isSaturated())
+                    f.doExec();
+                else
+                    externalSubmit(f);
             }
             long startTime = System.nanoTime(), ns = nanos;
             boolean timedOut = (ns < 0L);
@@ -2735,14 +2752,22 @@ static final class InvokeAnyRoot<E> extends ForkJoinTask<E> {
         final AtomicInteger count;              // in case all throw
         InvokeAnyRoot(int n) { count = new AtomicInteger(n); }
         final void tryComplete(Callable<E> c) { // called by InvokeAnyTasks
-            if (c != null && !isDone()) {       // raciness OK
-                try {
-                    complete(c.call());
-                } catch (Throwable ex) {
-                    if (count.getAndDecrement() <= 1)
-                        trySetThrown(ex);
+            Throwable ex = null;
+            boolean failed = false;
+            if (c != null) {                    // raciness OK
+                if (isCancelled())
+                    failed = true;
+                else if (!isDone()) {
+                    try {
+                        complete(c.call());
+                    } catch (Throwable tx) {
+                        ex = tx;
+                        failed = true;
+                    }
                 }
             }
+            if (failed && count.getAndDecrement() <= 1)
+                trySetThrown(ex != null ? ex : new CancellationException());
         }
         public final boolean exec()         { return false; } // never forked
         public final E getRawResult()       { return result; }