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/publicdomain/zero/1.0/
 */

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 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:
     * {@code Future<?> f = new FutureTask<Void>(runnable, null)}
     * @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);
    }

    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 ready to run */
        private static final int READY     = 0;
        /** 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 (s == RAN)
                    return;
                if (s == CANCELLED) {
                    // aggressively release to set runner to null,
                    // in case we are racing with a cancel request
                    // that will try to interrupt runner
                    releaseShared(0);
                    return;
                }
                if (compareAndSetState(s, RAN)) {
                    result = v;
                    releaseShared(0);
                    done();
                    return;
                }
            }
        }

        void innerSetException(Throwable t) {
            for (;;) {
                int s = getState();
                if (s == RAN)
                    return;
                if (s == CANCELLED) {
                    // aggressively release to set runner to null,
                    // in case we are racing with a cancel request
                    // that will try to interrupt runner
                    releaseShared(0);
                    return;
                }
                if (compareAndSetState(s, RAN)) {
                    exception = t;
                    releaseShared(0);
                    done();
                    return;
                }
            }
        }

        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(READY, RUNNING))
                return;

            runner = Thread.currentThread();
            if (getState() == RUNNING) { // recheck after setting thread
                V result;
                try {
                    result = callable.call();
                } catch (Throwable ex) {
                    setException(ex);
                    return;
                }
                set(result);
            } else {
                releaseShared(0); // cancel
            }
        }

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