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.54 by dl, Wed Aug 11 19:44:30 2010 UTC vs.
Revision 1.64 by jsr166, Mon Sep 20 20:42:37 2010 UTC

#	Line 7 | Line 7
7		package jsr166y;
8
9		import java.util.concurrent.*;
10	–
10		import java.io.Serializable;
11		import java.util.Collection;
12		import java.util.Collections;
#	Line 28 | Line 27 | import java.util.WeakHashMap;
27		* start other subtasks.  As indicated by the name of this class,
28		* many programs using {@code ForkJoinTask} employ only methods
29		* {@link #fork} and {@link #join}, or derivatives such as {@link
30	<	* #invokeAll}.  However, this class also provides a number of other
31	<	* methods that can come into play in advanced usages, as well as
32	<	* extension mechanics that allow support of new forms of fork/join
33	<	* processing.
30	>	* #invokeAll(ForkJoinTask...) invokeAll}.  However, this class also
31	>	* provides a number of other methods that can come into play in
32	>	* advanced usages, as well as extension mechanics that allow
33	>	* support of new forms of fork/join processing.
34		*
35		* <p>A {@code ForkJoinTask} is a lightweight form of {@link Future}.
36		* The efficiency of {@code ForkJoinTask}s stems from a set of
#	Line 100 | Line 99 | import java.util.WeakHashMap;
99		* ForkJoinTasks (as may be determined using method {@link
100		* #inForkJoinPool}).  Attempts to invoke them in other contexts
101		* result in exceptions or errors, possibly including
102	<	* ClassCastException.
102	>	* {@code ClassCastException}.
103		*
104		* <p>Most base support methods are {@code final}, to prevent
105		* overriding of implementations that are intrinsically tied to the
#	Line 153 | Line 152 | public abstract class ForkJoinTask<V> im
152		* single int to minimize footprint and to ensure atomicity (via
153		* CAS).  Status is initially zero, and takes on nonnegative
154		* values until completed, upon which status holds value
155	<	* COMPLETED. CANCELLED, or EXCEPTIONAL. Tasks undergoing blocking
155	>	* NORMAL, CANCELLED, or EXCEPTIONAL. Tasks undergoing blocking
156		* waits by other threads have the SIGNAL bit set.  Completion of
157		* a stolen task with SIGNAL set awakens any waiters via
158		* notifyAll. Even though suboptimal for some purposes, we use
#	Line 205 | Line 204 | public abstract class ForkJoinTask<V> im
204		}
205
206		/**
207	<	* Record exception and set exceptional completion
207	>	* Records exception and sets exceptional completion.
208	>	*
209		* @return status on exit
210		*/
211		private void setExceptionalCompletion(Throwable rex) {
#	Line 214 | Line 214 | public abstract class ForkJoinTask<V> im
214		}
215
216		/**
217	<	* Blocks a worker thread until completion. Called only by pool.
217	>	* Blocks a worker thread until completion. Called only by
218	>	* pool. Currently unused -- pool-based waits use timeout
219	>	* version below.
220		*/
221		final void internalAwaitDone() {
222		int s;         // the odd construction reduces lock bias effects
223		while ((s = status) >= 0) {
224		try {
225	<	synchronized(this) {
225	>	synchronized (this) {
226		if (UNSAFE.compareAndSwapInt(this, statusOffset, s,SIGNAL))
227		wait();
228		}
#	Line 231 | Line 233 | public abstract class ForkJoinTask<V> im
233		}
234
235		/**
236	+	* Blocks a worker thread until completed or timed out.  Called
237	+	* only by pool.
238	+	*
239	+	* @return status on exit
240	+	*/
241	+	final int internalAwaitDone(long millis) {
242	+	int s;
243	+	if ((s = status) >= 0) {
244	+	try {
245	+	synchronized (this) {
246	+	if (UNSAFE.compareAndSwapInt(this, statusOffset, s,SIGNAL))
247	+	wait(millis, 0);
248	+	}
249	+	} catch (InterruptedException ie) {
250	+	cancelIfTerminating();
251	+	}
252	+	s = status;
253	+	}
254	+	return s;
255	+	}
256	+
257	+	/**
258		* Blocks a non-worker-thread until completion.
259		*/
260		private void externalAwaitDone() {
261		int s;
262		while ((s = status) >= 0) {
263	<	synchronized(this) {
263	>	synchronized (this) {
264		if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)){
265		boolean interrupted = false;
266		while (status >= 0) {
#	Line 314 | Line 338 | public abstract class ForkJoinTask<V> im
338
339		/**
340		* Commences performing this task, awaits its completion if
341	<	* necessary, and return its result, or throws an (unchecked)
342	<	* exception if the underlying computation did so.
341	>	* necessary, and returns its result, or throws an (unchecked)
342	>	* {@code RuntimeException} or {@code Error} if the underlying
343	>	* computation did so.
344		*
345		* @return the computed result
346		*/
#	Line 330 | Line 355 | public abstract class ForkJoinTask<V> im
355		/**
356		* Forks the given tasks, returning when {@code isDone} holds for
357		* each task or an (unchecked) exception is encountered, in which
358	<	* case the exception is rethrown.  If either task encounters an
359	<	* exception, the other one may be, but is not guaranteed to be,
360	<	* cancelled.  If both tasks throw an exception, then this method
361	<	* throws one of them.  The individual status of each task may be
362	<	* checked using {@link #getException()} and related methods.
358	>	* case the exception is rethrown. If more than one task
359	>	* encounters an exception, then this method throws any one of
360	>	* these exceptions. If any task encounters an exception, the
361	>	* other may be cancelled. However, the execution status of
362	>	* individual tasks is not guaranteed upon exceptional return. The
363	>	* status of each task may be obtained using {@link
364	>	* #getException()} and related methods to check if they have been
365	>	* cancelled, completed normally or exceptionally, or left
366	>	* unprocessed.
367		*
368		* <p>This method may be invoked only from within {@code
369		* ForkJoinTask} computations (as may be determined using method
#	Line 355 | Line 384 | public abstract class ForkJoinTask<V> im
384		/**
385		* Forks the given tasks, returning when {@code isDone} holds for
386		* each task or an (unchecked) exception is encountered, in which
387	<	* case the exception is rethrown. If any task encounters an
388	<	* exception, others may be, but are not guaranteed to be,
389	<	* cancelled.  If more than one task encounters an exception, then
390	<	* this method throws any one of these exceptions.  The individual
391	<	* status of each task may be checked using {@link #getException()}
392	<	* and related methods.
387	>	* case the exception is rethrown. If more than one task
388	>	* encounters an exception, then this method throws any one of
389	>	* these exceptions. If any task encounters an exception, others
390	>	* may be cancelled. However, the execution status of individual
391	>	* tasks is not guaranteed upon exceptional return. The status of
392	>	* each task may be obtained using {@link #getException()} and
393	>	* related methods to check if they have been cancelled, completed
394	>	* normally or exceptionally, or left unprocessed.
395		*
396		* <p>This method may be invoked only from within {@code
397		* ForkJoinTask} computations (as may be determined using method
#	Line 407 | Line 438 | public abstract class ForkJoinTask<V> im
438		/**
439		* Forks all tasks in the specified collection, returning when
440		* {@code isDone} holds for each task or an (unchecked) exception
441	<	* is encountered.  If any task encounters an exception, others
442	<	* may be, but are not guaranteed to be, cancelled.  If more than
443	<	* one task encounters an exception, then this method throws any
444	<	* one of these exceptions.  The individual status of each task
445	<	* may be checked using {@link #getException()} and related
446	<	* methods.  The behavior of this operation is undefined if the
447	<	* specified collection is modified while the operation is in
448	<	* progress.
441	>	* is encountered, in which case the exception is rethrown. If
442	>	* more than one task encounters an exception, then this method
443	>	* throws any one of these exceptions. If any task encounters an
444	>	* exception, others may be cancelled. However, the execution
445	>	* status of individual tasks is not guaranteed upon exceptional
446	>	* return. The status of each task may be obtained using {@link
447	>	* #getException()} and related methods to check if they have been
448	>	* cancelled, completed normally or exceptionally, or left
449	>	* unprocessed.
450		*
451		* <p>This method may be invoked only from within {@code
452		* ForkJoinTask} computations (as may be determined using method
#	Line 510 | Line 542 | public abstract class ForkJoinTask<V> im
542		}
543
544		/**
545	<	* Cancels ignoring exceptions if worker is terminating
545	>	* Cancels if current thread is a terminating worker thread,
546	>	* ignoring any exceptions thrown by cancel.
547		*/
548		final void cancelIfTerminating() {
549		Thread t = Thread.currentThread();
#	Line 587 | Line 620 | public abstract class ForkJoinTask<V> im
620
621		/**
622		* Completes this task, and if not already aborted or cancelled,
623	<	* returning a {@code null} result upon {@code join} and related
624	<	* operations. This method may be used to provide results for
625	<	* asynchronous tasks, or to provide alternative handling for
626	<	* tasks that would not otherwise complete normally. Its use in
627	<	* other situations is discouraged. This method is
628	<	* overridable, but overridden versions must invoke {@code super}
629	<	* implementation to maintain guarantees.
623	>	* returning the given value as the result of subsequent
624	>	* invocations of {@code join} and related operations. This method
625	>	* may be used to provide results for asynchronous tasks, or to
626	>	* provide alternative handling for tasks that would not otherwise
627	>	* complete normally. Its use in other situations is
628	>	* discouraged. This method is overridable, but overridden
629	>	* versions must invoke {@code super} implementation to maintain
630	>	* guarantees.
631		*
632		* @param value the result value for this task
633		*/
#	Line 607 | Line 641 | public abstract class ForkJoinTask<V> im
641		setCompletion(NORMAL);
642		}
643
644	+	/**
645	+	* Waits if necessary for the computation to complete, and then
646	+	* retrieves its result.
647	+	*
648	+	* @return the computed result
649	+	* @throws CancellationException if the computation was cancelled
650	+	* @throws ExecutionException if the computation threw an
651	+	* exception
652	+	* @throws InterruptedException if the current thread is not a
653	+	* member of a ForkJoinPool and was interrupted while waiting
654	+	*/
655		public final V get() throws InterruptedException, ExecutionException {
656	<	quietlyJoin();
657	<	if (Thread.interrupted())
658	<	throw new InterruptedException();
659	<	int s = status;
656	>	int s;
657	>	if (Thread.currentThread() instanceof ForkJoinWorkerThread) {
658	>	quietlyJoin();
659	>	s = status;
660	>	}
661	>	else {
662	>	while ((s = status) >= 0) {
663	>	synchronized (this) { // interruptible form of awaitDone
664	>	if (UNSAFE.compareAndSwapInt(this, statusOffset,
665	>	s, SIGNAL)) {
666	>	while (status >= 0)
667	>	wait();
668	>	}
669	>	}
670	>	}
671	>	}
672		if (s < NORMAL) {
673		Throwable ex;
674		if (s == CANCELLED)
#	Line 622 | Line 679 | public abstract class ForkJoinTask<V> im
679		return getRawResult();
680		}
681
682	+	/**
683	+	* Waits if necessary for at most the given time for the computation
684	+	* to complete, and then retrieves its result, if available.
685	+	*
686	+	* @param timeout the maximum time to wait
687	+	* @param unit the time unit of the timeout argument
688	+	* @return the computed result
689	+	* @throws CancellationException if the computation was cancelled
690	+	* @throws ExecutionException if the computation threw an
691	+	* exception
692	+	* @throws InterruptedException if the current thread is not a
693	+	* member of a ForkJoinPool and was interrupted while waiting
694	+	* @throws TimeoutException if the wait timed out
695	+	*/
696		public final V get(long timeout, TimeUnit unit)
697		throws InterruptedException, ExecutionException, TimeoutException {
698		Thread t = Thread.currentThread();
#	Line 645 | Line 716 | public abstract class ForkJoinTask<V> im
716		boolean dec = false; // true if pool count decremented
717		long nanos = unit.toNanos(timeout);
718		for (;;) {
719	<	if (Thread.interrupted() && pool == null) {
719	>	if (pool == null && Thread.interrupted()) {
720		interrupted = true;
721		break;
722		}
#	Line 663 | Line 734 | public abstract class ForkJoinTask<V> im
734		long ms = nt / 1000000;
735		int ns = (int) (nt % 1000000);
736		try {
737	<	synchronized(this) {
737	>	synchronized (this) {
738		if (status >= 0)
739		wait(ms, ns);
740		}
#	Line 730 | Line 801 | public abstract class ForkJoinTask<V> im
801		/**
802		* Commences performing this task and awaits its completion if
803		* necessary, without returning its result or throwing its
804	<	* exception. This method may be useful when processing
734	<	* collections of tasks when some have been cancelled or otherwise
735	<	* known to have aborted.
804	>	* exception.
805		*/
806		public final void quietlyInvoke() {
807		if (status >= 0) {
#	Line 1069 | Line 1138 | public abstract class ForkJoinTask<V> im
1138		private static final long serialVersionUID = -7721805057305804111L;
1139
1140		/**
1141	<	* Saves the state to a stream.
1141	>	* Saves the state to a stream (that is, serializes it).
1142		*
1143		* @serialData the current run status and the exception thrown
1144		* during execution, or {@code null} if none
#	Line 1082 | Line 1151 | public abstract class ForkJoinTask<V> im
1151		}
1152
1153		/**
1154	<	* Reconstitutes the instance from a stream.
1154	>	* Reconstitutes the instance from a stream (that is, deserializes it).
1155		*
1156		* @param s the stream
1157		*/

Diff Legend

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