1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326
/* * @(#)FutureTask.java 1.14 06/07/13 * * Copyright 2006 Sun Microsystems, Inc. All rights reserved. * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms. */ 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: * <tt>Future<?> f = new FutureTask<Object>(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 (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; result = null; 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(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; } } } }