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

Comparing jsr166/src/jsr166e/ForkJoinTask.java (file contents):
Revision 1.3 by dl, Wed Oct 31 12:49:13 2012 UTC vs.
Revision 1.13 by dl, Wed Jun 19 14:55:40 2013 UTC

# Line 33 | Line 33 | import java.lang.reflect.Constructor;
33   * <p>A "main" {@code ForkJoinTask} begins execution when it is
34   * explicitly submitted to a {@link ForkJoinPool}, or, if not already
35   * engaged in a ForkJoin computation, commenced in the {@link
36 < * ForkJoinPool#commonPool} via {@link #fork}, {@link #invoke}, or
36 > * ForkJoinPool#commonPool()} via {@link #fork}, {@link #invoke}, or
37   * related methods.  Once started, it will usually in turn start other
38   * subtasks.  As indicated by the name of this class, many programs
39   * using {@code ForkJoinTask} employ only methods {@link #fork} and
# Line 55 | Line 55 | import java.lang.reflect.Constructor;
55   * minimize other blocking synchronization apart from joining other
56   * tasks or using synchronizers such as Phasers that are advertised to
57   * cooperate with fork/join scheduling. Subdividable tasks should also
58 < * not perform blocking IO, and should ideally access variables that
58 > * not perform blocking I/O, and should ideally access variables that
59   * are completely independent of those accessed by other running
60   * tasks. These guidelines are loosely enforced by not permitting
61   * checked exceptions such as {@code IOExceptions} to be
# Line 73 | Line 73 | import java.lang.reflect.Constructor;
73   * <p>It is possible to define and use ForkJoinTasks that may block,
74   * but doing do requires three further considerations: (1) Completion
75   * of few if any <em>other</em> tasks should be dependent on a task
76 < * that blocks on external synchronization or IO. Event-style async
76 > * that blocks on external synchronization or I/O. Event-style async
77   * tasks that are never joined (for example, those subclassing {@link
78   * CountedCompleter}) often fall into this category.  (2) To minimize
79   * resource impact, tasks should be small; ideally performing only the
# Line 136 | Line 136 | import java.lang.reflect.Constructor;
136   * supports other methods and techniques (for example the use of
137   * {@link Phaser}, {@link #helpQuiesce}, and {@link #complete}) that
138   * may be of use in constructing custom subclasses for problems that
139 < * are not statically structured as DAGs. To support such usages a
139 > * are not statically structured as DAGs. To support such usages, a
140   * ForkJoinTask may be atomically <em>tagged</em> with a {@code short}
141   * value using {@link #setForkJoinTaskTag} or {@link
142   * #compareAndSetForkJoinTaskTag} and checked using {@link
# Line 285 | Line 285 | public abstract class ForkJoinTask<V> im
285       */
286      private int externalAwaitDone() {
287          int s;
288 <        boolean interrupted = false;
289 <        if ((s = status) >= 0 && ForkJoinPool.tryUnsubmitFromCommonPool(this))
290 <            s = doExec();
291 <        while (s >= 0) {
292 <            if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
293 <                synchronized (this) {
294 <                    if (status >= 0) {
295 <                        try {
296 <                            wait();
297 <                        } catch (InterruptedException ie) {
298 <                            interrupted = true;
288 >        ForkJoinPool cp = ForkJoinPool.common;
289 >        if ((s = status) >= 0) {
290 >            if (cp != null) {
291 >                if (this instanceof CountedCompleter)
292 >                    s = cp.externalHelpComplete((CountedCompleter<?>)this);
293 >                else if (cp.tryExternalUnpush(this))
294 >                    s = doExec();
295 >            }
296 >            if (s >= 0 && (s = status) >= 0) {
297 >                boolean interrupted = false;
298 >                do {
299 >                    if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
300 >                        synchronized (this) {
301 >                            if (status >= 0) {
302 >                                try {
303 >                                    wait();
304 >                                } catch (InterruptedException ie) {
305 >                                    interrupted = true;
306 >                                }
307 >                            }
308 >                            else
309 >                                notifyAll();
310                          }
311                      }
312 <                    else
313 <                        notifyAll();
314 <                }
312 >                } while ((s = status) >= 0);
313 >                if (interrupted)
314 >                    Thread.currentThread().interrupt();
315              }
305            s = status;
316          }
307        if (interrupted)
308            Thread.currentThread().interrupt();
317          return s;
318      }
319  
# Line 313 | Line 321 | public abstract class ForkJoinTask<V> im
321       * Blocks a non-worker-thread until completion or interruption.
322       */
323      private int externalInterruptibleAwaitDone() throws InterruptedException {
324 +        int s;
325 +        ForkJoinPool cp = ForkJoinPool.common;
326          if (Thread.interrupted())
327              throw new InterruptedException();
328 <        int s;
329 <        if ((s = status) >= 0 && ForkJoinPool.tryUnsubmitFromCommonPool(this))
330 <            s = doExec();
331 <        while (s >= 0) {
328 >        if ((s = status) >= 0 && cp != null) {
329 >            if (this instanceof CountedCompleter)
330 >                cp.externalHelpComplete((CountedCompleter<?>)this);
331 >            else if (cp.tryExternalUnpush(this))
332 >                doExec();
333 >        }
334 >        while ((s = status) >= 0) {
335              if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
336                  synchronized (this) {
337                      if (status >= 0)
# Line 327 | Line 340 | public abstract class ForkJoinTask<V> im
340                          notifyAll();
341                  }
342              }
330            s = status;
343          }
344          return s;
345      }
346  
347 +
348      /**
349       * Implementation for join, get, quietlyJoin. Directly handles
350       * only cases of already-completed, external wait, and
# Line 438 | Line 451 | public abstract class ForkJoinTask<V> im
451      }
452  
453      /**
454 <     * Records exception and possibly propagates
454 >     * Records exception and possibly propagates.
455       *
456       * @return status on exit
457       */
# Line 471 | Line 484 | public abstract class ForkJoinTask<V> im
484      }
485  
486      /**
487 <     * Removes exception node and clears status
487 >     * Removes exception node and clears status.
488       */
489      private void clearExceptionalCompletion() {
490          int h = System.identityHashCode(this);
# Line 603 | Line 616 | public abstract class ForkJoinTask<V> im
616      /**
617       * A version of "sneaky throw" to relay exceptions
618       */
619 <    static void rethrow(final Throwable ex) {
620 <        if (ex != null) {
621 <            if (ex instanceof Error)
609 <                throw (Error)ex;
610 <            if (ex instanceof RuntimeException)
611 <                throw (RuntimeException)ex;
612 <            throw uncheckedThrowable(ex, RuntimeException.class);
613 <        }
619 >    static void rethrow(Throwable ex) {
620 >        if (ex != null)
621 >            ForkJoinTask.<RuntimeException>uncheckedThrow(ex);
622      }
623  
624      /**
# Line 619 | Line 627 | public abstract class ForkJoinTask<V> im
627       * unchecked exceptions
628       */
629      @SuppressWarnings("unchecked") static <T extends Throwable>
630 <        T uncheckedThrowable(final Throwable t, final Class<T> c) {
631 <        return (T)t; // rely on vacuous cast
630 >        void uncheckedThrow(Throwable t) throws T {
631 >        throw (T)t; // rely on vacuous cast
632      }
633  
634      /**
# Line 638 | Line 646 | public abstract class ForkJoinTask<V> im
646      /**
647       * Arranges to asynchronously execute this task in the pool the
648       * current task is running in, if applicable, or using the {@link
649 <     * ForkJoinPool#commonPool} if not {@link #inForkJoinPool}.  While
649 >     * ForkJoinPool#commonPool()} if not {@link #inForkJoinPool}.  While
650       * it is not necessarily enforced, it is a usage error to fork a
651       * task more than once unless it has completed and been
652       * reinitialized.  Subsequent modifications to the state of this
# Line 655 | Line 663 | public abstract class ForkJoinTask<V> im
663          if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
664              ((ForkJoinWorkerThread)t).workQueue.push(this);
665          else
666 <            ForkJoinPool.submitToCommonPool(this);
666 >            ForkJoinPool.common.externalPush(this);
667          return this;
668      }
669  
# Line 831 | Line 839 | public abstract class ForkJoinTask<V> im
839       * <p>This method is designed to be invoked by <em>other</em>
840       * tasks. To terminate the current task, you can just return or
841       * throw an unchecked exception from its computation method, or
842 <     * invoke {@link #completeExceptionally}.
842 >     * invoke {@link #completeExceptionally(Throwable)}.
843       *
844       * @param mayInterruptIfRunning this value has no effect in the
845       * default implementation because interrupts are not used to
# Line 981 | Line 989 | public abstract class ForkJoinTask<V> im
989          if (Thread.interrupted())
990              throw new InterruptedException();
991          // Messy in part because we measure in nanosecs, but wait in millisecs
992 <        int s; long ns, ms;
993 <        if ((s = status) >= 0 && (ns = unit.toNanos(timeout)) > 0L) {
992 >        int s; long ms;
993 >        long ns = unit.toNanos(timeout);
994 >        ForkJoinPool cp;
995 >        if ((s = status) >= 0 && ns > 0L) {
996              long deadline = System.nanoTime() + ns;
997              ForkJoinPool p = null;
998              ForkJoinPool.WorkQueue w = null;
# Line 991 | Line 1001 | public abstract class ForkJoinTask<V> im
1001                  ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
1002                  p = wt.pool;
1003                  w = wt.workQueue;
1004 <                s = p.helpJoinOnce(w, this); // no retries on failure
1004 >                p.helpJoinOnce(w, this); // no retries on failure
1005 >            }
1006 >            else if ((cp = ForkJoinPool.common) != null) {
1007 >                if (this instanceof CountedCompleter)
1008 >                    cp.externalHelpComplete((CountedCompleter<?>)this);
1009 >                else if (cp.tryExternalUnpush(this))
1010 >                    doExec();
1011              }
1012              boolean canBlock = false;
1013              boolean interrupted = false;
1014              try {
1015                  while ((s = status) >= 0) {
1016 <                    if (w != null && w.runState < 0)
1016 >                    if (w != null && w.qlock < 0)
1017                          cancelIgnoringExceptions(this);
1018                      else if (!canBlock) {
1019 <                        if (p == null || p.tryCompensate(this, null))
1019 >                        if (p == null || p.tryCompensate(p.ctl))
1020                              canBlock = true;
1021                      }
1022                      else {
# Line 1076 | Line 1092 | public abstract class ForkJoinTask<V> im
1092              wt.pool.helpQuiescePool(wt.workQueue);
1093          }
1094          else
1095 <            ForkJoinPool.externalHelpQuiescePool();
1095 >            ForkJoinPool.quiesceCommonPool();
1096      }
1097  
1098      /**
# Line 1139 | Line 1155 | public abstract class ForkJoinTask<V> im
1155       */
1156      public boolean tryUnfork() {
1157          Thread t;
1158 <        return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1159 <            ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) :
1160 <            ForkJoinPool.tryUnsubmitFromCommonPool(this);
1158 >        return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1159 >                ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) :
1160 >                ForkJoinPool.common.tryExternalUnpush(this));
1161      }
1162  
1163      /**
# Line 1153 | Line 1169 | public abstract class ForkJoinTask<V> im
1169       * @return the number of tasks
1170       */
1171      public static int getQueuedTaskCount() {
1172 <        Thread t;
1173 <        return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1174 <            ((ForkJoinWorkerThread)t).workQueue.queueSize() :
1175 <            ForkJoinPool.getEstimatedSubmitterQueueLength();
1172 >        Thread t; ForkJoinPool.WorkQueue q;
1173 >        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
1174 >            q = ((ForkJoinWorkerThread)t).workQueue;
1175 >        else
1176 >            q = ForkJoinPool.commonSubmitterQueue();
1177 >        return (q == null) ? 0 : q.queueSize();
1178      }
1179  
1180      /**
# Line 1173 | Line 1191 | public abstract class ForkJoinTask<V> im
1191       * @return the surplus number of tasks, which may be negative
1192       */
1193      public static int getSurplusQueuedTaskCount() {
1194 <        /*
1177 <         * The aim of this method is to return a cheap heuristic guide
1178 <         * for task partitioning when programmers, frameworks, tools,
1179 <         * or languages have little or no idea about task granularity.
1180 <         * In essence by offering this method, we ask users only about
1181 <         * tradeoffs in overhead vs expected throughput and its
1182 <         * variance, rather than how finely to partition tasks.
1183 <         *
1184 <         * In a steady state strict (tree-structured) computation,
1185 <         * each thread makes available for stealing enough tasks for
1186 <         * other threads to remain active. Inductively, if all threads
1187 <         * play by the same rules, each thread should make available
1188 <         * only a constant number of tasks.
1189 <         *
1190 <         * The minimum useful constant is just 1. But using a value of
1191 <         * 1 would require immediate replenishment upon each steal to
1192 <         * maintain enough tasks, which is infeasible.  Further,
1193 <         * partitionings/granularities of offered tasks should
1194 <         * minimize steal rates, which in general means that threads
1195 <         * nearer the top of computation tree should generate more
1196 <         * than those nearer the bottom. In perfect steady state, each
1197 <         * thread is at approximately the same level of computation
1198 <         * tree. However, producing extra tasks amortizes the
1199 <         * uncertainty of progress and diffusion assumptions.
1200 <         *
1201 <         * So, users will want to use values larger, but not much
1202 <         * larger than 1 to both smooth over transient shortages and
1203 <         * hedge against uneven progress; as traded off against the
1204 <         * cost of extra task overhead. We leave the user to pick a
1205 <         * threshold value to compare with the results of this call to
1206 <         * guide decisions, but recommend values such as 3.
1207 <         *
1208 <         * When all threads are active, it is on average OK to
1209 <         * estimate surplus strictly locally. In steady-state, if one
1210 <         * thread is maintaining say 2 surplus tasks, then so are
1211 <         * others. So we can just use estimated queue length.
1212 <         * However, this strategy alone leads to serious mis-estimates
1213 <         * in some non-steady-state conditions (ramp-up, ramp-down,
1214 <         * other stalls). We can detect many of these by further
1215 <         * considering the number of "idle" threads, that are known to
1216 <         * have zero queued tasks, so compensate by a factor of
1217 <         * (#idle/#active) threads.
1218 <         */
1219 <        Thread t; ForkJoinWorkerThread wt;
1220 <        return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1221 <            (wt = (ForkJoinWorkerThread)t).workQueue.queueSize() - wt.pool.idlePerActive() :
1222 <            0;
1194 >        return ForkJoinPool.getSurplusQueuedTaskCount();
1195      }
1196  
1197      // Extension methods
# Line 1263 | Line 1235 | public abstract class ForkJoinTask<V> im
1235      /**
1236       * Returns, but does not unschedule or execute, a task queued by
1237       * the current thread but not yet executed, if one is immediately
1238 <     * available and the current thread is operating in a
1239 <     * ForkJoinPool. There is no guarantee that this task will
1240 <     * actually be polled or executed next. Conversely, this method
1241 <     * may return null even if a task exists but cannot be accessed
1270 <     * without contention with other threads.  This method is designed
1238 >     * available. There is no guarantee that this task will actually
1239 >     * be polled or executed next. Conversely, this method may return
1240 >     * null even if a task exists but cannot be accessed without
1241 >     * contention with other threads.  This method is designed
1242       * primarily to support extensions, and is unlikely to be useful
1243       * otherwise.
1244       *
1245       * @return the next task, or {@code null} if none are available
1246       */
1247      protected static ForkJoinTask<?> peekNextLocalTask() {
1248 <        Thread t;
1249 <        return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1250 <            ((ForkJoinWorkerThread)t).workQueue.peek() :
1251 <            null;
1248 >        Thread t; ForkJoinPool.WorkQueue q;
1249 >        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
1250 >            q = ((ForkJoinWorkerThread)t).workQueue;
1251 >        else
1252 >            q = ForkJoinPool.commonSubmitterQueue();
1253 >        return (q == null) ? null : q.peek();
1254      }
1255  
1256      /**
# Line 1353 | Line 1326 | public abstract class ForkJoinTask<V> im
1326       *
1327       * @param e the expected tag value
1328       * @param tag the new tag value
1329 <     * @return true if successful; i.e., the current value was
1329 >     * @return {@code true} if successful; i.e., the current value was
1330       * equal to e and is now tag.
1331       * @since 1.8
1332       */
# Line 1406 | Line 1379 | public abstract class ForkJoinTask<V> im
1379      }
1380  
1381      /**
1382 +     * Adaptor for Runnables in which failure forces worker exception
1383 +     */
1384 +    static final class RunnableExecuteAction extends ForkJoinTask<Void> {
1385 +        final Runnable runnable;
1386 +        RunnableExecuteAction(Runnable runnable) {
1387 +            if (runnable == null) throw new NullPointerException();
1388 +            this.runnable = runnable;
1389 +        }
1390 +        public final Void getRawResult() { return null; }
1391 +        public final void setRawResult(Void v) { }
1392 +        public final boolean exec() { runnable.run(); return true; }
1393 +        void internalPropagateException(Throwable ex) {
1394 +            rethrow(ex); // rethrow outside exec() catches.
1395 +        }
1396 +        private static final long serialVersionUID = 5232453952276885070L;
1397 +    }
1398 +
1399 +    /**
1400       * Adaptor for Callables
1401       */
1402      static final class AdaptedCallable<T> extends ForkJoinTask<T>
# Line 1502 | Line 1493 | public abstract class ForkJoinTask<V> im
1493      // Unsafe mechanics
1494      private static final sun.misc.Unsafe U;
1495      private static final long STATUS;
1496 +
1497      static {
1498          exceptionTableLock = new ReentrantLock();
1499          exceptionTableRefQueue = new ReferenceQueue<Object>();
1500          exceptionTable = new ExceptionNode[EXCEPTION_MAP_CAPACITY];
1501          try {
1502              U = getUnsafe();
1503 +            Class<?> k = ForkJoinTask.class;
1504              STATUS = U.objectFieldOffset
1505 <                (ForkJoinTask.class.getDeclaredField("status"));
1505 >                (k.getDeclaredField("status"));
1506          } catch (Exception e) {
1507              throw new Error(e);
1508          }
# Line 1525 | Line 1518 | public abstract class ForkJoinTask<V> im
1518      private static sun.misc.Unsafe getUnsafe() {
1519          try {
1520              return sun.misc.Unsafe.getUnsafe();
1521 <        } catch (SecurityException se) {
1522 <            try {
1523 <                return java.security.AccessController.doPrivileged
1524 <                    (new java.security
1525 <                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
1526 <                        public sun.misc.Unsafe run() throws Exception {
1527 <                            java.lang.reflect.Field f = sun.misc
1528 <                                .Unsafe.class.getDeclaredField("theUnsafe");
1529 <                            f.setAccessible(true);
1530 <                            return (sun.misc.Unsafe) f.get(null);
1531 <                        }});
1532 <            } catch (java.security.PrivilegedActionException e) {
1533 <                throw new RuntimeException("Could not initialize intrinsics",
1534 <                                           e.getCause());
1535 <            }
1521 >        } catch (SecurityException tryReflectionInstead) {}
1522 >        try {
1523 >            return java.security.AccessController.doPrivileged
1524 >            (new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1525 >                public sun.misc.Unsafe run() throws Exception {
1526 >                    Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class;
1527 >                    for (java.lang.reflect.Field f : k.getDeclaredFields()) {
1528 >                        f.setAccessible(true);
1529 >                        Object x = f.get(null);
1530 >                        if (k.isInstance(x))
1531 >                            return k.cast(x);
1532 >                    }
1533 >                    throw new NoSuchFieldError("the Unsafe");
1534 >                }});
1535 >        } catch (java.security.PrivilegedActionException e) {
1536 >            throw new RuntimeException("Could not initialize intrinsics",
1537 >                                       e.getCause());
1538          }
1539      }
1540   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines