ViewVC Help

View File | Revision Log | Show Annotations | Download File | Root Listing

root/jsr166/jsr166/src/jsr166y/ForkJoinPool.java

Comparing jsr166/src/jsr166y/ForkJoinPool.java (file contents):
Revision 1.67 by jsr166, Wed Sep 1 03:32:03 2010 UTC vs.
Revision 1.81 by jsr166, Mon Sep 20 20:42:36 2010 UTC

#	Line 6 | Line 6
6
7		package jsr166y;
8
9	–	import java.util.concurrent.*;
10	–
9		import java.util.ArrayList;
10		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;
18	–	import java.util.concurrent.atomic.AtomicInteger;
19	–	import java.util.concurrent.CountDownLatch;
25
26		/**
27		* An {@link ExecutorService} for running {@link ForkJoinTask}s.
#	Line 157 | Line 162 | public class ForkJoinPool extends Abstra
162		*      links to try to find such a task.
163		*
164		*   Compensating: Unless there are already enough live threads,
165	<	*      method helpMaintainParallelism() may create or or
165	>	*      method helpMaintainParallelism() may create or
166		*      re-activate a spare thread to compensate for blocked
167		*      joiners until they unblock.
168		*
#	Line 301 | Line 306 | public class ForkJoinPool extends Abstra
306		* about the same time as another is needlessly being created. We
307		* counteract this and related slop in part by requiring resumed
308		* spares to immediately recheck (in preStep) to see whether they
309	<	* they should re-suspend.
309	>	* should re-suspend.
310		*
311		* 6. Killing off unneeded workers. A timeout mechanism is used to
312		* shed unused workers: The oldest (first) event queue waiter uses
#	Line 318 | Line 323 | public class ForkJoinPool extends Abstra
323		* exactly #parallelism threads running, which is an impossible
324		* task. We always need to create one when the number of running
325		* threads would become zero and all workers are busy. Beyond
326	<	* this, we must rely on heuristics that work well in the the
327	<	* presence of transients phenomena such as GC stalls, dynamic
326	>	* this, we must rely on heuristics that work well in the
327	>	* presence of transient phenomena such as GC stalls, dynamic
328		* compilation, and wake-up lags. These transients are extremely
329		* common -- we are normally trying to fully saturate the CPUs on
330		* a machine, so almost any activity other than running tasks
#	Line 346 | Line 351 | public class ForkJoinPool extends Abstra
351		* "while ((local = field) != 0)") which are usually the simplest
352		* way to ensure the required read orderings (which are sometimes
353		* critical). Also several occurrences of the unusual "do {}
354	<	* while(!cas...)" which is the simplest way to force an update of
354	>	* while (!cas...)" which is the simplest way to force an update of
355		* a CAS'ed variable. There are also other coding oddities that
356		* help some methods perform reasonably even when interpreted (not
357		* compiled), at the expense of some messy constructions that
#	Line 430 | Line 435 | public class ForkJoinPool extends Abstra
435
436		/**
437		* The wakeup interval (in nanoseconds) for the oldest worker
438	<	* worker waiting for an event invokes tryShutdownUnusedWorker to shrink
439	<	* the number of workers.  The exact value does not matter too
440	<	* much, but should be long enough to slowly release resources
441	<	* during long periods without use without disrupting normal use.
438	>	* waiting for an event to invoke tryShutdownUnusedWorker to
439	>	* shrink the number of workers.  The exact value does not matter
440	>	* too much. It must be short enough to release resources during
441	>	* sustained periods of idleness, but not so short that threads
442	>	* are continually re-created.
443		*/
444		private static final long SHRINK_RATE_NANOS =
445		30L * 1000L * 1000L * 1000L; // 2 per minute
#	Line 516 | 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 605 | Line 611 | public class ForkJoinPool extends Abstra
611		* (rarely) necessary when other count updates lag.
612		*
613		* @param dr -- either zero or ONE_RUNNING
614	<	* @param dt == either zero or ONE_TOTAL
614	>	* @param dt -- either zero or ONE_TOTAL
615		*/
616		private void decrementWorkerCounts(int dr, int dt) {
617		for (;;) {
#	Line 674 | Line 680 | public class ForkJoinPool extends Abstra
680		}
681
682		/**
683	<	* Nulls out record of worker in workers array
683	>	* Nulls out record of worker in workers array.
684		*/
685		private void forgetWorker(ForkJoinWorkerThread w) {
686		int idx = w.poolIndex;
#	Line 787 | Line 793 | public class ForkJoinPool extends Abstra
793		(workerCounts & RUNNING_COUNT_MASK) <= 1);
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
792	<	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 ||
798	<	runState >= TERMINATING)
802	>	if (eventCount != ec || w.isTerminating())
803		break;
804		if (System.nanoTime() - startTime >= SHRINK_RATE_NANOS)
805		tryShutdownUnusedWorker(ec);
#	Line 807 | Line 811 | public class ForkJoinPool extends Abstra
811		// Maintaining parallelism
812
813		/**
814	<	* Pushes worker onto the spare stack
814	>	* Pushes worker onto the spare stack.
815		*/
816		final void pushSpare(ForkJoinWorkerThread w) {
817		int ns = (++w.spareCount << SPARE_COUNT_SHIFT) | (w.poolIndex + 1);
#	Line 832 | Line 836 | public class ForkJoinPool extends Abstra
836		UNSAFE.compareAndSwapInt(this, spareWaitersOffset,
837		sw, w.nextSpare)) {
838		int c; // increment running count before resume
839	<	do {} while(!UNSAFE.compareAndSwapInt
840	<	(this, workerCountsOffset,
841	<	c = workerCounts, c + ONE_RUNNING));
839	>	do {} while (!UNSAFE.compareAndSwapInt
840	>	(this, workerCountsOffset,
841	>	c = workerCounts, c + ONE_RUNNING));
842		if (w.tryUnsuspend())
843		LockSupport.unpark(w);
844		else   // back out if w was shutdown
#	Line 863 | 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) {
870	<	decrementWorkerCounts(ONE_RUNNING, ONE_TOTAL);
871	<	tryTerminate(false); // handle failure during shutdown
872	<	}
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 961 | 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 985 | Line 1000 | public class ForkJoinPool extends Abstra
1000		w.lastEventCount = ec;     // no need to wait
1001		break;
1002		}
1003	<	else if (!(inactivate |= active))
1003	>	else if (!(inactivate |= active))
1004		eventSync(w, wec);         // must inactivate before sync
1005		}
1006		else
#	Line 1090 | Line 1105 | public class ForkJoinPool extends Abstra
1105		return true;
1106		}
1107
1108	+
1109		/**
1110		* Actions on transition to TERMINATING
1111		*
#	Line 1107 | Line 1123 | public class ForkJoinPool extends Abstra
1123		c = eventCount, c+1);
1124		eventWaiters = 0L; // clobber lists
1125		spareWaiters = 0;
1126	<	ForkJoinWorkerThread[] ws = workers;
1111	<	int n = ws.length;
1112	<	for (int i = 0; i < n; ++i) {
1113	<	ForkJoinWorkerThread w = ws[i];
1126	>	for (ForkJoinWorkerThread w : workers) {
1127		if (w != null) {
1128		w.shutdown();
1129		if (passes > 0 && !w.isTerminated()) {
1130		w.cancelTasks();
1131		LockSupport.unpark(w);
1132	<	if (passes > 1) {
1132	>	if (passes > 1 && !w.isInterrupted()) {
1133		try {
1134		w.interrupt();
1135		} catch (SecurityException ignore) {
#	Line 1129 | Line 1142 | public class ForkJoinPool extends Abstra
1142		}
1143
1144		/**
1145	<	* Clear out and cancel submissions, ignoring exceptions
1145	>	* Clears out and cancels submissions, ignoring exceptions.
1146		*/
1147		private void cancelSubmissions() {
1148		ForkJoinTask<?> task;
#	Line 1144 | Line 1157 | public class ForkJoinPool extends Abstra
1157		// misc support for ForkJoinWorkerThread
1158
1159		/**
1160	<	* Returns pool number
1160	>	* Returns pool number.
1161		*/
1162		final int getPoolNumber() {
1163		return poolNumber;
1164		}
1165
1166		/**
1167	<	* Tries to accumulates steal count from a worker, clearing
1168	<	* the worker's value.
1167	>	* Tries to accumulate steal count from a worker, clearing
1168	>	* the worker's value if successful.
1169		*
1170		* @return true if worker steal count now zero
1171		*/
#	Line 1176 | Line 1189 | public class ForkJoinPool extends Abstra
1189		int pc = parallelism; // use parallelism, not rc
1190		int ac = runState;    // no mask -- artificially boosts during shutdown
1191		// Use exact results for small values, saturate past 4
1192	<	return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
1192	>	return ((pc <= ac) ? 0 :
1193	>	(pc >>> 1 <= ac) ? 1 :
1194	>	(pc >>> 2 <= ac) ? 3 :
1195	>	pc >>> 3);
1196		}
1197
1198		// Public and protected methods
#	Line 1226 | Line 1242 | public class ForkJoinPool extends Abstra
1242		* use {@link #defaultForkJoinWorkerThreadFactory}.
1243		* @param handler the handler for internal worker threads that
1244		* terminate due to unrecoverable errors encountered while executing
1245	<	* tasks. For default value, use <code>null</code>.
1245	>	* tasks. For default value, use {@code null}.
1246		* @param asyncMode if true,
1247		* establishes local first-in-first-out scheduling mode for forked
1248		* tasks that are never joined. This mode may be more appropriate
1249		* than default locally stack-based mode in applications in which
1250		* worker threads only process event-style asynchronous tasks.
1251	<	* For default value, use <code>false</code>.
1251	>	* For default value, use {@code false}.
1252		* @throws IllegalArgumentException if parallelism less than or
1253		*         equal to zero, or greater than implementation limit
1254		* @throws NullPointerException if the factory is null
#	Line 1442 | Line 1458 | public class ForkJoinPool extends Abstra
1458
1459		/**
1460		* Returns the number of worker threads that have started but not
1461	<	* yet terminated.  This result returned by this method may differ
1461	>	* yet terminated.  The result returned by this method may differ
1462		* from {@link #getParallelism} when threads are created to
1463		* maintain parallelism when others are cooperatively blocked.
1464		*
#	Line 1527 | Line 1543 | public class ForkJoinPool extends Abstra
1543		*/
1544		public long getQueuedTaskCount() {
1545		long count = 0;
1546	<	ForkJoinWorkerThread[] ws = workers;
1531	<	int n = ws.length;
1532	<	for (int i = 0; i < n; ++i) {
1533	<	ForkJoinWorkerThread w = ws[i];
1546	>	for (ForkJoinWorkerThread w : workers)
1547		if (w != null)
1548		count += w.getQueueSize();
1536	–	}
1549		return count;
1550		}
1551
#	Line 1588 | Line 1600 | public class ForkJoinPool extends Abstra
1600		*/
1601		protected int drainTasksTo(Collection<? super ForkJoinTask<?>> c) {
1602		int count = submissionQueue.drainTo(c);
1603	<	ForkJoinWorkerThread[] ws = workers;
1592	<	int n = ws.length;
1593	<	for (int i = 0; i < n; ++i) {
1594	<	ForkJoinWorkerThread w = ws[i];
1603	>	for (ForkJoinWorkerThread w : workers)
1604		if (w != null)
1605		count += w.drainTasksTo(c);
1597	–	}
1606		return count;
1607		}
1608
#	Line 1698 | Line 1706 | public class ForkJoinPool extends Abstra
1706		}
1707
1708		/**
1709	+	* Returns true if terminating or terminated. Used by ForkJoinWorkerThread.
1710	+	*/
1711	+	final boolean isAtLeastTerminating() {
1712	+	return runState >= TERMINATING;
1713	+	}
1714	+
1715	+	/**
1716		* Returns {@code true} if this pool has been shut down.
1717		*
1718		* @return {@code true} if this pool has been shut down
#	Line 1721 | Line 1736 | public class ForkJoinPool extends Abstra
1736		throws InterruptedException {
1737		try {
1738		return termination.awaitAdvanceInterruptibly(0, timeout, unit) > 0;
1739	<	} catch(TimeoutException ex) {
1739	>	} catch (TimeoutException ex) {
1740		return false;
1741		}
1742		}
#	Line 1851 | Line 1866 | public class ForkJoinPool extends Abstra
1866		private static final long eventCountOffset =
1867		objectFieldOffset("eventCount", ForkJoinPool.class);
1868		private static final long eventWaitersOffset =
1869	<	objectFieldOffset("eventWaiters",ForkJoinPool.class);
1869	>	objectFieldOffset("eventWaiters", ForkJoinPool.class);
1870		private static final long stealCountOffset =
1871	<	objectFieldOffset("stealCount",ForkJoinPool.class);
1871	>	objectFieldOffset("stealCount", ForkJoinPool.class);
1872		private static final long spareWaitersOffset =
1873	<	objectFieldOffset("spareWaiters",ForkJoinPool.class);
1873	>	objectFieldOffset("spareWaiters", ForkJoinPool.class);
1874
1875		private static long objectFieldOffset(String field, Class<?> klazz) {
1876		try {

Diff Legend

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