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

Comparing jsr166/src/jsr166y/ForkJoinPool.java (file contents):
Revision 1.84 by dl, Sat Nov 13 13:11:51 2010 UTC vs.
Revision 1.85 by dl, Sun Nov 21 13:55:04 2010 UTC

# Line 596 | Line 596 | public class ForkJoinPool extends Abstra
596      }
597  
598      /**
599 +     * Tries to increment running count part of workerCounts
600 +     */
601 +    final boolean tryIncrementRunningCount() {
602 +        int c;
603 +        return UNSAFE.compareAndSwapInt(this, workerCountsOffset,
604 +                                        c = workerCounts,
605 +                                        c + ONE_RUNNING);
606 +    }
607 +
608 +    /**
609       * Tries to decrement running count unless already zero
610       */
611      final boolean tryDecrementRunningCount() {
# Line 669 | Line 679 | public class ForkJoinPool extends Abstra
679                  for (k = 0; k < n && ws[k] != null; ++k)
680                      ;
681                  if (k == n)
682 <                    ws = Arrays.copyOf(ws, n << 1);
682 >                    ws = workers = Arrays.copyOf(ws, n << 1);
683              }
684              ws[k] = w;
685 <            workers = ws; // volatile array write ensures slot visibility
685 >            int c = eventCount; // advance event count to ensure visibility
686 >            UNSAFE.compareAndSwapInt(this, eventCountOffset, c, c+1);
687          } finally {
688              lock.unlock();
689          }
# Line 793 | Line 804 | public class ForkJoinPool extends Abstra
804                                     (workerCounts & RUNNING_COUNT_MASK) <= 1);
805                  long startTime = untimed ? 0 : System.nanoTime();
806                  Thread.interrupted();         // clear/ignore interrupt
807 <                if (eventCount != ec || w.isTerminating())
807 >                if (w.isTerminating() || eventCount != ec)
808                      break;                    // recheck after clear
809                  if (untimed)
810                      LockSupport.park(w);
# Line 831 | Line 842 | public class ForkJoinPool extends Abstra
842          if ((sw = spareWaiters) != 0 &&
843              (id = (sw & SPARE_ID_MASK) - 1) >= 0 &&
844              id < n && (w = ws[id]) != null &&
845 <            (workerCounts & RUNNING_COUNT_MASK) < parallelism &&
845 >            (runState >= TERMINATING ||
846 >             (workerCounts & RUNNING_COUNT_MASK) < parallelism) &&
847              spareWaiters == sw &&
848              UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
849                                       sw, w.nextSpare)) {
# Line 885 | Line 897 | public class ForkJoinPool extends Abstra
897                      break;
898                  }
899                  w.start(recordWorker(w), ueh);
900 <                if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc) {
889 <                    int c; // advance event count
890 <                    UNSAFE.compareAndSwapInt(this, eventCountOffset,
891 <                                             c = eventCount, c+1);
900 >                if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc)
901                      break; // add at most one unless total below target
893 –                }
902              }
903          }
904          if (eventWaiters != 0L)
# Line 974 | Line 982 | public class ForkJoinPool extends Abstra
982          int pc = parallelism;
983          while (w.runState == 0) {
984              int rs = runState;
985 <            if (rs >= TERMINATING) { // propagate shutdown
985 >            if (rs >= TERMINATING) {           // propagate shutdown
986                  w.shutdown();
987                  break;
988              }
989              if ((inactivate || (active && (rs & ACTIVE_COUNT_MASK) >= pc)) &&
990 <                UNSAFE.compareAndSwapInt(this, runStateOffset, rs, rs - 1))
990 >                UNSAFE.compareAndSwapInt(this, runStateOffset, rs, --rs)) {
991                  inactivate = active = w.active = false;
992 <            int wc = workerCounts;
992 >                if (rs == SHUTDOWN) {          // all inactive and shut down
993 >                    tryTerminate(false);
994 >                    continue;
995 >                }
996 >            }
997 >            int wc = workerCounts;             // try to suspend as spare
998              if ((wc & RUNNING_COUNT_MASK) > pc) {
999                  if (!(inactivate |= active) && // must inactivate to suspend
1000 <                    workerCounts == wc &&      // try to suspend as spare
1000 >                    workerCounts == wc &&
1001                      UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1002                                               wc, wc - ONE_RUNNING))
1003                      w.suspendAsSpare();
1004              }
1005              else if ((wc >>> TOTAL_COUNT_SHIFT) < pc)
1006                  helpMaintainParallelism();     // not enough workers
1007 <            else if (!ran) {
1007 >            else if (ran)
1008 >                break;
1009 >            else {
1010                  long h = eventWaiters;
1011                  int ec = eventCount;
1012                  if (h != 0L && (int)(h >>> EVENT_COUNT_SHIFT) != ec)
# Line 1003 | Line 1018 | public class ForkJoinPool extends Abstra
1018                  else if (!(inactivate |= active))
1019                      eventSync(w, wec);         // must inactivate before sync
1020              }
1006 –            else
1007 –                break;
1021          }
1022      }
1023  
# Line 1021 | Line 1034 | public class ForkJoinPool extends Abstra
1034                           boolean timed, long nanos) {
1035          long startTime = timed? System.nanoTime() : 0L;
1036          int retries = 2 + (parallelism >> 2); // #helpJoins before blocking
1037 +        boolean running = true;               // false when count decremented
1038          while (joinMe.status >= 0) {
1025 –            int wc;
1026 –            long nt = 0L;
1039              if (runState >= TERMINATING) {
1040                  joinMe.cancelIgnoringExceptions();
1041                  break;
1042              }
1043 <            worker.helpJoinTask(joinMe);
1043 >            running = worker.helpJoinTask(joinMe, running);
1044              if (joinMe.status < 0)
1045                  break;
1046 <            else if (retries > 0)
1046 >            if (retries > 0) {
1047                  --retries;
1048 <            else if (timed &&
1049 <                     (nt = nanos - (System.nanoTime() - startTime)) <= 0L)
1050 <                break;
1051 <            else if (((wc = workerCounts) & RUNNING_COUNT_MASK) != 0 &&
1052 <                     UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1053 <                                              wc, wc - ONE_RUNNING)) {
1054 <                int stat, c; long h;
1055 <                while ((stat = joinMe.status) >= 0 &&
1056 <                       (h = eventWaiters) != 0L && // help release others
1057 <                       (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
1048 >                continue;
1049 >            }
1050 >            int wc = workerCounts;
1051 >            if ((wc & RUNNING_COUNT_MASK) != 0) {
1052 >                if (running) {
1053 >                    if (!UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1054 >                                                  wc, wc - ONE_RUNNING))
1055 >                        continue;
1056 >                    running = false;
1057 >                }
1058 >                long h = eventWaiters;
1059 >                if (h != 0L && (int)(h >>> EVENT_COUNT_SHIFT) != eventCount)
1060                      releaseEventWaiters();
1061 <                if (stat >= 0) {
1062 <                    if ((workerCounts & RUNNING_COUNT_MASK) != 0) {
1063 <                        long ms; int ns;
1064 <                        if (!timed) {
1061 >                if (joinMe.status < 0)
1062 >                    break;
1063 >                if ((workerCounts & RUNNING_COUNT_MASK) != 0) {
1064 >                    long ms; int ns;
1065 >                    if (!timed) {
1066 >                        ms = JOIN_TIMEOUT_MILLIS;
1067 >                        ns = 0;
1068 >                    }
1069 >                    else { // at most JOIN_TIMEOUT_MILLIS per wait
1070 >                        long nt = nanos - (System.nanoTime() - startTime);
1071 >                        if (nt <= 0L)
1072 >                            break;
1073 >                        ms = nt / 1000000;
1074 >                        if (ms > JOIN_TIMEOUT_MILLIS) {
1075                              ms = JOIN_TIMEOUT_MILLIS;
1076                              ns = 0;
1077                          }
1078 <                        else { // at most JOIN_TIMEOUT_MILLIS per wait
1079 <                            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);
1078 >                        else
1079 >                            ns = (int) (nt % 1000000);
1080                      }
1081 <                    if (stat >= 0) // timeout or no running workers
1082 <                        helpMaintainParallelism();
1081 >                    if (joinMe.internalAwaitDone(ms, ns) < 0)
1082 >                        break;
1083                  }
1068 –                do {} while (!UNSAFE.compareAndSwapInt
1069 –                             (this, workerCountsOffset,
1070 –                              c = workerCounts, c + ONE_RUNNING));
1071 –                if (stat < 0)
1072 –                    break;   // else restart
1084              }
1085 +            helpMaintainParallelism();
1086 +        }
1087 +        if (!running) {
1088 +            int c;
1089 +            do {} while (!UNSAFE.compareAndSwapInt
1090 +                         (this, workerCountsOffset,
1091 +                          c = workerCounts, c + ONE_RUNNING));
1092          }
1093      }
1094  
# Line 1081 | Line 1099 | public class ForkJoinPool extends Abstra
1099          throws InterruptedException {
1100          while (!blocker.isReleasable()) {
1101              int wc = workerCounts;
1102 <            if ((wc & RUNNING_COUNT_MASK) != 0 &&
1103 <                UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1104 <                                         wc, wc - ONE_RUNNING)) {
1102 >            if ((wc & RUNNING_COUNT_MASK) == 0)
1103 >                helpMaintainParallelism();
1104 >            else if (UNSAFE.compareAndSwapInt(this, workerCountsOffset,
1105 >                                              wc, wc - ONE_RUNNING)) {
1106                  try {
1107                      while (!blocker.isReleasable()) {
1108                          long h = eventWaiters;
# Line 1133 | Line 1152 | public class ForkJoinPool extends Abstra
1152          return true;
1153      }
1154  
1136 –
1155      /**
1156       * Actions on transition to TERMINATING
1157       *

Diff Legend

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