ViewVC Help

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

root/jsr166/jsr166/src/jsr166e/ForkJoinPool.java

Comparing jsr166/src/jsr166e/ForkJoinPool.java (file contents):
Revision 1.8 by dl, Mon Oct 29 17:23:26 2012 UTC vs.
Revision 1.11 by dl, Wed Oct 31 12:49:13 2012 UTC

#	Line 692 | Line 692 | public class ForkJoinPool extends Abstra
692
693		/**
694		* Takes next task, if one exists, in LIFO order.  Call only
695	<	* by owner in unshared queues. (We do not have a shared
696	<	* version of this method because it is never needed.)
695	>	* by owner in unshared queues.
696		*/
697		final ForkJoinTask<?> pop() {
698		ForkJoinTask<?>[] a; ForkJoinTask<?> t; int m;
#	Line 711 | Line 710 | public class ForkJoinPool extends Abstra
710		return null;
711		}
712
713	+	final ForkJoinTask<?> sharedPop() {
714	+	ForkJoinTask<?> task = null;
715	+	if (runState == 0 && U.compareAndSwapInt(this, RUNSTATE, 0, 1)) {
716	+	try {
717	+	ForkJoinTask<?>[] a; int m;
718	+	if ((a = array) != null && (m = a.length - 1) >= 0) {
719	+	for (int s; (s = top - 1) - base >= 0;) {
720	+	long j = ((m & s) << ASHIFT) + ABASE;
721	+	ForkJoinTask<?> t =
722	+	(ForkJoinTask<?>)U.getObject(a, j);
723	+	if (t == null)
724	+	break;
725	+	if (U.compareAndSwapObject(a, j, t, null)) {
726	+	top = s;
727	+	task = t;
728	+	break;
729	+	}
730	+	}
731	+	}
732	+	} finally {
733	+	runState = 0;
734	+	}
735	+	}
736	+	return task;
737	+	}
738	+
739	+
740		/**
741		* Takes a task in FIFO order if b is base of queue and a task
742		* can be claimed without contention. Specialized versions
#	Line 884 | Line 910 | public class ForkJoinPool extends Abstra
910		return seed = r ^= r << 5;
911		}
912
913	<	// Execution methods
913	>	// Specialized execution methods
914
915		/**
916		* Pops and runs tasks until empty.
#	Line 963 | Line 989 | public class ForkJoinPool extends Abstra
989		}
990
991		/**
992	+	* Version of shared pop that takes top element only if it
993	+	* its root is the given CountedCompleter.
994	+	*/
995	+	final CountedCompleter<?> sharedPopCC(CountedCompleter<?> root) {
996	+	CountedCompleter<?> task = null;
997	+	if (runState == 0 && U.compareAndSwapInt(this, RUNSTATE, 0, 1)) {
998	+	try {
999	+	ForkJoinTask<?>[] a; int m;
1000	+	if ((a = array) != null && (m = a.length - 1) >= 0) {
1001	+	outer:for (int s; (s = top - 1) - base >= 0;) {
1002	+	long j = ((m & s) << ASHIFT) + ABASE;
1003	+	ForkJoinTask<?> t =
1004	+	(ForkJoinTask<?>)U.getObject(a, j);
1005	+	if (t == null || !(t instanceof CountedCompleter))
1006	+	break;
1007	+	CountedCompleter<?> cc = (CountedCompleter<?>)t;
1008	+	for (CountedCompleter<?> q = cc, p;;) {
1009	+	if (q == root) {
1010	+	if (U.compareAndSwapObject(a, j, cc, null)) {
1011	+	top = s;
1012	+	task = cc;
1013	+	break outer;
1014	+	}
1015	+	break;
1016	+	}
1017	+	if ((p = q.completer) == null)
1018	+	break outer;
1019	+	q = p;
1020	+	}
1021	+	}
1022	+	}
1023	+	} finally {
1024	+	runState = 0;
1025	+	}
1026	+	}
1027	+	return task;
1028	+	}
1029	+
1030	+	/**
1031		* Executes a top-level task and any local tasks remaining
1032		* after execution.
1033		*/
#	Line 1083 | Line 1148 | public class ForkJoinPool extends Abstra
1148		public static final ForkJoinWorkerThreadFactory
1149		defaultForkJoinWorkerThreadFactory;
1150
1086	–
1151		/** Property prefix for constructing common pool */
1152		private static final String propPrefix =
1153		"java.util.concurrent.ForkJoinPool.common.";
#	Line 1297 | Line 1361 | public class ForkJoinPool extends Abstra
1361		try {
1362		wait();
1363		} catch (InterruptedException ie) {
1364	<	Thread.currentThread().interrupt();
1364	>	try {
1365	>	Thread.currentThread().interrupt();
1366	>	} catch (SecurityException ignore) {
1367	>	}
1368		}
1369		}
1370		else
#	Line 1335 | Line 1402 | public class ForkJoinPool extends Abstra
1402		*/
1403		final String nextWorkerName() {
1404		int n;
1405	<	do {} while(!U.compareAndSwapInt(this, NEXTWORKERNUMBER,
1406	<	n = nextWorkerNumber, ++n));
1405	>	do {} while (!U.compareAndSwapInt(this, NEXTWORKERNUMBER,
1406	>	n = nextWorkerNumber, ++n));
1407		return workerNamePrefix.concat(Integer.toString(n));
1408		}
1409
#	Line 1382 | Line 1449 | public class ForkJoinPool extends Abstra
1449		synchronized (this) { notifyAll(); };
1450		}
1451		}
1385	–
1452		}
1453
1454		/**
#	Line 1399 | Line 1465 | public class ForkJoinPool extends Abstra
1465		if (wt != null && (w = wt.workQueue) != null) {
1466		w.runState = -1;                // ensure runState is set
1467		long steals = w.totalSteals + w.nsteals, sc;
1468	<	do {} while(!U.compareAndSwapLong(this, STEALCOUNT,
1469	<	sc = stealCount, sc + steals));
1468	>	do {} while (!U.compareAndSwapLong(this, STEALCOUNT,
1469	>	sc = stealCount, sc + steals));
1470		int idx = w.poolIndex;
1471		while (!U.compareAndSwapInt(this, MAINLOCK, 0, 1))
1472		tryAwaitMainLock();
#	Line 1431 | Line 1497 | public class ForkJoinPool extends Abstra
1497		}
1498
1499		if (ex != null)                     // rethrow
1500	<	U.throwException(ex);
1500	>	ForkJoinTask.rethrow(ex);
1501		}
1502
1503		// Submissions
#	Line 1517 | Line 1583 | public class ForkJoinPool extends Abstra
1583		* @return true if successful
1584		*/
1585		static boolean tryUnsubmitFromCommonPool(ForkJoinTask<?> task) {
1586	<	// Peek, looking for task and eligibility before
1587	<	// using trySharedUnpush to actually take it under lock
1588	<	ForkJoinPool p; WorkQueue[] ws; WorkQueue q;
1589	<	ForkJoinTask<?>[] a; int t, s, n;
1590	<	int k = submitters.get().seed & SQMASK;
1591	<	return ((p = commonPool) != null &&
1592	<	(ws = p.workQueues) != null &&
1593	<	ws.length > (k &= p.submitMask) &&
1594	<	(q = ws[k]) != null &&
1595	<	(a = q.array) != null &&
1596	<	(n = (t = q.top) - q.base) > 0 &&
1597	<	(n > 1 || (int)(p.ctl >> AC_SHIFT) < 0) &&
1598	<	(s = t - 1) >= 0 && s < a.length && a[s] == task &&
1599	<	q.trySharedUnpush(task));
1586	>	// If not oversaturating platform, peek, looking for task and
1587	>	// eligibility before using trySharedUnpush to actually take
1588	>	// it under lock
1589	>	ForkJoinPool p; WorkQueue[] ws; WorkQueue w, q;
1590	>	ForkJoinTask<?>[] a; int ac, s, m;
1591	>	if ((p = commonPool) != null && (ws = p.workQueues) != null) {
1592	>	int k = submitters.get().seed & p.submitMask & SQMASK;
1593	>	if ((m = ws.length - 1) >= k && (q = ws[k]) != null &&
1594	>	(ac = (int)(p.ctl >> AC_SHIFT)) <= 0) {
1595	>	if (ac == 0) { // double check if all workers active
1596	>	for (int i = 1; i <= m; i += 2) {
1597	>	if ((w = ws[i]) != null && w.parker != null) {
1598	>	ac = -1;
1599	>	break;
1600	>	}
1601	>	}
1602	>	}
1603	>	return (ac < 0 && (a = q.array) != null &&
1604	>	(s = q.top - 1) - q.base >= 0 &&
1605	>	s >= 0 && s < a.length &&
1606	>	a[s] == task &&
1607	>	q.trySharedUnpush(task));
1608	>	}
1609	>	}
1610	>	return false;
1611	>	}
1612	>
1613	>	/**
1614	>	* Tries to pop and run a task within same computation from common pool
1615	>	*/
1616	>	static void popAndExecCCFromCommonPool(CountedCompleter<?> cc) {
1617	>	ForkJoinPool p; WorkQueue[] ws; WorkQueue q, w; int m, ac;
1618	>	CountedCompleter<?> par, task;
1619	>	if ((p = commonPool) != null && (ws = p.workQueues) != null) {
1620	>	while ((par = cc.completer) != null) // find root
1621	>	cc = par;
1622	>	int k = submitters.get().seed & p.submitMask & SQMASK;
1623	>	if ((m = ws.length - 1) >= k && (q = ws[k]) != null &&
1624	>	(ac = (int)(p.ctl >> AC_SHIFT)) <= 0) {
1625	>	if (ac == 0) {
1626	>	for (int i = 1; i <= m; i += 2) {
1627	>	if ((w = ws[i]) != null && w.parker != null) {
1628	>	ac = -1;
1629	>	break;
1630	>	}
1631	>	}
1632	>	}
1633	>	if (ac < 0 && q.top - q.base > 0 &&
1634	>	(task = q.sharedPopCC(cc)) != null)
1635	>	task.exec();
1636	>	}
1637	>	}
1638		}
1639
1640		// Maintaining ctl counts
#	Line 2068 | Line 2172 | public class ForkJoinPool extends Abstra
2172		* Restricted version of helpQuiescePool for non-FJ callers
2173		*/
2174		static void externalHelpQuiescePool() {
2175	<	ForkJoinPool p; WorkQueue[] ws; WorkQueue w, q;
2176	<	ForkJoinTask<?> t; int b;
2175	>	ForkJoinPool p; WorkQueue[] ws; WorkQueue q, sq;
2176	>	ForkJoinTask<?>[] a; int b;
2177	>	ForkJoinTask<?> t = null;
2178		int k = submitters.get().seed & SQMASK;
2179		if ((p = commonPool) != null &&
2180		(ws = p.workQueues) != null &&
2181		ws.length > (k &= p.submitMask) &&
2182	<	(w = ws[k]) != null &&
2183	<	(q = p.findNonEmptyStealQueue(w)) != null &&
2184	<	(b = q.base) - q.top < 0 &&
2185	<	(t = q.pollAt(b)) != null)
2186	<	t.doExec();
2182	>	(q = ws[k]) != null) {
2183	>	while (q.top - q.base > 0) {
2184	>	if ((t = q.sharedPop()) != null)
2185	>	break;
2186	>	}
2187	>	if (t == null && (sq = p.findNonEmptyStealQueue(q)) != null &&
2188	>	(b = sq.base) - sq.top < 0)
2189	>	t = sq.pollAt(b);
2190	>	if (t != null)
2191	>	t.doExec();
2192	>	}
2193		}
2194
2195		/**
#	Line 2120 | Line 2231 | public class ForkJoinPool extends Abstra
2231		static int getEstimatedSubmitterQueueLength() {
2232		ForkJoinPool p; WorkQueue[] ws; WorkQueue q;
2233		int k = submitters.get().seed & SQMASK;
2234	<	return ((p = commonPool) != null &&
2124	<	p.runState >= 0 &&
2125	<	(ws = p.workQueues) != null &&
2234	>	return ((p = commonPool) != null && (ws = p.workQueues) != null &&
2235		ws.length > (k &= p.submitMask) &&
2236		(q = ws[k]) != null) ?
2237		q.queueSize() : 0;
#	Line 2148 | Line 2257 | public class ForkJoinPool extends Abstra
2257		for (long c;;) {
2258		if (((c = ctl) & STOP_BIT) != 0) {      // already terminating
2259		if ((short)(c >>> TC_SHIFT) == -parallelism) {
2260	<	synchronized(this) {
2260	>	synchronized (this) {
2261		notifyAll();                // signal when 0 workers
2262		}
2263		}
#	Line 2872 | Line 2981 | public class ForkJoinPool extends Abstra
2981		return true;
2982		long startTime = System.nanoTime();
2983		boolean terminated = false;
2984	<	synchronized(this) {
2984	>	synchronized (this) {
2985		for (long waitTime = nanos, millis = 0L;;) {
2986		if (terminated = isTerminated() ||
2987		waitTime <= 0L ||
#	Line 3057 | Line 3166 | public class ForkJoinPool extends Abstra
3166		defaultForkJoinWorkerThreadFactory :
3167		((ForkJoinWorkerThreadFactory)ClassLoader.
3168		getSystemClassLoader().loadClass(fp).newInstance());
3169	<	Thread.UncaughtExceptionHandler ueh = (up == null)? null :
3169	>	Thread.UncaughtExceptionHandler ueh = (up == null) ? null :
3170		((Thread.UncaughtExceptionHandler)ClassLoader.
3171		getSystemClassLoader().loadClass(up).newInstance());
3172		int par;

Diff Legend

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