ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/FutureTask.java
Revision: 1.68
Committed: Fri Jun 17 22:07:24 2011 UTC (12 years, 11 months ago) by jsr166
Branch: MAIN
Changes since 1.67: +38 -44 lines
Log Message:
only reset runner to null after state is decided

File Contents

# 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 jsr166 1.59 * http://creativecommons.org/publicdomain/zero/1.0/
5 tim 1.1 */
6    
7     package java.util.concurrent;
8 dl 1.62 import java.util.concurrent.locks.LockSupport;
9 dl 1.13
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 jsr166 1.64 * retrieved when the computation has completed; the {@code get}
16     * methods will block if the computation has not yet completed. Once
17 dl 1.8 * the computation has completed, the computation cannot be restarted
18 jsr166 1.64 * or cancelled (unless the computation is invoked using
19     * {@link #runAndReset}).
20 tim 1.1 *
21 jsr166 1.64 * <p>A {@code FutureTask} can be used to wrap a {@link Callable} or
22     * {@link Runnable} object. Because {@code FutureTask} implements
23     * {@code Runnable}, a {@code FutureTask} can be submitted to an
24     * {@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 jsr166 1.64 * {@code protected} functionality that may be useful when creating
28 dl 1.14 * customized task classes.
29     *
30 tim 1.1 * @since 1.5
31 dl 1.4 * @author Doug Lea
32 jsr166 1.64 * @param <V> The result type returned by this FutureTask's {@code get} methods
33 tim 1.1 */
34 peierls 1.39 public class FutureTask<V> implements RunnableFuture<V> {
35 dl 1.62 /*
36     * Revision notes: This differs from previous versions of this
37     * class that relied on AbstractQueuedSynchronizer, mainly to
38     * avoid surprising users about retaining interrupt status during
39     * cancellation races. Sync control in the current design relies
40     * on a "state" field updated via CAS to track completion, along
41     * with a simple Treiber stack to hold waiting threads.
42     *
43     * Style note: As usual, we bypass overhead of using
44     * AtomicXFieldUpdaters and instead directly use Unsafe intrinsics.
45     */
46    
47     /**
48 jsr166 1.67 * The run state of this task, initially UNDECIDED. The run state
49 dl 1.62 * transitions to NORMAL, EXCEPTIONAL, or CANCELLED (only) in
50 jsr166 1.67 * method setCompletion. During setCompletion, state may take on
51 dl 1.62 * transient values of COMPLETING (while outcome is being set) or
52 jsr166 1.67 * INTERRUPTING (while interrupting the runner). State values are
53     * ordered and set to powers of two to simplify checks.
54 dl 1.62 */
55     private volatile int state;
56 jsr166 1.67 private static final int UNDECIDED = 0x00;
57 dl 1.62 private static final int COMPLETING = 0x01;
58     private static final int INTERRUPTING = 0x02;
59     private static final int NORMAL = 0x04;
60     private static final int EXCEPTIONAL = 0x08;
61     private static final int CANCELLED = 0x10;
62    
63 jsr166 1.64 /** The underlying callable */
64     private final Callable<V> callable;
65 dl 1.62 /** The result to return or exception to throw from get() */
66     private Object outcome; // non-volatile, protected by state reads/writes
67     /** The thread running the callable; CASed during run() */
68     private volatile Thread runner;
69     /** Treiber stack of waiting threads */
70     private volatile WaitNode waiters;
71    
72     /**
73     * Sets completion status, unless already completed. If
74     * necessary, we first set state to COMPLETING or INTERRUPTING to
75 jsr166 1.64 * establish precedence. This intentionally stalls (just via
76 dl 1.62 * yields) in (uncommon) cases of concurrent calls during
77     * cancellation until state is set, to avoid surprising users
78     * during cancellation races.
79     *
80     * @param x the outcome
81     * @param mode the completion state value
82 jsr166 1.67 * @return true if this call caused transition from UNDECIDED to completed
83 dl 1.62 */
84     private boolean setCompletion(Object x, int mode) {
85     int next = ((mode == INTERRUPTING) ? // set up transient states
86 jsr166 1.68 (runner != null) ? INTERRUPTING : CANCELLED :
87 dl 1.62 (x != null) ? COMPLETING : mode);
88 jsr166 1.68 if (!UNSAFE.compareAndSwapInt(this, stateOffset,
89     UNDECIDED, next))
90     return false;
91     if (next == INTERRUPTING) {
92     Thread t = runner; // recheck
93     if (t != null)
94     t.interrupt();
95     state = CANCELLED;
96     }
97     else if (next == COMPLETING) {
98     outcome = x;
99     state = mode;
100 dl 1.62 }
101 jsr166 1.68 if (waiters != null)
102     releaseAll();
103     done();
104     return true;
105 dl 1.62 }
106    
107     /**
108 jsr166 1.64 * Returns result or throws exception for completed task.
109     *
110 dl 1.62 * @param s completed state value
111     */
112     private V report(int s) throws ExecutionException {
113     Object x = outcome;
114     if (s == NORMAL)
115     return (V)x;
116     if ((s & (CANCELLED | INTERRUPTING)) != 0)
117     throw new CancellationException();
118     throw new ExecutionException((Throwable)x);
119     }
120 dl 1.11
121 tim 1.1 /**
122 jsr166 1.64 * Creates a {@code FutureTask} that will, upon running, execute the
123     * given {@code Callable}.
124 tim 1.1 *
125     * @param callable the callable task
126 dl 1.9 * @throws NullPointerException if callable is null
127 tim 1.1 */
128     public FutureTask(Callable<V> callable) {
129 dl 1.9 if (callable == null)
130     throw new NullPointerException();
131 dl 1.62 this.callable = callable;
132 tim 1.1 }
133    
134     /**
135 jsr166 1.64 * Creates a {@code FutureTask} that will, upon running, execute the
136     * given {@code Runnable}, and arrange that {@code get} will return the
137 tim 1.1 * given result on successful completion.
138     *
139 jsr166 1.54 * @param runnable the runnable task
140 tim 1.1 * @param result the result to return on successful completion. If
141 dl 1.9 * you don't need a particular result, consider using
142 dl 1.16 * constructions of the form:
143 jsr166 1.58 * {@code Future<?> f = new FutureTask<Void>(runnable, null)}
144 dl 1.9 * @throws NullPointerException if runnable is null
145 tim 1.1 */
146 dl 1.15 public FutureTask(Runnable runnable, V result) {
147 dl 1.62 this.callable = Executors.callable(runnable, result);
148 dl 1.20 }
149    
150     public boolean isCancelled() {
151 dl 1.62 return (state & (CANCELLED | INTERRUPTING)) != 0;
152 dl 1.20 }
153 jsr166 1.35
154 dl 1.20 public boolean isDone() {
155 jsr166 1.67 return state != UNDECIDED;
156 dl 1.13 }
157    
158     public boolean cancel(boolean mayInterruptIfRunning) {
159 jsr166 1.67 return state == UNDECIDED &&
160 dl 1.62 setCompletion(null, mayInterruptIfRunning ?
161     INTERRUPTING : CANCELLED);
162 dl 1.13 }
163 jsr166 1.35
164 jsr166 1.43 /**
165     * @throws CancellationException {@inheritDoc}
166     */
167 dl 1.2 public V get() throws InterruptedException, ExecutionException {
168 jsr166 1.64 int s = state;
169     if (s <= COMPLETING)
170     s = awaitDone(false, 0L);
171     return report(s);
172 tim 1.1 }
173    
174 jsr166 1.43 /**
175     * @throws CancellationException {@inheritDoc}
176     */
177 dl 1.2 public V get(long timeout, TimeUnit unit)
178 tim 1.1 throws InterruptedException, ExecutionException, TimeoutException {
179 jsr166 1.64 int s = state;
180     if (s <= COMPLETING &&
181     (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
182 dl 1.62 throw new TimeoutException();
183     return report(s);
184 tim 1.1 }
185    
186     /**
187 dl 1.20 * Protected method invoked when this task transitions to state
188 jsr166 1.64 * {@code isDone} (whether normally or via cancellation). The
189 dl 1.20 * default implementation does nothing. Subclasses may override
190     * this method to invoke completion callbacks or perform
191     * bookkeeping. Note that you can query status inside the
192     * implementation of this method to determine whether this task
193     * has been cancelled.
194     */
195     protected void done() { }
196    
197     /**
198 jsr166 1.64 * Sets the result of this future to the given value unless
199 dl 1.29 * this future has already been set or has been cancelled.
200 jsr166 1.64 *
201     * <p>This method is invoked internally by the {@link #run} method
202 dl 1.40 * upon successful completion of the computation.
203 jsr166 1.64 *
204 tim 1.1 * @param v the value
205 jsr166 1.35 */
206 dl 1.2 protected void set(V v) {
207 dl 1.62 setCompletion(v, NORMAL);
208 tim 1.1 }
209    
210     /**
211 jsr166 1.64 * Causes this future to report an {@link ExecutionException}
212     * with the given throwable as its cause, unless this future has
213 dl 1.24 * already been set or has been cancelled.
214 jsr166 1.64 *
215     * <p>This method is invoked internally by the {@link #run} method
216 dl 1.40 * upon failure of the computation.
217 jsr166 1.64 *
218 jsr166 1.41 * @param t the cause of failure
219 jsr166 1.35 */
220 dl 1.2 protected void setException(Throwable t) {
221 dl 1.62 setCompletion(t, EXCEPTIONAL);
222 tim 1.1 }
223 jsr166 1.35
224 dl 1.24 public void run() {
225 jsr166 1.68 if (state != UNDECIDED ||
226     !UNSAFE.compareAndSwapObject(this, runnerOffset,
227     null, Thread.currentThread()))
228     return;
229    
230     try {
231 dl 1.62 V result;
232     try {
233     result = callable.call();
234     } catch (Throwable ex) {
235     setException(ex);
236     return;
237     }
238     set(result);
239 jsr166 1.68 } finally {
240     runner = null;
241     while (state == INTERRUPTING)
242     Thread.yield(); // wait out pending cancellation interrupt
243 dl 1.62 }
244 dl 1.24 }
245    
246     /**
247 dl 1.30 * Executes the computation without setting its result, and then
248 jsr166 1.64 * resets this future to initial state, failing to do so if the
249 dl 1.24 * computation encounters an exception or is cancelled. This is
250     * designed for use with tasks that intrinsically execute more
251     * than once.
252 jsr166 1.64 *
253 dl 1.24 * @return true if successfully run and reset
254     */
255     protected boolean runAndReset() {
256 jsr166 1.67 if (state != UNDECIDED ||
257 jsr166 1.66 !UNSAFE.compareAndSwapObject(this, runnerOffset,
258     null, Thread.currentThread()))
259 dl 1.62 return false;
260 jsr166 1.68
261 dl 1.62 try {
262 jsr166 1.68 try {
263     callable.call(); // don't set result
264     return (state == UNDECIDED);
265     } catch (Throwable ex) {
266     setException(ex);
267 dl 1.62 return false;
268 jsr166 1.68 }
269     } finally {
270     runner = null;
271     while (state == INTERRUPTING)
272     Thread.yield(); // wait out pending cancellation interrupt
273 dl 1.62 }
274 dl 1.14 }
275 dl 1.3
276 dl 1.14 /**
277 dl 1.62 * Simple linked list nodes to record waiting threads in a Treiber
278 jsr166 1.64 * stack. See other classes such as Phaser and SynchronousQueue
279 dl 1.62 * for more detailed explanation.
280 dl 1.20 */
281 dl 1.62 static final class WaitNode {
282     volatile Thread thread;
283     WaitNode next;
284     }
285 dl 1.42
286 dl 1.62 /**
287 jsr166 1.64 * Removes and signals all waiting threads.
288 dl 1.62 */
289     private void releaseAll() {
290     WaitNode q;
291     while ((q = waiters) != null) {
292     if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
293     for (;;) {
294     Thread t = q.thread;
295     if (t != null) {
296     q.thread = null;
297     LockSupport.unpark(t);
298     }
299     WaitNode next = q.next;
300     if (next == null)
301     return;
302     q.next = null; // unlink to help gc
303     q = next;
304     }
305     }
306 dl 1.24 }
307 dl 1.62 }
308 dl 1.24
309 dl 1.62 /**
310 jsr166 1.64 * Awaits completion or aborts on interrupt or timeout.
311     *
312 dl 1.62 * @param timed true if use timed waits
313 jsr166 1.64 * @param nanos time to wait, if timed
314 dl 1.62 * @return state upon completion
315     */
316     private int awaitDone(boolean timed, long nanos)
317     throws InterruptedException {
318 jsr166 1.63 long last = timed ? System.nanoTime() : 0L;
319 dl 1.62 WaitNode q = null;
320     boolean queued = false;
321 jsr166 1.64 for (;;) {
322 dl 1.62 if (Thread.interrupted()) {
323     removeWaiter(q);
324     throw new InterruptedException();
325     }
326 jsr166 1.64
327     int s = state;
328     if (s > COMPLETING) {
329 dl 1.62 if (q != null)
330     q.thread = null;
331     return s;
332     }
333     else if (q == null)
334     q = new WaitNode();
335     else if (!queued)
336     queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
337     q.next = waiters, q);
338     else if (q.thread == null)
339     q.thread = Thread.currentThread();
340     else if (timed) {
341     long now = System.nanoTime();
342     if ((nanos -= (now - last)) <= 0L) {
343     removeWaiter(q);
344     return state;
345 dl 1.50 }
346 dl 1.62 last = now;
347     LockSupport.parkNanos(this, nanos);
348 dl 1.50 }
349 dl 1.62 else
350     LockSupport.park(this);
351 dl 1.24 }
352 dl 1.62 }
353 dl 1.24
354 dl 1.62 /**
355 jsr166 1.64 * Tries to unlink a timed-out or interrupted wait node to avoid
356     * accumulating garbage. Internal nodes are simply unspliced
357 dl 1.62 * without CAS since it is harmless if they are traversed anyway
358     * by releasers or concurrent calls to removeWaiter.
359     */
360     private void removeWaiter(WaitNode node) {
361     if (node != null) {
362     node.thread = null;
363     WaitNode pred = null;
364     WaitNode q = waiters;
365     while (q != null) {
366     WaitNode next = node.next;
367     if (q != node) {
368     pred = q;
369     q = next;
370 dl 1.50 }
371 dl 1.62 else if (pred != null) {
372     pred.next = next;
373     break;
374 dl 1.50 }
375 dl 1.62 else if (UNSAFE.compareAndSwapObject(this, waitersOffset,
376     q, next))
377 jsr166 1.56 break;
378 dl 1.62 else { // restart on CAS failure
379     pred = null;
380     q = waiters;
381 jsr166 1.55 }
382 jsr166 1.56 }
383 dl 1.14 }
384 dl 1.62 }
385 dl 1.14
386 dl 1.62 // Unsafe mechanics
387     private static final sun.misc.Unsafe UNSAFE;
388     private static final long stateOffset;
389     private static final long runnerOffset;
390     private static final long waitersOffset;
391     static {
392     try {
393     UNSAFE = sun.misc.Unsafe.getUnsafe();
394     Class<?> k = FutureTask.class;
395     stateOffset = UNSAFE.objectFieldOffset
396     (k.getDeclaredField("state"));
397     runnerOffset = UNSAFE.objectFieldOffset
398     (k.getDeclaredField("runner"));
399     waitersOffset = UNSAFE.objectFieldOffset
400     (k.getDeclaredField("waiters"));
401     } catch (Exception e) {
402     throw new Error(e);
403 dl 1.14 }
404 dl 1.15 }
405 dl 1.62
406 dl 1.15 }