util/concurrent/FutureTask.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/licenses/publicdomain
 */

package java.util.concurrent;
import java.util.concurrent.*; // for javadoc
import java.util.concurrent.locks.*;

/**
 * A cancellable asynchronous computation.  This class provides a base
 * implementation of {@link Future}, with methods to start and cancel
 * a computation, query to see if the computation is complete, and
 * retrieve the result of the computation.  The result can only be
 * retrieved when the computation has completed; the <tt>get</tt>
 * method will block if the computation has not yet completed.  Once
 * the computation has completed, the computation cannot be restarted
 * or cancelled.
 *
 * <p>A <tt>FutureTask</tt> can be used to wrap a {@link Callable} or
 * {@link java.lang.Runnable} object.  Because <tt>FutureTask</tt>
 * implements <tt>Runnable</tt>, a <tt>FutureTask</tt> can be
 * submitted to an {@link Executor} for execution.
 *
 * <p>In addition to serving as a standalone class, this class provides
 * <tt>protected</tt> functionality that may be useful when creating
 * customized task classes.
 *
 * @since 1.5
 * @author Doug Lea
 * @param <V> The result type returned by this FutureTask's <tt>get</tt> method
 */
public class FutureTask<V> implements RunnableFuture<V> {
    /** Synchronization control for FutureTask */
    private final Sync sync;

    /**
     * Creates a <tt>FutureTask</tt> that will upon running, execute the
     * given <tt>Callable</tt>.
     *
     * @param  callable the callable task
     * @throws NullPointerException if callable is null
     */
    public FutureTask(Callable<V> callable) {
        if (callable == null)
            throw new NullPointerException();
        sync = new Sync(callable);
    }

    /**
     * Creates a <tt>FutureTask</tt> that will upon running, execute the
     * given <tt>Runnable</tt>, and arrange that <tt>get</tt> will return the
     * given result on successful completion.
     *
     * @param  runnable the runnable task
     * @param result the result to return on successful completion. If
     * you don't need a particular result, consider using
     * constructions of the form:
     * <tt>Future&lt;?&gt; f = new FutureTask&lt;Object&gt;(runnable, null)</tt>
     * @throws NullPointerException if runnable is null
     */
    public FutureTask(Runnable runnable, V result) {
        sync = new Sync(Executors.callable(runnable, result));
    }

    public boolean isCancelled() {
        return sync.innerIsCancelled();
    }

    public boolean isDone() {
        return sync.innerIsDone();
    }

    public boolean cancel(boolean mayInterruptIfRunning) {
        return sync.innerCancel(mayInterruptIfRunning);
    }

    public V get() throws InterruptedException, ExecutionException {
        return sync.innerGet();
    }

    public V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        return sync.innerGet(unit.toNanos(timeout));
    }

    /**
     * Protected method invoked when this task transitions to state
     * <tt>isDone</tt> (whether normally or via cancellation). The
     * default implementation does nothing.  Subclasses may override
     * this method to invoke completion callbacks or perform
     * bookkeeping. Note that you can query status inside the
     * implementation of this method to determine whether this task
     * has been cancelled.
     */
    protected void done() { }

    /**
     * Sets the result of this Future to the given value unless
     * this future has already been set or has been cancelled.
     * This method is invoked internally by the <tt>run</tt> method
     * upon successful completion of the computation.
     * @param v the value
     */
    protected void set(V v) {
        sync.innerSet(v);
    }

    /**
     * Causes this future to report an <tt>ExecutionException</tt>
     * with the given throwable as its cause, unless this Future has
     * already been set or has been cancelled.
     * This method is invoked internally by the <tt>run</tt> method
     * upon failure of the computation.
     * @param t the cause of failure
     */
    protected void setException(Throwable t) {
        sync.innerSetException(t);
    }

    public void run() {
        sync.innerRun();
    }

    /**
     * Executes the computation without setting its result, and then
     * resets this Future to initial state, failing to do so if the
     * computation encounters an exception or is cancelled.  This is
     * designed for use with tasks that intrinsically execute more
     * than once.
     * @return true if successfully run and reset
     */
    protected boolean runAndReset() {
        return sync.innerRunAndReset();
    }

    /**
     * Synchronization control for FutureTask. Note that this must be
     * a non-static inner class in order to invoke the protected
     * <tt>done</tt> method. For clarity, all inner class support
     * methods are same as outer, prefixed with "inner".
     *
     * Uses AQS sync state to represent run status
     */
    private final class Sync extends AbstractQueuedSynchronizer {
        /** State value representing that task is running */
        private static final int RUNNING   = 1;
        /** State value representing that task ran */
        private static final int RAN       = 2;
        /** State value representing that task was cancelled */
        private static final int CANCELLED = 4;

        /** The underlying callable */
        private final Callable<V> callable;
        /** The result to return from get() */
        private V result;
        /** The exception to throw from get() */
        private Throwable exception;

        /**
         * The thread running task. When nulled after set/cancel, this
         * indicates that the results are accessible.  Must be
         * volatile, to ensure visibility upon completion.
         */
        private volatile Thread runner;

        Sync(Callable<V> callable) {
            this.callable = callable;
        }

        private boolean ranOrCancelled(int state) {
            return (state & (RAN | CANCELLED)) != 0;
        }

        /**
         * Implements AQS base acquire to succeed if ran or cancelled
         */
        protected int tryAcquireShared(int ignore) {
            return innerIsDone()? 1 : -1;
        }

        /**
         * Implements AQS base release to always signal after setting
         * final done status by nulling runner thread.
         */
        protected boolean tryReleaseShared(int ignore) {
            runner = null;
            return true;
        }

        boolean innerIsCancelled() {
            return getState() == CANCELLED;
        }

        boolean innerIsDone() {
            return ranOrCancelled(getState()) && runner == null;
        }

        V innerGet() throws InterruptedException, ExecutionException {
            acquireSharedInterruptibly(0);
            if (getState() == CANCELLED)
                throw new CancellationException();
            if (exception != null)
                throw new ExecutionException(exception);
            return result;
        }

        V innerGet(long nanosTimeout) throws InterruptedException, ExecutionException, TimeoutException {
            if (!tryAcquireSharedNanos(0, nanosTimeout))
                throw new TimeoutException();
            if (getState() == CANCELLED)
                throw new CancellationException();
            if (exception != null)
                throw new ExecutionException(exception);
            return result;
        }

        void innerSet(V v) {
            for (;;) {
                int s = getState();
                if (ranOrCancelled(s))
                    return;
                if (compareAndSetState(s, RAN))
                    break;
            }
            result = v;
            releaseShared(0);
            done();
        }

        void innerSetException(Throwable t) {
            for (;;) {
                int s = getState();
                if (ranOrCancelled(s))
                    return;
                if (compareAndSetState(s, RAN))
                    break;
            }
            exception = t;
            result = null;
            releaseShared(0);
            done();
        }

        boolean innerCancel(boolean mayInterruptIfRunning) {
            for (;;) {
                int s = getState();
                if (ranOrCancelled(s))
                    return false;
                if (compareAndSetState(s, CANCELLED))
                    break;
            }
            if (mayInterruptIfRunning) {
                Thread r = runner;
                if (r != null)
                    r.interrupt();
            }
            releaseShared(0);
            done();
            return true;
        }

        void innerRun() {
            if (!compareAndSetState(0, RUNNING))
                return;
            try {
                runner = Thread.currentThread();
                if (getState() == RUNNING) // recheck after setting thread
                    innerSet(callable.call());
                else
                    releaseShared(0); // cancel
            } catch (Throwable ex) {
                innerSetException(ex);
            }
        }

        boolean innerRunAndReset() {
            if (!compareAndSetState(0, RUNNING))
                return false;
            try {
                runner = Thread.currentThread();
                if (getState() == RUNNING)
                    callable.call(); // don't set result
                runner = null;
                return compareAndSetState(RUNNING, 0);
            } catch (Throwable ex) {
                innerSetException(ex);
                return false;
            }
        }
    }
}
Revision:	1.41
Committed:	Sun Jun 12 00:36:30 2005 UTC (18 years, 11 months ago) by jsr166
Branch:	MAIN
Changes since 1.40:	+2 -1 lines
Log Message:	doc fixes
#	User	Rev	Content
1	tim	1.1	/*
2	dl	1.2	* Written by Doug Lea with assistance from members of JCP JSR-166
3	dl	1.23	* Expert Group and released to the public domain, as explained at
4			* http://creativecommons.org/licenses/publicdomain
5	tim	1.1	*/
6
7			package java.util.concurrent;
8	jsr166	1.41	import java.util.concurrent.*; // for javadoc
9	dl	1.13	import java.util.concurrent.locks.*;
10
11	tim	1.1	/**
12	dl	1.8	* A cancellable asynchronous computation. This class provides a base
13			* implementation of {@link Future}, with methods to start and cancel
14			* a computation, query to see if the computation is complete, and
15	dl	1.4	* retrieve the result of the computation. The result can only be
16			* retrieved when the computation has completed; the <tt>get</tt>
17			* method will block if the computation has not yet completed. Once
18	dl	1.8	* the computation has completed, the computation cannot be restarted
19			* or cancelled.
20	tim	1.1	*
21	dl	1.8	* <p>A <tt>FutureTask</tt> can be used to wrap a {@link Callable} or
22			* {@link java.lang.Runnable} object. Because <tt>FutureTask</tt>
23			* implements <tt>Runnable</tt>, a <tt>FutureTask</tt> can be
24			* submitted to an {@link Executor} for execution.
25	tim	1.1	*
26	dl	1.14	* <p>In addition to serving as a standalone class, this class provides
27			* <tt>protected</tt> functionality that may be useful when creating
28			* customized task classes.
29			*
30	tim	1.1	* @since 1.5
31	dl	1.4	* @author Doug Lea
32	dl	1.12	* @param <V> The result type returned by this FutureTask's <tt>get</tt> method
33	tim	1.1	*/
34	peierls	1.39	public class FutureTask<V> implements RunnableFuture<V> {
35	dl	1.24	/** Synchronization control for FutureTask */
36			private final Sync sync;
37	dl	1.11
38	tim	1.1	/**
39	dl	1.31	* Creates a <tt>FutureTask</tt> that will upon running, execute the
40	tim	1.1	* given <tt>Callable</tt>.
41			*
42			* @param callable the callable task
43	dl	1.9	* @throws NullPointerException if callable is null
44	tim	1.1	*/
45			public FutureTask(Callable<V> callable) {
46	dl	1.9	if (callable == null)
47			throw new NullPointerException();
48	dl	1.24	sync = new Sync(callable);
49	tim	1.1	}
50
51			/**
52	dl	1.31	* Creates a <tt>FutureTask</tt> that will upon running, execute the
53	tim	1.1	* given <tt>Runnable</tt>, and arrange that <tt>get</tt> will return the
54			* given result on successful completion.
55			*
56			* @param runnable the runnable task
57			* @param result the result to return on successful completion. If
58	dl	1.9	* you don't need a particular result, consider using
59	dl	1.16	* constructions of the form:
60			* <tt>Future<?> f = new FutureTask<Object>(runnable, null)</tt>
61	dl	1.9	* @throws NullPointerException if runnable is null
62	tim	1.1	*/
63	dl	1.15	public FutureTask(Runnable runnable, V result) {
64	dl	1.24	sync = new Sync(Executors.callable(runnable, result));
65	dl	1.20	}
66
67			public boolean isCancelled() {
68	dl	1.27	return sync.innerIsCancelled();
69	dl	1.20	}
70	jsr166	1.35
71	dl	1.20	public boolean isDone() {
72	dl	1.27	return sync.innerIsDone();
73	dl	1.13	}
74
75			public boolean cancel(boolean mayInterruptIfRunning) {
76	dl	1.27	return sync.innerCancel(mayInterruptIfRunning);
77	dl	1.13	}
78	jsr166	1.35
79	dl	1.2	public V get() throws InterruptedException, ExecutionException {
80	dl	1.27	return sync.innerGet();
81	tim	1.1	}
82
83	dl	1.2	public V get(long timeout, TimeUnit unit)
84	tim	1.1	throws InterruptedException, ExecutionException, TimeoutException {
85	dl	1.27	return sync.innerGet(unit.toNanos(timeout));
86	tim	1.1	}
87
88			/**
89	dl	1.20	* Protected method invoked when this task transitions to state
90			* <tt>isDone</tt> (whether normally or via cancellation). The
91			* default implementation does nothing. Subclasses may override
92			* this method to invoke completion callbacks or perform
93			* bookkeeping. Note that you can query status inside the
94			* implementation of this method to determine whether this task
95			* has been cancelled.
96			*/
97			protected void done() { }
98
99			/**
100			* Sets the result of this Future to the given value unless
101	dl	1.29	* this future has already been set or has been cancelled.
102	dl	1.40	* This method is invoked internally by the <tt>run</tt> method
103			* upon successful completion of the computation.
104	tim	1.1	* @param v the value
105	jsr166	1.35	*/
106	dl	1.2	protected void set(V v) {
107	dl	1.27	sync.innerSet(v);
108	tim	1.1	}
109
110			/**
111	dl	1.13	* Causes this future to report an <tt>ExecutionException</tt>
112	dl	1.20	* with the given throwable as its cause, unless this Future has
113	dl	1.24	* already been set or has been cancelled.
114	dl	1.40	* This method is invoked internally by the <tt>run</tt> method
115			* upon failure of the computation.
116	jsr166	1.41	* @param t the cause of failure
117	jsr166	1.35	*/
118	dl	1.2	protected void setException(Throwable t) {
119	dl	1.27	sync.innerSetException(t);
120	tim	1.1	}
121	jsr166	1.35
122	dl	1.24	public void run() {
123	dl	1.27	sync.innerRun();
124	dl	1.24	}
125
126			/**
127	dl	1.30	* Executes the computation without setting its result, and then
128			* resets this Future to initial state, failing to do so if the
129	dl	1.24	* computation encounters an exception or is cancelled. This is
130			* designed for use with tasks that intrinsically execute more
131			* than once.
132			* @return true if successfully run and reset
133			*/
134			protected boolean runAndReset() {
135	dl	1.27	return sync.innerRunAndReset();
136	dl	1.14	}
137	dl	1.3
138	dl	1.14	/**
139	dl	1.24	* Synchronization control for FutureTask. Note that this must be
140	dl	1.30	* a non-static inner class in order to invoke the protected
141	dl	1.24	* <tt>done</tt> method. For clarity, all inner class support
142	dl	1.27	* methods are same as outer, prefixed with "inner".
143	dl	1.24	*
144			* Uses AQS sync state to represent run status
145	dl	1.20	*/
146	dl	1.24	private final class Sync extends AbstractQueuedSynchronizer {
147			/** State value representing that task is running */
148			private static final int RUNNING = 1;
149			/** State value representing that task ran */
150			private static final int RAN = 2;
151			/** State value representing that task was cancelled */
152			private static final int CANCELLED = 4;
153
154			/** The underlying callable */
155			private final Callable<V> callable;
156			/** The result to return from get() */
157			private V result;
158			/** The exception to throw from get() */
159			private Throwable exception;
160
161	jsr166	1.35	/**
162	dl	1.24	* The thread running task. When nulled after set/cancel, this
163			* indicates that the results are accessible. Must be
164	dl	1.32	* volatile, to ensure visibility upon completion.
165	dl	1.24	*/
166			private volatile Thread runner;
167
168			Sync(Callable<V> callable) {
169			this.callable = callable;
170			}
171
172			private boolean ranOrCancelled(int state) {
173			return (state & (RAN \| CANCELLED)) != 0;
174			}
175
176			/**
177	dl	1.26	* Implements AQS base acquire to succeed if ran or cancelled
178	dl	1.24	*/
179	dl	1.26	protected int tryAcquireShared(int ignore) {
180	dl	1.27	return innerIsDone()? 1 : -1;
181	dl	1.24	}
182
183			/**
184			* Implements AQS base release to always signal after setting
185			* final done status by nulling runner thread.
186			*/
187	dl	1.25	protected boolean tryReleaseShared(int ignore) {
188	dl	1.24	runner = null;
189	jsr166	1.35	return true;
190	dl	1.24	}
191
192	dl	1.27	boolean innerIsCancelled() {
193	dl	1.24	return getState() == CANCELLED;
194			}
195	jsr166	1.35
196	dl	1.27	boolean innerIsDone() {
197	dl	1.24	return ranOrCancelled(getState()) && runner == null;
198			}
199
200	dl	1.27	V innerGet() throws InterruptedException, ExecutionException {
201	dl	1.24	acquireSharedInterruptibly(0);
202			if (getState() == CANCELLED)
203			throw new CancellationException();
204			if (exception != null)
205			throw new ExecutionException(exception);
206			return result;
207			}
208
209	dl	1.27	V innerGet(long nanosTimeout) throws InterruptedException, ExecutionException, TimeoutException {
210	dl	1.28	if (!tryAcquireSharedNanos(0, nanosTimeout))
211	jsr166	1.35	throw new TimeoutException();
212	dl	1.24	if (getState() == CANCELLED)
213			throw new CancellationException();
214			if (exception != null)
215			throw new ExecutionException(exception);
216			return result;
217			}
218
219	dl	1.27	void innerSet(V v) {
220	jsr166	1.38	for (;;) {
221			int s = getState();
222			if (ranOrCancelled(s))
223			return;
224			if (compareAndSetState(s, RAN))
225			break;
226			}
227	dl	1.24	result = v;
228			releaseShared(0);
229			done();
230			}
231
232	dl	1.27	void innerSetException(Throwable t) {
233	jsr166	1.38	for (;;) {
234			int s = getState();
235			if (ranOrCancelled(s))
236			return;
237			if (compareAndSetState(s, RAN))
238			break;
239			}
240	dl	1.24	exception = t;
241			result = null;
242			releaseShared(0);
243			done();
244			}
245
246	dl	1.27	boolean innerCancel(boolean mayInterruptIfRunning) {
247	jsr166	1.38	for (;;) {
248			int s = getState();
249			if (ranOrCancelled(s))
250			return false;
251			if (compareAndSetState(s, CANCELLED))
252			break;
253			}
254	dl	1.24	if (mayInterruptIfRunning) {
255			Thread r = runner;
256			if (r != null)
257			r.interrupt();
258			}
259			releaseShared(0);
260			done();
261	dl	1.14	return true;
262			}
263
264	dl	1.27	void innerRun() {
265	jsr166	1.35	if (!compareAndSetState(0, RUNNING))
266	dl	1.24	return;
267	dl	1.14	try {
268	dl	1.24	runner = Thread.currentThread();
269	dl	1.37	if (getState() == RUNNING) // recheck after setting thread
270	dl	1.36	innerSet(callable.call());
271			else
272	dl	1.37	releaseShared(0); // cancel
273	jsr166	1.34	} catch (Throwable ex) {
274	dl	1.27	innerSetException(ex);
275	jsr166	1.35	}
276	dl	1.14	}
277
278	dl	1.27	boolean innerRunAndReset() {
279	jsr166	1.35	if (!compareAndSetState(0, RUNNING))
280	dl	1.24	return false;
281	dl	1.15	try {
282	dl	1.24	runner = Thread.currentThread();
283	dl	1.36	if (getState() == RUNNING)
284			callable.call(); // don't set result
285	dl	1.24	runner = null;
286			return compareAndSetState(RUNNING, 0);
287	jsr166	1.34	} catch (Throwable ex) {
288	dl	1.27	innerSetException(ex);
289	dl	1.24	return false;
290	jsr166	1.35	}
291	dl	1.14	}
292	dl	1.15	}
293			}