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.58 by dl, Mon Sep 6 11:55:39 2010 UTC vs.
Revision 1.69 by dl, Mon Nov 22 12:24:34 2010 UTC

#	Line 6 | Line 6
6
7		package jsr166y;
8
9	–	import java.util.concurrent.*;
10	–
9		import java.io.Serializable;
10		import java.util.Collection;
11		import java.util.Collections;
#	Line 15 | Line 13 | import java.util.List;
13		import java.util.RandomAccess;
14		import java.util.Map;
15		import java.util.WeakHashMap;
16	+	import java.util.concurrent.Callable;
17	+	import java.util.concurrent.CancellationException;
18	+	import java.util.concurrent.ExecutionException;
19	+	import java.util.concurrent.Executor;
20	+	import java.util.concurrent.ExecutorService;
21	+	import java.util.concurrent.Future;
22	+	import java.util.concurrent.RejectedExecutionException;
23	+	import java.util.concurrent.RunnableFuture;
24	+	import java.util.concurrent.TimeUnit;
25	+	import java.util.concurrent.TimeoutException;
26
27		/**
28		* Abstract base class for tasks that run within a {@link ForkJoinPool}.
#	Line 28 | Line 36 | import java.util.WeakHashMap;
36		* start other subtasks.  As indicated by the name of this class,
37		* many programs using {@code ForkJoinTask} employ only methods
38		* {@link #fork} and {@link #join}, or derivatives such as {@link
39	<	* #invokeAll}.  However, this class also provides a number of other
40	<	* methods that can come into play in advanced usages, as well as
41	<	* extension mechanics that allow support of new forms of fork/join
42	<	* processing.
39	>	* #invokeAll(ForkJoinTask...) invokeAll}.  However, this class also
40	>	* provides a number of other methods that can come into play in
41	>	* advanced usages, as well as extension mechanics that allow
42	>	* support of new forms of fork/join processing.
43		*
44		* <p>A {@code ForkJoinTask} is a lightweight form of {@link Future}.
45		* The efficiency of {@code ForkJoinTask}s stems from a set of
#	Line 102 | Line 110 | import java.util.WeakHashMap;
110		* result in exceptions or errors, possibly including
111		* {@code ClassCastException}.
112		*
113	+	* <p>Method {@link #join} and its variants are appropriate for use
114	+	* only when completion dependencies are acyclic; that is, the
115	+	* parallel computation can be described as a directed acyclic graph
116	+	* (DAG). Otherwise, executions may encounter a form of deadlock as
117	+	* tasks cyclically wait for each other.  However, this framework
118	+	* supports other methods and techniques (for example the use of
119	+	* {@link Phaser}, {@link #helpQuiesce}, and {@link #complete}) that
120	+	* may be of use in constructing custom subclasses for problems that
121	+	* are not statically structured as DAGs.
122	+	*
123		* <p>Most base support methods are {@code final}, to prevent
124		* overriding of implementations that are intrinsically tied to the
125		* underlying lightweight task scheduling framework.  Developers
#	Line 153 | Line 171 | public abstract class ForkJoinTask<V> im
171		* single int to minimize footprint and to ensure atomicity (via
172		* CAS).  Status is initially zero, and takes on nonnegative
173		* values until completed, upon which status holds value
174	<	* NORMAL. CANCELLED, or EXCEPTIONAL. Tasks undergoing blocking
174	>	* NORMAL, CANCELLED, or EXCEPTIONAL. Tasks undergoing blocking
175		* waits by other threads have the SIGNAL bit set.  Completion of
176		* a stolen task with SIGNAL set awakens any waiters via
177		* notifyAll. Even though suboptimal for some purposes, we use
#	Line 206 | Line 224 | public abstract class ForkJoinTask<V> im
224
225		/**
226		* Records exception and sets exceptional completion.
227	<	*
227	>	*
228		* @return status on exit
229		*/
230		private void setExceptionalCompletion(Throwable rex) {
#	Line 223 | Line 241 | public abstract class ForkJoinTask<V> im
241		int s;         // the odd construction reduces lock bias effects
242		while ((s = status) >= 0) {
243		try {
244	<	synchronized(this) {
244	>	synchronized (this) {
245		if (UNSAFE.compareAndSwapInt(this, statusOffset, s,SIGNAL))
246		wait();
247		}
#	Line 239 | Line 257 | public abstract class ForkJoinTask<V> im
257		*
258		* @return status on exit
259		*/
260	<	final int internalAwaitDone(long millis) {
260	>	final int internalAwaitDone(long millis, int nanos) {
261		int s;
262		if ((s = status) >= 0) {
263		try {
264	<	synchronized(this) {
264	>	synchronized (this) {
265		if (UNSAFE.compareAndSwapInt(this, statusOffset, s,SIGNAL))
266	<	wait(millis, 0);
266	>	wait(millis, nanos);
267		}
268		} catch (InterruptedException ie) {
269		cancelIfTerminating();
#	Line 261 | Line 279 | public abstract class ForkJoinTask<V> im
279		private void externalAwaitDone() {
280		int s;
281		while ((s = status) >= 0) {
282	<	synchronized(this) {
283	<	if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)){
282	>	synchronized (this) {
283	>	if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)) {
284		boolean interrupted = false;
285		while (status >= 0) {
286		try {
#	Line 322 | Line 340 | public abstract class ForkJoinTask<V> im
340		}
341
342		/**
343	<	* Returns the result of the computation when it {@link #isDone is done}.
344	<	* This method differs from {@link #get()} in that
343	>	* Returns the result of the computation when it {@link #isDone is
344	>	* done}.  This method differs from {@link #get()} in that
345		* abnormal completion results in {@code RuntimeException} or
346	<	* {@code Error}, not {@code ExecutionException}.
346	>	* {@code Error}, not {@code ExecutionException}, and that
347	>	* interrupts of the calling thread do <em>not</em> cause the
348	>	* method to abruptly return by throwing {@code
349	>	* InterruptedException}.
350		*
351		* @return the computed result
352		*/
#	Line 367 | Line 388 | public abstract class ForkJoinTask<V> im
388		* unprocessed.
389		*
390		* <p>This method may be invoked only from within {@code
391	<	* ForkJoinTask} computations (as may be determined using method
391	>	* ForkJoinPool} computations (as may be determined using method
392		* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
393		* result in exceptions or errors, possibly including {@code
394		* ClassCastException}.
#	Line 395 | Line 416 | public abstract class ForkJoinTask<V> im
416		* normally or exceptionally, or left unprocessed.
417		*
418		* <p>This method may be invoked only from within {@code
419	<	* ForkJoinTask} computations (as may be determined using method
419	>	* ForkJoinPool} computations (as may be determined using method
420		* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
421		* result in exceptions or errors, possibly including {@code
422		* ClassCastException}.
#	Line 450 | Line 471 | public abstract class ForkJoinTask<V> im
471		* unprocessed.
472		*
473		* <p>This method may be invoked only from within {@code
474	<	* ForkJoinTask} computations (as may be determined using method
474	>	* ForkJoinPool} computations (as may be determined using method
475		* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
476		* result in exceptions or errors, possibly including {@code
477		* ClassCastException}.
#	Line 502 | Line 523 | public abstract class ForkJoinTask<V> im
523
524		/**
525		* Attempts to cancel execution of this task. This attempt will
526	<	* fail if the task has already completed, has already been
527	<	* cancelled, or could not be cancelled for some other reason. If
528	<	* successful, and this task has not started when cancel is
529	<	* called, execution of this task is suppressed, {@link
530	<	* #isCancelled} will report true, and {@link #join} will result
531	<	* in a {@code CancellationException} being thrown.
526	>	* fail if the task has already completed or could not be
527	>	* cancelled for some other reason. If successful, and this task
528	>	* has not started when {@code cancel} is called, execution of
529	>	* this task is suppressed. After this method returns
530	>	* successfully, unless there is an intervening call to {@link
531	>	* #reinitialize}, subsequent calls to {@link #isCancelled},
532	>	* {@link #isDone}, and {@code cancel} will return {@code true}
533	>	* and calls to {@link #join} and related methods will result in
534	>	* {@code CancellationException}.
535		*
536		* <p>This method may be overridden in subclasses, but if so, must
537	<	* still ensure that these minimal properties hold. In particular,
538	<	* the {@code cancel} method itself must not throw exceptions.
537	>	* still ensure that these properties hold. In particular, the
538	>	* {@code cancel} method itself must not throw exceptions.
539		*
540		* <p>This method is designed to be invoked by <em>other</em>
541		* tasks. To terminate the current task, you can just return or
542		* throw an unchecked exception from its computation method, or
543		* invoke {@link #completeExceptionally}.
544		*
545	<	* @param mayInterruptIfRunning this value is ignored in the
546	<	* default implementation because tasks are not
547	<	* cancelled via interruption
545	>	* @param mayInterruptIfRunning this value has no effect in the
546	>	* default implementation because interrupts are not used to
547	>	* control cancellation.
548		*
549		* @return {@code true} if this task is now cancelled
550		*/
#	Line 642 | Line 666 | public abstract class ForkJoinTask<V> im
666		setCompletion(NORMAL);
667		}
668
669	+	/**
670	+	* Waits if necessary for the computation to complete, and then
671	+	* retrieves its result.
672	+	*
673	+	* @return the computed result
674	+	* @throws CancellationException if the computation was cancelled
675	+	* @throws ExecutionException if the computation threw an
676	+	* exception
677	+	* @throws InterruptedException if the current thread is not a
678	+	* member of a ForkJoinPool and was interrupted while waiting
679	+	*/
680		public final V get() throws InterruptedException, ExecutionException {
681	<	quietlyJoin();
682	<	if (Thread.interrupted())
683	<	throw new InterruptedException();
684	<	int s = status;
681	>	int s;
682	>	if (Thread.currentThread() instanceof ForkJoinWorkerThread) {
683	>	quietlyJoin();
684	>	s = status;
685	>	}
686	>	else {
687	>	while ((s = status) >= 0) {
688	>	synchronized (this) { // interruptible form of awaitDone
689	>	if (UNSAFE.compareAndSwapInt(this, statusOffset,
690	>	s, SIGNAL)) {
691	>	while (status >= 0)
692	>	wait();
693	>	}
694	>	}
695	>	}
696	>	}
697		if (s < NORMAL) {
698		Throwable ex;
699		if (s == CANCELLED)
#	Line 657 | Line 704 | public abstract class ForkJoinTask<V> im
704		return getRawResult();
705		}
706
707	+	/**
708	+	* Waits if necessary for at most the given time for the computation
709	+	* to complete, and then retrieves its result, if available.
710	+	*
711	+	* @param timeout the maximum time to wait
712	+	* @param unit the time unit of the timeout argument
713	+	* @return the computed result
714	+	* @throws CancellationException if the computation was cancelled
715	+	* @throws ExecutionException if the computation threw an
716	+	* exception
717	+	* @throws InterruptedException if the current thread is not a
718	+	* member of a ForkJoinPool and was interrupted while waiting
719	+	* @throws TimeoutException if the wait timed out
720	+	*/
721		public final V get(long timeout, TimeUnit unit)
722		throws InterruptedException, ExecutionException, TimeoutException {
662	–	Thread t = Thread.currentThread();
663	–	ForkJoinPool pool;
664	–	if (t instanceof ForkJoinWorkerThread) {
665	–	ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
666	–	if (status >= 0 && w.unpushTask(this))
667	–	quietlyExec();
668	–	pool = w.pool;
669	–	}
670	–	else
671	–	pool = null;
672	–	/*
673	–	* Timed wait loop intermixes cases for FJ (pool != null) and
674	–	* non FJ threads. For FJ, decrement pool count but don't try
675	–	* for replacement; increment count on completion. For non-FJ,
676	–	* deal with interrupts. This is messy, but a little less so
677	–	* than is splitting the FJ and nonFJ cases.
678	–	*/
679	–	boolean interrupted = false;
680	–	boolean dec = false; // true if pool count decremented
723		long nanos = unit.toNanos(timeout);
724	<	for (;;) {
725	<	if (pool == null && Thread.interrupted()) {
726	<	interrupted = true;
727	<	break;
724	>	if (status >= 0) {
725	>	Thread t = Thread.currentThread();
726	>	if (t instanceof ForkJoinWorkerThread) {
727	>	ForkJoinWorkerThread w = (ForkJoinWorkerThread) t;
728	>	boolean completed = false; // timed variant of quietlyJoin
729	>	if (w.unpushTask(this)) {
730	>	try {
731	>	completed = exec();
732	>	} catch (Throwable rex) {
733	>	setExceptionalCompletion(rex);
734	>	}
735	>	}
736	>	if (completed)
737	>	setCompletion(NORMAL);
738	>	else if (status >= 0)
739	>	w.joinTask(this, true, nanos);
740		}
741	<	int s = status;
742	<	if (s < 0)
743	<	break;
690	<	if (UNSAFE.compareAndSwapInt(this, statusOffset, s, SIGNAL)) {
741	>	else if (Thread.interrupted())
742	>	throw new InterruptedException();
743	>	else {
744		long startTime = System.nanoTime();
745	<	long nt; // wait time
746	<	while (status >= 0 &&
745	>	int s; long nt;
746	>	while ((s = status) >= 0 &&
747		(nt = nanos - (System.nanoTime() - startTime)) > 0) {
748	<	if (pool != null && !dec)
749	<	dec = pool.tryDecrementRunningCount();
697	<	else {
748	>	if (UNSAFE.compareAndSwapInt(this, statusOffset, s,
749	>	SIGNAL)) {
750		long ms = nt / 1000000;
751		int ns = (int) (nt % 1000000);
752	<	try {
753	<	synchronized(this) {
754	<	if (status >= 0)
703	<	wait(ms, ns);
704	<	}
705	<	} catch (InterruptedException ie) {
706	<	if (pool != null)
707	<	cancelIfTerminating();
708	<	else {
709	<	interrupted = true;
710	<	break;
711	<	}
752	>	synchronized (this) {
753	>	if (status >= 0)
754	>	wait(ms, ns); // exit on IE throw
755		}
756		}
757		}
715	–	break;
758		}
759		}
718	–	if (pool != null && dec)
719	–	pool.incrementRunningCount();
720	–	if (interrupted)
721	–	throw new InterruptedException();
760		int es = status;
761		if (es != NORMAL) {
762		Throwable ex;
#	Line 755 | Line 793 | public abstract class ForkJoinTask<V> im
793		return;
794		}
795		}
796	<	w.joinTask(this);
796	>	w.joinTask(this, false, 0L);
797		}
798		}
799		else
#	Line 791 | Line 829 | public abstract class ForkJoinTask<V> im
829		* processed.
830		*
831		* <p>This method may be invoked only from within {@code
832	<	* ForkJoinTask} computations (as may be determined using method
832	>	* ForkJoinPool} computations (as may be determined using method
833		* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
834		* result in exceptions or errors, possibly including {@code
835		* ClassCastException}.
#	Line 810 | Line 848 | public abstract class ForkJoinTask<V> im
848		* under any other usage conditions are not guaranteed.
849		* This method may be useful when executing
850		* pre-constructed trees of subtasks in loops.
851	+	*
852	+	* <p>Upon completion of this method, {@code isDone()} reports
853	+	* {@code false}, and {@code getException()} reports {@code
854	+	* null}. However, the value returned by {@code getRawResult} is
855	+	* unaffected. To clear this value, you can invoke {@code
856	+	* setRawResult(null)}.
857		*/
858		public void reinitialize() {
859		if (status == EXCEPTIONAL)
#	Line 850 | Line 894 | public abstract class ForkJoinTask<V> im
894		* were not, stolen.
895		*
896		* <p>This method may be invoked only from within {@code
897	<	* ForkJoinTask} computations (as may be determined using method
897	>	* ForkJoinPool} computations (as may be determined using method
898		* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
899		* result in exceptions or errors, possibly including {@code
900		* ClassCastException}.
#	Line 869 | Line 913 | public abstract class ForkJoinTask<V> im
913		* fork other tasks.
914		*
915		* <p>This method may be invoked only from within {@code
916	<	* ForkJoinTask} computations (as may be determined using method
916	>	* ForkJoinPool} computations (as may be determined using method
917		* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
918		* result in exceptions or errors, possibly including {@code
919		* ClassCastException}.
#	Line 892 | Line 936 | public abstract class ForkJoinTask<V> im
936		* exceeded.
937		*
938		* <p>This method may be invoked only from within {@code
939	<	* ForkJoinTask} computations (as may be determined using method
939	>	* ForkJoinPool} computations (as may be determined using method
940		* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
941		* result in exceptions or errors, possibly including {@code
942		* ClassCastException}.
#	Line 950 | Line 994 | public abstract class ForkJoinTask<V> im
994		* otherwise.
995		*
996		* <p>This method may be invoked only from within {@code
997	<	* ForkJoinTask} computations (as may be determined using method
997	>	* ForkJoinPool} computations (as may be determined using method
998		* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
999		* result in exceptions or errors, possibly including {@code
1000		* ClassCastException}.
#	Line 969 | Line 1013 | public abstract class ForkJoinTask<V> im
1013		* be useful otherwise.
1014		*
1015		* <p>This method may be invoked only from within {@code
1016	<	* ForkJoinTask} computations (as may be determined using method
1016	>	* ForkJoinPool} computations (as may be determined using method
1017		* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1018		* result in exceptions or errors, possibly including {@code
1019		* ClassCastException}.
#	Line 992 | Line 1036 | public abstract class ForkJoinTask<V> im
1036		* otherwise.
1037		*
1038		* <p>This method may be invoked only from within {@code
1039	<	* ForkJoinTask} computations (as may be determined using method
1039	>	* ForkJoinPool} computations (as may be determined using method
1040		* {@link #inForkJoinPool}).  Attempts to invoke in other contexts
1041		* result in exceptions or errors, possibly including {@code
1042		* ClassCastException}.

Diff Legend

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