ViewVC Help
View File | Revision Log | Show Annotations | Download File | Root Listing
root/jsr166/jsr166/src/main/java/util/concurrent/FutureTask.java
Revision: 1.87
Committed: Sun Jun 19 07:46:18 2011 UTC (12 years, 11 months ago) by jsr166
Branch: MAIN
Changes since 1.86: +19 -19 lines
Log Message:
use try-finally to reset runner back to null; set/setException may throw

File Contents

# 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.LockSupport;
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 {@code get}
16 * methods will block if the computation has not yet completed. Once
17 * the computation has completed, the computation cannot be restarted
18 * or cancelled (unless the computation is invoked using
19 * {@link #runAndReset}).
20 *
21 * <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 *
26 * <p>In addition to serving as a standalone class, this class provides
27 * {@code protected} functionality that may be useful when creating
28 * customized task classes.
29 *
30 * @since 1.5
31 * @author Doug Lea
32 * @param <V> The result type returned by this FutureTask's {@code get} methods
33 */
34 public class FutureTask<V> implements RunnableFuture<V> {
35 /*
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 * The run state of this task, initially NEW. The run state
49 * transitions to a terminal state only in methods set,
50 * setException, and cancel. During completion, state may take on
51 * transient values of COMPLETING (while outcome is being set) or
52 * INTERRUPTING (only while interrupting the runner to satisfy a
53 * cancel(true)). Transitions from these intermediate to final
54 * states use cheaper ordered/lazy writes because values are unique
55 * and cannot be further modified.
56 *
57 * Possible state transitions:
58 * NEW -> COMPLETING -> NORMAL
59 * NEW -> COMPLETING -> EXCEPTIONAL
60 * NEW -> CANCELLED
61 * NEW -> INTERRUPTING -> INTERRUPTED
62 */
63 private volatile int state;
64 private static final int NEW = 0;
65 private static final int COMPLETING = 1;
66 private static final int NORMAL = 2;
67 private static final int EXCEPTIONAL = 3;
68 private static final int CANCELLED = 4;
69 private static final int INTERRUPTING = 5;
70 private static final int INTERRUPTED = 6;
71
72 /** The underlying callable; nulled out after running */
73 private Callable<V> callable;
74 /** The result to return or exception to throw from get() */
75 private Object outcome; // non-volatile, protected by state reads/writes
76 /** The thread running the callable; CASed during run() */
77 private volatile Thread runner;
78 /** Treiber stack of waiting threads */
79 private volatile WaitNode waiters;
80
81 /**
82 * Returns result or throws exception for completed task.
83 *
84 * @param s completed state value
85 */
86 private V report(int s) throws ExecutionException {
87 Object x = outcome;
88 if (s == NORMAL)
89 return (V)x;
90 if (s >= CANCELLED)
91 throw new CancellationException();
92 throw new ExecutionException((Throwable)x);
93 }
94
95 /**
96 * Creates a {@code FutureTask} that will, upon running, execute the
97 * given {@code Callable}.
98 *
99 * @param callable the callable task
100 * @throws NullPointerException if the callable is null
101 */
102 public FutureTask(Callable<V> callable) {
103 if (callable == null)
104 throw new NullPointerException();
105 this.callable = callable;
106 this.state = NEW; // ensure visibility of callable
107 }
108
109 /**
110 * Creates a {@code FutureTask} that will, upon running, execute the
111 * given {@code Runnable}, and arrange that {@code get} will return the
112 * given result on successful completion.
113 *
114 * @param runnable the runnable task
115 * @param result the result to return on successful completion. If
116 * you don't need a particular result, consider using
117 * constructions of the form:
118 * {@code Future<?> f = new FutureTask<Void>(runnable, null)}
119 * @throws NullPointerException if the runnable is null
120 */
121 public FutureTask(Runnable runnable, V result) {
122 this.callable = Executors.callable(runnable, result);
123 this.state = NEW; // ensure visibility of callable
124 }
125
126 public boolean isCancelled() {
127 return state >= CANCELLED;
128 }
129
130 public boolean isDone() {
131 return state != NEW;
132 }
133
134 public boolean cancel(boolean mayInterruptIfRunning) {
135 if (state != NEW)
136 return false;
137 if (mayInterruptIfRunning) {
138 if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, INTERRUPTING))
139 return false;
140 Thread t = runner;
141 if (t != null)
142 t.interrupt();
143 UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED); // final state
144 }
145 else if (!UNSAFE.compareAndSwapInt(this, stateOffset, NEW, CANCELLED))
146 return false;
147 finishCompletion();
148 return true;
149 }
150
151 /**
152 * @throws CancellationException {@inheritDoc}
153 */
154 public V get() throws InterruptedException, ExecutionException {
155 int s = state;
156 return report((s <= COMPLETING) ? awaitDone(false, 0L) : s);
157 }
158
159 /**
160 * @throws CancellationException {@inheritDoc}
161 */
162 public V get(long timeout, TimeUnit unit)
163 throws InterruptedException, ExecutionException, TimeoutException {
164 if (unit == null)
165 throw new NullPointerException();
166 int s = state;
167 if (s <= COMPLETING &&
168 (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
169 throw new TimeoutException();
170 return report(s);
171 }
172
173 /**
174 * Protected method invoked when this task transitions to state
175 * {@code isDone} (whether normally or via cancellation). The
176 * default implementation does nothing. Subclasses may override
177 * this method to invoke completion callbacks or perform
178 * bookkeeping. Note that you can query status inside the
179 * implementation of this method to determine whether this task
180 * has been cancelled.
181 */
182 protected void done() { }
183
184 /**
185 * Sets the result of this future to the given value unless
186 * this future has already been set or has been cancelled.
187 *
188 * <p>This method is invoked internally by the {@link #run} method
189 * upon successful completion of the computation.
190 *
191 * @param v the value
192 */
193 protected void set(V v) {
194 if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
195 outcome = v;
196 UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
197 finishCompletion();
198 }
199 }
200
201 /**
202 * Causes this future to report an {@link ExecutionException}
203 * with the given throwable as its cause, unless this future has
204 * already been set or has been cancelled.
205 *
206 * <p>This method is invoked internally by the {@link #run} method
207 * upon failure of the computation.
208 *
209 * @param t the cause of failure
210 */
211 protected void setException(Throwable t) {
212 if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
213 outcome = t;
214 UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
215 finishCompletion();
216 }
217 }
218
219 public void run() {
220 if (state != NEW ||
221 !UNSAFE.compareAndSwapObject(this, runnerOffset,
222 null, Thread.currentThread()))
223 return;
224 try {
225 Callable<V> c = callable;
226 if (c != null && state == NEW) {
227 V result;
228 try {
229 result = c.call();
230 } catch (Throwable ex) {
231 setException(ex);
232 return;
233 }
234 set(result);
235 }
236 } finally {
237 runner = null;
238 int s = state;
239 if (s >= INTERRUPTING)
240 handlePossibleCancellationInterrupt(s);
241 }
242 }
243
244 /**
245 * Executes the computation without setting its result, and then
246 * resets this future to initial state, failing to do so if the
247 * computation encounters an exception or is cancelled. This is
248 * designed for use with tasks that intrinsically execute more
249 * than once.
250 *
251 * @return true if successfully run and reset
252 */
253 protected boolean runAndReset() {
254 if (state != NEW ||
255 !UNSAFE.compareAndSwapObject(this, runnerOffset,
256 null, Thread.currentThread()))
257 return false;
258 try {
259 Callable<V> c = callable;
260 if (c != null && state == NEW) {
261 try {
262 c.call(); // don't set result
263 return state == NEW;
264 } catch (Throwable ex) {
265 setException(ex);
266 }
267 }
268 return false;
269 } finally {
270 runner = null;
271 int s = state;
272 if (s >= INTERRUPTING)
273 handlePossibleCancellationInterrupt(s);
274 }
275 }
276
277 /**
278 * Ensures that any interrupt from a possible cancel(true) does
279 * not leak into subsequent code.
280 */
281 private void handlePossibleCancellationInterrupt(int s) {
282 // It is possible for our interrupter to stall before getting a
283 // chance to interrupt us. Let's spin-wait patiently.
284 if (s == INTERRUPTING) {
285 while ((s = state) == INTERRUPTING)
286 Thread.yield(); // wait out pending interrupt
287 }
288 assert state == INTERRUPTED;
289 // Clear any interrupt we may have received.
290 Thread.interrupted(); // clear interrupt from cancel(true)
291 }
292
293 /**
294 * Simple linked list nodes to record waiting threads in a Treiber
295 * stack. See other classes such as Phaser and SynchronousQueue
296 * for more detailed explanation.
297 */
298 static final class WaitNode {
299 volatile Thread thread;
300 volatile WaitNode next;
301 WaitNode() { thread = Thread.currentThread(); }
302 }
303
304 /**
305 * Removes and signals all waiting threads, invokes done(), and
306 * nulls out callable.
307 */
308 private void finishCompletion() {
309 assert state > NEW;
310 for (WaitNode q; (q = waiters) != null;) {
311 if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
312 for (;;) {
313 Thread t = q.thread;
314 if (t != null) {
315 q.thread = null;
316 LockSupport.unpark(t);
317 }
318 WaitNode next = q.next;
319 if (next == null)
320 break;
321 q.next = null; // unlink to help gc
322 q = next;
323 }
324 break;
325 }
326 }
327
328 done();
329
330 callable = null; // to reduce footprint
331 }
332
333 /**
334 * Awaits completion or aborts on interrupt or timeout.
335 *
336 * @param timed true if use timed waits
337 * @param nanos time to wait, if timed
338 * @return state upon completion
339 */
340 private int awaitDone(boolean timed, long nanos)
341 throws InterruptedException {
342 long last = timed ? System.nanoTime() : 0L;
343 WaitNode q = null;
344 boolean queued = false;
345 for (;;) {
346 if (Thread.interrupted()) {
347 removeWaiter(q);
348 throw new InterruptedException();
349 }
350
351 int s = state;
352 if (s > COMPLETING) {
353 if (q != null)
354 q.thread = null;
355 return s;
356 }
357 else if (q == null)
358 q = new WaitNode();
359 else if (!queued)
360 queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
361 q.next = waiters, q);
362 else if (timed) {
363 long now = System.nanoTime();
364 if ((nanos -= (now - last)) <= 0L) {
365 removeWaiter(q);
366 return state;
367 }
368 last = now;
369 LockSupport.parkNanos(this, nanos);
370 }
371 else
372 LockSupport.park(this);
373 }
374 }
375
376 /**
377 * Tries to unlink a timed-out or interrupted wait node to avoid
378 * accumulating garbage. Internal nodes are simply unspliced
379 * without CAS since it is harmless if they are traversed anyway
380 * by releasers. To avoid effects of unsplicing from already
381 * removed nodes, the list is retraversed in case of an apparent
382 * race. This is slow when there are a lot of nodes, but we don't
383 * expect lists to be long enough to outweigh higher-overhead
384 * schemes.
385 */
386 private void removeWaiter(WaitNode node) {
387 if (node != null) {
388 node.thread = null;
389 retry:
390 for (;;) { // restart on removeWaiter race
391 for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
392 s = q.next;
393 if (q.thread != null)
394 pred = q;
395 else if (pred != null) {
396 pred.next = s;
397 if (pred.thread == null) // check for race
398 continue retry;
399 }
400 else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
401 q, s))
402 continue retry;
403 }
404 break;
405 }
406 }
407 }
408
409 // Unsafe mechanics
410 private static final sun.misc.Unsafe UNSAFE;
411 private static final long stateOffset;
412 private static final long runnerOffset;
413 private static final long waitersOffset;
414 static {
415 try {
416 UNSAFE = sun.misc.Unsafe.getUnsafe();
417 Class<?> k = FutureTask.class;
418 stateOffset = UNSAFE.objectFieldOffset
419 (k.getDeclaredField("state"));
420 runnerOffset = UNSAFE.objectFieldOffset
421 (k.getDeclaredField("runner"));
422 waitersOffset = UNSAFE.objectFieldOffset
423 (k.getDeclaredField("waiters"));
424 } catch (Exception e) {
425 throw new Error(e);
426 }
427 }
428
429 }