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.13 by dl, Wed Jun 19 14:55:40 2013 UTC

# 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 <        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 829 | 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 991 | public abstract class ForkJoinTask<V> im
991          // Messy in part because we measure in nanosecs, but wait in millisecs
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;
# Line 992 | Line 1003 | public abstract class ForkJoinTask<V> im
1003                  w = wt.workQueue;
1004                  p.helpJoinOnce(w, this); // no retries on failure
1005              }
1006 <            else
1007 <                ForkJoinPool.externalHelpJoin(this);
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 {
# Line 1001 | Line 1016 | public abstract class ForkJoinTask<V> im
1016                      if (w != null && w.qlock < 0)
1017                          cancelIgnoringExceptions(this);
1018                      else if (!canBlock) {
1019 <                        if (p == null || p.tryCompensate())
1019 >                        if (p == null || p.tryCompensate(p.ctl))
1020                              canBlock = true;
1021                      }
1022                      else {
# Line 1142 | Line 1157 | public abstract class ForkJoinTask<V> im
1157          Thread t;
1158          return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1159                  ((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) :
1160 <                ForkJoinPool.tryExternalUnpush(this));
1160 >                ForkJoinPool.common.tryExternalUnpush(this));
1161      }
1162  
1163      /**
# Line 1311 | 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 1364 | 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 1485 | 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      }
1505
1540   }

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines