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.11 by jsr166, Sun Jan 20 03:44:14 2013 UTC vs.
Revision 1.15 by jsr166, Mon Jul 22 16:52:31 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 java.util.concurrent.Phaser}, {@link #helpQuiesce}, and
138 < * {@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 286 | 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          }
306 –        if (interrupted)
307 –            Thread.currentThread().interrupt();
317          return s;
318      }
319  
# Line 313 | 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 469 | 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 601 | 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) {
606 <            if (ex instanceof Error)
607 <                throw (Error)ex;
608 <            if (ex instanceof RuntimeException)
609 <                throw (RuntimeException)ex;
619 >    static void rethrow(Throwable ex) {
620 >        if (ex != null)
621              ForkJoinTask.<RuntimeException>uncheckedThrow(ex);
611 –        }
622      }
623  
624      /**
# Line 618 | 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)
622 <            throw (T)t; // rely on vacuous cast
631 >        throw (T)t; // rely on vacuous cast
632      }
633  
634      /**
# Line 773 | 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 830 | 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 982 | 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 993 | 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 1002 | 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 1143 | 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 1312 | 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 1365 | 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 1412 | 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 1425 | 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 1438 | Line 1473 | public abstract class ForkJoinTask<V> im
1473      /**
1474       * Saves this task to a stream (that is, serializes it).
1475       *
1476 +     * @param s the stream
1477 +     * @throws java.io.IOException if an I/O error occurs
1478       * @serialData the current run status and the exception thrown
1479       * during execution, or {@code null} if none
1480       */
# Line 1449 | Line 1486 | public abstract class ForkJoinTask<V> im
1486  
1487      /**
1488       * Reconstitutes this task from a stream (that is, deserializes it).
1489 +     * @param s the stream
1490 +     * @throws ClassNotFoundException if the class of a serialized object
1491 +     *         could not be found
1492 +     * @throws java.io.IOException if an I/O error occurs
1493       */
1494      private void readObject(java.io.ObjectInputStream s)
1495          throws java.io.IOException, ClassNotFoundException {

Diff Legend

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