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.90 by dl, Sat Apr 21 11:45:20 2012 UTC vs.
Revision 1.92 by dl, Wed Oct 31 12:49:24 2012 UTC

#	Line 30 | Line 30 | import java.lang.reflect.Constructor;
30		* subtasks may be hosted by a small number of actual threads in a
31		* ForkJoinPool, at the price of some usage limitations.
32		*
33	<	* <p>A "main" {@code ForkJoinTask} begins execution when submitted
34	<	* to a {@link ForkJoinPool}.  Once started, it will usually in turn
35	<	* start other subtasks.  As indicated by the name of this class,
36	<	* many programs using {@code ForkJoinTask} employ only methods
37	<	* {@link #fork} and {@link #join}, or derivatives such as {@link
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
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
40	>	* {@link #join}, or derivatives such as {@link
41		* #invokeAll(ForkJoinTask...) invokeAll}.  However, this class also
42		* provides a number of other methods that can come into play in
43	<	* advanced usages, as well as extension mechanics that allow
44	<	* support of new forms of fork/join processing.
43	>	* advanced usages, as well as extension mechanics that allow support
44	>	* of new forms of fork/join processing.
45		*
46		* <p>A {@code ForkJoinTask} is a lightweight form of {@link Future}.
47		* The efficiency of {@code ForkJoinTask}s stems from a set of
#	Line 123 | Line 126 | import java.lang.reflect.Constructor;
126		* other actions.  Normally, a concrete ForkJoinTask subclass declares
127		* fields comprising its parameters, established in a constructor, and
128		* then defines a {@code compute} method that somehow uses the control
129	<	* methods supplied by this base class. While these methods have
127	<	* {@code public} access (to allow instances of different task
128	<	* subclasses to call each other's methods), some of them may only be
129	<	* called from within other ForkJoinTasks (as may be determined using
130	<	* method {@link #inForkJoinPool}).  Attempts to invoke them in other
131	<	* contexts result in exceptions or errors, possibly including {@code
132	<	* ClassCastException}.
129	>	* methods supplied by this base class.
130		*
131		* <p>Method {@link #join} and its variants are appropriate for use
132		* only when completion dependencies are acyclic; that is, the
#	Line 287 | Line 284 | public abstract class ForkJoinTask<V> im
284		* @return status upon completion
285		*/
286		private int externalAwaitDone() {
290	–	boolean interrupted = false;
287		int s;
288	<	while ((s = status) >= 0) {
288	>	boolean interrupted = false;
289	>	if ((s = status) >= 0 && ForkJoinPool.tryUnsubmitFromCommonPool(this))
290	>	s = doExec();
291	>	while (s >= 0) {
292		if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
293		synchronized (this) {
294		if (status >= 0) {
#	Line 303 | Line 302 | public abstract class ForkJoinTask<V> im
302		notifyAll();
303		}
304		}
305	+	s = status;
306		}
307		if (interrupted)
308		Thread.currentThread().interrupt();
#	Line 313 | Line 313 | public abstract class ForkJoinTask<V> im
313		* Blocks a non-worker-thread until completion or interruption.
314		*/
315		private int externalInterruptibleAwaitDone() throws InterruptedException {
316	–	int s;
316		if (Thread.interrupted())
317		throw new InterruptedException();
318	<	while ((s = status) >= 0) {
318	>	int s;
319	>	if ((s = status) >= 0 && ForkJoinPool.tryUnsubmitFromCommonPool(this))
320	>	s = doExec();
321	>	while (s >= 0) {
322		if (U.compareAndSwapInt(this, STATUS, s, s | SIGNAL)) {
323		synchronized (this) {
324		if (status >= 0)
#	Line 325 | Line 327 | public abstract class ForkJoinTask<V> im
327		notifyAll();
328		}
329		}
330	+	s = status;
331		}
332		return s;
333		}
#	Line 338 | Line 341 | public abstract class ForkJoinTask<V> im
341		*/
342		private int doJoin() {
343		int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
344	<	if ((s = status) >= 0) {
345	<	if (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)) {
346	<	if (!(w = (wt = (ForkJoinWorkerThread)t).workQueue).
347	<	tryUnpush(this) || (s = doExec()) >= 0)
348	<	s = wt.pool.awaitJoin(w, this);
349	<	}
347	<	else
348	<	s = externalAwaitDone();
349	<	}
350	<	return s;
344	>	return (s = status) < 0 ? s :
345	>	((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
346	>	(w = (wt = (ForkJoinWorkerThread)t).workQueue).
347	>	tryUnpush(this) && (s = doExec()) < 0 ? s :
348	>	wt.pool.awaitJoin(w, this) :
349	>	externalAwaitDone();
350		}
351
352		/**
#	Line 357 | Line 356 | public abstract class ForkJoinTask<V> im
356		*/
357		private int doInvoke() {
358		int s; Thread t; ForkJoinWorkerThread wt;
359	<	if ((s = doExec()) >= 0) {
360	<	if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
361	<	s = (wt = (ForkJoinWorkerThread)t).pool.awaitJoin(wt.workQueue,
362	<	this);
364	<	else
365	<	s = externalAwaitDone();
366	<	}
367	<	return s;
359	>	return (s = doExec()) < 0 ? s :
360	>	((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
361	>	(wt = (ForkJoinWorkerThread)t).pool.awaitJoin(wt.workQueue, this) :
362	>	externalAwaitDone();
363		}
364
365		// Exception table support
#	Line 606 | Line 601 | public abstract class ForkJoinTask<V> im
601		}
602
603		/**
604	+	* A version of "sneaky throw" to relay exceptions
605	+	*/
606	+	static void rethrow(final Throwable ex) {
607	+	if (ex != null) {
608	+	if (ex instanceof Error)
609	+	throw (Error)ex;
610	+	if (ex instanceof RuntimeException)
611	+	throw (RuntimeException)ex;
612	+	throw uncheckedThrowable(ex, RuntimeException.class);
613	+	}
614	+	}
615	+
616	+	/**
617	+	* The sneaky part of sneaky throw, relying on generics
618	+	* limitations to evade compiler complaints about rethrowing
619	+	* unchecked exceptions
620	+	*/
621	+	@SuppressWarnings("unchecked") static <T extends Throwable>
622	+	T uncheckedThrowable(final Throwable t, final Class<T> c) {
623	+	return (T)t; // rely on vacuous cast
624	+	}
625	+
626	+	/**
627		* Throws exception, if any, associated with the given status.
628		*/
629		private void reportException(int s) {
630	<	Throwable ex = ((s == CANCELLED) ?  new CancellationException() :
631	<	(s == EXCEPTIONAL) ? getThrowableException() :
632	<	null);
633	<	if (ex != null)
616	<	U.throwException(ex);
630	>	if (s == CANCELLED)
631	>	throw new CancellationException();
632	>	if (s == EXCEPTIONAL)
633	>	rethrow(getThrowableException());
634		}
635
636		// public methods
637
638		/**
639	<	* Arranges to asynchronously execute this task.  While it is not
640	<	* necessarily enforced, it is a usage error to fork a task more
641	<	* than once unless it has completed and been reinitialized.
642	<	* Subsequent modifications to the state of this task or any data
643	<	* it operates on are not necessarily consistently observable by
644	<	* any thread other than the one executing it unless preceded by a
645	<	* call to {@link #join} or related methods, or a call to {@link
646	<	* #isDone} returning {@code true}.
647	<	*
648	<	* <p>This method may be invoked only from within {@code
649	<	* ForkJoinPool} computations (as may be determined using method
633	<	* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
634	<	* result in exceptions or errors, possibly including {@code
635	<	* ClassCastException}.
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
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
645	>	* task or any data it operates on are not necessarily
646	>	* consistently observable by any thread other than the one
647	>	* executing it unless preceded by a call to {@link #join} or
648	>	* related methods, or a call to {@link #isDone} returning {@code
649	>	* true}.
650		*
651		* @return {@code this}, to simplify usage
652		*/
653		public final ForkJoinTask<V> fork() {
654	<	((ForkJoinWorkerThread)Thread.currentThread()).workQueue.push(this);
654	>	Thread t;
655	>	if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
656	>	((ForkJoinWorkerThread)t).workQueue.push(this);
657	>	else
658	>	ForkJoinPool.submitToCommonPool(this);
659		return this;
660		}
661
#	Line 687 | Line 705 | public abstract class ForkJoinTask<V> im
705		* cancelled, completed normally or exceptionally, or left
706		* unprocessed.
707		*
690	–	* <p>This method may be invoked only from within {@code
691	–	* ForkJoinPool} computations (as may be determined using method
692	–	* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
693	–	* result in exceptions or errors, possibly including {@code
694	–	* ClassCastException}.
695	–	*
708		* @param t1 the first task
709		* @param t2 the second task
710		* @throws NullPointerException if any task is null
#	Line 718 | Line 730 | public abstract class ForkJoinTask<V> im
730		* related methods to check if they have been cancelled, completed
731		* normally or exceptionally, or left unprocessed.
732		*
721	–	* <p>This method may be invoked only from within {@code
722	–	* ForkJoinPool} computations (as may be determined using method
723	–	* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
724	–	* result in exceptions or errors, possibly including {@code
725	–	* ClassCastException}.
726	–	*
733		* @param tasks the tasks
734		* @throws NullPointerException if any task is null
735		*/
#	Line 751 | Line 757 | public abstract class ForkJoinTask<V> im
757		}
758		}
759		if (ex != null)
760	<	U.throwException(ex);
760	>	rethrow(ex);
761		}
762
763		/**
#	Line 767 | Line 773 | public abstract class ForkJoinTask<V> im
773		* cancelled, completed normally or exceptionally, or left
774		* unprocessed.
775		*
770	–	* <p>This method may be invoked only from within {@code
771	–	* ForkJoinPool} computations (as may be determined using method
772	–	* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
773	–	* result in exceptions or errors, possibly including {@code
774	–	* ClassCastException}.
775	–	*
776		* @param tasks the collection of tasks
777		* @return the tasks argument, to simplify usage
778		* @throws NullPointerException if tasks or any element are null
#	Line 808 | Line 808 | public abstract class ForkJoinTask<V> im
808		}
809		}
810		if (ex != null)
811	<	U.throwException(ex);
811	>	rethrow(ex);
812		return tasks;
813		}
814
#	Line 1068 | Line 1068 | public abstract class ForkJoinTask<V> im
1068		* be of use in designs in which many tasks are forked, but none
1069		* are explicitly joined, instead executing them until all are
1070		* processed.
1071	–	*
1072	–	* <p>This method may be invoked only from within {@code
1073	–	* ForkJoinPool} computations (as may be determined using method
1074	–	* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1075	–	* result in exceptions or errors, possibly including {@code
1076	–	* ClassCastException}.
1071		*/
1072		public static void helpQuiesce() {
1073	<	ForkJoinWorkerThread wt =
1074	<	(ForkJoinWorkerThread)Thread.currentThread();
1075	<	wt.pool.helpQuiescePool(wt.workQueue);
1073	>	Thread t;
1074	>	if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
1075	>	ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
1076	>	wt.pool.helpQuiescePool(wt.workQueue);
1077	>	}
1078	>	else
1079	>	ForkJoinPool.externalHelpQuiescePool();
1080		}
1081
1082		/**
#	Line 1131 | Line 1129 | public abstract class ForkJoinTask<V> im
1129
1130		/**
1131		* Tries to unschedule this task for execution. This method will
1132	<	* typically succeed if this task is the most recently forked task
1133	<	* by the current thread, and has not commenced executing in
1134	<	* another thread.  This method may be useful when arranging
1135	<	* alternative local processing of tasks that could have been, but
1136	<	* were not, stolen.
1139	<	*
1140	<	* <p>This method may be invoked only from within {@code
1141	<	* ForkJoinPool} computations (as may be determined using method
1142	<	* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1143	<	* result in exceptions or errors, possibly including {@code
1144	<	* ClassCastException}.
1132	>	* typically (but is not guaranteed to) succeed if this task is
1133	>	* the most recently forked task by the current thread, and has
1134	>	* not commenced executing in another thread.  This method may be
1135	>	* useful when arranging alternative local processing of tasks
1136	>	* that could have been, but were not, stolen.
1137		*
1138		* @return {@code true} if unforked
1139		*/
1140		public boolean tryUnfork() {
1141	<	return ((ForkJoinWorkerThread)Thread.currentThread())
1142	<	.workQueue.tryUnpush(this);
1141	>	Thread t;
1142	>	return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1143	>	((ForkJoinWorkerThread)t).workQueue.tryUnpush(this) :
1144	>	ForkJoinPool.tryUnsubmitFromCommonPool(this);
1145		}
1146
1147		/**
#	Line 1156 | Line 1150 | public abstract class ForkJoinTask<V> im
1150		* value may be useful for heuristic decisions about whether to
1151		* fork other tasks.
1152		*
1159	–	* <p>This method may be invoked only from within {@code
1160	–	* ForkJoinPool} computations (as may be determined using method
1161	–	* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1162	–	* result in exceptions or errors, possibly including {@code
1163	–	* ClassCastException}.
1164	–	*
1153		* @return the number of tasks
1154		*/
1155		public static int getQueuedTaskCount() {
1156	<	return ((ForkJoinWorkerThread) Thread.currentThread())
1157	<	.workQueue.queueSize();
1156	>	Thread t;
1157	>	return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1158	>	((ForkJoinWorkerThread)t).workQueue.queueSize() :
1159	>	ForkJoinPool.getEstimatedSubmitterQueueLength();
1160		}
1161
1162		/**
1163		* Returns an estimate of how many more locally queued tasks are
1164		* held by the current worker thread than there are other worker
1165	<	* threads that might steal them.  This value may be useful for
1165	>	* threads that might steal them, or zero if this thread is not
1166	>	* operating in a ForkJoinPool. This value may be useful for
1167		* heuristic decisions about whether to fork other tasks. In many
1168		* usages of ForkJoinTasks, at steady state, each worker should
1169		* aim to maintain a small constant surplus (for example, 3) of
1170		* tasks, and to process computations locally if this threshold is
1171		* exceeded.
1172		*
1182	–	* <p>This method may be invoked only from within {@code
1183	–	* ForkJoinPool} computations (as may be determined using method
1184	–	* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1185	–	* result in exceptions or errors, possibly including {@code
1186	–	* ClassCastException}.
1187	–	*
1173		* @return the surplus number of tasks, which may be negative
1174		*/
1175		public static int getSurplusQueuedTaskCount() {
#	Line 1231 | Line 1216 | public abstract class ForkJoinTask<V> im
1216		* have zero queued tasks, so compensate by a factor of
1217		* (#idle/#active) threads.
1218		*/
1219	<	ForkJoinWorkerThread wt =
1220	<	(ForkJoinWorkerThread)Thread.currentThread();
1221	<	return wt.workQueue.queueSize() - wt.pool.idlePerActive();
1219	>	Thread t; ForkJoinWorkerThread wt;
1220	>	return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1221	>	(wt = (ForkJoinWorkerThread)t).workQueue.queueSize() - wt.pool.idlePerActive() :
1222	>	0;
1223		}
1224
1225		// Extension methods
#	Line 1277 | Line 1263 | public abstract class ForkJoinTask<V> im
1263		/**
1264		* Returns, but does not unschedule or execute, a task queued by
1265		* the current thread but not yet executed, if one is immediately
1266	<	* available. There is no guarantee that this task will actually
1267	<	* be polled or executed next. Conversely, this method may return
1268	<	* null even if a task exists but cannot be accessed without
1269	<	* contention with other threads.  This method is designed
1266	>	* available and the current thread is operating in a
1267	>	* ForkJoinPool. There is no guarantee that this task will
1268	>	* actually be polled or executed next. Conversely, this method
1269	>	* may return null even if a task exists but cannot be accessed
1270	>	* without contention with other threads.  This method is designed
1271		* primarily to support extensions, and is unlikely to be useful
1272		* otherwise.
1273		*
1287	–	* <p>This method may be invoked only from within {@code
1288	–	* ForkJoinPool} computations (as may be determined using method
1289	–	* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1290	–	* result in exceptions or errors, possibly including {@code
1291	–	* ClassCastException}.
1292	–	*
1274		* @return the next task, or {@code null} if none are available
1275		*/
1276		protected static ForkJoinTask<?> peekNextLocalTask() {
1277	<	return ((ForkJoinWorkerThread) Thread.currentThread()).workQueue.peek();
1277	>	Thread t;
1278	>	return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1279	>	((ForkJoinWorkerThread)t).workQueue.peek() :
1280	>	null;
1281		}
1282
1283		/**
1284		* Unschedules and returns, without executing, the next task
1285	<	* queued by the current thread but not yet executed.  This method
1286	<	* is designed primarily to support extensions, and is unlikely to
1287	<	* be useful otherwise.
1288	<	*
1305	<	* <p>This method may be invoked only from within {@code
1306	<	* ForkJoinPool} computations (as may be determined using method
1307	<	* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1308	<	* result in exceptions or errors, possibly including {@code
1309	<	* ClassCastException}.
1285	>	* queued by the current thread but not yet executed, if the
1286	>	* current thread is operating in a ForkJoinPool.  This method is
1287	>	* designed primarily to support extensions, and is unlikely to be
1288	>	* useful otherwise.
1289		*
1290		* @return the next task, or {@code null} if none are available
1291		*/
1292		protected static ForkJoinTask<?> pollNextLocalTask() {
1293	<	return ((ForkJoinWorkerThread) Thread.currentThread())
1294	<	.workQueue.nextLocalTask();
1293	>	Thread t;
1294	>	return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1295	>	((ForkJoinWorkerThread)t).workQueue.nextLocalTask() :
1296	>	null;
1297		}
1298
1299		/**
1300	<	* Unschedules and returns, without executing, the next task
1300	>	* If the current thread is operating in a ForkJoinPool,
1301	>	* unschedules and returns, without executing, the next task
1302		* queued by the current thread but not yet executed, if one is
1303		* available, or if not available, a task that was forked by some
1304		* other thread, if available. Availability may be transient, so a
1305	<	* {@code null} result does not necessarily imply quiescence
1306	<	* of the pool this task is operating in.  This method is designed
1305	>	* {@code null} result does not necessarily imply quiescence of
1306	>	* the pool this task is operating in.  This method is designed
1307		* primarily to support extensions, and is unlikely to be useful
1308		* otherwise.
1309		*
1328	–	* <p>This method may be invoked only from within {@code
1329	–	* ForkJoinPool} computations (as may be determined using method
1330	–	* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1331	–	* result in exceptions or errors, possibly including {@code
1332	–	* ClassCastException}.
1333	–	*
1310		* @return a task, or {@code null} if none are available
1311		*/
1312		protected static ForkJoinTask<?> pollTask() {
1313	<	ForkJoinWorkerThread wt =
1314	<	(ForkJoinWorkerThread)Thread.currentThread();
1315	<	return wt.pool.nextTaskFor(wt.workQueue);
1313	>	Thread t; ForkJoinWorkerThread wt;
1314	>	return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1315	>	(wt = (ForkJoinWorkerThread)t).pool.nextTaskFor(wt.workQueue) :
1316	>	null;
1317		}
1318
1319		// tag operations

Diff Legend

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