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.77 by dl, Tue Sep 7 14:43:31 2010 UTC vs.
Revision 1.83 by dl, Sun Oct 24 19:37:26 2010 UTC

# Line 11 | Line 11 | import java.util.Arrays;
11   import java.util.Collection;
12   import java.util.Collections;
13   import java.util.List;
14 + import java.util.concurrent.AbstractExecutorService;
15 + import java.util.concurrent.Callable;
16 + import java.util.concurrent.ExecutorService;
17 + import java.util.concurrent.Future;
18 + import java.util.concurrent.RejectedExecutionException;
19 + import java.util.concurrent.RunnableFuture;
20 + import java.util.concurrent.TimeUnit;
21 + import java.util.concurrent.TimeoutException;
22 + import java.util.concurrent.atomic.AtomicInteger;
23   import java.util.concurrent.locks.LockSupport;
24   import java.util.concurrent.locks.ReentrantLock;
16 – import java.util.concurrent.atomic.AtomicInteger;
17 – import java.util.concurrent.CountDownLatch;
25  
26   /**
27   * An {@link ExecutorService} for running {@link ForkJoinTask}s.
# Line 515 | Line 522 | public class ForkJoinPool extends Abstra
522       * Lifecycle control. The low word contains the number of workers
523       * that are (probably) executing tasks. This value is atomically
524       * incremented before a worker gets a task to run, and decremented
525 <     * when worker has no tasks and cannot find any.  Bits 16-18
525 >     * when a worker has no tasks and cannot find any.  Bits 16-18
526       * contain runLevel value. When all are zero, the pool is
527       * running. Level transitions are monotonic (running -> shutdown
528       * -> terminating -> terminated) so each transition adds a bit.
# Line 698 | Line 705 | public class ForkJoinPool extends Abstra
705       */
706      final void workerTerminated(ForkJoinWorkerThread w) {
707          forgetWorker(w);
708 <        decrementWorkerCounts(w.isTrimmed()? 0 : ONE_RUNNING, ONE_TOTAL);
708 >        decrementWorkerCounts(w.isTrimmed() ? 0 : ONE_RUNNING, ONE_TOTAL);
709          while (w.stealCount != 0) // collect final count
710              tryAccumulateStealCount(w);
711          tryTerminate(false);
# Line 784 | Line 791 | public class ForkJoinPool extends Abstra
791              if (tryAccumulateStealCount(w)) { // transfer while idle
792                  boolean untimed = (w.nextWaiter != 0L ||
793                                     (workerCounts & RUNNING_COUNT_MASK) <= 1);
794 <                long startTime = untimed? 0 : System.nanoTime();
794 >                long startTime = untimed ? 0 : System.nanoTime();
795                  Thread.interrupted();         // clear/ignore interrupt
796 <                if (eventCount != ec || w.runState != 0 ||
797 <                    runState >= TERMINATING)  // recheck after clear
791 <                    break;
796 >                if (eventCount != ec || w.isTerminating())
797 >                    break;                    // recheck after clear
798                  if (untimed)
799                      LockSupport.park(w);
800                  else {
801                      LockSupport.parkNanos(w, SHRINK_RATE_NANOS);
802 <                    if (eventCount != ec || w.runState != 0 ||
797 <                        runState >= TERMINATING)
802 >                    if (eventCount != ec || w.isTerminating())
803                          break;
804                      if (System.nanoTime() - startTime >= SHRINK_RATE_NANOS)
805                          tryShutdownUnusedWorker(ec);
# Line 862 | Line 867 | public class ForkJoinPool extends Abstra
867                       UNSAFE.compareAndSwapInt(this, workerCountsOffset, wc,
868                                                wc + (ONE_RUNNING|ONE_TOTAL))) {
869                  ForkJoinWorkerThread w = null;
870 +                Throwable fail = null;
871                  try {
872                      w = factory.newThread(this);
873 <                } finally { // adjust on null or exceptional factory return
874 <                    if (w == null) {
869 <                        decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
870 <                        tryTerminate(false); // handle failure during shutdown
871 <                    }
873 >                } catch (Throwable ex) {
874 >                    fail = ex;
875                  }
876 <                if (w == null)
876 >                if (w == null) { // null or exceptional factory return
877 >                    decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
878 >                    tryTerminate(false); // handle failure during shutdown
879 >                    // If originating from an external caller,
880 >                    // propagate exception, else ignore
881 >                    if (fail != null && runState < TERMINATING &&
882 >                        !(Thread.currentThread() instanceof
883 >                          ForkJoinWorkerThread))
884 >                        UNSAFE.throwException(fail);
885                      break;
886 +                }
887                  w.start(recordWorker(w), ueh);
888                  if ((workerCounts >>> TOTAL_COUNT_SHIFT) >= pc) {
889                      int c; // advance event count
# Line 960 | Line 972 | public class ForkJoinPool extends Abstra
972          boolean active = w.active;
973          boolean inactivate = false;
974          int pc = parallelism;
975 <        int rs;
976 <        while (w.runState == 0 && (rs = runState) < TERMINATING) {
975 >        while (w.runState == 0) {
976 >            int rs = runState;
977 >            if (rs >= TERMINATING) { // propagate shutdown
978 >                w.shutdown();
979 >                break;
980 >            }
981              if ((inactivate || (active && (rs & ACTIVE_COUNT_MASK) >= pc)) &&
982                  UNSAFE.compareAndSwapInt(this, runStateOffset, rs, rs - 1))
983                  inactivate = active = w.active = false;
# Line 998 | 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;
1030 +            }
1031              worker.helpJoinTask(joinMe);
1032              if (joinMe.status < 0)
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 1016 | 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));
# Line 1089 | Line 1133 | public class ForkJoinPool extends Abstra
1133          return true;
1134      }
1135  
1136 +
1137      /**
1138       * Actions on transition to TERMINATING
1139       *
# Line 1112 | Line 1157 | public class ForkJoinPool extends Abstra
1157                      if (passes > 0 && !w.isTerminated()) {
1158                          w.cancelTasks();
1159                          LockSupport.unpark(w);
1160 <                        if (passes > 1) {
1160 >                        if (passes > 1 && !w.isInterrupted()) {
1161                              try {
1162                                  w.interrupt();
1163                              } catch (SecurityException ignore) {
# Line 1279 | Line 1324 | public class ForkJoinPool extends Abstra
1324      // Execution methods
1325  
1326      /**
1327 <     * Common code for execute, invoke and submit
1327 >     * Submits task and creates, starts, or resumes some workers if necessary
1328       */
1329      private <T> void doSubmit(ForkJoinTask<T> task) {
1285 –        if (task == null)
1286 –            throw new NullPointerException();
1287 –        if (runState >= SHUTDOWN)
1288 –            throw new RejectedExecutionException();
1330          submissionQueue.offer(task);
1331          int c; // try to increment event count -- CAS failure OK
1332          UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
1333 <        helpMaintainParallelism(); // create, start, or resume some workers
1333 >        helpMaintainParallelism();
1334      }
1335  
1336      /**
# Line 1302 | Line 1343 | public class ForkJoinPool extends Abstra
1343       *         scheduled for execution
1344       */
1345      public <T> T invoke(ForkJoinTask<T> task) {
1346 <        doSubmit(task);
1347 <        return task.join();
1346 >        if (task == null)
1347 >            throw new NullPointerException();
1348 >        if (runState >= SHUTDOWN)
1349 >            throw new RejectedExecutionException();
1350 >        Thread t = Thread.currentThread();
1351 >        if ((t instanceof ForkJoinWorkerThread) &&
1352 >            ((ForkJoinWorkerThread)t).pool == this)
1353 >            return task.invoke();  // bypass submit if in same pool
1354 >        else {
1355 >            doSubmit(task);
1356 >            return task.join();
1357 >        }
1358 >    }
1359 >
1360 >    /**
1361 >     * Unless terminating, forks task if within an ongoing FJ
1362 >     * computation in the current pool, else submits as external task.
1363 >     */
1364 >    private <T> void forkOrSubmit(ForkJoinTask<T> task) {
1365 >        if (runState >= SHUTDOWN)
1366 >            throw new RejectedExecutionException();
1367 >        Thread t = Thread.currentThread();
1368 >        if ((t instanceof ForkJoinWorkerThread) &&
1369 >            ((ForkJoinWorkerThread)t).pool == this)
1370 >            task.fork();
1371 >        else
1372 >            doSubmit(task);
1373      }
1374  
1375      /**
# Line 1315 | Line 1381 | public class ForkJoinPool extends Abstra
1381       *         scheduled for execution
1382       */
1383      public void execute(ForkJoinTask<?> task) {
1384 <        doSubmit(task);
1384 >        if (task == null)
1385 >            throw new NullPointerException();
1386 >        forkOrSubmit(task);
1387      }
1388  
1389      // AbstractExecutorService methods
# Line 1326 | Line 1394 | public class ForkJoinPool extends Abstra
1394       *         scheduled for execution
1395       */
1396      public void execute(Runnable task) {
1397 +        if (task == null)
1398 +            throw new NullPointerException();
1399          ForkJoinTask<?> job;
1400          if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1401              job = (ForkJoinTask<?>) task;
1402          else
1403              job = ForkJoinTask.adapt(task, null);
1404 <        doSubmit(job);
1404 >        forkOrSubmit(job);
1405      }
1406  
1407      /**
# Line 1344 | Line 1414 | public class ForkJoinPool extends Abstra
1414       *         scheduled for execution
1415       */
1416      public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
1417 <        doSubmit(task);
1417 >        if (task == null)
1418 >            throw new NullPointerException();
1419 >        forkOrSubmit(task);
1420          return task;
1421      }
1422  
# Line 1354 | Line 1426 | public class ForkJoinPool extends Abstra
1426       *         scheduled for execution
1427       */
1428      public <T> ForkJoinTask<T> submit(Callable<T> task) {
1429 +        if (task == null)
1430 +            throw new NullPointerException();
1431          ForkJoinTask<T> job = ForkJoinTask.adapt(task);
1432 <        doSubmit(job);
1432 >        forkOrSubmit(job);
1433          return job;
1434      }
1435  
# Line 1365 | Line 1439 | public class ForkJoinPool extends Abstra
1439       *         scheduled for execution
1440       */
1441      public <T> ForkJoinTask<T> submit(Runnable task, T result) {
1442 +        if (task == null)
1443 +            throw new NullPointerException();
1444          ForkJoinTask<T> job = ForkJoinTask.adapt(task, result);
1445 <        doSubmit(job);
1445 >        forkOrSubmit(job);
1446          return job;
1447      }
1448  
# Line 1376 | Line 1452 | public class ForkJoinPool extends Abstra
1452       *         scheduled for execution
1453       */
1454      public ForkJoinTask<?> submit(Runnable task) {
1455 +        if (task == null)
1456 +            throw new NullPointerException();
1457          ForkJoinTask<?> job;
1458          if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1459              job = (ForkJoinTask<?>) task;
1460          else
1461              job = ForkJoinTask.adapt(task, null);
1462 <        doSubmit(job);
1462 >        forkOrSubmit(job);
1463          return job;
1464      }
1465  
# Line 1689 | Line 1767 | public class ForkJoinPool extends Abstra
1767      }
1768  
1769      /**
1770 +     * Returns true if terminating or terminated. Used by ForkJoinWorkerThread.
1771 +     */
1772 +    final boolean isAtLeastTerminating() {
1773 +        return runState >= TERMINATING;
1774 +    }
1775 +
1776 +    /**
1777       * Returns {@code true} if this pool has been shut down.
1778       *
1779       * @return {@code true} if this pool has been shut down

Diff Legend

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