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.81 by jsr166, Mon Sep 20 20:42:36 2010 UTC vs.
Revision 1.84 by dl, Sat Nov 13 13:11:51 2010 UTC

# Line 705 | 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 791 | 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.isTerminating())
797                      break;                    // recheck after clear
# 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;
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 1032 | 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 1100 | Line 1128 | public class ForkJoinPool extends Abstra
1128          // Finish now if all threads terminated; else in some subsequent call
1129          if ((workerCounts >>> TOTAL_COUNT_SHIFT) == 0) {
1130              advanceRunLevel(TERMINATED);
1131 <            termination.arrive();
1131 >            termination.forceTermination();
1132          }
1133          return true;
1134      }
# Line 1296 | 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) {
1302 –        if (task == null)
1303 –            throw new NullPointerException();
1304 –        if (runState >= SHUTDOWN)
1305 –            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 1319 | 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 1332 | 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 1343 | 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 1361 | 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 1371 | 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 1382 | 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 1393 | 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 1735 | Line 1796 | public class ForkJoinPool extends Abstra
1796      public boolean awaitTermination(long timeout, TimeUnit unit)
1797          throws InterruptedException {
1798          try {
1799 <            return termination.awaitAdvanceInterruptibly(0, timeout, unit) > 0;
1799 >            termination.awaitAdvanceInterruptibly(0, timeout, unit);
1800          } catch (TimeoutException ex) {
1801              return false;
1802          }
1803 +        return true;
1804      }
1805  
1806      /**

Diff Legend

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