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.168 by dl, Tue Jan 1 15:10:39 2013 UTC vs.
Revision 1.189 by jsr166, Sat Sep 12 19:16:45 2015 UTC

# Line 687 | Line 687 | public class ForkJoinPool extends Abstra
687              return (n >= 0) ? 0 : -n; // ignore transient negative
688          }
689  
690 <       /**
690 >        /**
691           * Provides a more accurate estimate of whether this queue has
692           * any tasks than does queueSize, by checking whether a
693           * near-empty queue has at least one unclaimed task.
# Line 708 | Line 708 | public class ForkJoinPool extends Abstra
708           * shared-queue version is embedded in method externalPush.)
709           *
710           * @param task the task. Caller must ensure non-null.
711 <         * @throw RejectedExecutionException if array cannot be resized
711 >         * @throws RejectedExecutionException if array cannot be resized
712           */
713          final void push(ForkJoinTask<?> task) {
714              ForkJoinTask<?>[] a; ForkJoinPool p;
# Line 725 | Line 725 | public class ForkJoinPool extends Abstra
725              }
726          }
727  
728 <       /**
728 >        /**
729           * Initializes or doubles the capacity of array. Call either
730           * by owner or with lock held -- it is OK for base, but not
731           * top, to move while resizings are in progress.
# Line 907 | Line 907 | public class ForkJoinPool extends Abstra
907           * or any other cancelled task. Returns (true) on any CAS
908           * or consistency check failure so caller can retry.
909           *
910 <         * @return false if no progress can be made, else true;
910 >         * @return false if no progress can be made, else true
911           */
912          final boolean tryRemoveAndExec(ForkJoinTask<?> task) {
913              boolean stat = true, removed = false, empty = true;
# Line 952 | Line 952 | public class ForkJoinPool extends Abstra
952  
953          /**
954           * Polls for and executes the given task or any other task in
955 <         * its CountedCompleter computation
955 >         * its CountedCompleter computation.
956           */
957          final boolean pollAndExecCC(ForkJoinTask<?> root) {
958              ForkJoinTask<?>[] a; int b; Object o;
# Line 1026 | Line 1026 | public class ForkJoinPool extends Abstra
1026          private static final int ABASE;
1027          private static final int ASHIFT;
1028          static {
1029 –            int s;
1029              try {
1030                  U = getUnsafe();
1031                  Class<?> k = WorkQueue.class;
# Line 1034 | 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              }
1041 –            if ((s & (s-1)) != 0)
1042 –                throw new Error("data type scale not a power of two");
1043 –            ASHIFT = 31 - Integer.numberOfLeadingZeros(s);
1043          }
1044      }
1045  
# Line 1132 | Line 1131 | public class ForkJoinPool extends Abstra
1131       */
1132      private static final int SEED_INCREMENT = 0x61c88647;
1133  
1134 <    /**
1134 >    /*
1135       * Bits and masks for control variables
1136       *
1137       * Field ctl is a long packed with:
# Line 1245 | Line 1244 | public class ForkJoinPool extends Abstra
1244      volatile Object pad10, pad11, pad12, pad13, pad14, pad15, pad16, pad17;
1245      volatile Object pad18, pad19, pad1a, pad1b;
1246  
1247 <    /*
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.
# Line 1522 | Line 1521 | public class ForkJoinPool extends Abstra
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;
1524 >                n |= n >>> 1;
1525 >                n |= n >>> 2;
1526 >                n |= n >>> 4;
1527 >                n |= n >>> 8;
1528 >                n |= n >>> 16;
1529 >                n = (n + 1) << 1;
1530                  WorkQueue[] nws = ((ws = workQueues) == null || ws.length == 0 ?
1531                                     new WorkQueue[n] : null);
1532                  if (((ps = plock) & PL_LOCK) != 0 ||
# Line 1932 | Line 1935 | public class ForkJoinPool extends Abstra
1935       * @param task the task to join
1936       * @param mode if shared, exit upon completing any task
1937       * if all workers are active
1935 –     *
1938       */
1939      private int helpComplete(ForkJoinTask<?> task, int mode) {
1940          WorkQueue[] ws; WorkQueue q; int m, n, s, u;
# Line 2129 | Line 2131 | public class ForkJoinPool extends Abstra
2131                      w.runSubtask(t);
2132                  }
2133              }
2134 <            else if (active) {       // decrement active count without queuing
2134 >            else if (active) {      // decrement active count without queuing
2135                  long nc = (c = ctl) - AC_UNIT;
2136                  if ((int)(nc >> AC_SHIFT) + (config & SMASK) == 0)
2137 <                    return;          // bypass decrement-then-increment
2137 >                    return;         // bypass decrement-then-increment
2138                  if (U.compareAndSwapLong(this, CTL, c, nc))
2139                      active = false;
2140              }
# Line 2187 | Line 2189 | public class ForkJoinPool extends Abstra
2189       * producing extra tasks amortizes the uncertainty of progress and
2190       * diffusion assumptions.
2191       *
2192 <     * So, users will want to use values larger, but not much larger
2192 >     * So, users will want to use values larger (but not much larger)
2193       * than 1 to both smooth over transient shortages and hedge
2194       * against uneven progress; as traded off against the cost of
2195       * extra task overhead. We leave the user to pick a threshold
# Line 2468 | Line 2470 | public class ForkJoinPool extends Abstra
2470       *         java.lang.RuntimePermission}{@code ("modifyThread")}
2471       */
2472      public ForkJoinPool() {
2473 <        this(Runtime.getRuntime().availableProcessors(),
2473 >        this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
2474               defaultForkJoinWorkerThreadFactory, null, false);
2475      }
2476  
# Line 2556 | Line 2558 | public class ForkJoinPool extends Abstra
2558       * ongoing processing are automatically terminated upon program
2559       * {@link System#exit}.  Any program that relies on asynchronous
2560       * task processing to complete before program termination should
2561 <     * invoke {@link #quiesceCommonPool}, or the timeout-based {@code
2562 <     * commonPool().}{@link #awaitQuiescence}, before exit.
2561 >     * invoke {@code commonPool().}{@link #awaitQuiescence}, before
2562 >     * exit.
2563       *
2564       * @return the common pool instance
2565 +     * @since 1.8
2566       */
2567      public static ForkJoinPool commonPool() {
2568          // assert common != null : "static init error";
# Line 2686 | Line 2689 | public class ForkJoinPool extends Abstra
2689          // In previous versions of this class, this method constructed
2690          // a task to run ForkJoinTask.invokeAll, but now external
2691          // invocation of multiple tasks is at least as efficient.
2692 <        List<ForkJoinTask<T>> fs = new ArrayList<ForkJoinTask<T>>(tasks.size());
2690 <        // Workaround needed because method wasn't declared with
2691 <        // wildcards in return type but should have been.
2692 <        @SuppressWarnings({"unchecked", "rawtypes"})
2693 <            List<Future<T>> futures = (List<Future<T>>) (List) fs;
2692 >        ArrayList<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
2693  
2694          boolean done = false;
2695          try {
2696              for (Callable<T> t : tasks) {
2697                  ForkJoinTask<T> f = new ForkJoinTask.AdaptedCallable<T>(t);
2698 +                futures.add(f);
2699                  externalPush(f);
2700 –                fs.add(f);
2700              }
2701 <            for (ForkJoinTask<T> f : fs)
2702 <                f.quietlyJoin();
2701 >            for (int i = 0, size = futures.size(); i < size; i++)
2702 >                ((ForkJoinTask<?>)futures.get(i)).quietlyJoin();
2703              done = true;
2704              return futures;
2705          } finally {
2706              if (!done)
2707 <                for (ForkJoinTask<T> f : fs)
2708 <                    f.cancel(false);
2707 >                for (int i = 0, size = futures.size(); i < size; i++)
2708 >                    futures.get(i).cancel(false);
2709          }
2710      }
2711  
# Line 2742 | Line 2741 | public class ForkJoinPool extends Abstra
2741       * Returns the targeted parallelism level of the common pool.
2742       *
2743       * @return the targeted parallelism level of the common pool
2744 +     * @since 1.8
2745       */
2746      public static int getCommonPoolParallelism() {
2747          return commonParallelism;
# Line 3002 | Line 3002 | public class ForkJoinPool extends Abstra
3002       * Possibly initiates an orderly shutdown in which previously
3003       * submitted tasks are executed, but no new tasks will be
3004       * accepted. Invocation has no effect on execution state if this
3005 <     * is the {@link #commonPool}, and no additional effect if
3005 >     * is the {@link #commonPool()}, and no additional effect if
3006       * already shut down.  Tasks that are in the process of being
3007       * submitted concurrently during the course of this method may or
3008       * may not be rejected.
# Line 3020 | Line 3020 | public class ForkJoinPool extends Abstra
3020      /**
3021       * Possibly attempts to cancel and/or stop all tasks, and reject
3022       * all subsequently submitted tasks.  Invocation has no effect on
3023 <     * execution state if this is the {@link #commonPool}, and no
3023 >     * execution state if this is the {@link #commonPool()}, and no
3024       * additional effect if already shut down. Otherwise, tasks that
3025       * are in the process of being submitted or executed concurrently
3026       * during the course of this method may or may not be
# Line 3170 | Line 3170 | public class ForkJoinPool extends Abstra
3170  
3171      /**
3172       * Waits and/or attempts to assist performing tasks indefinitely
3173 <     * until the {@link #commonPool()} {@link #isQuiescent}
3173 >     * until the {@link #commonPool()} {@link #isQuiescent}.
3174       */
3175 <    public static void quiesceCommonPool() {
3175 >    static void quiesceCommonPool() {
3176          common.awaitQuiescence(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
3177      }
3178  
# Line 3329 | Line 3329 | public class ForkJoinPool extends Abstra
3329      private static final long QLOCK;
3330  
3331      static {
3332 <        int s; // initialize field offsets for CAS etc
3332 >        // initialize field offsets for CAS etc
3333          try {
3334              U = getUnsafe();
3335              Class<?> k = ForkJoinPool.class;
# Line 3349 | Line 3349 | public class ForkJoinPool extends Abstra
3349                  (wk.getDeclaredField("qlock"));
3350              Class<?> ak = ForkJoinTask[].class;
3351              ABASE = U.arrayBaseOffset(ak);
3352 <            s = U.arrayIndexScale(ak);
3353 <            ASHIFT = 31 - Integer.numberOfLeadingZeros(s);
3352 >            int scale = U.arrayIndexScale(ak);
3353 >            if ((scale & (scale - 1)) != 0)
3354 >                throw new Error("data type scale not a power of two");
3355 >            ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
3356          } catch (Exception e) {
3357              throw new Error(e);
3358          }
3357 –        if ((s & (s-1)) != 0)
3358 –            throw new Error("data type scale not a power of two");
3359  
3360          submitters = new ThreadLocal<Submitter>();
3361          ForkJoinWorkerThreadFactory fac = defaultForkJoinWorkerThreadFactory =
# Line 3409 | Line 3409 | public class ForkJoinPool extends Abstra
3409      private static sun.misc.Unsafe getUnsafe() {
3410          try {
3411              return sun.misc.Unsafe.getUnsafe();
3412 <        } catch (SecurityException se) {
3413 <            try {
3414 <                return java.security.AccessController.doPrivileged
3415 <                    (new java.security
3416 <                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
3417 <                        public sun.misc.Unsafe run() throws Exception {
3418 <                            java.lang.reflect.Field f = sun.misc
3419 <                                .Unsafe.class.getDeclaredField("theUnsafe");
3420 <                            f.setAccessible(true);
3421 <                            return (sun.misc.Unsafe) f.get(null);
3422 <                        }});
3423 <            } catch (java.security.PrivilegedActionException e) {
3424 <                throw new RuntimeException("Could not initialize intrinsics",
3425 <                                           e.getCause());
3426 <            }
3412 >        } catch (SecurityException tryReflectionInstead) {}
3413 >        try {
3414 >            return java.security.AccessController.doPrivileged
3415 >            (new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
3416 >                public sun.misc.Unsafe run() throws Exception {
3417 >                    Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class;
3418 >                    for (java.lang.reflect.Field f : k.getDeclaredFields()) {
3419 >                        f.setAccessible(true);
3420 >                        Object x = f.get(null);
3421 >                        if (k.isInstance(x))
3422 >                            return k.cast(x);
3423 >                    }
3424 >                    throw new NoSuchFieldError("the Unsafe");
3425 >                }});
3426 >        } catch (java.security.PrivilegedActionException e) {
3427 >            throw new RuntimeException("Could not initialize intrinsics",
3428 >                                       e.getCause());
3429          }
3430      }
3429 –
3431   }

Diff Legend

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