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 javax.annotation.Nullable; 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 /** 172 * Waiter links form a Treiber stack, in the {@link #waiters} field. 173 */ 174 private static final class Waiter { 175 static final Waiter TOMBSTONE = new Waiter(false /* ignored param */); 176 177 @Nullable volatile Thread thread; 178 @Nullable volatile Waiter next; 179 180 /** 181 * Constructor for the TOMBSTONE, avoids use of ATOMIC_HELPER in case this class is loaded 182 * before the ATOMIC_HELPER. Apparently this is possible on some android platforms. 183 */ 184 Waiter(boolean unused) {} 185 186 Waiter() { 187 // avoid volatile write, write is made visible by subsequent CAS on waiters field 188 ATOMIC_HELPER.putThread(this, Thread.currentThread()); 189 } 190 191 // non-volatile write to the next field. Should be made visible by subsequent CAS on waiters 192 // field. 193 void setNext(Waiter next) { 194 ATOMIC_HELPER.putNext(this, next); 195 } 196 197 void unpark() { 198 // This is racy with removeWaiter. The consequence of the race is that we may spuriously call 199 // unpark even though the thread has already removed itself from the list. But even if we did 200 // use a CAS, that race would still exist (it would just be ever so slightly smaller). 201 Thread w = thread; 202 if (w != null) { 203 thread = null; 204 LockSupport.unpark(w); 205 } 206 } 207 } 208 209 /** 210 * Marks the given node as 'deleted' (null waiter) and then scans the list to unlink all deleted 211 * nodes. This is an O(n) operation in the common case (and O(n^2) in the worst), but we are saved 212 * by two things. 213 * <ul> 214 * <li>This is only called when a waiting thread times out or is interrupted. Both of which should 215 * be rare. 216 * <li>The waiters list should be very short. 217 * </ul> 218 */ 219 private void removeWaiter(Waiter node) { 220 node.thread = null; // mark as 'deleted' 221 restart: 222 while (true) { 223 Waiter pred = null; 224 Waiter curr = waiters; 225 if (curr == Waiter.TOMBSTONE) { 226 return; // give up if someone is calling complete 227 } 228 Waiter succ; 229 while (curr != null) { 230 succ = curr.next; 231 if (curr.thread != null) { // we aren't unlinking this node, update pred. 232 pred = curr; 233 } else if (pred != null) { // We are unlinking this node and it has a predecessor. 234 pred.next = succ; 235 if (pred.thread == null) { // We raced with another node that unlinked pred. Restart. 236 continue restart; 237 } 238 } else if (!ATOMIC_HELPER.casWaiters(this, curr, succ)) { // We are unlinking head 239 continue restart; // We raced with an add or complete 240 } 241 curr = succ; 242 } 243 break; 244 } 245 } 246 247 /** Listeners also form a stack through the {@link #listeners} field. */ 248 private static final class Listener { 249 static final Listener TOMBSTONE = new Listener(null, null); 250 final Runnable task; 251 final Executor executor; 252 253 // writes to next are made visible by subsequent CAS's on the listeners field 254 @Nullable Listener next; 255 256 Listener(Runnable task, Executor executor) { 257 this.task = task; 258 this.executor = executor; 259 } 260 } 261 262 /** A special value to represent {@code null}. */ 263 private static final Object NULL = new Object(); 264 265 /** A special value to represent failure, when {@link #setException} is called successfully. */ 266 private static final class Failure { 267 static final Failure FALLBACK_INSTANCE = 268 new Failure( 269 new Throwable("Failure occurred while trying to finish a future.") { 270 @Override 271 public synchronized Throwable fillInStackTrace() { 272 return this; // no stack trace 273 } 274 }); 275 final Throwable exception; 276 277 Failure(Throwable exception) { 278 this.exception = checkNotNull(exception); 279 } 280 } 281 282 /** A special value to represent cancellation and the 'wasInterrupted' bit. */ 283 private static final class Cancellation { 284 // constants to use when GENERATE_CANCELLATION_CAUSES = false 285 static final Cancellation CAUSELESS_INTERRUPTED; 286 static final Cancellation CAUSELESS_CANCELLED; 287 288 static { 289 if (GENERATE_CANCELLATION_CAUSES) { 290 CAUSELESS_CANCELLED = null; 291 CAUSELESS_INTERRUPTED = null; 292 } else { 293 CAUSELESS_CANCELLED = new Cancellation(false, null); 294 CAUSELESS_INTERRUPTED = new Cancellation(true, null); 295 } 296 } 297 298 final boolean wasInterrupted; 299 @Nullable final Throwable cause; 300 301 Cancellation(boolean wasInterrupted, @Nullable Throwable cause) { 302 this.wasInterrupted = wasInterrupted; 303 this.cause = cause; 304 } 305 } 306 307 /** A special value that encodes the 'setFuture' state. */ 308 private static final class SetFuture<V> implements Runnable { 309 final AbstractFuture<V> owner; 310 final ListenableFuture<? extends V> future; 311 312 SetFuture(AbstractFuture<V> owner, ListenableFuture<? extends V> future) { 313 this.owner = owner; 314 this.future = future; 315 } 316 317 @Override 318 public void run() { 319 if (owner.value != this) { 320 // nothing to do, we must have been cancelled, don't bother inspecting the future. 321 return; 322 } 323 Object valueToSet = getFutureValue(future); 324 if (ATOMIC_HELPER.casValue(owner, this, valueToSet)) { 325 complete(owner); 326 } 327 } 328 } 329 330 // TODO(lukes): investigate using the @Contended annotation on these fields when jdk8 is 331 // available. 332 /** 333 * This field encodes the current state of the future. 334 * 335 * <p>The valid values are: 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 argument. 343 * </ul> 344 */ 345 private volatile Object value; 346 347 /** All listeners. */ 348 private volatile Listener listeners; 349 350 /** All waiting threads. */ 351 private volatile Waiter waiters; 352 353 /** 354 * Constructor for use by subclasses. 355 */ 356 protected AbstractFuture() {} 357 358 // Gets and Timed Gets 359 // 360 // * Be responsive to interruption 361 // * Don't create Waiter nodes if you aren't going to park, this helps reduce contention on the 362 // waiters field. 363 // * Future completion is defined by when #value becomes non-null/non SetFuture 364 // * Future completion can be observed if the waiters field contains a TOMBSTONE 365 366 // Timed Get 367 // There are a few design constraints to consider 368 // * We want to be responsive to small timeouts, unpark() has non trivial latency overheads (I 369 // have observed 12 micros on 64 bit linux systems to wake up a parked thread). So if the 370 // timeout is small we shouldn't park(). This needs to be traded off with the cpu overhead of 371 // spinning, so we use SPIN_THRESHOLD_NANOS which is what AbstractQueuedSynchronizer uses for 372 // similar purposes. 373 // * We want to behave reasonably for timeouts of 0 374 // * We are more responsive to completion than timeouts. This is because parkNanos depends on 375 // system scheduling and as such we could either miss our deadline, or unpark() could be delayed 376 // so that it looks like we timed out even though we didn't. For comparison FutureTask respects 377 // completion preferably and AQS is non-deterministic (depends on where in the queue the waiter 378 // is). If we wanted to be strict about it, we could store the unpark() time in the Waiter node 379 // and we could use that to make a decision about whether or not we timed out prior to being 380 // unparked. 381 382 /** 383 * {@inheritDoc} 384 * 385 * <p>The default {@link AbstractFuture} implementation throws {@code InterruptedException} if the 386 * current thread is interrupted during the call, even if the value is already available. 387 * 388 * @throws CancellationException {@inheritDoc} 389 */ 390 @CanIgnoreReturnValue 391 @Override 392 public V get(long timeout, TimeUnit unit) 393 throws InterruptedException, TimeoutException, ExecutionException { 394 // NOTE: if timeout < 0, remainingNanos will be < 0 and we will fall into the while(true) loop 395 // at the bottom and throw a timeoutexception. 396 long remainingNanos = unit.toNanos(timeout); // we rely on the implicit null check on unit. 397 if (Thread.interrupted()) { 398 throw new InterruptedException(); 399 } 400 Object localValue = value; 401 if (localValue != null & !(localValue instanceof SetFuture)) { 402 return getDoneValue(localValue); 403 } 404 // we delay calling nanoTime until we know we will need to either park or spin 405 final long endNanos = remainingNanos > 0 ? System.nanoTime() + remainingNanos : 0; 406 long_wait_loop: 407 if (remainingNanos >= SPIN_THRESHOLD_NANOS) { 408 Waiter oldHead = waiters; 409 if (oldHead != Waiter.TOMBSTONE) { 410 Waiter node = new Waiter(); 411 do { 412 node.setNext(oldHead); 413 if (ATOMIC_HELPER.casWaiters(this, oldHead, node)) { 414 while (true) { 415 LockSupport.parkNanos(this, remainingNanos); 416 // Check interruption first, if we woke up due to interruption we need to honor that. 417 if (Thread.interrupted()) { 418 removeWaiter(node); 419 throw new InterruptedException(); 420 } 421 422 // Otherwise re-read and check doneness. If we loop then it must have been a spurious 423 // wakeup 424 localValue = value; 425 if (localValue != null & !(localValue instanceof SetFuture)) { 426 return getDoneValue(localValue); 427 } 428 429 // timed out? 430 remainingNanos = endNanos - System.nanoTime(); 431 if (remainingNanos < SPIN_THRESHOLD_NANOS) { 432 // Remove the waiter, one way or another we are done parking this thread. 433 removeWaiter(node); 434 break long_wait_loop; // jump down to the busy wait loop 435 } 436 } 437 } 438 oldHead = waiters; // re-read and loop. 439 } while (oldHead != Waiter.TOMBSTONE); 440 } 441 // re-read value, if we get here then we must have observed a TOMBSTONE while trying to add a 442 // waiter. 443 return getDoneValue(value); 444 } 445 // If we get here then we have remainingNanos < SPIN_THRESHOLD_NANOS and there is no node on the 446 // waiters list 447 while (remainingNanos > 0) { 448 localValue = value; 449 if (localValue != null & !(localValue instanceof SetFuture)) { 450 return getDoneValue(localValue); 451 } 452 if (Thread.interrupted()) { 453 throw new InterruptedException(); 454 } 455 remainingNanos = endNanos - System.nanoTime(); 456 } 457 458 String futureToString = toString(); 459 // It's confusing to see a completed future in a timeout message; if isDone() returns false, 460 // then we know it must have given a pending toString value earlier. If not, then the future 461 // completed after the timeout expired, and the message might be success. 462 if (isDone()) { 463 throw new TimeoutException( 464 "Waited " + timeout + " " + Ascii.toLowerCase(unit.toString()) 465 + " but future completed as timeout expired"); 466 } 467 throw new TimeoutException( 468 "Waited " + timeout + " " + Ascii.toLowerCase(unit.toString()) + " for " + futureToString); 469 } 470 471 /** 472 * {@inheritDoc} 473 * 474 * <p>The default {@link AbstractFuture} implementation throws {@code InterruptedException} if the 475 * current thread is interrupted during the call, even if the value is already available. 476 * 477 * @throws CancellationException {@inheritDoc} 478 */ 479 @CanIgnoreReturnValue 480 @Override 481 public V get() throws InterruptedException, ExecutionException { 482 if (Thread.interrupted()) { 483 throw new InterruptedException(); 484 } 485 Object localValue = value; 486 if (localValue != null & !(localValue instanceof SetFuture)) { 487 return getDoneValue(localValue); 488 } 489 Waiter oldHead = waiters; 490 if (oldHead != Waiter.TOMBSTONE) { 491 Waiter node = new Waiter(); 492 do { 493 node.setNext(oldHead); 494 if (ATOMIC_HELPER.casWaiters(this, oldHead, node)) { 495 // we are on the stack, now wait for completion. 496 while (true) { 497 LockSupport.park(this); 498 // Check interruption first, if we woke up due to interruption we need to honor that. 499 if (Thread.interrupted()) { 500 removeWaiter(node); 501 throw new InterruptedException(); 502 } 503 // Otherwise re-read and check doneness. If we loop then it must have been a spurious 504 // wakeup 505 localValue = value; 506 if (localValue != null & !(localValue instanceof SetFuture)) { 507 return getDoneValue(localValue); 508 } 509 } 510 } 511 oldHead = waiters; // re-read and loop. 512 } while (oldHead != Waiter.TOMBSTONE); 513 } 514 // re-read value, if we get here then we must have observed a TOMBSTONE while trying to add a 515 // waiter. 516 return getDoneValue(value); 517 } 518 519 /** 520 * Unboxes {@code obj}. Assumes that obj is not {@code null} or a {@link SetFuture}. 521 */ 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(@Nullable 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: while (true) { 838 future.releaseWaiters(); 839 // We call this before the listeners in order to avoid needing to manage a separate stack data 840 // structure for them. 841 // afterDone() should be generally fast and only used for cleanup work... but in theory can 842 // also be recursive and create StackOverflowErrors 843 future.afterDone(); 844 // push the current set of listeners onto next 845 next = future.clearListeners(next); 846 future = null; 847 while (next != null) { 848 Listener curr = next; 849 next = next.next; 850 Runnable task = curr.task; 851 if (task instanceof SetFuture) { 852 SetFuture<?> setFuture = (SetFuture<?>) task; 853 // We unwind setFuture specifically to avoid StackOverflowErrors in the case of long 854 // chains of SetFutures 855 // Handling this special case is important because there is no way to pass an executor to 856 // setFuture, so a user couldn't break the chain by doing this themselves. It is also 857 // potentially common if someone writes a recursive Futures.transformAsync transformer. 858 future = setFuture.owner; 859 if (future.value == setFuture) { 860 Object valueToSet = getFutureValue(setFuture.future); 861 if (ATOMIC_HELPER.casValue(future, setFuture, valueToSet)) { 862 continue outer; 863 } 864 } 865 // other wise the future we were trying to set is already done. 866 } else { 867 executeListener(task, curr.executor); 868 } 869 } 870 break; 871 } 872 } 873 874 /** 875 * Callback method that is called exactly once after the future is completed. 876 * 877 * <p>If {@link #interruptTask} is also run during completion, {@link #afterDone} runs after it. 878 * 879 * <p>The default implementation of this method in {@code AbstractFuture} does nothing. This is 880 * intended for very lightweight cleanup work, for example, timing statistics or clearing fields. 881 * If your task does anything heavier consider, just using a listener with an executor. 882 * 883 * @since 20.0 884 */ 885 @Beta 886 @ForOverride 887 protected void afterDone() {} 888 889 /** 890 * Returns the exception that this {@code Future} completed with. This includes completion through 891 * a call to {@link #setException} or {@link #setFuture setFuture}{@code (failedFuture)} but not 892 * cancellation. 893 * 894 * @throws RuntimeException if the {@code Future} has not failed 895 */ 896 final Throwable trustedGetException() { 897 return ((Failure) value).exception; 898 } 899 900 /** 901 * If this future has been cancelled (and possibly interrupted), cancels (and possibly interrupts) 902 * the given future (if available). 903 */ 904 final void maybePropagateCancellationTo(@Nullable Future<?> related) { 905 if (related != null & isCancelled()) { 906 related.cancel(wasInterrupted()); 907 } 908 } 909 910 /** Releases all threads in the {@link #waiters} list, and clears the list. */ 911 private void releaseWaiters() { 912 Waiter head; 913 do { 914 head = waiters; 915 } while (!ATOMIC_HELPER.casWaiters(this, head, Waiter.TOMBSTONE)); 916 for (Waiter currentWaiter = head; 917 currentWaiter != null; 918 currentWaiter = currentWaiter.next) { 919 currentWaiter.unpark(); 920 } 921 } 922 923 /** 924 * Clears the {@link #listeners} list and prepends its contents to {@code onto}, least recently 925 * added first. 926 */ 927 private Listener clearListeners(Listener onto) { 928 // We need to 929 // 1. atomically swap the listeners with TOMBSTONE, this is because addListener uses that to 930 // to synchronize with us 931 // 2. reverse the linked list, because despite our rather clear contract, people depend on us 932 // executing listeners in the order they were added 933 // 3. push all the items onto 'onto' and return the new head of the stack 934 Listener head; 935 do { 936 head = listeners; 937 } while (!ATOMIC_HELPER.casListeners(this, head, Listener.TOMBSTONE)); 938 Listener reversedList = onto; 939 while (head != null) { 940 Listener tmp = head; 941 head = head.next; 942 tmp.next = reversedList; 943 reversedList = tmp; 944 } 945 return reversedList; 946 } 947 948 // TODO(user) move this up into FluentFuture, or parts as a default method on ListenableFuture? 949 @Override 950 public String toString() { 951 StringBuilder builder = new StringBuilder().append(super.toString()).append("[status="); 952 if (isCancelled()) { 953 builder.append("CANCELLED"); 954 } else if (isDone()) { 955 addDoneString(builder); 956 } else { 957 String pendingDescription; 958 try { 959 pendingDescription = pendingToString(); 960 } catch (RuntimeException e) { 961 // Don't call getMessage or toString() on the exception, in case the exception thrown by the 962 // subclass is implemented with bugs similar to the subclass. 963 pendingDescription = "Exception thrown from implementation: " + e.getClass(); 964 } 965 // The future may complete during or before the call to getPendingToString, so we use null 966 // as a signal that we should try checking if the future is done again. 967 if (!isNullOrEmpty(pendingDescription)) { 968 builder.append("PENDING, info=[").append(pendingDescription).append("]"); 969 } else if (isDone()) { 970 addDoneString(builder); 971 } else { 972 builder.append("PENDING"); 973 } 974 } 975 return builder.append("]").toString(); 976 } 977 978 /** 979 * Provide a human-readable explanation of why this future has not yet completed. 980 * 981 * @return null if an explanation cannot be provided because the future is done. 982 * @since 23.0 983 */ 984 @Nullable 985 protected String pendingToString() { 986 Object localValue = value; 987 if (localValue instanceof SetFuture) { 988 return "setFuture=[" + ((SetFuture) localValue).future + "]"; 989 } else if (this instanceof ScheduledFuture) { 990 return "remaining delay=[" 991 + ((ScheduledFuture) this).getDelay(TimeUnit.MILLISECONDS) 992 + " ms]"; 993 } 994 return null; 995 } 996 997 private void addDoneString(StringBuilder builder) { 998 try { 999 V value = getDone(this); 1000 builder.append("SUCCESS, result=[").append(value).append("]"); 1001 } catch (ExecutionException e) { 1002 builder.append("FAILURE, cause=[").append(e.getCause()).append("]"); 1003 } catch (CancellationException e) { 1004 builder.append("CANCELLED"); // shouldn't be reachable 1005 } catch (RuntimeException e) { 1006 builder.append("UNKNOWN, cause=[").append(e.getClass()).append(" thrown from get()]"); 1007 } 1008 } 1009 1010 /** 1011 * Submits the given runnable to the given {@link Executor} catching and logging all 1012 * {@linkplain RuntimeException runtime exceptions} thrown by the executor. 1013 */ 1014 private static void executeListener(Runnable runnable, Executor executor) { 1015 try { 1016 executor.execute(runnable); 1017 } catch (RuntimeException e) { 1018 // Log it and keep going -- bad runnable and/or executor. Don't punish the other runnables if 1019 // we're given a bad one. We only catch RuntimeException because we want Errors to propagate 1020 // up. 1021 log.log( 1022 Level.SEVERE, 1023 "RuntimeException while executing runnable " + runnable + " with executor " + executor, 1024 e); 1025 } 1026 } 1027 1028 private abstract static class AtomicHelper { 1029 /** Non volatile write of the thread to the {@link Waiter#thread} field. */ 1030 abstract void putThread(Waiter waiter, Thread newValue); 1031 1032 /** Non volatile write of the waiter to the {@link Waiter#next} field. */ 1033 abstract void putNext(Waiter waiter, Waiter newValue); 1034 1035 /** Performs a CAS operation on the {@link #waiters} field. */ 1036 abstract boolean casWaiters(AbstractFuture<?> future, Waiter expect, Waiter update); 1037 1038 /** Performs a CAS operation on the {@link #listeners} field. */ 1039 abstract boolean casListeners(AbstractFuture<?> future, Listener expect, Listener update); 1040 1041 /** Performs a CAS operation on the {@link #value} field. */ 1042 abstract boolean casValue(AbstractFuture<?> future, Object expect, Object update); 1043 } 1044 1045 /** 1046 * {@link AtomicHelper} based on {@link sun.misc.Unsafe}. 1047 * 1048 * <p>Static initialization of this class will fail if the {@link sun.misc.Unsafe} object cannot 1049 * be accessed. 1050 */ 1051 private static final class UnsafeAtomicHelper extends AtomicHelper { 1052 static final sun.misc.Unsafe UNSAFE; 1053 static final long LISTENERS_OFFSET; 1054 static final long WAITERS_OFFSET; 1055 static final long VALUE_OFFSET; 1056 static final long WAITER_THREAD_OFFSET; 1057 static final long WAITER_NEXT_OFFSET; 1058 1059 static { 1060 sun.misc.Unsafe unsafe = null; 1061 try { 1062 unsafe = sun.misc.Unsafe.getUnsafe(); 1063 } catch (SecurityException tryReflectionInstead) { 1064 try { 1065 unsafe = 1066 AccessController.doPrivileged( 1067 new PrivilegedExceptionAction<sun.misc.Unsafe>() { 1068 @Override 1069 public sun.misc.Unsafe run() throws Exception { 1070 Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class; 1071 for (java.lang.reflect.Field f : k.getDeclaredFields()) { 1072 f.setAccessible(true); 1073 Object x = f.get(null); 1074 if (k.isInstance(x)) { 1075 return k.cast(x); 1076 } 1077 } 1078 throw new NoSuchFieldError("the Unsafe"); 1079 } 1080 }); 1081 } catch (PrivilegedActionException e) { 1082 throw new RuntimeException("Could not initialize intrinsics", e.getCause()); 1083 } 1084 } 1085 try { 1086 Class<?> abstractFuture = AbstractFuture.class; 1087 WAITERS_OFFSET = unsafe.objectFieldOffset(abstractFuture.getDeclaredField("waiters")); 1088 LISTENERS_OFFSET = unsafe.objectFieldOffset(abstractFuture.getDeclaredField("listeners")); 1089 VALUE_OFFSET = unsafe.objectFieldOffset(abstractFuture.getDeclaredField("value")); 1090 WAITER_THREAD_OFFSET = unsafe.objectFieldOffset(Waiter.class.getDeclaredField("thread")); 1091 WAITER_NEXT_OFFSET = unsafe.objectFieldOffset(Waiter.class.getDeclaredField("next")); 1092 UNSAFE = unsafe; 1093 } catch (Exception e) { 1094 throwIfUnchecked(e); 1095 throw new RuntimeException(e); 1096 } 1097 } 1098 1099 @Override 1100 void putThread(Waiter waiter, Thread newValue) { 1101 UNSAFE.putObject(waiter, WAITER_THREAD_OFFSET, newValue); 1102 } 1103 1104 @Override 1105 void putNext(Waiter waiter, Waiter newValue) { 1106 UNSAFE.putObject(waiter, WAITER_NEXT_OFFSET, newValue); 1107 } 1108 1109 /** Performs a CAS operation on the {@link #waiters} field. */ 1110 @Override 1111 boolean casWaiters(AbstractFuture<?> future, Waiter expect, Waiter update) { 1112 return UNSAFE.compareAndSwapObject(future, WAITERS_OFFSET, expect, update); 1113 } 1114 1115 /** Performs a CAS operation on the {@link #listeners} field. */ 1116 @Override 1117 boolean casListeners(AbstractFuture<?> future, Listener expect, Listener update) { 1118 return UNSAFE.compareAndSwapObject(future, LISTENERS_OFFSET, expect, update); 1119 } 1120 1121 /** Performs a CAS operation on the {@link #value} field. */ 1122 @Override 1123 boolean casValue(AbstractFuture<?> future, Object expect, Object update) { 1124 return UNSAFE.compareAndSwapObject(future, VALUE_OFFSET, expect, update); 1125 } 1126 } 1127 1128 /** {@link AtomicHelper} based on {@link AtomicReferenceFieldUpdater}. */ 1129 private static final class SafeAtomicHelper extends AtomicHelper { 1130 final AtomicReferenceFieldUpdater<Waiter, Thread> waiterThreadUpdater; 1131 final AtomicReferenceFieldUpdater<Waiter, Waiter> waiterNextUpdater; 1132 final AtomicReferenceFieldUpdater<AbstractFuture, Waiter> waitersUpdater; 1133 final AtomicReferenceFieldUpdater<AbstractFuture, Listener> listenersUpdater; 1134 final AtomicReferenceFieldUpdater<AbstractFuture, Object> valueUpdater; 1135 1136 SafeAtomicHelper( 1137 AtomicReferenceFieldUpdater<Waiter, Thread> waiterThreadUpdater, 1138 AtomicReferenceFieldUpdater<Waiter, Waiter> waiterNextUpdater, 1139 AtomicReferenceFieldUpdater<AbstractFuture, Waiter> waitersUpdater, 1140 AtomicReferenceFieldUpdater<AbstractFuture, Listener> listenersUpdater, 1141 AtomicReferenceFieldUpdater<AbstractFuture, Object> valueUpdater) { 1142 this.waiterThreadUpdater = waiterThreadUpdater; 1143 this.waiterNextUpdater = waiterNextUpdater; 1144 this.waitersUpdater = waitersUpdater; 1145 this.listenersUpdater = listenersUpdater; 1146 this.valueUpdater = valueUpdater; 1147 } 1148 1149 @Override 1150 void putThread(Waiter waiter, Thread newValue) { 1151 waiterThreadUpdater.lazySet(waiter, newValue); 1152 } 1153 1154 @Override 1155 void putNext(Waiter waiter, Waiter newValue) { 1156 waiterNextUpdater.lazySet(waiter, newValue); 1157 } 1158 1159 @Override 1160 boolean casWaiters(AbstractFuture<?> future, Waiter expect, Waiter update) { 1161 return waitersUpdater.compareAndSet(future, expect, update); 1162 } 1163 1164 @Override 1165 boolean casListeners(AbstractFuture<?> future, Listener expect, Listener update) { 1166 return listenersUpdater.compareAndSet(future, expect, update); 1167 } 1168 1169 @Override 1170 boolean casValue(AbstractFuture<?> future, Object expect, Object update) { 1171 return valueUpdater.compareAndSet(future, expect, update); 1172 } 1173 } 1174 1175 /** 1176 * {@link AtomicHelper} based on {@code synchronized} and volatile writes. 1177 * 1178 * <p>This is an implementation of last resort for when certain basic VM features are broken (like 1179 * AtomicReferenceFieldUpdater). 1180 */ 1181 private static final class SynchronizedHelper extends AtomicHelper { 1182 @Override 1183 void putThread(Waiter waiter, Thread newValue) { 1184 waiter.thread = newValue; 1185 } 1186 1187 @Override 1188 void putNext(Waiter waiter, Waiter newValue) { 1189 waiter.next = newValue; 1190 } 1191 1192 @Override 1193 boolean casWaiters(AbstractFuture<?> future, Waiter expect, Waiter update) { 1194 synchronized (future) { 1195 if (future.waiters == expect) { 1196 future.waiters = update; 1197 return true; 1198 } 1199 return false; 1200 } 1201 } 1202 1203 @Override 1204 boolean casListeners(AbstractFuture<?> future, Listener expect, Listener update) { 1205 synchronized (future) { 1206 if (future.listeners == expect) { 1207 future.listeners = update; 1208 return true; 1209 } 1210 return false; 1211 } 1212 } 1213 1214 @Override 1215 boolean casValue(AbstractFuture<?> future, Object expect, Object update) { 1216 synchronized (future) { 1217 if (future.value == expect) { 1218 future.value = update; 1219 return true; 1220 } 1221 return false; 1222 } 1223 } 1224 } 1225 1226 private static CancellationException cancellationExceptionWithCause( 1227 @Nullable String message, @Nullable Throwable cause) { 1228 CancellationException exception = new CancellationException(message); 1229 exception.initCause(cause); 1230 return exception; 1231 } 1232}