001/* 002 * Copyright (C) 2007 The Guava Authors 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except 005 * in compliance with the License. You may obtain a copy of the License at 006 * 007 * http://www.apache.org/licenses/LICENSE-2.0 008 * 009 * Unless required by applicable law or agreed to in writing, software distributed under the License 010 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express 011 * or implied. See the License for the specific language governing permissions and limitations under 012 * the License. 013 */ 014 015package com.google.common.util.concurrent; 016 017import static com.google.common.base.Preconditions.checkNotNull; 018import static com.google.common.base.Strings.isNullOrEmpty; 019import static com.google.common.base.Throwables.throwIfUnchecked; 020import static com.google.common.util.concurrent.Futures.getDone; 021import static com.google.common.util.concurrent.MoreExecutors.directExecutor; 022import static java.util.concurrent.atomic.AtomicReferenceFieldUpdater.newUpdater; 023 024import com.google.common.annotations.Beta; 025import com.google.common.annotations.GwtCompatible; 026import com.google.common.base.Ascii; 027import com.google.errorprone.annotations.CanIgnoreReturnValue; 028import com.google.errorprone.annotations.DoNotMock; 029import com.google.errorprone.annotations.ForOverride; 030import com.google.j2objc.annotations.ReflectionSupport; 031import java.security.AccessController; 032import java.security.PrivilegedActionException; 033import java.security.PrivilegedExceptionAction; 034import java.util.concurrent.CancellationException; 035import java.util.concurrent.ExecutionException; 036import java.util.concurrent.Executor; 037import java.util.concurrent.Future; 038import java.util.concurrent.ScheduledFuture; 039import java.util.concurrent.TimeUnit; 040import java.util.concurrent.TimeoutException; 041import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; 042import java.util.concurrent.locks.LockSupport; 043import java.util.logging.Level; 044import java.util.logging.Logger; 045import org.checkerframework.checker.nullness.compatqual.NullableDecl; 046 047/** 048 * An abstract implementation of {@link ListenableFuture}, intended for advanced users only. More 049 * common ways to create a {@code ListenableFuture} include instantiating a {@link SettableFuture}, 050 * submitting a task to a {@link ListeningExecutorService}, and deriving a {@code Future} from an 051 * existing one, typically using methods like {@link Futures#transform(ListenableFuture, 052 * com.google.common.base.Function, java.util.concurrent.Executor) Futures.transform} and {@link 053 * Futures#catching(ListenableFuture, Class, com.google.common.base.Function, 054 * java.util.concurrent.Executor) Futures.catching}. 055 * 056 * <p>This class implements all methods in {@code ListenableFuture}. Subclasses should provide a way 057 * to set the result of the computation through the protected methods {@link #set(Object)}, {@link 058 * #setFuture(ListenableFuture)} and {@link #setException(Throwable)}. Subclasses may also override 059 * {@link #afterDone()}, which will be invoked automatically when the future completes. Subclasses 060 * should rarely override other methods. 061 * 062 * @author Sven Mawson 063 * @author Luke Sandberg 064 * @since 1.0 065 */ 066@SuppressWarnings("ShortCircuitBoolean") // we use non-short circuiting comparisons intentionally 067@DoNotMock("Use Futures.immediate*Future or SettableFuture") 068@GwtCompatible(emulated = true) 069@ReflectionSupport(value = ReflectionSupport.Level.FULL) 070public abstract class AbstractFuture<V> extends FluentFuture<V> { 071 // NOTE: Whenever both tests are cheap and functional, it's faster to use &, | instead of &&, || 072 073 private static final boolean GENERATE_CANCELLATION_CAUSES = 074 Boolean.parseBoolean( 075 System.getProperty("guava.concurrent.generate_cancellation_cause", "false")); 076 077 /** 078 * A less abstract subclass of AbstractFuture. This can be used to optimize setFuture by ensuring 079 * that {@link #get} calls exactly the implementation of {@link AbstractFuture#get}. 080 */ 081 abstract static class TrustedFuture<V> extends AbstractFuture<V> { 082 @CanIgnoreReturnValue 083 @Override 084 public final V get() throws InterruptedException, ExecutionException { 085 return super.get(); 086 } 087 088 @CanIgnoreReturnValue 089 @Override 090 public final V get(long timeout, TimeUnit unit) 091 throws InterruptedException, ExecutionException, TimeoutException { 092 return super.get(timeout, unit); 093 } 094 095 @Override 096 public final boolean isDone() { 097 return super.isDone(); 098 } 099 100 @Override 101 public final boolean isCancelled() { 102 return super.isCancelled(); 103 } 104 105 @Override 106 public final void addListener(Runnable listener, Executor executor) { 107 super.addListener(listener, executor); 108 } 109 110 @CanIgnoreReturnValue 111 @Override 112 public final boolean cancel(boolean mayInterruptIfRunning) { 113 return super.cancel(mayInterruptIfRunning); 114 } 115 } 116 117 // Logger to log exceptions caught when running listeners. 118 private static final Logger log = Logger.getLogger(AbstractFuture.class.getName()); 119 120 // A heuristic for timed gets. If the remaining timeout is less than this, spin instead of 121 // blocking. This value is what AbstractQueuedSynchronizer uses. 122 private static final long SPIN_THRESHOLD_NANOS = 1000L; 123 124 private static final AtomicHelper ATOMIC_HELPER; 125 126 static { 127 AtomicHelper helper; 128 Throwable thrownUnsafeFailure = null; 129 Throwable thrownAtomicReferenceFieldUpdaterFailure = null; 130 131 try { 132 helper = new UnsafeAtomicHelper(); 133 } catch (Throwable unsafeFailure) { 134 thrownUnsafeFailure = unsafeFailure; 135 // catch absolutely everything and fall through to our 'SafeAtomicHelper' 136 // The access control checks that ARFU does means the caller class has to be AbstractFuture 137 // instead of SafeAtomicHelper, so we annoyingly define these here 138 try { 139 helper = 140 new SafeAtomicHelper( 141 newUpdater(Waiter.class, Thread.class, "thread"), 142 newUpdater(Waiter.class, Waiter.class, "next"), 143 newUpdater(AbstractFuture.class, Waiter.class, "waiters"), 144 newUpdater(AbstractFuture.class, Listener.class, "listeners"), 145 newUpdater(AbstractFuture.class, Object.class, "value")); 146 } catch (Throwable atomicReferenceFieldUpdaterFailure) { 147 // Some Android 5.0.x Samsung devices have bugs in JDK reflection APIs that cause 148 // getDeclaredField to throw a NoSuchFieldException when the field is definitely there. 149 // For these users fallback to a suboptimal implementation, based on synchronized. This will 150 // be a definite performance hit to those users. 151 thrownAtomicReferenceFieldUpdaterFailure = atomicReferenceFieldUpdaterFailure; 152 helper = new SynchronizedHelper(); 153 } 154 } 155 ATOMIC_HELPER = helper; 156 157 // Prevent rare disastrous classloading in first call to LockSupport.park. 158 // See: https://bugs.openjdk.java.net/browse/JDK-8074773 159 @SuppressWarnings("unused") 160 Class<?> ensureLoaded = LockSupport.class; 161 162 // Log after all static init is finished; if an installed logger uses any Futures methods, it 163 // shouldn't break in cases where reflection is missing/broken. 164 if (thrownAtomicReferenceFieldUpdaterFailure != null) { 165 log.log(Level.SEVERE, "UnsafeAtomicHelper is broken!", thrownUnsafeFailure); 166 log.log( 167 Level.SEVERE, "SafeAtomicHelper is broken!", thrownAtomicReferenceFieldUpdaterFailure); 168 } 169 } 170 171 /** Waiter links form a Treiber stack, in the {@link #waiters} field. */ 172 private static final class Waiter { 173 static final Waiter TOMBSTONE = new Waiter(false /* ignored param */); 174 175 @NullableDecl volatile Thread thread; 176 @NullableDecl volatile Waiter next; 177 178 /** 179 * Constructor for the TOMBSTONE, avoids use of ATOMIC_HELPER in case this class is loaded 180 * before the ATOMIC_HELPER. Apparently this is possible on some android platforms. 181 */ 182 Waiter(boolean unused) {} 183 184 Waiter() { 185 // avoid volatile write, write is made visible by subsequent CAS on waiters field 186 ATOMIC_HELPER.putThread(this, Thread.currentThread()); 187 } 188 189 // non-volatile write to the next field. Should be made visible by subsequent CAS on waiters 190 // field. 191 void setNext(Waiter next) { 192 ATOMIC_HELPER.putNext(this, next); 193 } 194 195 void unpark() { 196 // This is racy with removeWaiter. The consequence of the race is that we may spuriously call 197 // unpark even though the thread has already removed itself from the list. But even if we did 198 // use a CAS, that race would still exist (it would just be ever so slightly smaller). 199 Thread w = thread; 200 if (w != null) { 201 thread = null; 202 LockSupport.unpark(w); 203 } 204 } 205 } 206 207 /** 208 * Marks the given node as 'deleted' (null waiter) and then scans the list to unlink all deleted 209 * nodes. This is an O(n) operation in the common case (and O(n^2) in the worst), but we are saved 210 * by two things. 211 * 212 * <ul> 213 * <li>This is only called when a waiting thread times out or is interrupted. Both of which 214 * should be rare. 215 * <li>The waiters list should be very short. 216 * </ul> 217 */ 218 private void removeWaiter(Waiter node) { 219 node.thread = null; // mark as 'deleted' 220 restart: 221 while (true) { 222 Waiter pred = null; 223 Waiter curr = waiters; 224 if (curr == Waiter.TOMBSTONE) { 225 return; // give up if someone is calling complete 226 } 227 Waiter succ; 228 while (curr != null) { 229 succ = curr.next; 230 if (curr.thread != null) { // we aren't unlinking this node, update pred. 231 pred = curr; 232 } else if (pred != null) { // We are unlinking this node and it has a predecessor. 233 pred.next = succ; 234 if (pred.thread == null) { // We raced with another node that unlinked pred. Restart. 235 continue restart; 236 } 237 } else if (!ATOMIC_HELPER.casWaiters(this, curr, succ)) { // We are unlinking head 238 continue restart; // We raced with an add or complete 239 } 240 curr = succ; 241 } 242 break; 243 } 244 } 245 246 /** Listeners also form a stack through the {@link #listeners} field. */ 247 private static final class Listener { 248 static final Listener TOMBSTONE = new Listener(null, null); 249 final Runnable task; 250 final Executor executor; 251 252 // writes to next are made visible by subsequent CAS's on the listeners field 253 @NullableDecl Listener next; 254 255 Listener(Runnable task, Executor executor) { 256 this.task = task; 257 this.executor = executor; 258 } 259 } 260 261 /** A special value to represent {@code null}. */ 262 private static final Object NULL = new Object(); 263 264 /** A special value to represent failure, when {@link #setException} is called successfully. */ 265 private static final class Failure { 266 static final Failure FALLBACK_INSTANCE = 267 new Failure( 268 new Throwable("Failure occurred while trying to finish a future.") { 269 @Override 270 public synchronized Throwable fillInStackTrace() { 271 return this; // no stack trace 272 } 273 }); 274 final Throwable exception; 275 276 Failure(Throwable exception) { 277 this.exception = checkNotNull(exception); 278 } 279 } 280 281 /** A special value to represent cancellation and the 'wasInterrupted' bit. */ 282 private static final class Cancellation { 283 // constants to use when GENERATE_CANCELLATION_CAUSES = false 284 static final Cancellation CAUSELESS_INTERRUPTED; 285 static final Cancellation CAUSELESS_CANCELLED; 286 287 static { 288 if (GENERATE_CANCELLATION_CAUSES) { 289 CAUSELESS_CANCELLED = null; 290 CAUSELESS_INTERRUPTED = null; 291 } else { 292 CAUSELESS_CANCELLED = new Cancellation(false, null); 293 CAUSELESS_INTERRUPTED = new Cancellation(true, null); 294 } 295 } 296 297 final boolean wasInterrupted; 298 @NullableDecl final Throwable cause; 299 300 Cancellation(boolean wasInterrupted, @NullableDecl Throwable cause) { 301 this.wasInterrupted = wasInterrupted; 302 this.cause = cause; 303 } 304 } 305 306 /** A special value that encodes the 'setFuture' state. */ 307 private static final class SetFuture<V> implements Runnable { 308 final AbstractFuture<V> owner; 309 final ListenableFuture<? extends V> future; 310 311 SetFuture(AbstractFuture<V> owner, ListenableFuture<? extends V> future) { 312 this.owner = owner; 313 this.future = future; 314 } 315 316 @Override 317 public void run() { 318 if (owner.value != this) { 319 // nothing to do, we must have been cancelled, don't bother inspecting the future. 320 return; 321 } 322 Object valueToSet = getFutureValue(future); 323 if (ATOMIC_HELPER.casValue(owner, this, valueToSet)) { 324 complete(owner); 325 } 326 } 327 } 328 329 // TODO(lukes): investigate using the @Contended annotation on these fields when jdk8 is 330 // available. 331 /** 332 * This field encodes the current state of the future. 333 * 334 * <p>The valid values are: 335 * 336 * <ul> 337 * <li>{@code null} initial state, nothing has happened. 338 * <li>{@link Cancellation} terminal state, {@code cancel} was called. 339 * <li>{@link Failure} terminal state, {@code setException} was called. 340 * <li>{@link SetFuture} intermediate state, {@code setFuture} was called. 341 * <li>{@link #NULL} terminal state, {@code set(null)} was called. 342 * <li>Any other non-null value, terminal state, {@code set} was called with a non-null 343 * argument. 344 * </ul> 345 */ 346 private volatile Object value; 347 348 /** All listeners. */ 349 private volatile Listener listeners; 350 351 /** All waiting threads. */ 352 private volatile Waiter waiters; 353 354 /** Constructor for use by subclasses. */ 355 protected AbstractFuture() {} 356 357 // Gets and Timed Gets 358 // 359 // * Be responsive to interruption 360 // * Don't create Waiter nodes if you aren't going to park, this helps reduce contention on the 361 // waiters field. 362 // * Future completion is defined by when #value becomes non-null/non SetFuture 363 // * Future completion can be observed if the waiters field contains a TOMBSTONE 364 365 // Timed Get 366 // There are a few design constraints to consider 367 // * We want to be responsive to small timeouts, unpark() has non trivial latency overheads (I 368 // have observed 12 micros on 64 bit linux systems to wake up a parked thread). So if the 369 // timeout is small we shouldn't park(). This needs to be traded off with the cpu overhead of 370 // spinning, so we use SPIN_THRESHOLD_NANOS which is what AbstractQueuedSynchronizer uses for 371 // similar purposes. 372 // * We want to behave reasonably for timeouts of 0 373 // * We are more responsive to completion than timeouts. This is because parkNanos depends on 374 // system scheduling and as such we could either miss our deadline, or unpark() could be delayed 375 // so that it looks like we timed out even though we didn't. For comparison FutureTask respects 376 // completion preferably and AQS is non-deterministic (depends on where in the queue the waiter 377 // is). If we wanted to be strict about it, we could store the unpark() time in the Waiter node 378 // and we could use that to make a decision about whether or not we timed out prior to being 379 // unparked. 380 381 /** 382 * {@inheritDoc} 383 * 384 * <p>The default {@link AbstractFuture} implementation throws {@code InterruptedException} if the 385 * current thread is interrupted during the call, even if the value is already available. 386 * 387 * @throws CancellationException {@inheritDoc} 388 */ 389 @CanIgnoreReturnValue 390 @Override 391 public V get(long timeout, TimeUnit unit) 392 throws InterruptedException, TimeoutException, ExecutionException { 393 // NOTE: if timeout < 0, remainingNanos will be < 0 and we will fall into the while(true) loop 394 // at the bottom and throw a timeoutexception. 395 long remainingNanos = unit.toNanos(timeout); // we rely on the implicit null check on unit. 396 if (Thread.interrupted()) { 397 throw new InterruptedException(); 398 } 399 Object localValue = value; 400 if (localValue != null & !(localValue instanceof SetFuture)) { 401 return getDoneValue(localValue); 402 } 403 // we delay calling nanoTime until we know we will need to either park or spin 404 final long endNanos = remainingNanos > 0 ? System.nanoTime() + remainingNanos : 0; 405 long_wait_loop: 406 if (remainingNanos >= SPIN_THRESHOLD_NANOS) { 407 Waiter oldHead = waiters; 408 if (oldHead != Waiter.TOMBSTONE) { 409 Waiter node = new Waiter(); 410 do { 411 node.setNext(oldHead); 412 if (ATOMIC_HELPER.casWaiters(this, oldHead, node)) { 413 while (true) { 414 LockSupport.parkNanos(this, remainingNanos); 415 // Check interruption first, if we woke up due to interruption we need to honor that. 416 if (Thread.interrupted()) { 417 removeWaiter(node); 418 throw new InterruptedException(); 419 } 420 421 // Otherwise re-read and check doneness. If we loop then it must have been a spurious 422 // wakeup 423 localValue = value; 424 if (localValue != null & !(localValue instanceof SetFuture)) { 425 return getDoneValue(localValue); 426 } 427 428 // timed out? 429 remainingNanos = endNanos - System.nanoTime(); 430 if (remainingNanos < SPIN_THRESHOLD_NANOS) { 431 // Remove the waiter, one way or another we are done parking this thread. 432 removeWaiter(node); 433 break long_wait_loop; // jump down to the busy wait loop 434 } 435 } 436 } 437 oldHead = waiters; // re-read and loop. 438 } while (oldHead != Waiter.TOMBSTONE); 439 } 440 // re-read value, if we get here then we must have observed a TOMBSTONE while trying to add a 441 // waiter. 442 return getDoneValue(value); 443 } 444 // If we get here then we have remainingNanos < SPIN_THRESHOLD_NANOS and there is no node on the 445 // waiters list 446 while (remainingNanos > 0) { 447 localValue = value; 448 if (localValue != null & !(localValue instanceof SetFuture)) { 449 return getDoneValue(localValue); 450 } 451 if (Thread.interrupted()) { 452 throw new InterruptedException(); 453 } 454 remainingNanos = endNanos - System.nanoTime(); 455 } 456 457 String futureToString = toString(); 458 // It's confusing to see a completed future in a timeout message; if isDone() returns false, 459 // then we know it must have given a pending toString value earlier. If not, then the future 460 // completed after the timeout expired, and the message might be success. 461 if (isDone()) { 462 throw new TimeoutException( 463 "Waited " 464 + timeout 465 + " " 466 + Ascii.toLowerCase(unit.toString()) 467 + " but future completed as timeout expired"); 468 } 469 throw new TimeoutException( 470 "Waited " + timeout + " " + Ascii.toLowerCase(unit.toString()) + " for " + futureToString); 471 } 472 473 /** 474 * {@inheritDoc} 475 * 476 * <p>The default {@link AbstractFuture} implementation throws {@code InterruptedException} if the 477 * current thread is interrupted during the call, even if the value is already available. 478 * 479 * @throws CancellationException {@inheritDoc} 480 */ 481 @CanIgnoreReturnValue 482 @Override 483 public V get() throws InterruptedException, ExecutionException { 484 if (Thread.interrupted()) { 485 throw new InterruptedException(); 486 } 487 Object localValue = value; 488 if (localValue != null & !(localValue instanceof SetFuture)) { 489 return getDoneValue(localValue); 490 } 491 Waiter oldHead = waiters; 492 if (oldHead != Waiter.TOMBSTONE) { 493 Waiter node = new Waiter(); 494 do { 495 node.setNext(oldHead); 496 if (ATOMIC_HELPER.casWaiters(this, oldHead, node)) { 497 // we are on the stack, now wait for completion. 498 while (true) { 499 LockSupport.park(this); 500 // Check interruption first, if we woke up due to interruption we need to honor that. 501 if (Thread.interrupted()) { 502 removeWaiter(node); 503 throw new InterruptedException(); 504 } 505 // Otherwise re-read and check doneness. If we loop then it must have been a spurious 506 // wakeup 507 localValue = value; 508 if (localValue != null & !(localValue instanceof SetFuture)) { 509 return getDoneValue(localValue); 510 } 511 } 512 } 513 oldHead = waiters; // re-read and loop. 514 } while (oldHead != Waiter.TOMBSTONE); 515 } 516 // re-read value, if we get here then we must have observed a TOMBSTONE while trying to add a 517 // waiter. 518 return getDoneValue(value); 519 } 520 521 /** Unboxes {@code obj}. Assumes that obj is not {@code null} or a {@link SetFuture}. */ 522 private V getDoneValue(Object obj) throws ExecutionException { 523 // While this seems like it might be too branch-y, simple benchmarking proves it to be 524 // unmeasurable (comparing done AbstractFutures with immediateFuture) 525 if (obj instanceof Cancellation) { 526 throw cancellationExceptionWithCause("Task was cancelled.", ((Cancellation) obj).cause); 527 } else if (obj instanceof Failure) { 528 throw new ExecutionException(((Failure) obj).exception); 529 } else if (obj == NULL) { 530 return null; 531 } else { 532 @SuppressWarnings("unchecked") // this is the only other option 533 V asV = (V) obj; 534 return asV; 535 } 536 } 537 538 @Override 539 public boolean isDone() { 540 final Object localValue = value; 541 return localValue != null & !(localValue instanceof SetFuture); 542 } 543 544 @Override 545 public boolean isCancelled() { 546 final Object localValue = value; 547 return localValue instanceof Cancellation; 548 } 549 550 /** 551 * {@inheritDoc} 552 * 553 * <p>If a cancellation attempt succeeds on a {@code Future} that had previously been {@linkplain 554 * #setFuture set asynchronously}, then the cancellation will also be propagated to the delegate 555 * {@code Future} that was supplied in the {@code setFuture} call. 556 * 557 * <p>Rather than override this method to perform additional cancellation work or cleanup, 558 * subclasses should override {@link #afterDone}, consulting {@link #isCancelled} and {@link 559 * #wasInterrupted} as necessary. This ensures that the work is done even if the future is 560 * cancelled without a call to {@code cancel}, such as by calling {@code 561 * setFuture(cancelledFuture)}. 562 */ 563 @CanIgnoreReturnValue 564 @Override 565 public boolean cancel(boolean mayInterruptIfRunning) { 566 Object localValue = value; 567 boolean rValue = false; 568 if (localValue == null | localValue instanceof SetFuture) { 569 // Try to delay allocating the exception. At this point we may still lose the CAS, but it is 570 // certainly less likely. 571 Object valueToSet = 572 GENERATE_CANCELLATION_CAUSES 573 ? new Cancellation( 574 mayInterruptIfRunning, new CancellationException("Future.cancel() was called.")) 575 : (mayInterruptIfRunning 576 ? Cancellation.CAUSELESS_INTERRUPTED 577 : Cancellation.CAUSELESS_CANCELLED); 578 AbstractFuture<?> abstractFuture = this; 579 while (true) { 580 if (ATOMIC_HELPER.casValue(abstractFuture, localValue, valueToSet)) { 581 rValue = true; 582 // We call interuptTask before calling complete(), which is consistent with 583 // FutureTask 584 if (mayInterruptIfRunning) { 585 abstractFuture.interruptTask(); 586 } 587 complete(abstractFuture); 588 if (localValue instanceof SetFuture) { 589 // propagate cancellation to the future set in setfuture, this is racy, and we don't 590 // care if we are successful or not. 591 ListenableFuture<?> futureToPropagateTo = ((SetFuture) localValue).future; 592 if (futureToPropagateTo instanceof TrustedFuture) { 593 // If the future is a TrustedFuture then we specifically avoid calling cancel() 594 // this has 2 benefits 595 // 1. for long chains of futures strung together with setFuture we consume less stack 596 // 2. we avoid allocating Cancellation objects at every level of the cancellation 597 // chain 598 // We can only do this for TrustedFuture, because TrustedFuture.cancel is final and 599 // does nothing but delegate to this method. 600 AbstractFuture<?> trusted = (AbstractFuture<?>) futureToPropagateTo; 601 localValue = trusted.value; 602 if (localValue == null | localValue instanceof SetFuture) { 603 abstractFuture = trusted; 604 continue; // loop back up and try to complete the new future 605 } 606 } else { 607 // not a TrustedFuture, call cancel directly. 608 futureToPropagateTo.cancel(mayInterruptIfRunning); 609 } 610 } 611 break; 612 } 613 // obj changed, reread 614 localValue = abstractFuture.value; 615 if (!(localValue instanceof SetFuture)) { 616 // obj cannot be null at this point, because value can only change from null to non-null. 617 // So if value changed (and it did since we lost the CAS), then it cannot be null and 618 // since it isn't a SetFuture, then the future must be done and we should exit the loop 619 break; 620 } 621 } 622 } 623 return rValue; 624 } 625 626 /** 627 * Subclasses can override this method to implement interruption of the future's computation. The 628 * method is invoked automatically by a successful call to {@link #cancel(boolean) cancel(true)}. 629 * 630 * <p>The default implementation does nothing. 631 * 632 * <p>This method is likely to be deprecated. Prefer to override {@link #afterDone}, checking 633 * {@link #wasInterrupted} to decide whether to interrupt your task. 634 * 635 * @since 10.0 636 */ 637 protected void interruptTask() {} 638 639 /** 640 * Returns true if this future was cancelled with {@code mayInterruptIfRunning} set to {@code 641 * true}. 642 * 643 * @since 14.0 644 */ 645 protected final boolean wasInterrupted() { 646 final Object localValue = value; 647 return (localValue instanceof Cancellation) && ((Cancellation) localValue).wasInterrupted; 648 } 649 650 /** 651 * {@inheritDoc} 652 * 653 * @since 10.0 654 */ 655 @Override 656 public void addListener(Runnable listener, Executor executor) { 657 checkNotNull(listener, "Runnable was null."); 658 checkNotNull(executor, "Executor was null."); 659 Listener oldHead = listeners; 660 if (oldHead != Listener.TOMBSTONE) { 661 Listener newNode = new Listener(listener, executor); 662 do { 663 newNode.next = oldHead; 664 if (ATOMIC_HELPER.casListeners(this, oldHead, newNode)) { 665 return; 666 } 667 oldHead = listeners; // re-read 668 } while (oldHead != Listener.TOMBSTONE); 669 } 670 // If we get here then the Listener TOMBSTONE was set, which means the future is done, call 671 // the listener. 672 executeListener(listener, executor); 673 } 674 675 /** 676 * Sets the result of this {@code Future} unless this {@code Future} has already been cancelled or 677 * set (including {@linkplain #setFuture set asynchronously}). When a call to this method returns, 678 * the {@code Future} is guaranteed to be {@linkplain #isDone done} <b>only if</b> the call was 679 * accepted (in which case it returns {@code true}). If it returns {@code false}, the {@code 680 * Future} may have previously been set asynchronously, in which case its result may not be known 681 * yet. That result, though not yet known, cannot be overridden by a call to a {@code set*} 682 * method, only by a call to {@link #cancel}. 683 * 684 * @param value the value to be used as the result 685 * @return true if the attempt was accepted, completing the {@code Future} 686 */ 687 @CanIgnoreReturnValue 688 protected boolean set(@NullableDecl V value) { 689 Object valueToSet = value == null ? NULL : value; 690 if (ATOMIC_HELPER.casValue(this, null, valueToSet)) { 691 complete(this); 692 return true; 693 } 694 return false; 695 } 696 697 /** 698 * Sets the failed result of this {@code Future} unless this {@code Future} has already been 699 * cancelled or set (including {@linkplain #setFuture set asynchronously}). When a call to this 700 * method returns, the {@code Future} is guaranteed to be {@linkplain #isDone done} <b>only if</b> 701 * the call was accepted (in which case it returns {@code true}). If it returns {@code false}, the 702 * {@code Future} may have previously been set asynchronously, in which case its result may not be 703 * known yet. That result, though not yet known, cannot be overridden by a call to a {@code set*} 704 * method, only by a call to {@link #cancel}. 705 * 706 * @param throwable the exception to be used as the failed result 707 * @return true if the attempt was accepted, completing the {@code Future} 708 */ 709 @CanIgnoreReturnValue 710 protected boolean setException(Throwable throwable) { 711 Object valueToSet = new Failure(checkNotNull(throwable)); 712 if (ATOMIC_HELPER.casValue(this, null, valueToSet)) { 713 complete(this); 714 return true; 715 } 716 return false; 717 } 718 719 /** 720 * Sets the result of this {@code Future} to match the supplied input {@code Future} once the 721 * supplied {@code Future} is done, unless this {@code Future} has already been cancelled or set 722 * (including "set asynchronously," defined below). 723 * 724 * <p>If the supplied future is {@linkplain #isDone done} when this method is called and the call 725 * is accepted, then this future is guaranteed to have been completed with the supplied future by 726 * the time this method returns. If the supplied future is not done and the call is accepted, then 727 * the future will be <i>set asynchronously</i>. Note that such a result, though not yet known, 728 * cannot be overridden by a call to a {@code set*} method, only by a call to {@link #cancel}. 729 * 730 * <p>If the call {@code setFuture(delegate)} is accepted and this {@code Future} is later 731 * cancelled, cancellation will be propagated to {@code delegate}. Additionally, any call to 732 * {@code setFuture} after any cancellation will propagate cancellation to the supplied {@code 733 * Future}. 734 * 735 * <p>Note that, even if the supplied future is cancelled and it causes this future to complete, 736 * it will never trigger interruption behavior. In particular, it will not cause this future to 737 * invoke the {@link #interruptTask} method, and the {@link #wasInterrupted} method will not 738 * return {@code true}. 739 * 740 * @param future the future to delegate to 741 * @return true if the attempt was accepted, indicating that the {@code Future} was not previously 742 * cancelled or set. 743 * @since 19.0 744 */ 745 @Beta 746 @CanIgnoreReturnValue 747 protected boolean setFuture(ListenableFuture<? extends V> future) { 748 checkNotNull(future); 749 Object localValue = value; 750 if (localValue == null) { 751 if (future.isDone()) { 752 Object value = getFutureValue(future); 753 if (ATOMIC_HELPER.casValue(this, null, value)) { 754 complete(this); 755 return true; 756 } 757 return false; 758 } 759 SetFuture valueToSet = new SetFuture<V>(this, future); 760 if (ATOMIC_HELPER.casValue(this, null, valueToSet)) { 761 // the listener is responsible for calling completeWithFuture, directExecutor is appropriate 762 // since all we are doing is unpacking a completed future which should be fast. 763 try { 764 future.addListener(valueToSet, directExecutor()); 765 } catch (Throwable t) { 766 // addListener has thrown an exception! SetFuture.run can't throw any exceptions so this 767 // must have been caused by addListener itself. The most likely explanation is a 768 // misconfigured mock. Try to switch to Failure. 769 Failure failure; 770 try { 771 failure = new Failure(t); 772 } catch (Throwable oomMostLikely) { 773 failure = Failure.FALLBACK_INSTANCE; 774 } 775 // Note: The only way this CAS could fail is if cancel() has raced with us. That is ok. 776 boolean unused = ATOMIC_HELPER.casValue(this, valueToSet, failure); 777 } 778 return true; 779 } 780 localValue = value; // we lost the cas, fall through and maybe cancel 781 } 782 // The future has already been set to something. If it is cancellation we should cancel the 783 // incoming future. 784 if (localValue instanceof Cancellation) { 785 // we don't care if it fails, this is best-effort. 786 future.cancel(((Cancellation) localValue).wasInterrupted); 787 } 788 return false; 789 } 790 791 /** 792 * Returns a value that satisfies the contract of the {@link #value} field based on the state of 793 * given future. 794 * 795 * <p>This is approximately the inverse of {@link #getDoneValue(Object)} 796 */ 797 private static Object getFutureValue(ListenableFuture<?> future) { 798 Object valueToSet; 799 if (future instanceof TrustedFuture) { 800 // Break encapsulation for TrustedFuture instances since we know that subclasses cannot 801 // override .get() (since it is final) and therefore this is equivalent to calling .get() 802 // and unpacking the exceptions like we do below (just much faster because it is a single 803 // field read instead of a read, several branches and possibly creating exceptions). 804 Object v = ((AbstractFuture<?>) future).value; 805 if (v instanceof Cancellation) { 806 // If the other future was interrupted, clear the interrupted bit while preserving the cause 807 // this will make it consistent with how non-trustedfutures work which cannot propagate the 808 // wasInterrupted bit 809 Cancellation c = (Cancellation) v; 810 if (c.wasInterrupted) { 811 v = 812 c.cause != null 813 ? new Cancellation(/* wasInterrupted= */ false, c.cause) 814 : Cancellation.CAUSELESS_CANCELLED; 815 } 816 } 817 return v; 818 } else { 819 // Otherwise calculate valueToSet by calling .get() 820 try { 821 Object v = getDone(future); 822 valueToSet = v == null ? NULL : v; 823 } catch (ExecutionException exception) { 824 valueToSet = new Failure(exception.getCause()); 825 } catch (CancellationException cancellation) { 826 valueToSet = new Cancellation(false, cancellation); 827 } catch (Throwable t) { 828 valueToSet = new Failure(t); 829 } 830 } 831 return valueToSet; 832 } 833 834 /** Unblocks all threads and runs all listeners. */ 835 private static void complete(AbstractFuture<?> future) { 836 Listener next = null; 837 outer: 838 while (true) { 839 future.releaseWaiters(); 840 // We call this before the listeners in order to avoid needing to manage a separate stack data 841 // structure for them. 842 // afterDone() should be generally fast and only used for cleanup work... but in theory can 843 // also be recursive and create StackOverflowErrors 844 future.afterDone(); 845 // push the current set of listeners onto next 846 next = future.clearListeners(next); 847 future = null; 848 while (next != null) { 849 Listener curr = next; 850 next = next.next; 851 Runnable task = curr.task; 852 if (task instanceof SetFuture) { 853 SetFuture<?> setFuture = (SetFuture<?>) task; 854 // We unwind setFuture specifically to avoid StackOverflowErrors in the case of long 855 // chains of SetFutures 856 // Handling this special case is important because there is no way to pass an executor to 857 // setFuture, so a user couldn't break the chain by doing this themselves. It is also 858 // potentially common if someone writes a recursive Futures.transformAsync transformer. 859 future = setFuture.owner; 860 if (future.value == setFuture) { 861 Object valueToSet = getFutureValue(setFuture.future); 862 if (ATOMIC_HELPER.casValue(future, setFuture, valueToSet)) { 863 continue outer; 864 } 865 } 866 // other wise the future we were trying to set is already done. 867 } else { 868 executeListener(task, curr.executor); 869 } 870 } 871 break; 872 } 873 } 874 875 /** 876 * Callback method that is called exactly once after the future is completed. 877 * 878 * <p>If {@link #interruptTask} is also run during completion, {@link #afterDone} runs after it. 879 * 880 * <p>The default implementation of this method in {@code AbstractFuture} does nothing. This is 881 * intended for very lightweight cleanup work, for example, timing statistics or clearing fields. 882 * If your task does anything heavier consider, just using a listener with an executor. 883 * 884 * @since 20.0 885 */ 886 @Beta 887 @ForOverride 888 protected void afterDone() {} 889 890 /** 891 * Returns the exception that this {@code Future} completed with. This includes completion through 892 * a call to {@link #setException} or {@link #setFuture setFuture}{@code (failedFuture)} but not 893 * cancellation. 894 * 895 * @throws RuntimeException if the {@code Future} has not failed 896 */ 897 final Throwable trustedGetException() { 898 return ((Failure) value).exception; 899 } 900 901 /** 902 * If this future has been cancelled (and possibly interrupted), cancels (and possibly interrupts) 903 * the given future (if available). 904 */ 905 final void maybePropagateCancellationTo(@NullableDecl Future<?> related) { 906 if (related != null & isCancelled()) { 907 related.cancel(wasInterrupted()); 908 } 909 } 910 911 /** Releases all threads in the {@link #waiters} list, and clears the list. */ 912 private void releaseWaiters() { 913 Waiter head; 914 do { 915 head = waiters; 916 } while (!ATOMIC_HELPER.casWaiters(this, head, Waiter.TOMBSTONE)); 917 for (Waiter currentWaiter = head; currentWaiter != null; currentWaiter = currentWaiter.next) { 918 currentWaiter.unpark(); 919 } 920 } 921 922 /** 923 * Clears the {@link #listeners} list and prepends its contents to {@code onto}, least recently 924 * added first. 925 */ 926 private Listener clearListeners(Listener onto) { 927 // We need to 928 // 1. atomically swap the listeners with TOMBSTONE, this is because addListener uses that to 929 // to synchronize with us 930 // 2. reverse the linked list, because despite our rather clear contract, people depend on us 931 // executing listeners in the order they were added 932 // 3. push all the items onto 'onto' and return the new head of the stack 933 Listener head; 934 do { 935 head = listeners; 936 } while (!ATOMIC_HELPER.casListeners(this, head, Listener.TOMBSTONE)); 937 Listener reversedList = onto; 938 while (head != null) { 939 Listener tmp = head; 940 head = head.next; 941 tmp.next = reversedList; 942 reversedList = tmp; 943 } 944 return reversedList; 945 } 946 947 // TODO(user) move this up into FluentFuture, or parts as a default method on ListenableFuture? 948 @Override 949 public String toString() { 950 StringBuilder builder = new StringBuilder().append(super.toString()).append("[status="); 951 if (isCancelled()) { 952 builder.append("CANCELLED"); 953 } else if (isDone()) { 954 addDoneString(builder); 955 } else { 956 String pendingDescription; 957 try { 958 pendingDescription = pendingToString(); 959 } catch (RuntimeException e) { 960 // Don't call getMessage or toString() on the exception, in case the exception thrown by the 961 // subclass is implemented with bugs similar to the subclass. 962 pendingDescription = "Exception thrown from implementation: " + e.getClass(); 963 } 964 // The future may complete during or before the call to getPendingToString, so we use null 965 // as a signal that we should try checking if the future is done again. 966 if (!isNullOrEmpty(pendingDescription)) { 967 builder.append("PENDING, info=[").append(pendingDescription).append("]"); 968 } else if (isDone()) { 969 addDoneString(builder); 970 } else { 971 builder.append("PENDING"); 972 } 973 } 974 return builder.append("]").toString(); 975 } 976 977 /** 978 * Provide a human-readable explanation of why this future has not yet completed. 979 * 980 * @return null if an explanation cannot be provided because the future is done. 981 * @since 23.0 982 */ 983 @NullableDecl 984 protected String pendingToString() { 985 Object localValue = value; 986 if (localValue instanceof SetFuture) { 987 return "setFuture=[" + ((SetFuture) localValue).future + "]"; 988 } else if (this instanceof ScheduledFuture) { 989 return "remaining delay=[" 990 + ((ScheduledFuture) this).getDelay(TimeUnit.MILLISECONDS) 991 + " ms]"; 992 } 993 return null; 994 } 995 996 private void addDoneString(StringBuilder builder) { 997 try { 998 V value = getDone(this); 999 builder.append("SUCCESS, result=[").append(value).append("]"); 1000 } catch (ExecutionException e) { 1001 builder.append("FAILURE, cause=[").append(e.getCause()).append("]"); 1002 } catch (CancellationException e) { 1003 builder.append("CANCELLED"); // shouldn't be reachable 1004 } catch (RuntimeException e) { 1005 builder.append("UNKNOWN, cause=[").append(e.getClass()).append(" thrown from get()]"); 1006 } 1007 } 1008 1009 /** 1010 * Submits the given runnable to the given {@link Executor} catching and logging all {@linkplain 1011 * RuntimeException runtime exceptions} thrown by the executor. 1012 */ 1013 private static void executeListener(Runnable runnable, Executor executor) { 1014 try { 1015 executor.execute(runnable); 1016 } catch (RuntimeException e) { 1017 // Log it and keep going -- bad runnable and/or executor. Don't punish the other runnables if 1018 // we're given a bad one. We only catch RuntimeException because we want Errors to propagate 1019 // up. 1020 log.log( 1021 Level.SEVERE, 1022 "RuntimeException while executing runnable " + runnable + " with executor " + executor, 1023 e); 1024 } 1025 } 1026 1027 private abstract static class AtomicHelper { 1028 /** Non volatile write of the thread to the {@link Waiter#thread} field. */ 1029 abstract void putThread(Waiter waiter, Thread newValue); 1030 1031 /** Non volatile write of the waiter to the {@link Waiter#next} field. */ 1032 abstract void putNext(Waiter waiter, Waiter newValue); 1033 1034 /** Performs a CAS operation on the {@link #waiters} field. */ 1035 abstract boolean casWaiters(AbstractFuture<?> future, Waiter expect, Waiter update); 1036 1037 /** Performs a CAS operation on the {@link #listeners} field. */ 1038 abstract boolean casListeners(AbstractFuture<?> future, Listener expect, Listener update); 1039 1040 /** Performs a CAS operation on the {@link #value} field. */ 1041 abstract boolean casValue(AbstractFuture<?> future, Object expect, Object update); 1042 } 1043 1044 /** 1045 * {@link AtomicHelper} based on {@link sun.misc.Unsafe}. 1046 * 1047 * <p>Static initialization of this class will fail if the {@link sun.misc.Unsafe} object cannot 1048 * be accessed. 1049 */ 1050 private static final class UnsafeAtomicHelper extends AtomicHelper { 1051 static final sun.misc.Unsafe UNSAFE; 1052 static final long LISTENERS_OFFSET; 1053 static final long WAITERS_OFFSET; 1054 static final long VALUE_OFFSET; 1055 static final long WAITER_THREAD_OFFSET; 1056 static final long WAITER_NEXT_OFFSET; 1057 1058 static { 1059 sun.misc.Unsafe unsafe = null; 1060 try { 1061 unsafe = sun.misc.Unsafe.getUnsafe(); 1062 } catch (SecurityException tryReflectionInstead) { 1063 try { 1064 unsafe = 1065 AccessController.doPrivileged( 1066 new PrivilegedExceptionAction<sun.misc.Unsafe>() { 1067 @Override 1068 public sun.misc.Unsafe run() throws Exception { 1069 Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class; 1070 for (java.lang.reflect.Field f : k.getDeclaredFields()) { 1071 f.setAccessible(true); 1072 Object x = f.get(null); 1073 if (k.isInstance(x)) { 1074 return k.cast(x); 1075 } 1076 } 1077 throw new NoSuchFieldError("the Unsafe"); 1078 } 1079 }); 1080 } catch (PrivilegedActionException e) { 1081 throw new RuntimeException("Could not initialize intrinsics", e.getCause()); 1082 } 1083 } 1084 try { 1085 Class<?> abstractFuture = AbstractFuture.class; 1086 WAITERS_OFFSET = unsafe.objectFieldOffset(abstractFuture.getDeclaredField("waiters")); 1087 LISTENERS_OFFSET = unsafe.objectFieldOffset(abstractFuture.getDeclaredField("listeners")); 1088 VALUE_OFFSET = unsafe.objectFieldOffset(abstractFuture.getDeclaredField("value")); 1089 WAITER_THREAD_OFFSET = unsafe.objectFieldOffset(Waiter.class.getDeclaredField("thread")); 1090 WAITER_NEXT_OFFSET = unsafe.objectFieldOffset(Waiter.class.getDeclaredField("next")); 1091 UNSAFE = unsafe; 1092 } catch (Exception e) { 1093 throwIfUnchecked(e); 1094 throw new RuntimeException(e); 1095 } 1096 } 1097 1098 @Override 1099 void putThread(Waiter waiter, Thread newValue) { 1100 UNSAFE.putObject(waiter, WAITER_THREAD_OFFSET, newValue); 1101 } 1102 1103 @Override 1104 void putNext(Waiter waiter, Waiter newValue) { 1105 UNSAFE.putObject(waiter, WAITER_NEXT_OFFSET, newValue); 1106 } 1107 1108 /** Performs a CAS operation on the {@link #waiters} field. */ 1109 @Override 1110 boolean casWaiters(AbstractFuture<?> future, Waiter expect, Waiter update) { 1111 return UNSAFE.compareAndSwapObject(future, WAITERS_OFFSET, expect, update); 1112 } 1113 1114 /** Performs a CAS operation on the {@link #listeners} field. */ 1115 @Override 1116 boolean casListeners(AbstractFuture<?> future, Listener expect, Listener update) { 1117 return UNSAFE.compareAndSwapObject(future, LISTENERS_OFFSET, expect, update); 1118 } 1119 1120 /** Performs a CAS operation on the {@link #value} field. */ 1121 @Override 1122 boolean casValue(AbstractFuture<?> future, Object expect, Object update) { 1123 return UNSAFE.compareAndSwapObject(future, VALUE_OFFSET, expect, update); 1124 } 1125 } 1126 1127 /** {@link AtomicHelper} based on {@link AtomicReferenceFieldUpdater}. */ 1128 private static final class SafeAtomicHelper extends AtomicHelper { 1129 final AtomicReferenceFieldUpdater<Waiter, Thread> waiterThreadUpdater; 1130 final AtomicReferenceFieldUpdater<Waiter, Waiter> waiterNextUpdater; 1131 final AtomicReferenceFieldUpdater<AbstractFuture, Waiter> waitersUpdater; 1132 final AtomicReferenceFieldUpdater<AbstractFuture, Listener> listenersUpdater; 1133 final AtomicReferenceFieldUpdater<AbstractFuture, Object> valueUpdater; 1134 1135 SafeAtomicHelper( 1136 AtomicReferenceFieldUpdater<Waiter, Thread> waiterThreadUpdater, 1137 AtomicReferenceFieldUpdater<Waiter, Waiter> waiterNextUpdater, 1138 AtomicReferenceFieldUpdater<AbstractFuture, Waiter> waitersUpdater, 1139 AtomicReferenceFieldUpdater<AbstractFuture, Listener> listenersUpdater, 1140 AtomicReferenceFieldUpdater<AbstractFuture, Object> valueUpdater) { 1141 this.waiterThreadUpdater = waiterThreadUpdater; 1142 this.waiterNextUpdater = waiterNextUpdater; 1143 this.waitersUpdater = waitersUpdater; 1144 this.listenersUpdater = listenersUpdater; 1145 this.valueUpdater = valueUpdater; 1146 } 1147 1148 @Override 1149 void putThread(Waiter waiter, Thread newValue) { 1150 waiterThreadUpdater.lazySet(waiter, newValue); 1151 } 1152 1153 @Override 1154 void putNext(Waiter waiter, Waiter newValue) { 1155 waiterNextUpdater.lazySet(waiter, newValue); 1156 } 1157 1158 @Override 1159 boolean casWaiters(AbstractFuture<?> future, Waiter expect, Waiter update) { 1160 return waitersUpdater.compareAndSet(future, expect, update); 1161 } 1162 1163 @Override 1164 boolean casListeners(AbstractFuture<?> future, Listener expect, Listener update) { 1165 return listenersUpdater.compareAndSet(future, expect, update); 1166 } 1167 1168 @Override 1169 boolean casValue(AbstractFuture<?> future, Object expect, Object update) { 1170 return valueUpdater.compareAndSet(future, expect, update); 1171 } 1172 } 1173 1174 /** 1175 * {@link AtomicHelper} based on {@code synchronized} and volatile writes. 1176 * 1177 * <p>This is an implementation of last resort for when certain basic VM features are broken (like 1178 * AtomicReferenceFieldUpdater). 1179 */ 1180 private static final class SynchronizedHelper extends AtomicHelper { 1181 @Override 1182 void putThread(Waiter waiter, Thread newValue) { 1183 waiter.thread = newValue; 1184 } 1185 1186 @Override 1187 void putNext(Waiter waiter, Waiter newValue) { 1188 waiter.next = newValue; 1189 } 1190 1191 @Override 1192 boolean casWaiters(AbstractFuture<?> future, Waiter expect, Waiter update) { 1193 synchronized (future) { 1194 if (future.waiters == expect) { 1195 future.waiters = update; 1196 return true; 1197 } 1198 return false; 1199 } 1200 } 1201 1202 @Override 1203 boolean casListeners(AbstractFuture<?> future, Listener expect, Listener update) { 1204 synchronized (future) { 1205 if (future.listeners == expect) { 1206 future.listeners = update; 1207 return true; 1208 } 1209 return false; 1210 } 1211 } 1212 1213 @Override 1214 boolean casValue(AbstractFuture<?> future, Object expect, Object update) { 1215 synchronized (future) { 1216 if (future.value == expect) { 1217 future.value = update; 1218 return true; 1219 } 1220 return false; 1221 } 1222 } 1223 } 1224 1225 private static CancellationException cancellationExceptionWithCause( 1226 @NullableDecl String message, @NullableDecl Throwable cause) { 1227 CancellationException exception = new CancellationException(message); 1228 exception.initCause(cause); 1229 return exception; 1230 } 1231}