ViewVC Help

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

root/jsr166/jsr166/src/jsr166y/ForkJoinTask.java

Comparing jsr166/src/jsr166y/ForkJoinTask.java (file contents):
Revision 1.92 by dl, Wed Oct 31 12:49:24 2012 UTC vs.
Revision 1.103 by jsr166, Tue Oct 13 19:45:34 2015 UTC

#	Line 33 | Line 33 | import java.lang.reflect.Constructor;
33		* <p>A "main" {@code ForkJoinTask} begins execution when it is
34		* explicitly submitted to a {@link ForkJoinPool}, or, if not already
35		* engaged in a ForkJoin computation, commenced in the {@link
36	<	* ForkJoinPool#commonPool} via {@link #fork}, {@link #invoke}, or
36	>	* ForkJoinPool#commonPool()} via {@link #fork}, {@link #invoke}, or
37		* related methods.  Once started, it will usually in turn start other
38		* subtasks.  As indicated by the name of this class, many programs
39		* using {@code ForkJoinTask} employ only methods {@link #fork} and
#	Line 55 | Line 55 | import java.lang.reflect.Constructor;
55		* minimize other blocking synchronization apart from joining other
56		* tasks or using synchronizers such as Phasers that are advertised to
57		* cooperate with fork/join scheduling. Subdividable tasks should also
58	<	* not perform blocking IO, and should ideally access variables that
58	>	* not perform blocking I/O, and should ideally access variables that
59		* are completely independent of those accessed by other running
60		* tasks. These guidelines are loosely enforced by not permitting
61		* checked exceptions such as {@code IOExceptions} to be
#	Line 73 | Line 73 | import java.lang.reflect.Constructor;
73		* <p>It is possible to define and use ForkJoinTasks that may block,
74		* but doing do requires three further considerations: (1) Completion
75		* of few if any <em>other</em> tasks should be dependent on a task
76	<	* that blocks on external synchronization or IO. Event-style async
76	>	* that blocks on external synchronization or I/O. Event-style async
77		* tasks that are never joined (for example, those subclassing {@link
78		* CountedCompleter}) often fall into this category.  (2) To minimize
79		* resource impact, tasks should be small; ideally performing only the
#	Line 285 | Line 285 | public abstract class ForkJoinTask<V> im
285		*/
286		private int externalAwaitDone() {
287		int s;
288	+	ForkJoinPool.externalHelpJoin(this);
289		boolean interrupted = false;
290	<	if ((s = status) >= 0 && ForkJoinPool.tryUnsubmitFromCommonPool(this))
290	<	s = doExec();
291	<	while (s >= 0) {
290	>	while ((s = status) >= 0) {
291		if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
292		synchronized (this) {
293		if (status >= 0) {
#	Line 302 | Line 301 | public abstract class ForkJoinTask<V> im
301		notifyAll();
302		}
303		}
305	–	s = status;
304		}
305		if (interrupted)
306		Thread.currentThread().interrupt();
#	Line 313 | Line 311 | public abstract class ForkJoinTask<V> im
311		* Blocks a non-worker-thread until completion or interruption.
312		*/
313		private int externalInterruptibleAwaitDone() throws InterruptedException {
314	+	int s;
315		if (Thread.interrupted())
316		throw new InterruptedException();
317	<	int s;
318	<	if ((s = status) >= 0 && ForkJoinPool.tryUnsubmitFromCommonPool(this))
320	<	s = doExec();
321	<	while (s >= 0) {
317	>	ForkJoinPool.externalHelpJoin(this);
318	>	while ((s = status) >= 0) {
319		if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
320		synchronized (this) {
321		if (status >= 0)
#	Line 327 | Line 324 | public abstract class ForkJoinTask<V> im
324		notifyAll();
325		}
326		}
330	–	s = status;
327		}
328		return s;
329		}
330
331	+
332		/**
333		* Implementation for join, get, quietlyJoin. Directly handles
334		* only cases of already-completed, external wait, and
#	Line 398 | Line 395 | public abstract class ForkJoinTask<V> im
395		final Throwable ex;
396		ExceptionNode next;
397		final long thrower;  // use id not ref to avoid weak cycles
398	+	final int hashCode;  // store task hashCode before weak ref disappears
399		ExceptionNode(ForkJoinTask<?> task, Throwable ex, ExceptionNode next) {
400		super(task, exceptionTableRefQueue);
401		this.ex = ex;
402		this.next = next;
403		this.thrower = Thread.currentThread().getId();
404	+	this.hashCode = System.identityHashCode(task);
405		}
406		}
407
#	Line 438 | Line 437 | public abstract class ForkJoinTask<V> im
437		}
438
439		/**
440	<	* Records exception and possibly propagates
440	>	* Records exception and possibly propagates.
441		*
442		* @return status on exit
443		*/
#	Line 471 | Line 470 | public abstract class ForkJoinTask<V> im
470		}
471
472		/**
473	<	* Removes exception node and clears status
473	>	* Removes exception node and clears status.
474		*/
475		private void clearExceptionalCompletion() {
476		int h = System.identityHashCode(this);
#	Line 564 | Line 563 | public abstract class ForkJoinTask<V> im
563		private static void expungeStaleExceptions() {
564		for (Object x; (x = exceptionTableRefQueue.poll()) != null;) {
565		if (x instanceof ExceptionNode) {
566	<	ForkJoinTask<?> key = ((ExceptionNode)x).get();
566	>	int hashCode = ((ExceptionNode)x).hashCode;
567		ExceptionNode[] t = exceptionTable;
568	<	int i = System.identityHashCode(key) & (t.length - 1);
568	>	int i = hashCode & (t.length - 1);
569		ExceptionNode e = t[i];
570		ExceptionNode pred = null;
571		while (e != null) {
#	Line 609 | Line 608 | public abstract class ForkJoinTask<V> im
608		throw (Error)ex;
609		if (ex instanceof RuntimeException)
610		throw (RuntimeException)ex;
611	<	throw uncheckedThrowable(ex, RuntimeException.class);
611	>	ForkJoinTask.<RuntimeException>uncheckedThrow(ex);
612		}
613		}
614
#	Line 619 | Line 618 | public abstract class ForkJoinTask<V> im
618		* unchecked exceptions
619		*/
620		@SuppressWarnings("unchecked") static <T extends Throwable>
621	<	T uncheckedThrowable(final Throwable t, final Class<T> c) {
622	<	return (T)t; // rely on vacuous cast
621	>	void uncheckedThrow(Throwable t) throws T {
622	>	if (t != null)
623	>	throw (T)t; // rely on vacuous cast
624		}
625
626		/**
#	Line 638 | Line 638 | public abstract class ForkJoinTask<V> im
638		/**
639		* Arranges to asynchronously execute this task in the pool the
640		* current task is running in, if applicable, or using the {@link
641	<	* ForkJoinPool#commonPool} if not {@link #inForkJoinPool}.  While
641	>	* ForkJoinPool#commonPool()} if not {@link #inForkJoinPool}.  While
642		* it is not necessarily enforced, it is a usage error to fork a
643		* task more than once unless it has completed and been
644		* reinitialized.  Subsequent modifications to the state of this
#	Line 655 | Line 655 | public abstract class ForkJoinTask<V> im
655		if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
656		((ForkJoinWorkerThread)t).workQueue.push(this);
657		else
658	<	ForkJoinPool.submitToCommonPool(this);
658	>	ForkJoinPool.common.externalPush(this);
659		return this;
660		}
661
#	Line 981 | Line 981 | public abstract class ForkJoinTask<V> im
981		if (Thread.interrupted())
982		throw new InterruptedException();
983		// Messy in part because we measure in nanosecs, but wait in millisecs
984	<	int s; long ns, ms;
985	<	if ((s = status) >= 0 && (ns = unit.toNanos(timeout)) > 0L) {
984	>	int s; long ms;
985	>	long ns = unit.toNanos(timeout);
986	>	if ((s = status) >= 0 && ns > 0L) {
987		long deadline = System.nanoTime() + ns;
988		ForkJoinPool p = null;
989		ForkJoinPool.WorkQueue w = null;
#	Line 991 | Line 992 | public abstract class ForkJoinTask<V> im
992		ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
993		p = wt.pool;
994		w = wt.workQueue;
995	<	s = p.helpJoinOnce(w, this); // no retries on failure
995	>	p.helpJoinOnce(w, this); // no retries on failure
996		}
997	+	else
998	+	ForkJoinPool.externalHelpJoin(this);
999		boolean canBlock = false;
1000		boolean interrupted = false;
1001		try {
1002		while ((s = status) >= 0) {
1003	<	if (w != null && w.runState < 0)
1003	>	if (w != null && w.qlock < 0)
1004		cancelIgnoringExceptions(this);
1005		else if (!canBlock) {
1006	<	if (p == null || p.tryCompensate(this, null))
1006	>	if (p == null || p.tryCompensate())
1007		canBlock = true;
1008		}
1009		else {
#	Line 1064 | Line 1067 | public abstract class ForkJoinTask<V> im
1067
1068		/**
1069		* Possibly executes tasks until the pool hosting the current task
1070	<	* {@link ForkJoinPool#isQuiescent is quiescent}. This method may
1071	<	* be of use in designs in which many tasks are forked, but none
1072	<	* are explicitly joined, instead executing them until all are
1073	<	* processed.
1070	>	* {@linkplain ForkJoinPool#isQuiescent is quiescent}.  This
1071	>	* method may be of use in designs in which many tasks are forked,
1072	>	* but none are explicitly joined, instead executing them until
1073	>	* all are processed.
1074		*/
1075		public static void helpQuiesce() {
1076		Thread t;
#	Line 1076 | Line 1079 | public abstract class ForkJoinTask<V> im
1079		wt.pool.helpQuiescePool(wt.workQueue);
1080		}
1081		else
1082	<	ForkJoinPool.externalHelpQuiescePool();
1082	>	ForkJoinPool.quiesceCommonPool();
1083		}
1084
1085		/**
#	Line 1139 | Line 1142 | public abstract class ForkJoinTask<V> im
1142		*/
1143		public boolean tryUnfork() {
1144		Thread t;
1145	<	return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1146	<	((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) :
1147	<	ForkJoinPool.tryUnsubmitFromCommonPool(this);
1145	>	return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1146	>	((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) :
1147	>	ForkJoinPool.tryExternalUnpush(this));
1148		}
1149
1150		/**
#	Line 1153 | Line 1156 | public abstract class ForkJoinTask<V> im
1156		* @return the number of tasks
1157		*/
1158		public static int getQueuedTaskCount() {
1159	<	Thread t;
1160	<	return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1161	<	((ForkJoinWorkerThread)t).workQueue.queueSize() :
1162	<	ForkJoinPool.getEstimatedSubmitterQueueLength();
1159	>	Thread t; ForkJoinPool.WorkQueue q;
1160	>	if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
1161	>	q = ((ForkJoinWorkerThread)t).workQueue;
1162	>	else
1163	>	q = ForkJoinPool.commonSubmitterQueue();
1164	>	return (q == null) ? 0 : q.queueSize();
1165		}
1166
1167		/**
#	Line 1173 | Line 1178 | public abstract class ForkJoinTask<V> im
1178		* @return the surplus number of tasks, which may be negative
1179		*/
1180		public static int getSurplusQueuedTaskCount() {
1181	<	/*
1177	<	* The aim of this method is to return a cheap heuristic guide
1178	<	* for task partitioning when programmers, frameworks, tools,
1179	<	* or languages have little or no idea about task granularity.
1180	<	* In essence by offering this method, we ask users only about
1181	<	* tradeoffs in overhead vs expected throughput and its
1182	<	* variance, rather than how finely to partition tasks.
1183	<	*
1184	<	* In a steady state strict (tree-structured) computation,
1185	<	* each thread makes available for stealing enough tasks for
1186	<	* other threads to remain active. Inductively, if all threads
1187	<	* play by the same rules, each thread should make available
1188	<	* only a constant number of tasks.
1189	<	*
1190	<	* The minimum useful constant is just 1. But using a value of
1191	<	* 1 would require immediate replenishment upon each steal to
1192	<	* maintain enough tasks, which is infeasible.  Further,
1193	<	* partitionings/granularities of offered tasks should
1194	<	* minimize steal rates, which in general means that threads
1195	<	* nearer the top of computation tree should generate more
1196	<	* than those nearer the bottom. In perfect steady state, each
1197	<	* thread is at approximately the same level of computation
1198	<	* tree. However, producing extra tasks amortizes the
1199	<	* uncertainty of progress and diffusion assumptions.
1200	<	*
1201	<	* So, users will want to use values larger, but not much
1202	<	* larger than 1 to both smooth over transient shortages and
1203	<	* hedge against uneven progress; as traded off against the
1204	<	* cost of extra task overhead. We leave the user to pick a
1205	<	* threshold value to compare with the results of this call to
1206	<	* guide decisions, but recommend values such as 3.
1207	<	*
1208	<	* When all threads are active, it is on average OK to
1209	<	* estimate surplus strictly locally. In steady-state, if one
1210	<	* thread is maintaining say 2 surplus tasks, then so are
1211	<	* others. So we can just use estimated queue length.
1212	<	* However, this strategy alone leads to serious mis-estimates
1213	<	* in some non-steady-state conditions (ramp-up, ramp-down,
1214	<	* other stalls). We can detect many of these by further
1215	<	* considering the number of "idle" threads, that are known to
1216	<	* have zero queued tasks, so compensate by a factor of
1217	<	* (#idle/#active) threads.
1218	<	*/
1219	<	Thread t; ForkJoinWorkerThread wt;
1220	<	return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1221	<	(wt = (ForkJoinWorkerThread)t).workQueue.queueSize() - wt.pool.idlePerActive() :
1222	<	0;
1181	>	return ForkJoinPool.getSurplusQueuedTaskCount();
1182		}
1183
1184		// Extension methods
#	Line 1263 | Line 1222 | public abstract class ForkJoinTask<V> im
1222		/**
1223		* Returns, but does not unschedule or execute, a task queued by
1224		* the current thread but not yet executed, if one is immediately
1225	<	* available and the current thread is operating in a
1226	<	* ForkJoinPool. There is no guarantee that this task will
1227	<	* actually be polled or executed next. Conversely, this method
1228	<	* may return null even if a task exists but cannot be accessed
1270	<	* without contention with other threads.  This method is designed
1225	>	* available. There is no guarantee that this task will actually
1226	>	* be polled or executed next. Conversely, this method may return
1227	>	* null even if a task exists but cannot be accessed without
1228	>	* contention with other threads.  This method is designed
1229		* primarily to support extensions, and is unlikely to be useful
1230		* otherwise.
1231		*
1232		* @return the next task, or {@code null} if none are available
1233		*/
1234		protected static ForkJoinTask<?> peekNextLocalTask() {
1235	<	Thread t;
1236	<	return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1237	<	((ForkJoinWorkerThread)t).workQueue.peek() :
1238	<	null;
1235	>	Thread t; ForkJoinPool.WorkQueue q;
1236	>	if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
1237	>	q = ((ForkJoinWorkerThread)t).workQueue;
1238	>	else
1239	>	q = ForkJoinPool.commonSubmitterQueue();
1240	>	return (q == null) ? null : q.peek();
1241		}
1242
1243		/**
#	Line 1368 | Line 1328 | public abstract class ForkJoinTask<V> im
1328		}
1329
1330		/**
1331	<	* Adaptor for Runnables. This implements RunnableFuture
1331	>	* Adapter for Runnables. This implements RunnableFuture
1332		* to be compliant with AbstractExecutorService constraints
1333		* when used in ForkJoinPool.
1334		*/
#	Line 1389 | Line 1349 | public abstract class ForkJoinTask<V> im
1349		}
1350
1351		/**
1352	<	* Adaptor for Runnables without results
1352	>	* Adapter for Runnables without results
1353		*/
1354		static final class AdaptedRunnableAction extends ForkJoinTask<Void>
1355		implements RunnableFuture<Void> {
#	Line 1406 | Line 1366 | public abstract class ForkJoinTask<V> im
1366		}
1367
1368		/**
1369	<	* Adaptor for Callables
1369	>	* Adapter for Callables
1370		*/
1371		static final class AdaptedCallable<T> extends ForkJoinTask<T>
1372		implements RunnableFuture<T> {
#	Line 1502 | Line 1462 | public abstract class ForkJoinTask<V> im
1462		// Unsafe mechanics
1463		private static final sun.misc.Unsafe U;
1464		private static final long STATUS;
1465	+
1466		static {
1467		exceptionTableLock = new ReentrantLock();
1468		exceptionTableRefQueue = new ReferenceQueue<Object>();
1469		exceptionTable = new ExceptionNode[EXCEPTION_MAP_CAPACITY];
1470		try {
1471		U = getUnsafe();
1472	+	Class<?> k = ForkJoinTask.class;
1473		STATUS = U.objectFieldOffset
1474	<	(ForkJoinTask.class.getDeclaredField("status"));
1474	>	(k.getDeclaredField("status"));
1475		} catch (Exception e) {
1476		throw new Error(e);
1477		}
#	Line 1525 | Line 1487 | public abstract class ForkJoinTask<V> im
1487		private static sun.misc.Unsafe getUnsafe() {
1488		try {
1489		return sun.misc.Unsafe.getUnsafe();
1490	<	} catch (SecurityException se) {
1491	<	try {
1492	<	return java.security.AccessController.doPrivileged
1493	<	(new java.security
1494	<	.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1495	<	public sun.misc.Unsafe run() throws Exception {
1496	<	java.lang.reflect.Field f = sun.misc
1497	<	.Unsafe.class.getDeclaredField("theUnsafe");
1498	<	f.setAccessible(true);
1499	<	return (sun.misc.Unsafe) f.get(null);
1500	<	}});
1501	<	} catch (java.security.PrivilegedActionException e) {
1502	<	throw new RuntimeException("Could not initialize intrinsics",
1503	<	e.getCause());
1504	<	}
1490	>	} catch (SecurityException tryReflectionInstead) {}
1491	>	try {
1492	>	return java.security.AccessController.doPrivileged
1493	>	(new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1494	>	public sun.misc.Unsafe run() throws Exception {
1495	>	Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class;
1496	>	for (java.lang.reflect.Field f : k.getDeclaredFields()) {
1497	>	f.setAccessible(true);
1498	>	Object x = f.get(null);
1499	>	if (k.isInstance(x))
1500	>	return k.cast(x);
1501	>	}
1502	>	throw new NoSuchFieldError("the Unsafe");
1503	>	}});
1504	>	} catch (java.security.PrivilegedActionException e) {
1505	>	throw new RuntimeException("Could not initialize intrinsics",
1506	>	e.getCause());
1507		}
1508		}
1509		}

Diff Legend

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