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

Comparing jsr166/src/jsr166y/ForkJoinTask.java (file contents):
Revision 1.59 by jsr166, Tue Sep 7 07:51:13 2010 UTC vs.
Revision 1.68 by jsr166, Sun Nov 21 19:04:45 2010 UTC

# Line 6 | Line 6
6  
7   package jsr166y;
8  
9 – import java.util.concurrent.*;
10 –
9   import java.io.Serializable;
10   import java.util.Collection;
11   import java.util.Collections;
# Line 15 | Line 13 | import java.util.List;
13   import java.util.RandomAccess;
14   import java.util.Map;
15   import java.util.WeakHashMap;
16 + import java.util.concurrent.Callable;
17 + import java.util.concurrent.CancellationException;
18 + import java.util.concurrent.ExecutionException;
19 + import java.util.concurrent.Executor;
20 + import java.util.concurrent.ExecutorService;
21 + import java.util.concurrent.Future;
22 + import java.util.concurrent.RejectedExecutionException;
23 + import java.util.concurrent.RunnableFuture;
24 + import java.util.concurrent.TimeUnit;
25 + import java.util.concurrent.TimeoutException;
26  
27   /**
28   * Abstract base class for tasks that run within a {@link ForkJoinPool}.
# Line 28 | Line 36 | import java.util.WeakHashMap;
36   * start other subtasks.  As indicated by the name of this class,
37   * many programs using {@code ForkJoinTask} employ only methods
38   * {@link #fork} and {@link #join}, or derivatives such as {@link
39 < * #invokeAll}.  However, this class also provides a number of other
40 < * methods that can come into play in advanced usages, as well as
41 < * extension mechanics that allow support of new forms of fork/join
42 < * processing.
39 > * #invokeAll(ForkJoinTask...) invokeAll}.  However, this class also
40 > * provides a number of other methods that can come into play in
41 > * advanced usages, as well as extension mechanics that allow
42 > * support of new forms of fork/join processing.
43   *
44   * <p>A {@code ForkJoinTask} is a lightweight form of {@link Future}.
45   * The efficiency of {@code ForkJoinTask}s stems from a set of
# Line 206 | Line 214 | public abstract class ForkJoinTask<V> im
214  
215      /**
216       * Records exception and sets exceptional completion.
217 <     *
217 >     *
218       * @return status on exit
219       */
220      private void setExceptionalCompletion(Throwable rex) {
# Line 223 | Line 231 | public abstract class ForkJoinTask<V> im
231          int s;         // the odd construction reduces lock bias effects
232          while ((s = status) >= 0) {
233              try {
234 <                synchronized(this) {
234 >                synchronized (this) {
235                      if (UNSAFE.compareAndSwapInt(this, statusOffset, s,SIGNAL))
236                          wait();
237                  }
# Line 239 | Line 247 | public abstract class ForkJoinTask<V> im
247       *
248       * @return status on exit
249       */
250 <    final int internalAwaitDone(long millis) {
250 >    final int internalAwaitDone(long millis, int nanos) {
251          int s;
252          if ((s = status) >= 0) {
253              try {
254 <                synchronized(this) {
254 >                synchronized (this) {
255                      if (UNSAFE.compareAndSwapInt(this, statusOffset, s,SIGNAL))
256 <                        wait(millis, 0);
256 >                        wait(millis, nanos);
257                  }
258              } catch (InterruptedException ie) {
259                  cancelIfTerminating();
# Line 261 | Line 269 | public abstract class ForkJoinTask<V> im
269      private void externalAwaitDone() {
270          int s;
271          while ((s = status) >= 0) {
272 <            synchronized(this) {
273 <                if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)){
272 >            synchronized (this) {
273 >                if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)) {
274                      boolean interrupted = false;
275                      while (status >= 0) {
276                          try {
# Line 642 | Line 650 | public abstract class ForkJoinTask<V> im
650          setCompletion(NORMAL);
651      }
652  
653 +    /**
654 +     * Waits if necessary for the computation to complete, and then
655 +     * retrieves its result.
656 +     *
657 +     * @return the computed result
658 +     * @throws CancellationException if the computation was cancelled
659 +     * @throws ExecutionException if the computation threw an
660 +     * exception
661 +     * @throws InterruptedException if the current thread is not a
662 +     * member of a ForkJoinPool and was interrupted while waiting
663 +     */
664      public final V get() throws InterruptedException, ExecutionException {
665 <        quietlyJoin();
666 <        if (Thread.interrupted())
667 <            throw new InterruptedException();
668 <        int s = status;
665 >        int s;
666 >        if (Thread.currentThread() instanceof ForkJoinWorkerThread) {
667 >            quietlyJoin();
668 >            s = status;
669 >        }
670 >        else {
671 >            while ((s = status) >= 0) {
672 >                synchronized (this) { // interruptible form of awaitDone
673 >                    if (UNSAFE.compareAndSwapInt(this, statusOffset,
674 >                                                 s, SIGNAL)) {
675 >                        while (status >= 0)
676 >                            wait();
677 >                    }
678 >                }
679 >            }
680 >        }
681          if (s < NORMAL) {
682              Throwable ex;
683              if (s == CANCELLED)
# Line 657 | Line 688 | public abstract class ForkJoinTask<V> im
688          return getRawResult();
689      }
690  
691 +    /**
692 +     * Waits if necessary for at most the given time for the computation
693 +     * to complete, and then retrieves its result, if available.
694 +     *
695 +     * @param timeout the maximum time to wait
696 +     * @param unit the time unit of the timeout argument
697 +     * @return the computed result
698 +     * @throws CancellationException if the computation was cancelled
699 +     * @throws ExecutionException if the computation threw an
700 +     * exception
701 +     * @throws InterruptedException if the current thread is not a
702 +     * member of a ForkJoinPool and was interrupted while waiting
703 +     * @throws TimeoutException if the wait timed out
704 +     */
705      public final V get(long timeout, TimeUnit unit)
706          throws InterruptedException, ExecutionException, TimeoutException {
662 –        Thread t = Thread.currentThread();
663 –        ForkJoinPool pool;
664 –        if (t instanceof ForkJoinWorkerThread) {
665 –            ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
666 –            if (status >= 0 && w.unpushTask(this))
667 –                quietlyExec();
668 –            pool = w.pool;
669 –        }
670 –        else
671 –            pool = null;
672 –        /*
673 –         * Timed wait loop intermixes cases for FJ (pool != null) and
674 –         * non FJ threads. For FJ, decrement pool count but don't try
675 –         * for replacement; increment count on completion. For non-FJ,
676 –         * deal with interrupts. This is messy, but a little less so
677 –         * than is splitting the FJ and nonFJ cases.
678 –         */
679 –        boolean interrupted = false;
680 –        boolean dec = false; // true if pool count decremented
707          long nanos = unit.toNanos(timeout);
708 <        for (;;) {
709 <            if (pool == null && Thread.interrupted()) {
710 <                interrupted = true;
711 <                break;
708 >        if (status >= 0) {
709 >            Thread t = Thread.currentThread();
710 >            if (t instanceof ForkJoinWorkerThread) {
711 >                ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
712 >                boolean completed = false; // timed variant of quietlyJoin
713 >                if (w.unpushTask(this)) {
714 >                    try {
715 >                        completed = exec();
716 >                    } catch (Throwable rex) {
717 >                        setExceptionalCompletion(rex);
718 >                    }
719 >                }
720 >                if (completed)
721 >                    setCompletion(NORMAL);
722 >                else if (status >= 0)
723 >                    w.joinTask(this, true, nanos);
724              }
725 <            int s = status;
726 <            if (s < 0)
727 <                break;
690 <            if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)) {
725 >            else if (Thread.interrupted())
726 >                throw new InterruptedException();
727 >            else {
728                  long startTime = System.nanoTime();
729 <                long nt; // wait time
730 <                while (status >= 0 &&
729 >                int s; long nt;
730 >                while ((s = status) >= 0 &&
731                         (nt = nanos - (System.nanoTime() - startTime)) > 0) {
732 <                    if (pool != null && !dec)
733 <                        dec = pool.tryDecrementRunningCount();
697 <                    else {
732 >                    if (UNSAFE.compareAndSwapInt(this, statusOffset, s,
733 >                                                 SIGNAL)) {
734                          long ms = nt / 1000000;
735                          int ns = (int) (nt % 1000000);
736 <                        try {
737 <                            synchronized(this) {
738 <                                if (status >= 0)
703 <                                    wait(ms, ns);
704 <                            }
705 <                        } catch (InterruptedException ie) {
706 <                            if (pool != null)
707 <                                cancelIfTerminating();
708 <                            else {
709 <                                interrupted = true;
710 <                                break;
711 <                            }
736 >                        synchronized (this) {
737 >                            if (status >= 0)
738 >                                wait(ms, ns); // exit on IE throw
739                          }
740                      }
741                  }
715 –                break;
742              }
743          }
718 –        if (pool != null && dec)
719 –            pool.incrementRunningCount();
720 –        if (interrupted)
721 –            throw new InterruptedException();
744          int es = status;
745          if (es != NORMAL) {
746              Throwable ex;
# Line 755 | Line 777 | public abstract class ForkJoinTask<V> im
777                          return;
778                      }
779                  }
780 <                w.joinTask(this);
780 >                w.joinTask(this, false, 0L);
781              }
782          }
783          else
# Line 810 | Line 832 | public abstract class ForkJoinTask<V> im
832       * under any other usage conditions are not guaranteed.
833       * This method may be useful when executing
834       * pre-constructed trees of subtasks in loops.
835 +     *
836 +     * <p>Upon completion of this method, {@code isDone()} reports
837 +     * {@code false}, and {@code getException()} reports {@code
838 +     * null}. However, the value returned by {@code getRawResult} is
839 +     * unaffected. To clear this value, you can invoke {@code
840 +     * setRawResult(null)}.
841       */
842      public void reinitialize() {
843          if (status == EXCEPTIONAL)

Diff Legend

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