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 (till 6280605 is fixed)
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.
 *
 * <p> Memory visibility effects: State changes to the resulting value
 * or exception <a
 * href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
 * actions following a call to <tt>get()</tt>.
 *
 * @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);
    }

    /**
     * @throws CancellationException {@inheritDoc}
     */
    public V get() throws InterruptedException, ExecutionException {
        return sync.innerGet();
    }

    /**
     * @throws CancellationException {@inheritDoc}
     */
    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);
    }

    // The following (duplicated) doc comment can be removed once
    //
    // 6270645: Javadoc comments should be inherited from most derived
    //          superinterface or superclass
    // is fixed.
    /**
     * Sets this Future to the result of its computation
     * unless it has been cancelled.
     */
    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 {
        private static final long serialVersionUID = -7828117401763700385L;

        /** 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.45
Committed:	Fri Sep 2 01:03:08 2005 UTC (18 years, 9 months ago) by brian
Branch:	MAIN
Changes since 1.44:	+5 -0 lines
Log Message:	Happens-before markup
#	Content
1	/*
2	* Written by Doug Lea with assistance from members of JCP JSR-166
3	* Expert Group and released to the public domain, as explained at
4	* http://creativecommons.org/licenses/publicdomain
5	*/
6
7	package java.util.concurrent;
8	import java.util.concurrent.*; // for javadoc (till 6280605 is fixed)
9	import java.util.concurrent.locks.*;
10
11	/**
12	* 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	* 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	* the computation has completed, the computation cannot be restarted
19	* or cancelled.
20	*
21	* <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	*
26	* <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	* <p> Memory visibility effects: State changes to the resulting value
31	* or exception <a
32	* href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
33	* actions following a call to <tt>get()</tt>.
34	*
35	* @since 1.5
36	* @author Doug Lea
37	* @param <V> The result type returned by this FutureTask's <tt>get</tt> method
38	*/
39	public class FutureTask<V> implements RunnableFuture<V> {
40	/** Synchronization control for FutureTask */
41	private final Sync sync;
42
43	/**
44	* Creates a <tt>FutureTask</tt> that will upon running, execute the
45	* given <tt>Callable</tt>.
46	*
47	* @param callable the callable task
48	* @throws NullPointerException if callable is null
49	*/
50	public FutureTask(Callable<V> callable) {
51	if (callable == null)
52	throw new NullPointerException();
53	sync = new Sync(callable);
54	}
55
56	/**
57	* Creates a <tt>FutureTask</tt> that will upon running, execute the
58	* given <tt>Runnable</tt>, and arrange that <tt>get</tt> will return the
59	* given result on successful completion.
60	*
61	* @param runnable the runnable task
62	* @param result the result to return on successful completion. If
63	* you don't need a particular result, consider using
64	* constructions of the form:
65	* <tt>Future<?> f = new FutureTask<Object>(runnable, null)</tt>
66	* @throws NullPointerException if runnable is null
67	*/
68	public FutureTask(Runnable runnable, V result) {
69	sync = new Sync(Executors.callable(runnable, result));
70	}
71
72	public boolean isCancelled() {
73	return sync.innerIsCancelled();
74	}
75
76	public boolean isDone() {
77	return sync.innerIsDone();
78	}
79
80	public boolean cancel(boolean mayInterruptIfRunning) {
81	return sync.innerCancel(mayInterruptIfRunning);
82	}
83
84	/**
85	* @throws CancellationException {@inheritDoc}
86	*/
87	public V get() throws InterruptedException, ExecutionException {
88	return sync.innerGet();
89	}
90
91	/**
92	* @throws CancellationException {@inheritDoc}
93	*/
94	public V get(long timeout, TimeUnit unit)
95	throws InterruptedException, ExecutionException, TimeoutException {
96	return sync.innerGet(unit.toNanos(timeout));
97	}
98
99	/**
100	* Protected method invoked when this task transitions to state
101	* <tt>isDone</tt> (whether normally or via cancellation). The
102	* default implementation does nothing. Subclasses may override
103	* this method to invoke completion callbacks or perform
104	* bookkeeping. Note that you can query status inside the
105	* implementation of this method to determine whether this task
106	* has been cancelled.
107	*/
108	protected void done() { }
109
110	/**
111	* Sets the result of this Future to the given value unless
112	* this future has already been set or has been cancelled.
113	* This method is invoked internally by the <tt>run</tt> method
114	* upon successful completion of the computation.
115	* @param v the value
116	*/
117	protected void set(V v) {
118	sync.innerSet(v);
119	}
120
121	/**
122	* Causes this future to report an <tt>ExecutionException</tt>
123	* with the given throwable as its cause, unless this Future has
124	* already been set or has been cancelled.
125	* This method is invoked internally by the <tt>run</tt> method
126	* upon failure of the computation.
127	* @param t the cause of failure
128	*/
129	protected void setException(Throwable t) {
130	sync.innerSetException(t);
131	}
132
133	// The following (duplicated) doc comment can be removed once
134	//
135	// 6270645: Javadoc comments should be inherited from most derived
136	// superinterface or superclass
137	// is fixed.
138	/**
139	* Sets this Future to the result of its computation
140	* unless it has been cancelled.
141	*/
142	public void run() {
143	sync.innerRun();
144	}
145
146	/**
147	* Executes the computation without setting its result, and then
148	* resets this Future to initial state, failing to do so if the
149	* computation encounters an exception or is cancelled. This is
150	* designed for use with tasks that intrinsically execute more
151	* than once.
152	* @return true if successfully run and reset
153	*/
154	protected boolean runAndReset() {
155	return sync.innerRunAndReset();
156	}
157
158	/**
159	* Synchronization control for FutureTask. Note that this must be
160	* a non-static inner class in order to invoke the protected
161	* <tt>done</tt> method. For clarity, all inner class support
162	* methods are same as outer, prefixed with "inner".
163	*
164	* Uses AQS sync state to represent run status
165	*/
166	private final class Sync extends AbstractQueuedSynchronizer {
167	private static final long serialVersionUID = -7828117401763700385L;
168
169	/** State value representing that task is running */
170	private static final int RUNNING = 1;
171	/** State value representing that task ran */
172	private static final int RAN = 2;
173	/** State value representing that task was cancelled */
174	private static final int CANCELLED = 4;
175
176	/** The underlying callable */
177	private final Callable<V> callable;
178	/** The result to return from get() */
179	private V result;
180	/** The exception to throw from get() */
181	private Throwable exception;
182
183	/**
184	* The thread running task. When nulled after set/cancel, this
185	* indicates that the results are accessible. Must be
186	* volatile, to ensure visibility upon completion.
187	*/
188	private volatile Thread runner;
189
190	Sync(Callable<V> callable) {
191	this.callable = callable;
192	}
193
194	private boolean ranOrCancelled(int state) {
195	return (state & (RAN \| CANCELLED)) != 0;
196	}
197
198	/**
199	* Implements AQS base acquire to succeed if ran or cancelled
200	*/
201	protected int tryAcquireShared(int ignore) {
202	return innerIsDone()? 1 : -1;
203	}
204
205	/**
206	* Implements AQS base release to always signal after setting
207	* final done status by nulling runner thread.
208	*/
209	protected boolean tryReleaseShared(int ignore) {
210	runner = null;
211	return true;
212	}
213
214	boolean innerIsCancelled() {
215	return getState() == CANCELLED;
216	}
217
218	boolean innerIsDone() {
219	return ranOrCancelled(getState()) && runner == null;
220	}
221
222	V innerGet() throws InterruptedException, ExecutionException {
223	acquireSharedInterruptibly(0);
224	if (getState() == CANCELLED)
225	throw new CancellationException();
226	if (exception != null)
227	throw new ExecutionException(exception);
228	return result;
229	}
230
231	V innerGet(long nanosTimeout) throws InterruptedException, ExecutionException, TimeoutException {
232	if (!tryAcquireSharedNanos(0, nanosTimeout))
233	throw new TimeoutException();
234	if (getState() == CANCELLED)
235	throw new CancellationException();
236	if (exception != null)
237	throw new ExecutionException(exception);
238	return result;
239	}
240
241	void innerSet(V v) {
242	for (;;) {
243	int s = getState();
244	if (ranOrCancelled(s))
245	return;
246	if (compareAndSetState(s, RAN))
247	break;
248	}
249	result = v;
250	releaseShared(0);
251	done();
252	}
253
254	void innerSetException(Throwable t) {
255	for (;;) {
256	int s = getState();
257	if (ranOrCancelled(s))
258	return;
259	if (compareAndSetState(s, RAN))
260	break;
261	}
262	exception = t;
263	result = null;
264	releaseShared(0);
265	done();
266	}
267
268	boolean innerCancel(boolean mayInterruptIfRunning) {
269	for (;;) {
270	int s = getState();
271	if (ranOrCancelled(s))
272	return false;
273	if (compareAndSetState(s, CANCELLED))
274	break;
275	}
276	if (mayInterruptIfRunning) {
277	Thread r = runner;
278	if (r != null)
279	r.interrupt();
280	}
281	releaseShared(0);
282	done();
283	return true;
284	}
285
286	void innerRun() {
287	if (!compareAndSetState(0, RUNNING))
288	return;
289	try {
290	runner = Thread.currentThread();
291	if (getState() == RUNNING) // recheck after setting thread
292	innerSet(callable.call());
293	else
294	releaseShared(0); // cancel
295	} catch (Throwable ex) {
296	innerSetException(ex);
297	}
298	}
299
300	boolean innerRunAndReset() {
301	if (!compareAndSetState(0, RUNNING))
302	return false;
303	try {
304	runner = Thread.currentThread();
305	if (getState() == RUNNING)
306	callable.call(); // don't set result
307	runner = null;
308	return compareAndSetState(RUNNING, 0);
309	} catch (Throwable ex) {
310	innerSetException(ex);
311	return false;
312	}
313	}
314	}
315	}