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.9 by dl, Tue Jan 1 15:10:32 2013 UTC vs.
Revision 1.14 by jsr166, Sun Jul 21 06:32:28 2013 UTC

# Line 134 | Line 134 | import java.lang.reflect.Constructor;
134   * (DAG). Otherwise, executions may encounter a form of deadlock as
135   * tasks cyclically wait for each other.  However, this framework
136   * supports other methods and techniques (for example the use of
137 < * {@link Phaser}, {@link #helpQuiesce}, and {@link #complete}) that
137 > * {@link java.util.concurrent.Phaser 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 <        ForkJoinPool.externalHelpJoin(this);
289 <        boolean interrupted = false;
290 <        while ((s = status) >= 0) {
291 <            if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
292 <                synchronized (this) {
293 <                    if (status >= 0) {
294 <                        try {
295 <                            wait();
296 <                        } catch (InterruptedException ie) {
297 <                            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              }
316          }
305        if (interrupted)
306            Thread.currentThread().interrupt();
317          return s;
318      }
319  
# Line 312 | Line 322 | public abstract class ForkJoinTask<V> im
322       */
323      private int externalInterruptibleAwaitDone() throws InterruptedException {
324          int s;
325 +        ForkJoinPool cp = ForkJoinPool.common;
326          if (Thread.interrupted())
327              throw new InterruptedException();
328 <        ForkJoinPool.externalHelpJoin(this);
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) {
# Line 468 | 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 600 | 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) {
605 <            if (ex instanceof Error)
606 <                throw (Error)ex;
607 <            if (ex instanceof RuntimeException)
608 <                throw (RuntimeException)ex;
619 >    static void rethrow(Throwable ex) {
620 >        if (ex != null)
621              ForkJoinTask.<RuntimeException>uncheckedThrow(ex);
610        }
622      }
623  
624      /**
# Line 617 | Line 628 | public abstract class ForkJoinTask<V> im
628       */
629      @SuppressWarnings("unchecked") static <T extends Throwable>
630          void uncheckedThrow(Throwable t) throws T {
631 <        if (t != null)
621 <            throw (T)t; // rely on vacuous cast
631 >        throw (T)t; // rely on vacuous cast
632      }
633  
634      /**
# Line 772 | Line 782 | public abstract class ForkJoinTask<V> im
782       * unprocessed.
783       *
784       * @param tasks the collection of tasks
785 +     * @param <T> the type of the values returned from the tasks
786       * @return the tasks argument, to simplify usage
787       * @throws NullPointerException if tasks or any element are null
788       */
# Line 829 | Line 840 | public abstract class ForkJoinTask<V> im
840       * <p>This method is designed to be invoked by <em>other</em>
841       * tasks. To terminate the current task, you can just return or
842       * throw an unchecked exception from its computation method, or
843 <     * invoke {@link #completeExceptionally}.
843 >     * invoke {@link #completeExceptionally(Throwable)}.
844       *
845       * @param mayInterruptIfRunning this value has no effect in the
846       * default implementation because interrupts are not used to
# Line 981 | Line 992 | public abstract class ForkJoinTask<V> im
992          // Messy in part because we measure in nanosecs, but wait in millisecs
993          int s; long ms;
994          long ns = unit.toNanos(timeout);
995 +        ForkJoinPool cp;
996          if ((s = status) >= 0 && ns > 0L) {
997              long deadline = System.nanoTime() + ns;
998              ForkJoinPool p = null;
# Line 992 | Line 1004 | public abstract class ForkJoinTask<V> im
1004                  w = wt.workQueue;
1005                  p.helpJoinOnce(w, this); // no retries on failure
1006              }
1007 <            else
1008 <                ForkJoinPool.externalHelpJoin(this);
1007 >            else if ((cp = ForkJoinPool.common) != null) {
1008 >                if (this instanceof CountedCompleter)
1009 >                    cp.externalHelpComplete((CountedCompleter<?>)this);
1010 >                else if (cp.tryExternalUnpush(this))
1011 >                    doExec();
1012 >            }
1013              boolean canBlock = false;
1014              boolean interrupted = false;
1015              try {
# Line 1001 | Line 1017 | public abstract class ForkJoinTask<V> im
1017                      if (w != null && w.qlock < 0)
1018                          cancelIgnoringExceptions(this);
1019                      else if (!canBlock) {
1020 <                        if (p == null || p.tryCompensate())
1020 >                        if (p == null || p.tryCompensate(p.ctl))
1021                              canBlock = true;
1022                      }
1023                      else {
# Line 1142 | Line 1158 | public abstract class ForkJoinTask<V> im
1158          Thread t;
1159          return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1160                  ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) :
1161 <                ForkJoinPool.tryExternalUnpush(this));
1161 >                ForkJoinPool.common.tryExternalUnpush(this));
1162      }
1163  
1164      /**
# Line 1311 | Line 1327 | public abstract class ForkJoinTask<V> im
1327       *
1328       * @param e the expected tag value
1329       * @param tag the new tag value
1330 <     * @return true if successful; i.e., the current value was
1330 >     * @return {@code true} if successful; i.e., the current value was
1331       * equal to e and is now tag.
1332       * @since 1.8
1333       */
# Line 1364 | Line 1380 | public abstract class ForkJoinTask<V> im
1380      }
1381  
1382      /**
1383 +     * Adaptor for Runnables in which failure forces worker exception
1384 +     */
1385 +    static final class RunnableExecuteAction extends ForkJoinTask<Void> {
1386 +        final Runnable runnable;
1387 +        RunnableExecuteAction(Runnable runnable) {
1388 +            if (runnable == null) throw new NullPointerException();
1389 +            this.runnable = runnable;
1390 +        }
1391 +        public final Void getRawResult() { return null; }
1392 +        public final void setRawResult(Void v) { }
1393 +        public final boolean exec() { runnable.run(); return true; }
1394 +        void internalPropagateException(Throwable ex) {
1395 +            rethrow(ex); // rethrow outside exec() catches.
1396 +        }
1397 +        private static final long serialVersionUID = 5232453952276885070L;
1398 +    }
1399 +
1400 +    /**
1401       * Adaptor for Callables
1402       */
1403      static final class AdaptedCallable<T> extends ForkJoinTask<T>
# Line 1411 | Line 1445 | public abstract class ForkJoinTask<V> im
1445       *
1446       * @param runnable the runnable action
1447       * @param result the result upon completion
1448 +     * @param <T> the type of the result
1449       * @return the task
1450       */
1451      public static <T> ForkJoinTask<T> adapt(Runnable runnable, T result) {
# Line 1424 | Line 1459 | public abstract class ForkJoinTask<V> im
1459       * encountered into {@code RuntimeException}.
1460       *
1461       * @param callable the callable action
1462 +     * @param <T> the type of the callable's result
1463       * @return the task
1464       */
1465      public static <T> ForkJoinTask<T> adapt(Callable<? extends T> callable) {
# Line 1485 | Line 1521 | public abstract class ForkJoinTask<V> im
1521      private static sun.misc.Unsafe getUnsafe() {
1522          try {
1523              return sun.misc.Unsafe.getUnsafe();
1524 <        } catch (SecurityException se) {
1525 <            try {
1526 <                return java.security.AccessController.doPrivileged
1527 <                    (new java.security
1528 <                     .PrivilegedExceptionAction<sun.misc.Unsafe>() {
1529 <                        public sun.misc.Unsafe run() throws Exception {
1530 <                            java.lang.reflect.Field f = sun.misc
1531 <                                .Unsafe.class.getDeclaredField("theUnsafe");
1532 <                            f.setAccessible(true);
1533 <                            return (sun.misc.Unsafe) f.get(null);
1534 <                        }});
1535 <            } catch (java.security.PrivilegedActionException e) {
1536 <                throw new RuntimeException("Could not initialize intrinsics",
1537 <                                           e.getCause());
1538 <            }
1524 >        } catch (SecurityException tryReflectionInstead) {}
1525 >        try {
1526 >            return java.security.AccessController.doPrivileged
1527 >            (new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1528 >                public sun.misc.Unsafe run() throws Exception {
1529 >                    Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class;
1530 >                    for (java.lang.reflect.Field f : k.getDeclaredFields()) {
1531 >                        f.setAccessible(true);
1532 >                        Object x = f.get(null);
1533 >                        if (k.isInstance(x))
1534 >                            return k.cast(x);
1535 >                    }
1536 >                    throw new NoSuchFieldError("the Unsafe");
1537 >                }});
1538 >        } catch (java.security.PrivilegedActionException e) {
1539 >            throw new RuntimeException("Could not initialize intrinsics",
1540 >                                       e.getCause());
1541          }
1542      }
1505
1543   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines