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.11 by jsr166, Tue Jul 21 18:11:44 2009 UTC vs.
Revision 1.15 by jsr166, Fri Jul 24 22:05:22 2009 UTC

# Line 81 | Line 81 | import java.lang.reflect.*;
81   * class. While these methods have {@code public} access (to allow
82   * instances of different task subclasses to call each others
83   * methods), some of them may only be called from within other
84 < * ForkJoinTasks. Attempts to invoke them in other contexts result in
85 < * exceptions or errors possibly including ClassCastException.
84 > * ForkJoinTasks (as may be determined using method {@link
85 > * #inForkJoinPool}).  Attempts to invoke them in other contexts
86 > * result in exceptions or errors, possibly including
87 > * ClassCastException.
88   *
89   * <p>Most base support methods are {@code final} because their
90   * implementations are intrinsically tied to the underlying
# Line 107 | Line 109 | import java.lang.reflect.*;
109   * in general sensible to serialize tasks only before or after, but
110   * not during execution. Serialization is not relied on during
111   * execution itself.
112 + *
113 + * @since 1.7
114 + * @author Doug Lea
115   */
116   public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
117  
# Line 157 | Line 162 | public abstract class ForkJoinTask<V> im
162       */
163      static ForkJoinWorkerThread getWorker() {
164          Thread t = Thread.currentThread();
165 <        return ((t instanceof ForkJoinWorkerThread)?
166 <                (ForkJoinWorkerThread)t : null);
165 >        return ((t instanceof ForkJoinWorkerThread) ?
166 >                (ForkJoinWorkerThread) t : null);
167      }
168  
169      final boolean casStatus(int cmp, int val) {
# Line 184 | Line 189 | public abstract class ForkJoinTask<V> im
189          ForkJoinPool pool = getPool();
190          if (pool != null) {
191              int s; // Clear signal bits while setting completion status
192 <            do;while ((s = status) >= 0 && !casStatus(s, completion));
192 >            do {} while ((s = status) >= 0 && !casStatus(s, completion));
193  
194              if ((s & SIGNAL_MASK) != 0) {
195                  if ((s &= INTERNAL_SIGNAL_MASK) != 0)
196                      pool.updateRunningCount(s);
197 <                synchronized(this) { notifyAll(); }
197 >                synchronized (this) { notifyAll(); }
198              }
199          }
200          else
# Line 202 | Line 207 | public abstract class ForkJoinTask<V> im
207       */
208      private void externallySetCompletion(int completion) {
209          int s;
210 <        do;while ((s = status) >= 0 &&
211 <                  !casStatus(s, (s & SIGNAL_MASK) | completion));
212 <        synchronized(this) { notifyAll(); }
210 >        do {} while ((s = status) >= 0 &&
211 >                     !casStatus(s, (s & SIGNAL_MASK) | completion));
212 >        synchronized (this) { notifyAll(); }
213      }
214  
215      /**
216 <     * Sets status to indicate normal completion
216 >     * Sets status to indicate normal completion.
217       */
218      final void setNormalCompletion() {
219          // Try typical fast case -- single CAS, no signal, not already done.
# Line 220 | Line 225 | public abstract class ForkJoinTask<V> im
225      // internal waiting and notification
226  
227      /**
228 <     * Performs the actual monitor wait for awaitDone
228 >     * Performs the actual monitor wait for awaitDone.
229       */
230      private void doAwaitDone() {
231          // Minimize lock bias and in/de-flation effects by maximizing
232          // chances of waiting inside sync
233          try {
234              while (status >= 0)
235 <                synchronized(this) { if (status >= 0) wait(); }
235 >                synchronized (this) { if (status >= 0) wait(); }
236          } catch (InterruptedException ie) {
237              onInterruptedWait();
238          }
239      }
240  
241      /**
242 <     * Performs the actual monitor wait for awaitDone
242 >     * Performs the actual timed monitor wait for awaitDone.
243       */
244      private void doAwaitDone(long startTime, long nanos) {
245 <        synchronized(this) {
245 >        synchronized (this) {
246              try {
247                  while (status >= 0) {
248                      long nt = nanos - System.nanoTime() - startTime;
249                      if (nt <= 0)
250                          break;
251 <                    wait(nt / 1000000, (int)(nt % 1000000));
251 >                    wait(nt / 1000000, (int) (nt % 1000000));
252                  }
253              } catch (InterruptedException ie) {
254                  onInterruptedWait();
# Line 259 | Line 264 | public abstract class ForkJoinTask<V> im
264       *
265       * @return status upon exit
266       */
267 <    private int awaitDone(ForkJoinWorkerThread w, boolean maintainParallelism) {
268 <        ForkJoinPool pool = w == null? null : w.pool;
267 >    private int awaitDone(ForkJoinWorkerThread w,
268 >                          boolean maintainParallelism) {
269 >        ForkJoinPool pool = (w == null) ? null : w.pool;
270          int s;
271          while ((s = status) >= 0) {
272 <            if (casStatus(s, pool == null? s|EXTERNAL_SIGNAL : s+1)) {
272 >            if (casStatus(s, (pool == null) ? s|EXTERNAL_SIGNAL : s+1)) {
273                  if (pool == null || !pool.preJoin(this, maintainParallelism))
274                      doAwaitDone();
275                  if (((s = status) & INTERNAL_SIGNAL_MASK) != 0)
# Line 276 | Line 282 | public abstract class ForkJoinTask<V> im
282  
283      /**
284       * Timed version of awaitDone
285 +     *
286       * @return status upon exit
287       */
288      private int awaitDone(ForkJoinWorkerThread w, long nanos) {
289 <        ForkJoinPool pool = w == null? null : w.pool;
289 >        ForkJoinPool pool = (w == null) ? null : w.pool;
290          int s;
291          while ((s = status) >= 0) {
292 <            if (casStatus(s, pool == null? s|EXTERNAL_SIGNAL : s+1)) {
292 >            if (casStatus(s, (pool == null) ? s|EXTERNAL_SIGNAL : s+1)) {
293                  long startTime = System.nanoTime();
294                  if (pool == null || !pool.preJoin(this, false))
295                      doAwaitDone(startTime, nanos);
# Line 304 | Line 311 | public abstract class ForkJoinTask<V> im
311       */
312      private void adjustPoolCountsOnUnblock(ForkJoinPool pool) {
313          int s;
314 <        do;while ((s = status) < 0 && !casStatus(s, s & COMPLETION_MASK));
314 >        do {} while ((s = status) < 0 && !casStatus(s, s & COMPLETION_MASK));
315          if (pool != null && (s &= INTERNAL_SIGNAL_MASK) != 0)
316              pool.updateRunningCount(s);
317      }
# Line 359 | Line 366 | public abstract class ForkJoinTask<V> im
366  
367      /**
368       * Returns result or throws exception using j.u.c.Future conventions.
369 <     * Only call when isDone known to be true.
369 >     * Only call when {@code isDone} known to be true.
370       */
371      private V reportFutureResult()
372          throws ExecutionException, InterruptedException {
# Line 425 | Line 432 | public abstract class ForkJoinTask<V> im
432              try {
433                  if (!exec())
434                      return;
435 <            } catch(Throwable rex) {
435 >            } catch (Throwable rex) {
436                  setDoneExceptionally(rex);
437                  return;
438              }
# Line 457 | Line 464 | public abstract class ForkJoinTask<V> im
464      final void cancelIgnoringExceptions() {
465          try {
466              cancel(false);
467 <        } catch(Throwable ignore) {
467 >        } catch (Throwable ignore) {
468          }
469      }
470  
# Line 469 | Line 476 | public abstract class ForkJoinTask<V> im
476          ForkJoinTask<?> t;
477          while ((s = status) >= 0 && (t = w.scanWhileJoining(this)) != null)
478              t.quietlyExec();
479 <        return (s >= 0)? awaitDone(w, false) : s; // block if no work
479 >        return (s >= 0) ? awaitDone(w, false) : s; // block if no work
480      }
481  
482      // public methods
# Line 479 | Line 486 | public abstract class ForkJoinTask<V> im
486       * necessarily enforced, it is a usage error to fork a task more
487       * than once unless it has completed and been reinitialized.  This
488       * method may be invoked only from within ForkJoinTask
489 <     * computations. Attempts to invoke in other contexts result in
490 <     * exceptions or errors possibly including ClassCastException.
489 >     * computations (as may be determined using method {@link
490 >     * #inForkJoinPool}). Attempts to invoke in other contexts result
491 >     * in exceptions or errors, possibly including ClassCastException.
492       */
493      public final void fork() {
494 <        ((ForkJoinWorkerThread)(Thread.currentThread())).pushTask(this);
494 >        ((ForkJoinWorkerThread) Thread.currentThread())
495 >            .pushTask(this);
496      }
497  
498      /**
# Line 519 | Line 528 | public abstract class ForkJoinTask<V> im
528      /**
529       * Forks both tasks, returning when {@code isDone} holds for
530       * both of them or an exception is encountered. This method may be
531 <     * invoked only from within ForkJoinTask computations. Attempts to
532 <     * invoke in other contexts result in exceptions or errors
531 >     * invoked only from within ForkJoinTask computations (as may be
532 >     * determined using method {@link #inForkJoinPool}). Attempts to
533 >     * invoke in other contexts result in exceptions or errors,
534       * possibly including ClassCastException.
535       *
536       * @param t1 one task
# Line 538 | Line 548 | public abstract class ForkJoinTask<V> im
548       * Forks the given tasks, returning when {@code isDone} holds
549       * for all of them. If any task encounters an exception, others
550       * may be cancelled.  This method may be invoked only from within
551 <     * ForkJoinTask computations. Attempts to invoke in other contexts
552 <     * result in exceptions or errors possibly including ClassCastException.
551 >     * ForkJoinTask computations (as may be determined using method
552 >     * {@link #inForkJoinPool}). Attempts to invoke in other contexts
553 >     * result in exceptions or errors, possibly including
554 >     * ClassCastException.
555       *
556       * @param tasks the array of tasks
557       * @throws NullPointerException if tasks or any element are null
# Line 582 | Line 594 | public abstract class ForkJoinTask<V> im
594       * Forks all tasks in the collection, returning when
595       * {@code isDone} holds for all of them. If any task
596       * encounters an exception, others may be cancelled.  This method
597 <     * may be invoked only from within ForkJoinTask
598 <     * computations. Attempts to invoke in other contexts result in
599 <     * exceptions or errors possibly including ClassCastException.
597 >     * may be invoked only from within ForkJoinTask computations (as
598 >     * may be determined using method {@link
599 >     * #inForkJoinPool}). Attempts to invoke in other contexts result
600 >     * in exceptions or errors, possibly including ClassCastException.
601       *
602       * @param tasks the collection of tasks
603       * @throws NullPointerException if tasks or any element are null
604       * @throws RuntimeException or Error if any task did so
605       */
606      public static void invokeAll(Collection<? extends ForkJoinTask<?>> tasks) {
607 <        if (!(tasks instanceof List)) {
608 <            invokeAll(tasks.toArray(new ForkJoinTask[tasks.size()]));
607 >        if (!(tasks instanceof List<?>)) {
608 >            invokeAll(tasks.toArray(new ForkJoinTask<?>[tasks.size()]));
609              return;
610          }
611 +        @SuppressWarnings("unchecked")
612          List<? extends ForkJoinTask<?>> ts =
613 <            (List<? extends ForkJoinTask<?>>)tasks;
613 >            (List<? extends ForkJoinTask<?>>) tasks;
614          Throwable ex = null;
615          int last = ts.size() - 1;
616          for (int i = last; i >= 0; --i) {
# Line 671 | Line 685 | public abstract class ForkJoinTask<V> im
685       *
686       * @param mayInterruptIfRunning this value is ignored in the
687       * default implementation because tasks are not in general
688 <     * cancelled via interruption.
688 >     * cancelled via interruption
689       *
690       * @return true if this task is now cancelled
691       */
# Line 721 | Line 735 | public abstract class ForkJoinTask<V> im
735       */
736      public void completeExceptionally(Throwable ex) {
737          setDoneExceptionally((ex instanceof RuntimeException) ||
738 <                             (ex instanceof Error)? ex :
738 >                             (ex instanceof Error) ? ex :
739                               new RuntimeException(ex));
740      }
741  
# Line 740 | Line 754 | public abstract class ForkJoinTask<V> im
754      public void complete(V value) {
755          try {
756              setRawResult(value);
757 <        } catch(Throwable rex) {
757 >        } catch (Throwable rex) {
758              setDoneExceptionally(rex);
759              return;
760          }
# Line 770 | Line 784 | public abstract class ForkJoinTask<V> im
784       * current task and that of any other task that might be executed
785       * while helping. (This usually holds for pure divide-and-conquer
786       * tasks). This method may be invoked only from within
787 <     * ForkJoinTask computations. Attempts to invoke in other contexts
788 <     * result in exceptions or errors possibly including ClassCastException.
787 >     * ForkJoinTask computations (as may be determined using method
788 >     * {@link #inForkJoinPool}). Attempts to invoke in other contexts
789 >     * result in exceptions or errors, possibly including
790 >     * ClassCastException.
791       *
792       * @return the computed result
793       */
794      public final V helpJoin() {
795 <        ForkJoinWorkerThread w = (ForkJoinWorkerThread)(Thread.currentThread());
795 >        ForkJoinWorkerThread w = (ForkJoinWorkerThread) Thread.currentThread();
796          if (status < 0 || !w.unpushTask(this) || !tryExec())
797              reportException(busyJoin(w));
798          return getRawResult();
# Line 785 | Line 801 | public abstract class ForkJoinTask<V> im
801      /**
802       * Possibly executes other tasks until this task is ready.  This
803       * method may be invoked only from within ForkJoinTask
804 <     * computations. Attempts to invoke in other contexts result in
805 <     * exceptions or errors possibly including ClassCastException.
804 >     * computations (as may be determined using method {@link
805 >     * #inForkJoinPool}). Attempts to invoke in other contexts result
806 >     * in exceptions or errors, possibly including ClassCastException.
807       */
808      public final void quietlyHelpJoin() {
809          if (status >= 0) {
810              ForkJoinWorkerThread w =
811 <                (ForkJoinWorkerThread)(Thread.currentThread());
811 >                (ForkJoinWorkerThread) Thread.currentThread();
812              if (!w.unpushTask(this) || !tryQuietlyInvoke())
813                  busyJoin(w);
814          }
# Line 830 | Line 847 | public abstract class ForkJoinTask<V> im
847       * joined, instead executing them until all are processed.
848       */
849      public static void helpQuiesce() {
850 <        ((ForkJoinWorkerThread)(Thread.currentThread())).
851 <            helpQuiescePool();
850 >        ((ForkJoinWorkerThread) Thread.currentThread())
851 >            .helpQuiescePool();
852      }
853  
854      /**
# Line 852 | Line 869 | public abstract class ForkJoinTask<V> im
869  
870      /**
871       * Returns the pool hosting the current task execution, or null
872 <     * if this task is executing outside of any pool.
872 >     * if this task is executing outside of any ForkJoinPool.
873       *
874       * @return the pool, or null if none
875       */
876      public static ForkJoinPool getPool() {
877          Thread t = Thread.currentThread();
878 <        return ((t instanceof ForkJoinWorkerThread)?
879 <                ((ForkJoinWorkerThread)t).pool : null);
878 >        return (t instanceof ForkJoinWorkerThread) ?
879 >            ((ForkJoinWorkerThread) t).pool : null;
880 >    }
881 >
882 >    /**
883 >     * Returns {@code true} if the current thread is executing as a
884 >     * ForkJoinPool computation.
885 >     *
886 >     * @return {@code true} if the current thread is executing as a
887 >     * ForkJoinPool computation, or false otherwise
888 >     */
889 >    public static boolean inForkJoinPool() {
890 >        return Thread.currentThread() instanceof ForkJoinWorkerThread;
891      }
892  
893      /**
# Line 869 | Line 897 | public abstract class ForkJoinTask<V> im
897       * another thread.  This method may be useful when arranging
898       * alternative local processing of tasks that could have been, but
899       * were not, stolen. This method may be invoked only from within
900 <     * ForkJoinTask computations. Attempts to invoke in other contexts
901 <     * result in exceptions or errors possibly including ClassCastException.
900 >     * ForkJoinTask computations (as may be determined using method
901 >     * {@link #inForkJoinPool}). Attempts to invoke in other contexts
902 >     * result in exceptions or errors, possibly including
903 >     * ClassCastException.
904       *
905       * @return true if unforked
906       */
907      public boolean tryUnfork() {
908 <        return ((ForkJoinWorkerThread)(Thread.currentThread())).unpushTask(this);
908 >        return ((ForkJoinWorkerThread) Thread.currentThread())
909 >            .unpushTask(this);
910      }
911  
912      /**
# Line 887 | Line 918 | public abstract class ForkJoinTask<V> im
918       * @return the number of tasks
919       */
920      public static int getQueuedTaskCount() {
921 <        return ((ForkJoinWorkerThread)(Thread.currentThread())).
922 <            getQueueSize();
921 >        return ((ForkJoinWorkerThread) Thread.currentThread())
922 >            .getQueueSize();
923      }
924  
925      /**
# Line 904 | Line 935 | public abstract class ForkJoinTask<V> im
935       * @return the surplus number of tasks, which may be negative
936       */
937      public static int getSurplusQueuedTaskCount() {
938 <        return ((ForkJoinWorkerThread)(Thread.currentThread()))
938 >        return ((ForkJoinWorkerThread) Thread.currentThread())
939              .getEstimatedSurplusTaskCount();
940      }
941  
# Line 951 | Line 982 | public abstract class ForkJoinTask<V> im
982       * be polled or executed next.  This method is designed primarily
983       * to support extensions, and is unlikely to be useful otherwise.
984       * This method may be invoked only from within ForkJoinTask
985 <     * computations. Attempts to invoke in other contexts result in
986 <     * exceptions or errors possibly including ClassCastException.
985 >     * computations (as may be determined using method {@link
986 >     * #inForkJoinPool}). Attempts to invoke in other contexts result
987 >     * in exceptions or errors, possibly including ClassCastException.
988       *
989       * @return the next task, or null if none are available
990       */
991      protected static ForkJoinTask<?> peekNextLocalTask() {
992 <        return ((ForkJoinWorkerThread)(Thread.currentThread())).peekTask();
992 >        return ((ForkJoinWorkerThread) Thread.currentThread())
993 >            .peekTask();
994      }
995  
996      /**
# Line 965 | Line 998 | public abstract class ForkJoinTask<V> im
998       * queued by the current thread but not yet executed.  This method
999       * is designed primarily to support extensions, and is unlikely to
1000       * be useful otherwise.  This method may be invoked only from
1001 <     * within ForkJoinTask computations. Attempts to invoke in other
1002 <     * contexts result in exceptions or errors possibly including
1001 >     * within ForkJoinTask computations (as may be determined using
1002 >     * method {@link #inForkJoinPool}). Attempts to invoke in other
1003 >     * contexts result in exceptions or errors, possibly including
1004       * ClassCastException.
1005       *
1006       * @return the next task, or null if none are available
1007       */
1008      protected static ForkJoinTask<?> pollNextLocalTask() {
1009 <        return ((ForkJoinWorkerThread)(Thread.currentThread())).pollLocalTask();
1009 >        return ((ForkJoinWorkerThread) Thread.currentThread())
1010 >            .pollLocalTask();
1011      }
1012  
1013      /**
# Line 984 | Line 1019 | public abstract class ForkJoinTask<V> im
1019       * of the pool this task is operating in.  This method is designed
1020       * primarily to support extensions, and is unlikely to be useful
1021       * otherwise.  This method may be invoked only from within
1022 <     * ForkJoinTask computations. Attempts to invoke in other contexts
1023 <     * result in exceptions or errors possibly including
1022 >     * ForkJoinTask computations (as may be determined using method
1023 >     * {@link #inForkJoinPool}). Attempts to invoke in other contexts
1024 >     * result in exceptions or errors, possibly including
1025       * ClassCastException.
1026       *
1027       * @return a task, or null if none are available
1028       */
1029      protected static ForkJoinTask<?> pollTask() {
1030 <        return ((ForkJoinWorkerThread)(Thread.currentThread())).
1031 <            pollTask();
1030 >        return ((ForkJoinWorkerThread) Thread.currentThread())
1031 >            .pollTask();
1032      }
1033  
1034      // Serialization support
# Line 1024 | Line 1060 | public abstract class ForkJoinTask<V> im
1060          status |= EXTERNAL_SIGNAL; // conservatively set external signal
1061          Object ex = s.readObject();
1062          if (ex != null)
1063 <            setDoneExceptionally((Throwable)ex);
1063 >            setDoneExceptionally((Throwable) ex);
1064      }
1065  
1066      // Temporary Unsafe mechanics for preliminary release

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines