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.71 by jsr166, Mon Sep 6 21:36:43 2010 UTC vs.
Revision 1.82 by dl, Sun Oct 10 11:56:11 2010 UTC

#	Line 6 | Line 6
6
7		package jsr166y;
8
9	–	import java.util.concurrent.*;
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;
17	–	import java.util.concurrent.atomic.AtomicInteger;
18	–	import java.util.concurrent.CountDownLatch;
25
26		/**
27		* An {@link ExecutorService} for running {@link ForkJoinTask}s.
#	Line 300 | 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 429 | 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 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 604 | 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 786 | 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
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 806 | 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 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 1003 | Line 1019 | public class ForkJoinPool extends Abstra
1019		int retries = 2 + (parallelism >> 2); // #helpJoins before blocking
1020		while (joinMe.status >= 0) {
1021		int wc;
1022	+	if (runState >= TERMINATING) {
1023	+	joinMe.cancelIgnoringExceptions();
1024	+	break;
1025	+	}
1026		worker.helpJoinTask(joinMe);
1027		if (joinMe.status < 0)
1028		break;
#	Line 1089 | Line 1109 | public class ForkJoinPool extends Abstra
1109		return true;
1110		}
1111
1112	+
1113		/**
1114		* Actions on transition to TERMINATING
1115		*
#	Line 1112 | Line 1133 | public class ForkJoinPool extends Abstra
1133		if (passes > 0 && !w.isTerminated()) {
1134		w.cancelTasks();
1135		LockSupport.unpark(w);
1136	<	if (passes > 1) {
1136	>	if (passes > 1 && !w.isInterrupted()) {
1137		try {
1138		w.interrupt();
1139		} catch (SecurityException ignore) {
#	Line 1125 | Line 1146 | public class ForkJoinPool extends Abstra
1146		}
1147
1148		/**
1149	<	* Clear out and cancel submissions, ignoring exceptions
1149	>	* Clears out and cancels submissions, ignoring exceptions.
1150		*/
1151		private void cancelSubmissions() {
1152		ForkJoinTask<?> task;
#	Line 1140 | Line 1161 | public class ForkJoinPool extends Abstra
1161		// misc support for ForkJoinWorkerThread
1162
1163		/**
1164	<	* Returns pool number
1164	>	* Returns pool number.
1165		*/
1166		final int getPoolNumber() {
1167		return poolNumber;
1168		}
1169
1170		/**
1171	<	* Tries to accumulates steal count from a worker, clearing
1172	<	* the worker's value.
1171	>	* Tries to accumulate steal count from a worker, clearing
1172	>	* the worker's value if successful.
1173		*
1174		* @return true if worker steal count now zero
1175		*/
#	Line 1172 | Line 1193 | public class ForkJoinPool extends Abstra
1193		int pc = parallelism; // use parallelism, not rc
1194		int ac = runState;    // no mask -- artificially boosts during shutdown
1195		// Use exact results for small values, saturate past 4
1196	<	return pc <= ac? 0 : pc >>> 1 <= ac? 1 : pc >>> 2 <= ac? 3 : pc >>> 3;
1196	>	return ((pc <= ac) ? 0 :
1197	>	(pc >>> 1 <= ac) ? 1 :
1198	>	(pc >>> 2 <= ac) ? 3 :
1199	>	pc >>> 3);
1200		}
1201
1202		// Public and protected methods
#	Line 1222 | Line 1246 | public class ForkJoinPool extends Abstra
1246		* use {@link #defaultForkJoinWorkerThreadFactory}.
1247		* @param handler the handler for internal worker threads that
1248		* terminate due to unrecoverable errors encountered while executing
1249	<	* tasks. For default value, use <code>null</code>.
1249	>	* tasks. For default value, use {@code null}.
1250		* @param asyncMode if true,
1251		* establishes local first-in-first-out scheduling mode for forked
1252		* tasks that are never joined. This mode may be more appropriate
1253		* than default locally stack-based mode in applications in which
1254		* worker threads only process event-style asynchronous tasks.
1255	<	* For default value, use <code>false</code>.
1255	>	* For default value, use {@code false}.
1256		* @throws IllegalArgumentException if parallelism less than or
1257		*         equal to zero, or greater than implementation limit
1258		* @throws NullPointerException if the factory is null
#	Line 1276 | Line 1300 | public class ForkJoinPool extends Abstra
1300		// Execution methods
1301
1302		/**
1303	<	* Common code for execute, invoke and submit
1303	>	* Submits task and creates, starts, or resumes some workers if necessary
1304		*/
1305		private <T> void doSubmit(ForkJoinTask<T> task) {
1282	–	if (task == null)
1283	–	throw new NullPointerException();
1284	–	if (runState >= SHUTDOWN)
1285	–	throw new RejectedExecutionException();
1306		submissionQueue.offer(task);
1307		int c; // try to increment event count -- CAS failure OK
1308		UNSAFE.compareAndSwapInt(this, eventCountOffset, c = eventCount, c+1);
1309	<	helpMaintainParallelism(); // create, start, or resume some workers
1309	>	helpMaintainParallelism();
1310		}
1311
1312		/**
#	Line 1299 | Line 1319 | public class ForkJoinPool extends Abstra
1319		*         scheduled for execution
1320		*/
1321		public <T> T invoke(ForkJoinTask<T> task) {
1322	<	doSubmit(task);
1323	<	return task.join();
1322	>	if (task == null)
1323	>	throw new NullPointerException();
1324	>	if (runState >= SHUTDOWN)
1325	>	throw new RejectedExecutionException();
1326	>	Thread t = Thread.currentThread();
1327	>	if ((t instanceof ForkJoinWorkerThread) &&
1328	>	((ForkJoinWorkerThread)t).pool == this)
1329	>	return task.invoke();  // bypass submit if in same pool
1330	>	else {
1331	>	doSubmit(task);
1332	>	return task.join();
1333	>	}
1334	>	}
1335	>
1336	>	/**
1337	>	* Unless terminating, forks task if within an ongoing FJ
1338	>	* computation in the current pool, else submits as external task.
1339	>	*/
1340	>	private <T> void forkOrSubmit(ForkJoinTask<T> task) {
1341	>	if (runState >= SHUTDOWN)
1342	>	throw new RejectedExecutionException();
1343	>	Thread t = Thread.currentThread();
1344	>	if ((t instanceof ForkJoinWorkerThread) &&
1345	>	((ForkJoinWorkerThread)t).pool == this)
1346	>	task.fork();
1347	>	else
1348	>	doSubmit(task);
1349		}
1350
1351		/**
#	Line 1312 | Line 1357 | public class ForkJoinPool extends Abstra
1357		*         scheduled for execution
1358		*/
1359		public void execute(ForkJoinTask<?> task) {
1360	<	doSubmit(task);
1360	>	if (task == null)
1361	>	throw new NullPointerException();
1362	>	forkOrSubmit(task);
1363		}
1364
1365		// AbstractExecutorService methods
#	Line 1323 | Line 1370 | public class ForkJoinPool extends Abstra
1370		*         scheduled for execution
1371		*/
1372		public void execute(Runnable task) {
1373	+	if (task == null)
1374	+	throw new NullPointerException();
1375		ForkJoinTask<?> job;
1376		if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1377		job = (ForkJoinTask<?>) task;
1378		else
1379		job = ForkJoinTask.adapt(task, null);
1380	<	doSubmit(job);
1380	>	forkOrSubmit(job);
1381		}
1382
1383		/**
#	Line 1341 | Line 1390 | public class ForkJoinPool extends Abstra
1390		*         scheduled for execution
1391		*/
1392		public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
1393	<	doSubmit(task);
1393	>	if (task == null)
1394	>	throw new NullPointerException();
1395	>	forkOrSubmit(task);
1396		return task;
1397		}
1398
#	Line 1351 | Line 1402 | public class ForkJoinPool extends Abstra
1402		*         scheduled for execution
1403		*/
1404		public <T> ForkJoinTask<T> submit(Callable<T> task) {
1405	+	if (task == null)
1406	+	throw new NullPointerException();
1407		ForkJoinTask<T> job = ForkJoinTask.adapt(task);
1408	<	doSubmit(job);
1408	>	forkOrSubmit(job);
1409		return job;
1410		}
1411
#	Line 1362 | Line 1415 | public class ForkJoinPool extends Abstra
1415		*         scheduled for execution
1416		*/
1417		public <T> ForkJoinTask<T> submit(Runnable task, T result) {
1418	+	if (task == null)
1419	+	throw new NullPointerException();
1420		ForkJoinTask<T> job = ForkJoinTask.adapt(task, result);
1421	<	doSubmit(job);
1421	>	forkOrSubmit(job);
1422		return job;
1423		}
1424
#	Line 1373 | Line 1428 | public class ForkJoinPool extends Abstra
1428		*         scheduled for execution
1429		*/
1430		public ForkJoinTask<?> submit(Runnable task) {
1431	+	if (task == null)
1432	+	throw new NullPointerException();
1433		ForkJoinTask<?> job;
1434		if (task instanceof ForkJoinTask<?>) // avoid re-wrap
1435		job = (ForkJoinTask<?>) task;
1436		else
1437		job = ForkJoinTask.adapt(task, null);
1438	<	doSubmit(job);
1438	>	forkOrSubmit(job);
1439		return job;
1440		}
1441
#	Line 1438 | Line 1495 | public class ForkJoinPool extends Abstra
1495
1496		/**
1497		* Returns the number of worker threads that have started but not
1498	<	* yet terminated.  This result returned by this method may differ
1498	>	* yet terminated.  The result returned by this method may differ
1499		* from {@link #getParallelism} when threads are created to
1500		* maintain parallelism when others are cooperatively blocked.
1501		*
#	Line 1686 | Line 1743 | public class ForkJoinPool extends Abstra
1743		}
1744
1745		/**
1746	+	* Returns true if terminating or terminated. Used by ForkJoinWorkerThread.
1747	+	*/
1748	+	final boolean isAtLeastTerminating() {
1749	+	return runState >= TERMINATING;
1750	+	}
1751	+
1752	+	/**
1753		* Returns {@code true} if this pool has been shut down.
1754		*
1755		* @return {@code true} if this pool has been shut down
#	Line 1839 | Line 1903 | public class ForkJoinPool extends Abstra
1903		private static final long eventCountOffset =
1904		objectFieldOffset("eventCount", ForkJoinPool.class);
1905		private static final long eventWaitersOffset =
1906	<	objectFieldOffset("eventWaiters",ForkJoinPool.class);
1906	>	objectFieldOffset("eventWaiters", ForkJoinPool.class);
1907		private static final long stealCountOffset =
1908	<	objectFieldOffset("stealCount",ForkJoinPool.class);
1908	>	objectFieldOffset("stealCount", ForkJoinPool.class);
1909		private static final long spareWaitersOffset =
1910	<	objectFieldOffset("spareWaiters",ForkJoinPool.class);
1910	>	objectFieldOffset("spareWaiters", ForkJoinPool.class);
1911
1912		private static long objectFieldOffset(String field, Class<?> klazz) {
1913		try {

Diff Legend

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