ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/ForkJoinPool.java
(Generate patch)

Comparing jsr166/src/main/java/util/concurrent/ForkJoinPool.java (file contents):
Revision 1.42 by jsr166, Fri Oct 22 05:18:30 2010 UTC vs.
Revision 1.43 by dl, Sun Oct 24 19:36:22 2010 UTC

# Line 1014 | Line 1014 | public class ForkJoinPool extends Abstra
1014       *
1015       * @param joinMe the task to join
1016       * @param worker the current worker thread
1017 +     * @param timed true if wait should time out
1018 +     * @param nanos timeout value if timed
1019       */
1020 <    final void awaitJoin(ForkJoinTask<?> joinMe, ForkJoinWorkerThread worker) {
1020 >    final void awaitJoin(ForkJoinTask<?> joinMe, ForkJoinWorkerThread worker,
1021 >                         boolean timed, long nanos) {
1022 >        long startTime = timed? System.nanoTime() : 0L;
1023          int retries = 2 + (parallelism >> 2); // #helpJoins before blocking
1024          while (joinMe.status >= 0) {
1025              int wc;
1026 +            long nt = 0L;
1027              if (runState >= TERMINATING) {
1028                  joinMe.cancelIgnoringExceptions();
1029                  break;
# Line 1028 | Line 1033 | public class ForkJoinPool extends Abstra
1033                  break;
1034              else if (retries > 0)
1035                  --retries;
1036 +            else if (timed &&
1037 +                     (nt = nanos - (System.nanoTime() - startTime)) <= 0L)
1038 +                break;
1039              else if (((wc = workerCounts) & RUNNING_COUNT_MASK) != 0 &&
1040                       UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1041                                                wc, wc - ONE_RUNNING)) {
# Line 1036 | Line 1044 | public class ForkJoinPool extends Abstra
1044                         (h = eventWaiters) != 0L && // help release others
1045                         (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
1046                      releaseEventWaiters();
1047 <                if (stat >= 0 &&
1048 <                    ((workerCounts & RUNNING_COUNT_MASK) == 0 ||
1049 <                     (stat =
1050 <                      joinMe.internalAwaitDone(JOIN_TIMEOUT_MILLIS)) >= 0))
1051 <                    helpMaintainParallelism(); // timeout or no running workers
1047 >                if (stat >= 0) {
1048 >                    if ((workerCounts & RUNNING_COUNT_MASK) != 0) {
1049 >                        long ms; int ns;
1050 >                        if (!timed) {
1051 >                            ms = JOIN_TIMEOUT_MILLIS;
1052 >                            ns = 0;
1053 >                        }
1054 >                        else { // at most JOIN_TIMEOUT_MILLIS per wait
1055 >                            ms = nt / 1000000;
1056 >                            if (ms > JOIN_TIMEOUT_MILLIS) {
1057 >                                ms = JOIN_TIMEOUT_MILLIS;
1058 >                                ns = 0;
1059 >                            }
1060 >                            else
1061 >                                ns = (int) (nt % 1000000);
1062 >                        }
1063 >                        stat = joinMe.internalAwaitDone(ms, ns);
1064 >                    }
1065 >                    if (stat >= 0) // timeout or no running workers
1066 >                        helpMaintainParallelism();
1067 >                }
1068                  do {} while (!UNSAFE.compareAndSwapInt
1069                               (this, workerCountsOffset,
1070                                c = workerCounts, c + ONE_RUNNING));

Diff Legend

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