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