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.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 Future<V>, Runnable {
    /** Synchronization control for FutureTask */
    private final Sync sync;

    /**
     * Constructs 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);
    }

    /**
     * Constructs 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.doIsCancelled();
    }
    
    public boolean isDone() {
        return sync.doIsDone();
    }

    public boolean cancel(boolean mayInterruptIfRunning) {
        return sync.doCancel(mayInterruptIfRunning);
    }
    
    /**
     * Waits if necessary for execution to complete, and then
     * retrieves its result.
     *
     * @return computed result
     * @throws CancellationException if underlying computation was
     * cancelled
     * @throws ExecutionException if underlying computation threw an
     * exception
     * @throws InterruptedException if current thread was interrupted
     * while waiting
     */
    public V get() throws InterruptedException, ExecutionException {
        return sync.doGet();
    }

    /**
     * Waits if necessary for at most the given time for execution to
     * complete, and then retrieves its result, if available.
     *
     * @param timeout the maximum time to wait
     * @param unit the time unit of the timeout argument
     * @return computed result
     * @throws CancellationException if underlying computation was
     * cancelled
     * @throws ExecutionException if underlying computation threw an
     * exception
     * @throws InterruptedException if current thread was interrupted
     * while waiting
     * @throws TimeoutException if the wait timed out
     */
    public V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        return sync.doGet(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
     * @param v the value
     */ 
    protected void set(V v) {
        sync.doSet(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.
     * @param t the cause of failure.
     */ 
    protected void setException(Throwable t) {
        sync.doSetException(t);
    }
    
    /**
     * Sets this Future to the result of computation unless
     * it has been cancelled.
     */
    public void run() {
        sync.doRun();
    }

    /**
     * Executes the computation without setting 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.doRunAndReset();
    }

    /**
     * Synchronization control for FutureTask. Note that this must be
     * a non-static inner class in order to invoke protected
     * <tt>done</tt> method. For clarity, all inner class support
     * methods are same as outer, prefixed with "do".
     *
     * 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 serve as write barrier on 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 Done/cancelled
         */
        protected int tryAcquireShared(boolean b, int ignore) {
            return doIsDone()? 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 doIsCancelled() {
            return getState() == CANCELLED;
        }
        
        boolean doIsDone() {
            return ranOrCancelled(getState()) && runner == null;
        }

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

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

        void doSet(V v) {
            int s = getState();
            if (ranOrCancelled(s) || !compareAndSetState(s, RAN))
                return;
            result = v;
            releaseShared(0);
            done();
        }

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

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

        void doRun() {
            if (!compareAndSetState(0, RUNNING)) 
                return;
            try {
                runner = Thread.currentThread();
                doSet(callable.call());
            } catch(Throwable ex) {
                doSetException(ex);
            } 
        }

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