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.util.concurrent.NullnessCasts.uncheckedNull; 019import static java.lang.Integer.toHexString; 020import static java.lang.System.identityHashCode; 021import static java.util.Objects.requireNonNull; 022import static java.util.concurrent.TimeUnit.MILLISECONDS; 023import static java.util.concurrent.TimeUnit.NANOSECONDS; 024import static java.util.concurrent.atomic.AtomicReferenceFieldUpdater.newUpdater; 025 026import com.google.common.annotations.GwtCompatible; 027import com.google.common.base.Strings; 028import com.google.common.util.concurrent.internal.InternalFutureFailureAccess; 029import com.google.common.util.concurrent.internal.InternalFutures; 030import com.google.errorprone.annotations.CanIgnoreReturnValue; 031import com.google.errorprone.annotations.ForOverride; 032import com.google.j2objc.annotations.ReflectionSupport; 033import com.google.j2objc.annotations.RetainedLocalRef; 034import java.lang.reflect.Field; 035import java.security.AccessController; 036import java.security.PrivilegedActionException; 037import java.security.PrivilegedExceptionAction; 038import java.util.Locale; 039import java.util.concurrent.CancellationException; 040import java.util.concurrent.ExecutionException; 041import java.util.concurrent.Executor; 042import java.util.concurrent.Future; 043import java.util.concurrent.ScheduledFuture; 044import java.util.concurrent.TimeUnit; 045import java.util.concurrent.TimeoutException; 046import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; 047import java.util.concurrent.locks.LockSupport; 048import java.util.logging.Level; 049import org.jspecify.annotations.Nullable; 050import sun.misc.Unsafe; 051 052/** 053 * An abstract implementation of {@link ListenableFuture}, intended for advanced users only. More 054 * common ways to create a {@code ListenableFuture} include instantiating a {@link SettableFuture}, 055 * submitting a task to a {@link ListeningExecutorService}, and deriving a {@code Future} from an 056 * existing one, typically using methods like {@link Futures#transform(ListenableFuture, 057 * com.google.common.base.Function, java.util.concurrent.Executor) Futures.transform} and {@link 058 * Futures#catching(ListenableFuture, Class, com.google.common.base.Function, 059 * java.util.concurrent.Executor) Futures.catching}. 060 * 061 * <p>This class implements all methods in {@code ListenableFuture}. Subclasses should provide a way 062 * to set the result of the computation through the protected methods {@link #set(Object)}, {@link 063 * #setFuture(ListenableFuture)} and {@link #setException(Throwable)}. Subclasses may also override 064 * {@link #afterDone()}, which will be invoked automatically when the future completes. Subclasses 065 * should rarely override other methods. 066 * 067 * @author Sven Mawson 068 * @author Luke Sandberg 069 * @since 1.0 070 */ 071@SuppressWarnings({ 072 // Whenever both tests are cheap and functional, it's faster to use &, | instead of &&, || 073 "ShortCircuitBoolean", 074 "nullness", // TODO(b/147136275): Remove once our checker understands & and |. 075}) 076@GwtCompatible(emulated = true) 077@ReflectionSupport(value = ReflectionSupport.Level.FULL) 078public abstract class AbstractFuture<V extends @Nullable Object> extends InternalFutureFailureAccess 079 implements ListenableFuture<V> { 080 static final boolean GENERATE_CANCELLATION_CAUSES; 081 082 static { 083 // System.getProperty may throw if the security policy does not permit access. 084 boolean generateCancellationCauses; 085 try { 086 generateCancellationCauses = 087 Boolean.parseBoolean( 088 System.getProperty("guava.concurrent.generate_cancellation_cause", "false")); 089 } catch (SecurityException e) { 090 generateCancellationCauses = false; 091 } 092 GENERATE_CANCELLATION_CAUSES = generateCancellationCauses; 093 } 094 095 /** 096 * Tag interface marking trusted subclasses. This enables some optimizations. The implementation 097 * of this interface must also be an AbstractFuture and must not override or expose for overriding 098 * any of the public methods of ListenableFuture. 099 */ 100 interface Trusted<V extends @Nullable Object> extends ListenableFuture<V> {} 101 102 /** 103 * A less abstract subclass of AbstractFuture. This can be used to optimize setFuture by ensuring 104 * that {@link #get} calls exactly the implementation of {@link AbstractFuture#get}. 105 */ 106 abstract static class TrustedFuture<V extends @Nullable Object> extends AbstractFuture<V> 107 implements Trusted<V> { 108 @CanIgnoreReturnValue 109 @Override 110 @ParametricNullness 111 public final V get() throws InterruptedException, ExecutionException { 112 return super.get(); 113 } 114 115 @CanIgnoreReturnValue 116 @Override 117 @ParametricNullness 118 public final V get(long timeout, TimeUnit unit) 119 throws InterruptedException, ExecutionException, TimeoutException { 120 return super.get(timeout, unit); 121 } 122 123 @Override 124 public final boolean isDone() { 125 return super.isDone(); 126 } 127 128 @Override 129 public final boolean isCancelled() { 130 return super.isCancelled(); 131 } 132 133 @Override 134 public final void addListener(Runnable listener, Executor executor) { 135 super.addListener(listener, executor); 136 } 137 138 @CanIgnoreReturnValue 139 @Override 140 public final boolean cancel(boolean mayInterruptIfRunning) { 141 return super.cancel(mayInterruptIfRunning); 142 } 143 } 144 145 static final LazyLogger log = new LazyLogger(AbstractFuture.class); 146 147 // A heuristic for timed gets. If the remaining timeout is less than this, spin instead of 148 // blocking. This value is what AbstractQueuedSynchronizer uses. 149 private static final long SPIN_THRESHOLD_NANOS = 1000L; 150 151 private static final AtomicHelper ATOMIC_HELPER; 152 153 static { 154 AtomicHelper helper; 155 Throwable thrownUnsafeFailure = null; 156 Throwable thrownAtomicReferenceFieldUpdaterFailure = null; 157 158 try { 159 helper = new UnsafeAtomicHelper(); 160 } catch (Exception | Error unsafeFailure) { // sneaky checked exception 161 thrownUnsafeFailure = unsafeFailure; 162 // Catch absolutely everything and fall through to AtomicReferenceFieldUpdaterAtomicHelper. 163 try { 164 helper = new AtomicReferenceFieldUpdaterAtomicHelper(); 165 } catch (Exception // sneaky checked exception 166 | Error atomicReferenceFieldUpdaterFailure) { 167 // Some Android 5.0.x Samsung devices have bugs in JDK reflection APIs that cause 168 // getDeclaredField to throw a NoSuchFieldException when the field is definitely there. 169 // For these users fallback to a suboptimal implementation, based on synchronized. This 170 // will be a definite performance hit to those users. 171 thrownAtomicReferenceFieldUpdaterFailure = atomicReferenceFieldUpdaterFailure; 172 helper = new SynchronizedHelper(); 173 } 174 } 175 ATOMIC_HELPER = helper; 176 177 // Prevent rare disastrous classloading in first call to LockSupport.park. 178 // See: https://bugs.openjdk.java.net/browse/JDK-8074773 179 @SuppressWarnings("unused") 180 Class<?> ensureLoaded = LockSupport.class; 181 182 // Log after all static init is finished; if an installed logger uses any Futures methods, it 183 // shouldn't break in cases where reflection is missing/broken. 184 if (thrownAtomicReferenceFieldUpdaterFailure != null) { 185 log.get().log(Level.SEVERE, "UnsafeAtomicHelper is broken!", thrownUnsafeFailure); 186 log.get() 187 .log( 188 Level.SEVERE, 189 "AtomicReferenceFieldUpdaterAtomicHelper is broken!", 190 thrownAtomicReferenceFieldUpdaterFailure); 191 } 192 } 193 194 /** Waiter links form a Treiber stack, in the {@link #waiters} field. */ 195 private static final class Waiter { 196 static final Waiter TOMBSTONE = new Waiter(false /* ignored param */); 197 198 volatile @Nullable Thread thread; 199 volatile @Nullable Waiter next; 200 201 /** 202 * Constructor for the TOMBSTONE, avoids use of ATOMIC_HELPER in case this class is loaded 203 * before the ATOMIC_HELPER. Apparently this is possible on some android platforms. 204 */ 205 Waiter(boolean unused) {} 206 207 Waiter() { 208 // avoid volatile write, write is made visible by subsequent CAS on waiters field 209 ATOMIC_HELPER.putThread(this, Thread.currentThread()); 210 } 211 212 // non-volatile write to the next field. Should be made visible by subsequent CAS on waiters 213 // field. 214 void setNext(@Nullable Waiter next) { 215 ATOMIC_HELPER.putNext(this, next); 216 } 217 218 void unpark() { 219 // This is racy with removeWaiter. The consequence of the race is that we may spuriously call 220 // unpark even though the thread has already removed itself from the list. But even if we did 221 // use a CAS, that race would still exist (it would just be ever so slightly smaller). 222 Thread w = thread; 223 if (w != null) { 224 thread = null; 225 LockSupport.unpark(w); 226 } 227 } 228 } 229 230 /** 231 * Marks the given node as 'deleted' (null waiter) and then scans the list to unlink all deleted 232 * nodes. This is an O(n) operation in the common case (and O(n^2) in the worst), but we are saved 233 * by two things. 234 * 235 * <ul> 236 * <li>This is only called when a waiting thread times out or is interrupted. Both of which 237 * should be rare. 238 * <li>The waiters list should be very short. 239 * </ul> 240 */ 241 private void removeWaiter(Waiter node) { 242 node.thread = null; // mark as 'deleted' 243 restart: 244 while (true) { 245 Waiter pred = null; 246 Waiter curr = waiters; 247 if (curr == Waiter.TOMBSTONE) { 248 return; // give up if someone is calling complete 249 } 250 Waiter succ; 251 while (curr != null) { 252 succ = curr.next; 253 if (curr.thread != null) { // we aren't unlinking this node, update pred. 254 pred = curr; 255 } else if (pred != null) { // We are unlinking this node and it has a predecessor. 256 pred.next = succ; 257 if (pred.thread == null) { // We raced with another node that unlinked pred. Restart. 258 continue restart; 259 } 260 } else if (!ATOMIC_HELPER.casWaiters(this, curr, succ)) { // We are unlinking head 261 continue restart; // We raced with an add or complete 262 } 263 curr = succ; 264 } 265 break; 266 } 267 } 268 269 /** Listeners also form a stack through the {@link #listeners} field. */ 270 private static final class Listener { 271 static final Listener TOMBSTONE = new Listener(); 272 // null only for TOMBSTONE 273 final @Nullable Runnable task; 274 // null only for TOMBSTONE 275 final @Nullable Executor executor; 276 277 // writes to next are made visible by subsequent CAS's on the listeners field 278 @Nullable Listener next; 279 280 Listener(Runnable task, Executor executor) { 281 this.task = task; 282 this.executor = executor; 283 } 284 285 Listener() { 286 this.task = null; 287 this.executor = null; 288 } 289 } 290 291 /** A special value to represent {@code null}. */ 292 private static final Object NULL = new Object(); 293 294 /** A special value to represent failure, when {@link #setException} is called successfully. */ 295 private static final class Failure { 296 static final Failure FALLBACK_INSTANCE = 297 new Failure( 298 new Throwable("Failure occurred while trying to finish a future.") { 299 @Override 300 public synchronized Throwable fillInStackTrace() { 301 return this; // no stack trace 302 } 303 }); 304 final Throwable exception; 305 306 Failure(Throwable exception) { 307 this.exception = checkNotNull(exception); 308 } 309 } 310 311 /** A special value to represent cancellation and the 'wasInterrupted' bit. */ 312 private static final class Cancellation { 313 // constants to use when GENERATE_CANCELLATION_CAUSES = false 314 static final @Nullable Cancellation CAUSELESS_INTERRUPTED; 315 static final @Nullable Cancellation CAUSELESS_CANCELLED; 316 317 static { 318 if (GENERATE_CANCELLATION_CAUSES) { 319 CAUSELESS_CANCELLED = null; 320 CAUSELESS_INTERRUPTED = null; 321 } else { 322 CAUSELESS_CANCELLED = new Cancellation(false, null); 323 CAUSELESS_INTERRUPTED = new Cancellation(true, null); 324 } 325 } 326 327 final boolean wasInterrupted; 328 final @Nullable Throwable cause; 329 330 Cancellation(boolean wasInterrupted, @Nullable Throwable cause) { 331 this.wasInterrupted = wasInterrupted; 332 this.cause = cause; 333 } 334 } 335 336 /** A special value that encodes the 'setFuture' state. */ 337 private static final class SetFuture<V extends @Nullable Object> implements Runnable { 338 final AbstractFuture<V> owner; 339 final ListenableFuture<? extends V> future; 340 341 SetFuture(AbstractFuture<V> owner, ListenableFuture<? extends V> future) { 342 this.owner = owner; 343 this.future = future; 344 } 345 346 @Override 347 public void run() { 348 if (owner.value != this) { 349 // nothing to do, we must have been cancelled, don't bother inspecting the future. 350 return; 351 } 352 Object valueToSet = getFutureValue(future); 353 if (ATOMIC_HELPER.casValue(owner, this, valueToSet)) { 354 complete( 355 owner, 356 /* 357 * Interruption doesn't propagate through a SetFuture chain (see getFutureValue), so 358 * don't invoke interruptTask. 359 */ 360 false); 361 } 362 } 363 } 364 365 // TODO(lukes): investigate using the @Contended annotation on these fields when jdk8 is 366 // available. 367 /** 368 * This field encodes the current state of the future. 369 * 370 * <p>The valid values are: 371 * 372 * <ul> 373 * <li>{@code null} initial state, nothing has happened. 374 * <li>{@link Cancellation} terminal state, {@code cancel} was called. 375 * <li>{@link Failure} terminal state, {@code setException} was called. 376 * <li>{@link SetFuture} intermediate state, {@code setFuture} was called. 377 * <li>{@link #NULL} terminal state, {@code set(null)} was called. 378 * <li>Any other non-null value, terminal state, {@code set} was called with a non-null 379 * argument. 380 * </ul> 381 */ 382 private volatile @Nullable Object value; 383 384 /** All listeners. */ 385 private volatile @Nullable Listener listeners; 386 387 /** All waiting threads. */ 388 private volatile @Nullable Waiter waiters; 389 390 /** Constructor for use by subclasses. */ 391 protected AbstractFuture() {} 392 393 // Gets and Timed Gets 394 // 395 // * Be responsive to interruption 396 // * Don't create Waiter nodes if you aren't going to park, this helps reduce contention on the 397 // waiters field. 398 // * Future completion is defined by when #value becomes non-null/non SetFuture 399 // * Future completion can be observed if the waiters field contains a TOMBSTONE 400 401 // Timed Get 402 // There are a few design constraints to consider 403 // * We want to be responsive to small timeouts, unpark() has non trivial latency overheads (I 404 // have observed 12 micros on 64-bit linux systems to wake up a parked thread). So if the 405 // timeout is small we shouldn't park(). This needs to be traded off with the cpu overhead of 406 // spinning, so we use SPIN_THRESHOLD_NANOS which is what AbstractQueuedSynchronizer uses for 407 // similar purposes. 408 // * We want to behave reasonably for timeouts of 0 409 // * We are more responsive to completion than timeouts. This is because parkNanos depends on 410 // system scheduling and as such we could either miss our deadline, or unpark() could be delayed 411 // so that it looks like we timed out even though we didn't. For comparison FutureTask respects 412 // completion preferably and AQS is non-deterministic (depends on where in the queue the waiter 413 // is). If we wanted to be strict about it, we could store the unpark() time in the Waiter node 414 // and we could use that to make a decision about whether or not we timed out prior to being 415 // unparked. 416 417 /** 418 * {@inheritDoc} 419 * 420 * <p>The default {@link AbstractFuture} implementation throws {@code InterruptedException} if the 421 * current thread is interrupted during the call, even if the value is already available. 422 * 423 * @throws CancellationException {@inheritDoc} 424 */ 425 @SuppressWarnings("LabelledBreakTarget") // TODO(b/345814817): Maybe fix? 426 @CanIgnoreReturnValue 427 @Override 428 @ParametricNullness 429 public V get(long timeout, TimeUnit unit) 430 throws InterruptedException, TimeoutException, ExecutionException { 431 // NOTE: if timeout < 0, remainingNanos will be < 0 and we will fall into the while(true) loop 432 // at the bottom and throw a timeoutexception. 433 final long timeoutNanos = unit.toNanos(timeout); // we rely on the implicit null check on unit. 434 long remainingNanos = timeoutNanos; 435 if (Thread.interrupted()) { 436 throw new InterruptedException(); 437 } 438 @RetainedLocalRef Object localValue = value; 439 if (localValue != null & !(localValue instanceof SetFuture)) { 440 return getDoneValue(localValue); 441 } 442 // we delay calling nanoTime until we know we will need to either park or spin 443 final long endNanos = remainingNanos > 0 ? System.nanoTime() + remainingNanos : 0; 444 long_wait_loop: 445 if (remainingNanos >= SPIN_THRESHOLD_NANOS) { 446 Waiter oldHead = waiters; 447 if (oldHead != Waiter.TOMBSTONE) { 448 Waiter node = new Waiter(); 449 do { 450 node.setNext(oldHead); 451 if (ATOMIC_HELPER.casWaiters(this, oldHead, node)) { 452 while (true) { 453 OverflowAvoidingLockSupport.parkNanos(this, remainingNanos); 454 // Check interruption first, if we woke up due to interruption we need to honor that. 455 if (Thread.interrupted()) { 456 removeWaiter(node); 457 throw new InterruptedException(); 458 } 459 460 // Otherwise re-read and check doneness. If we loop then it must have been a spurious 461 // wakeup 462 localValue = value; 463 if (localValue != null & !(localValue instanceof SetFuture)) { 464 return getDoneValue(localValue); 465 } 466 467 // timed out? 468 remainingNanos = endNanos - System.nanoTime(); 469 if (remainingNanos < SPIN_THRESHOLD_NANOS) { 470 // Remove the waiter, one way or another we are done parking this thread. 471 removeWaiter(node); 472 break long_wait_loop; // jump down to the busy wait loop 473 } 474 } 475 } 476 oldHead = waiters; // re-read and loop. 477 } while (oldHead != Waiter.TOMBSTONE); 478 } 479 // re-read value, if we get here then we must have observed a TOMBSTONE while trying to add a 480 // waiter. 481 // requireNonNull is safe because value is always set before TOMBSTONE. 482 return getDoneValue(requireNonNull(value)); 483 } 484 // If we get here then we have remainingNanos < SPIN_THRESHOLD_NANOS and there is no node on the 485 // waiters list 486 while (remainingNanos > 0) { 487 localValue = value; 488 if (localValue != null & !(localValue instanceof SetFuture)) { 489 return getDoneValue(localValue); 490 } 491 if (Thread.interrupted()) { 492 throw new InterruptedException(); 493 } 494 remainingNanos = endNanos - System.nanoTime(); 495 } 496 497 String futureToString = toString(); 498 final String unitString = unit.toString().toLowerCase(Locale.ROOT); 499 String message = "Waited " + timeout + " " + unit.toString().toLowerCase(Locale.ROOT); 500 // Only report scheduling delay if larger than our spin threshold - otherwise it's just noise 501 if (remainingNanos + SPIN_THRESHOLD_NANOS < 0) { 502 // We over-waited for our timeout. 503 message += " (plus "; 504 long overWaitNanos = -remainingNanos; 505 long overWaitUnits = unit.convert(overWaitNanos, NANOSECONDS); 506 long overWaitLeftoverNanos = overWaitNanos - unit.toNanos(overWaitUnits); 507 boolean shouldShowExtraNanos = 508 overWaitUnits == 0 || overWaitLeftoverNanos > SPIN_THRESHOLD_NANOS; 509 if (overWaitUnits > 0) { 510 message += overWaitUnits + " " + unitString; 511 if (shouldShowExtraNanos) { 512 message += ","; 513 } 514 message += " "; 515 } 516 if (shouldShowExtraNanos) { 517 message += overWaitLeftoverNanos + " nanoseconds "; 518 } 519 520 message += "delay)"; 521 } 522 // It's confusing to see a completed future in a timeout message; if isDone() returns false, 523 // then we know it must have given a pending toString value earlier. If not, then the future 524 // completed after the timeout expired, and the message might be success. 525 if (isDone()) { 526 throw new TimeoutException(message + " but future completed as timeout expired"); 527 } 528 throw new TimeoutException(message + " for " + futureToString); 529 } 530 531 /** 532 * {@inheritDoc} 533 * 534 * <p>The default {@link AbstractFuture} implementation throws {@code InterruptedException} if the 535 * current thread is interrupted during the call, even if the value is already available. 536 * 537 * @throws CancellationException {@inheritDoc} 538 */ 539 @CanIgnoreReturnValue 540 @Override 541 @ParametricNullness 542 public V get() throws InterruptedException, ExecutionException { 543 if (Thread.interrupted()) { 544 throw new InterruptedException(); 545 } 546 @RetainedLocalRef Object localValue = value; 547 if (localValue != null & !(localValue instanceof SetFuture)) { 548 return getDoneValue(localValue); 549 } 550 Waiter oldHead = waiters; 551 if (oldHead != Waiter.TOMBSTONE) { 552 Waiter node = new Waiter(); 553 do { 554 node.setNext(oldHead); 555 if (ATOMIC_HELPER.casWaiters(this, oldHead, node)) { 556 // we are on the stack, now wait for completion. 557 while (true) { 558 LockSupport.park(this); 559 // Check interruption first, if we woke up due to interruption we need to honor that. 560 if (Thread.interrupted()) { 561 removeWaiter(node); 562 throw new InterruptedException(); 563 } 564 // Otherwise re-read and check doneness. If we loop then it must have been a spurious 565 // wakeup 566 localValue = value; 567 if (localValue != null & !(localValue instanceof SetFuture)) { 568 return getDoneValue(localValue); 569 } 570 } 571 } 572 oldHead = waiters; // re-read and loop. 573 } while (oldHead != Waiter.TOMBSTONE); 574 } 575 // re-read value, if we get here then we must have observed a TOMBSTONE while trying to add a 576 // waiter. 577 // requireNonNull is safe because value is always set before TOMBSTONE. 578 return getDoneValue(requireNonNull(value)); 579 } 580 581 /** Unboxes {@code obj}. Assumes that obj is not {@code null} or a {@link SetFuture}. */ 582 @ParametricNullness 583 private V getDoneValue(Object obj) throws ExecutionException { 584 // While this seems like it might be too branch-y, simple benchmarking proves it to be 585 // unmeasurable (comparing done AbstractFutures with immediateFuture) 586 if (obj instanceof Cancellation) { 587 Cancellation cancellation = (Cancellation) obj; 588 Throwable cause = cancellation.cause; 589 throw cancellationExceptionWithCause("Task was cancelled.", cause); 590 } else if (obj instanceof Failure) { 591 Failure failure = (Failure) obj; 592 Throwable exception = failure.exception; 593 throw new ExecutionException(exception); 594 } else if (obj == NULL) { 595 /* 596 * It's safe to return null because we would only have stored it in the first place if it were 597 * a valid value for V. 598 */ 599 return uncheckedNull(); 600 } else { 601 @SuppressWarnings("unchecked") // this is the only other option 602 V asV = (V) obj; 603 return asV; 604 } 605 } 606 607 @Override 608 public boolean isDone() { 609 @RetainedLocalRef Object localValue = value; 610 return localValue != null & !(localValue instanceof SetFuture); 611 } 612 613 @Override 614 public boolean isCancelled() { 615 @RetainedLocalRef Object localValue = value; 616 return localValue instanceof Cancellation; 617 } 618 619 /** 620 * {@inheritDoc} 621 * 622 * <p>If a cancellation attempt succeeds on a {@code Future} that had previously been {@linkplain 623 * #setFuture set asynchronously}, then the cancellation will also be propagated to the delegate 624 * {@code Future} that was supplied in the {@code setFuture} call. 625 * 626 * <p>Rather than override this method to perform additional cancellation work or cleanup, 627 * subclasses should override {@link #afterDone}, consulting {@link #isCancelled} and {@link 628 * #wasInterrupted} as necessary. This ensures that the work is done even if the future is 629 * cancelled without a call to {@code cancel}, such as by calling {@code 630 * setFuture(cancelledFuture)}. 631 * 632 * <p>Beware of completing a future while holding a lock. Its listeners may do slow work or 633 * acquire other locks, risking deadlocks. 634 */ 635 @CanIgnoreReturnValue 636 @Override 637 public boolean cancel(boolean mayInterruptIfRunning) { 638 @RetainedLocalRef Object localValue = value; 639 boolean rValue = false; 640 if (localValue == null | localValue instanceof SetFuture) { 641 // Try to delay allocating the exception. At this point we may still lose the CAS, but it is 642 // certainly less likely. 643 Object valueToSet = 644 GENERATE_CANCELLATION_CAUSES 645 ? new Cancellation( 646 mayInterruptIfRunning, new CancellationException("Future.cancel() was called.")) 647 /* 648 * requireNonNull is safe because we've initialized these if 649 * !GENERATE_CANCELLATION_CAUSES. 650 * 651 * TODO(cpovirk): Maybe it would be cleaner to define a CancellationSupplier interface 652 * with two implementations, one that contains causeless Cancellation instances and 653 * the other of which creates new Cancellation instances each time it's called? Yet 654 * another alternative is to fill in a non-null value for each of the fields no matter 655 * what and to just not use it if !GENERATE_CANCELLATION_CAUSES. 656 */ 657 : requireNonNull( 658 mayInterruptIfRunning 659 ? Cancellation.CAUSELESS_INTERRUPTED 660 : Cancellation.CAUSELESS_CANCELLED); 661 AbstractFuture<?> abstractFuture = this; 662 while (true) { 663 if (ATOMIC_HELPER.casValue(abstractFuture, localValue, valueToSet)) { 664 rValue = true; 665 complete(abstractFuture, mayInterruptIfRunning); 666 if (localValue instanceof SetFuture) { 667 // propagate cancellation to the future set in setfuture, this is racy, and we don't 668 // care if we are successful or not. 669 ListenableFuture<?> futureToPropagateTo = ((SetFuture) localValue).future; 670 if (futureToPropagateTo instanceof Trusted) { 671 // If the future is a TrustedFuture then we specifically avoid calling cancel() 672 // this has 2 benefits 673 // 1. for long chains of futures strung together with setFuture we consume less stack 674 // 2. we avoid allocating Cancellation objects at every level of the cancellation 675 // chain 676 // We can only do this for TrustedFuture, because TrustedFuture.cancel is final and 677 // does nothing but delegate to this method. 678 AbstractFuture<?> trusted = (AbstractFuture<?>) futureToPropagateTo; 679 localValue = trusted.value; 680 if (localValue == null | localValue instanceof SetFuture) { 681 abstractFuture = trusted; 682 continue; // loop back up and try to complete the new future 683 } 684 } else { 685 // not a TrustedFuture, call cancel directly. 686 futureToPropagateTo.cancel(mayInterruptIfRunning); 687 } 688 } 689 break; 690 } 691 // obj changed, reread 692 localValue = abstractFuture.value; 693 if (!(localValue instanceof SetFuture)) { 694 // obj cannot be null at this point, because value can only change from null to non-null. 695 // So if value changed (and it did since we lost the CAS), then it cannot be null and 696 // since it isn't a SetFuture, then the future must be done and we should exit the loop 697 break; 698 } 699 } 700 } 701 return rValue; 702 } 703 704 /** 705 * Subclasses can override this method to implement interruption of the future's computation. The 706 * method is invoked automatically by a successful call to {@link #cancel(boolean) cancel(true)}. 707 * 708 * <p>The default implementation does nothing. 709 * 710 * <p>This method is likely to be deprecated. Prefer to override {@link #afterDone}, checking 711 * {@link #wasInterrupted} to decide whether to interrupt your task. 712 * 713 * @since 10.0 714 */ 715 protected void interruptTask() {} 716 717 /** 718 * Returns true if this future was cancelled with {@code mayInterruptIfRunning} set to {@code 719 * true}. 720 * 721 * @since 14.0 722 */ 723 protected final boolean wasInterrupted() { 724 @RetainedLocalRef Object localValue = value; 725 return (localValue instanceof Cancellation) && ((Cancellation) localValue).wasInterrupted; 726 } 727 728 /** 729 * {@inheritDoc} 730 * 731 * @since 10.0 732 */ 733 @Override 734 public void addListener(Runnable listener, Executor executor) { 735 checkNotNull(listener, "Runnable was null."); 736 checkNotNull(executor, "Executor was null."); 737 // Checking isDone and listeners != TOMBSTONE may seem redundant, but our contract for 738 // addListener says that listeners execute 'immediate' if the future isDone(). However, our 739 // protocol for completing a future is to assign the value field (which sets isDone to true) and 740 // then to release waiters, followed by executing afterDone(), followed by releasing listeners. 741 // That means that it is possible to observe that the future isDone and that your listeners 742 // don't execute 'immediately'. By checking isDone here we avoid that. 743 // A corollary to all that is that we don't need to check isDone inside the loop because if we 744 // get into the loop we know that we weren't done when we entered and therefore we aren't under 745 // an obligation to execute 'immediately'. 746 if (!isDone()) { 747 Listener oldHead = listeners; 748 if (oldHead != Listener.TOMBSTONE) { 749 Listener newNode = new Listener(listener, executor); 750 do { 751 newNode.next = oldHead; 752 if (ATOMIC_HELPER.casListeners(this, oldHead, newNode)) { 753 return; 754 } 755 oldHead = listeners; // re-read 756 } while (oldHead != Listener.TOMBSTONE); 757 } 758 } 759 // If we get here then the Listener TOMBSTONE was set, which means the future is done, call 760 // the listener. 761 executeListener(listener, executor); 762 } 763 764 /** 765 * Sets the result of this {@code Future} unless this {@code Future} has already been cancelled or 766 * set (including {@linkplain #setFuture set asynchronously}). When a call to this method returns, 767 * the {@code Future} is guaranteed to be {@linkplain #isDone done} <b>only if</b> the call was 768 * accepted (in which case it returns {@code true}). If it returns {@code false}, the {@code 769 * Future} may have previously been set asynchronously, in which case its result may not be known 770 * yet. That result, though not yet known, cannot be overridden by a call to a {@code set*} 771 * method, only by a call to {@link #cancel}. 772 * 773 * <p>Beware of completing a future while holding a lock. Its listeners may do slow work or 774 * acquire other locks, risking deadlocks. 775 * 776 * @param value the value to be used as the result 777 * @return true if the attempt was accepted, completing the {@code Future} 778 */ 779 @CanIgnoreReturnValue 780 protected boolean set(@ParametricNullness V value) { 781 Object valueToSet = value == null ? NULL : value; 782 if (ATOMIC_HELPER.casValue(this, null, valueToSet)) { 783 complete(this, /* callInterruptTask= */ false); 784 return true; 785 } 786 return false; 787 } 788 789 /** 790 * Sets the failed result of this {@code Future} unless this {@code Future} has already been 791 * cancelled or set (including {@linkplain #setFuture set asynchronously}). When a call to this 792 * method returns, the {@code Future} is guaranteed to be {@linkplain #isDone done} <b>only if</b> 793 * the call was accepted (in which case it returns {@code true}). If it returns {@code false}, the 794 * {@code Future} may have previously been set asynchronously, in which case its result may not be 795 * known yet. That result, though not yet known, cannot be overridden by a call to a {@code set*} 796 * method, only by a call to {@link #cancel}. 797 * 798 * <p>Beware of completing a future while holding a lock. Its listeners may do slow work or 799 * acquire other locks, risking deadlocks. 800 * 801 * @param throwable the exception to be used as the failed result 802 * @return true if the attempt was accepted, completing the {@code Future} 803 */ 804 @CanIgnoreReturnValue 805 protected boolean setException(Throwable throwable) { 806 Object valueToSet = new Failure(checkNotNull(throwable)); 807 if (ATOMIC_HELPER.casValue(this, null, valueToSet)) { 808 complete(this, /* callInterruptTask= */ false); 809 return true; 810 } 811 return false; 812 } 813 814 /** 815 * Sets the result of this {@code Future} to match the supplied input {@code Future} once the 816 * supplied {@code Future} is done, unless this {@code Future} has already been cancelled or set 817 * (including "set asynchronously," defined below). 818 * 819 * <p>If the supplied future is {@linkplain #isDone done} when this method is called and the call 820 * is accepted, then this future is guaranteed to have been completed with the supplied future by 821 * the time this method returns. If the supplied future is not done and the call is accepted, then 822 * the future will be <i>set asynchronously</i>. Note that such a result, though not yet known, 823 * cannot be overridden by a call to a {@code set*} method, only by a call to {@link #cancel}. 824 * 825 * <p>If the call {@code setFuture(delegate)} is accepted and this {@code Future} is later 826 * cancelled, cancellation will be propagated to {@code delegate}. Additionally, any call to 827 * {@code setFuture} after any cancellation will propagate cancellation to the supplied {@code 828 * Future}. 829 * 830 * <p>Note that, even if the supplied future is cancelled and it causes this future to complete, 831 * it will never trigger interruption behavior. In particular, it will not cause this future to 832 * invoke the {@link #interruptTask} method, and the {@link #wasInterrupted} method will not 833 * return {@code true}. 834 * 835 * <p>Beware of completing a future while holding a lock. Its listeners may do slow work or 836 * acquire other locks, risking deadlocks. 837 * 838 * @param future the future to delegate to 839 * @return true if the attempt was accepted, indicating that the {@code Future} was not previously 840 * cancelled or set. 841 * @since 19.0 842 */ 843 @CanIgnoreReturnValue 844 @SuppressWarnings("Interruption") // We are propagating an interrupt from a caller. 845 protected boolean setFuture(ListenableFuture<? extends V> future) { 846 checkNotNull(future); 847 @RetainedLocalRef Object localValue = value; 848 if (localValue == null) { 849 if (future.isDone()) { 850 Object value = getFutureValue(future); 851 if (ATOMIC_HELPER.casValue(this, null, value)) { 852 complete( 853 this, 854 /* 855 * Interruption doesn't propagate through a SetFuture chain (see getFutureValue), so 856 * don't invoke interruptTask. 857 */ 858 false); 859 return true; 860 } 861 return false; 862 } 863 SetFuture<V> valueToSet = new SetFuture<>(this, future); 864 if (ATOMIC_HELPER.casValue(this, null, valueToSet)) { 865 // the listener is responsible for calling completeWithFuture, directExecutor is appropriate 866 // since all we are doing is unpacking a completed future which should be fast. 867 try { 868 future.addListener(valueToSet, DirectExecutor.INSTANCE); 869 } catch (Throwable t) { 870 // Any Exception is either a RuntimeException or sneaky checked exception. 871 // 872 // addListener has thrown an exception! SetFuture.run can't throw any exceptions so this 873 // must have been caused by addListener itself. The most likely explanation is a 874 // misconfigured mock. Try to switch to Failure. 875 Failure failure; 876 try { 877 failure = new Failure(t); 878 } catch (Exception | Error oomMostLikely) { // sneaky checked exception 879 failure = Failure.FALLBACK_INSTANCE; 880 } 881 // Note: The only way this CAS could fail is if cancel() has raced with us. That is ok. 882 boolean unused = ATOMIC_HELPER.casValue(this, valueToSet, failure); 883 } 884 return true; 885 } 886 localValue = value; // we lost the cas, fall through and maybe cancel 887 } 888 // The future has already been set to something. If it is cancellation we should cancel the 889 // incoming future. 890 if (localValue instanceof Cancellation) { 891 // we don't care if it fails, this is best-effort. 892 future.cancel(((Cancellation) localValue).wasInterrupted); 893 } 894 return false; 895 } 896 897 /** 898 * Returns a value that satisfies the contract of the {@link #value} field based on the state of 899 * given future. 900 * 901 * <p>This is approximately the inverse of {@link #getDoneValue(Object)} 902 */ 903 private static Object getFutureValue(ListenableFuture<?> future) { 904 if (future instanceof Trusted) { 905 // Break encapsulation for TrustedFuture instances since we know that subclasses cannot 906 // override .get() (since it is final) and therefore this is equivalent to calling .get() 907 // and unpacking the exceptions like we do below (just much faster because it is a single 908 // field read instead of a read, several branches and possibly creating exceptions). 909 Object v = ((AbstractFuture<?>) future).value; 910 if (v instanceof Cancellation) { 911 // If the other future was interrupted, clear the interrupted bit while preserving the cause 912 // this will make it consistent with how non-trustedfutures work which cannot propagate the 913 // wasInterrupted bit 914 Cancellation c = (Cancellation) v; 915 if (c.wasInterrupted) { 916 v = 917 c.cause != null 918 ? new Cancellation(/* wasInterrupted= */ false, c.cause) 919 : Cancellation.CAUSELESS_CANCELLED; 920 } 921 } 922 // requireNonNull is safe as long as we call this method only on completed futures. 923 return requireNonNull(v); 924 } 925 if (future instanceof InternalFutureFailureAccess) { 926 Throwable throwable = 927 InternalFutures.tryInternalFastPathGetFailure((InternalFutureFailureAccess) future); 928 if (throwable != null) { 929 return new Failure(throwable); 930 } 931 } 932 boolean wasCancelled = future.isCancelled(); 933 // Don't allocate a CancellationException if it's not necessary 934 if (!GENERATE_CANCELLATION_CAUSES & wasCancelled) { 935 /* 936 * requireNonNull is safe because we've initialized CAUSELESS_CANCELLED if 937 * !GENERATE_CANCELLATION_CAUSES. 938 */ 939 return requireNonNull(Cancellation.CAUSELESS_CANCELLED); 940 } 941 // Otherwise calculate the value by calling .get() 942 try { 943 Object v = getUninterruptibly(future); 944 if (wasCancelled) { 945 return new Cancellation( 946 false, 947 new IllegalArgumentException( 948 "get() did not throw CancellationException, despite reporting " 949 + "isCancelled() == true: " 950 + future)); 951 } 952 return v == null ? NULL : v; 953 } catch (ExecutionException exception) { 954 if (wasCancelled) { 955 return new Cancellation( 956 false, 957 new IllegalArgumentException( 958 "get() did not throw CancellationException, despite reporting " 959 + "isCancelled() == true: " 960 + future, 961 exception)); 962 } 963 return new Failure(exception.getCause()); 964 } catch (CancellationException cancellation) { 965 if (!wasCancelled) { 966 return new Failure( 967 new IllegalArgumentException( 968 "get() threw CancellationException, despite reporting isCancelled() == false: " 969 + future, 970 cancellation)); 971 } 972 return new Cancellation(false, cancellation); 973 } catch (Exception | Error t) { // sneaky checked exception 974 return new Failure(t); 975 } 976 } 977 978 /** 979 * An inlined private copy of {@link Uninterruptibles#getUninterruptibly} used to break an 980 * internal dependency on other /util/concurrent classes. 981 */ 982 @ParametricNullness 983 private static <V extends @Nullable Object> V getUninterruptibly(Future<V> future) 984 throws ExecutionException { 985 boolean interrupted = false; 986 try { 987 while (true) { 988 try { 989 return future.get(); 990 } catch (InterruptedException e) { 991 interrupted = true; 992 } 993 } 994 } finally { 995 if (interrupted) { 996 Thread.currentThread().interrupt(); 997 } 998 } 999 } 1000 1001 /** Unblocks all threads and runs all listeners. */ 1002 private static void complete(AbstractFuture<?> param, boolean callInterruptTask) { 1003 // Declare a "true" local variable so that the Checker Framework will infer nullness. 1004 AbstractFuture<?> future = param; 1005 1006 Listener next = null; 1007 outer: 1008 while (true) { 1009 future.releaseWaiters(); 1010 /* 1011 * We call interruptTask() immediately before afterDone() so that migrating between the two 1012 * can be a no-op. 1013 */ 1014 if (callInterruptTask) { 1015 future.interruptTask(); 1016 /* 1017 * Interruption doesn't propagate through a SetFuture chain (see getFutureValue), so don't 1018 * invoke interruptTask on any subsequent futures. 1019 */ 1020 callInterruptTask = false; 1021 } 1022 // We call this before the listeners in order to avoid needing to manage a separate stack data 1023 // structure for them. Also, some implementations rely on this running prior to listeners 1024 // so that the cleanup work is visible to listeners. 1025 // afterDone() should be generally fast and only used for cleanup work... but in theory can 1026 // also be recursive and create StackOverflowErrors 1027 future.afterDone(); 1028 // push the current set of listeners onto next 1029 next = future.clearListeners(next); 1030 future = null; 1031 while (next != null) { 1032 Listener curr = next; 1033 next = next.next; 1034 /* 1035 * requireNonNull is safe because the listener stack never contains TOMBSTONE until after 1036 * clearListeners. 1037 */ 1038 Runnable task = requireNonNull(curr.task); 1039 if (task instanceof SetFuture) { 1040 SetFuture<?> setFuture = (SetFuture<?>) task; 1041 // We unwind setFuture specifically to avoid StackOverflowErrors in the case of long 1042 // chains of SetFutures 1043 // Handling this special case is important because there is no way to pass an executor to 1044 // setFuture, so a user couldn't break the chain by doing this themselves. It is also 1045 // potentially common if someone writes a recursive Futures.transformAsync transformer. 1046 future = setFuture.owner; 1047 if (future.value == setFuture) { 1048 Object valueToSet = getFutureValue(setFuture.future); 1049 if (ATOMIC_HELPER.casValue(future, setFuture, valueToSet)) { 1050 continue outer; 1051 } 1052 } 1053 // otherwise the future we were trying to set is already done. 1054 } else { 1055 /* 1056 * requireNonNull is safe because the listener stack never contains TOMBSTONE until after 1057 * clearListeners. 1058 */ 1059 executeListener(task, requireNonNull(curr.executor)); 1060 } 1061 } 1062 break; 1063 } 1064 } 1065 1066 /** 1067 * Callback method that is called exactly once after the future is completed. 1068 * 1069 * <p>If {@link #interruptTask} is also run during completion, {@link #afterDone} runs after it. 1070 * 1071 * <p>The default implementation of this method in {@code AbstractFuture} does nothing. This is 1072 * intended for very lightweight cleanup work, for example, timing statistics or clearing fields. 1073 * If your task does anything heavier consider, just using a listener with an executor. 1074 * 1075 * @since 20.0 1076 */ 1077 @ForOverride 1078 protected void afterDone() {} 1079 1080 // TODO(b/114236866): Inherit doc from InternalFutureFailureAccess. Also, -link to its URL. 1081 /** 1082 * Usually returns {@code null} but, if this {@code Future} has failed, may <i>optionally</i> 1083 * return the cause of the failure. "Failure" means specifically "completed with an exception"; it 1084 * does not include "was cancelled." To be explicit: If this method returns a non-null value, 1085 * then: 1086 * 1087 * <ul> 1088 * <li>{@code isDone()} must return {@code true} 1089 * <li>{@code isCancelled()} must return {@code false} 1090 * <li>{@code get()} must not block, and it must throw an {@code ExecutionException} with the 1091 * return value of this method as its cause 1092 * </ul> 1093 * 1094 * <p>This method is {@code protected} so that classes like {@code 1095 * com.google.common.util.concurrent.SettableFuture} do not expose it to their users as an 1096 * instance method. In the unlikely event that you need to call this method, call {@link 1097 * InternalFutures#tryInternalFastPathGetFailure(InternalFutureFailureAccess)}. 1098 * 1099 * @since 27.0 1100 */ 1101 @Override 1102 /* 1103 * We should annotate the superclass, InternalFutureFailureAccess, to say that its copy of this 1104 * method returns @Nullable, too. However, we're not sure if we want to make any changes to that 1105 * class, since it's in a separate artifact that we planned to release only a single version of. 1106 */ 1107 protected final @Nullable Throwable tryInternalFastPathGetFailure() { 1108 if (this instanceof Trusted) { 1109 @RetainedLocalRef Object localValue = value; 1110 if (localValue instanceof Failure) { 1111 return ((Failure) localValue).exception; 1112 } 1113 } 1114 return null; 1115 } 1116 1117 /** 1118 * If this future has been cancelled (and possibly interrupted), cancels (and possibly interrupts) 1119 * the given future (if available). 1120 */ 1121 final void maybePropagateCancellationTo(@Nullable Future<?> related) { 1122 if (related != null & isCancelled()) { 1123 related.cancel(wasInterrupted()); 1124 } 1125 } 1126 1127 /** Releases all threads in the {@link #waiters} list, and clears the list. */ 1128 private void releaseWaiters() { 1129 Waiter head = ATOMIC_HELPER.gasWaiters(this, Waiter.TOMBSTONE); 1130 for (Waiter currentWaiter = head; currentWaiter != null; currentWaiter = currentWaiter.next) { 1131 currentWaiter.unpark(); 1132 } 1133 } 1134 1135 /** 1136 * Clears the {@link #listeners} list and prepends its contents to {@code onto}, least recently 1137 * added first. 1138 */ 1139 private @Nullable Listener clearListeners(@Nullable Listener onto) { 1140 // We need to 1141 // 1. atomically swap the listeners with TOMBSTONE, this is because addListener uses that 1142 // to synchronize with us 1143 // 2. reverse the linked list, because despite our rather clear contract, people depend on us 1144 // executing listeners in the order they were added 1145 // 3. push all the items onto 'onto' and return the new head of the stack 1146 Listener head = ATOMIC_HELPER.gasListeners(this, Listener.TOMBSTONE); 1147 Listener reversedList = onto; 1148 while (head != null) { 1149 Listener tmp = head; 1150 head = head.next; 1151 tmp.next = reversedList; 1152 reversedList = tmp; 1153 } 1154 return reversedList; 1155 } 1156 1157 // TODO(user): move parts into a default method on ListenableFuture? 1158 @Override 1159 public String toString() { 1160 // TODO(cpovirk): Presize to something plausible? 1161 StringBuilder builder = new StringBuilder(); 1162 if (getClass().getName().startsWith("com.google.common.util.concurrent.")) { 1163 builder.append(getClass().getSimpleName()); 1164 } else { 1165 builder.append(getClass().getName()); 1166 } 1167 builder.append('@').append(toHexString(identityHashCode(this))).append("[status="); 1168 if (isCancelled()) { 1169 builder.append("CANCELLED"); 1170 } else if (isDone()) { 1171 addDoneString(builder); 1172 } else { 1173 addPendingString(builder); // delegates to addDoneString if future completes midway 1174 } 1175 return builder.append("]").toString(); 1176 } 1177 1178 /** 1179 * Provide a human-readable explanation of why this future has not yet completed. 1180 * 1181 * @return null if an explanation cannot be provided (e.g. because the future is done). 1182 * @since 23.0 1183 */ 1184 protected @Nullable String pendingToString() { 1185 // TODO(diamondm) consider moving this into addPendingString so it's always in the output 1186 if (this instanceof ScheduledFuture) { 1187 return "remaining delay=[" + ((ScheduledFuture) this).getDelay(MILLISECONDS) + " ms]"; 1188 } 1189 return null; 1190 } 1191 1192 @SuppressWarnings("CatchingUnchecked") // sneaky checked exception 1193 private void addPendingString(StringBuilder builder) { 1194 // Capture current builder length so it can be truncated if this future ends up completing while 1195 // the toString is being calculated 1196 int truncateLength = builder.length(); 1197 1198 builder.append("PENDING"); 1199 1200 @RetainedLocalRef Object localValue = value; 1201 if (localValue instanceof SetFuture) { 1202 builder.append(", setFuture=["); 1203 appendUserObject(builder, ((SetFuture) localValue).future); 1204 builder.append("]"); 1205 } else { 1206 String pendingDescription; 1207 try { 1208 pendingDescription = Strings.emptyToNull(pendingToString()); 1209 } catch (Exception | StackOverflowError e) { 1210 // Any Exception is either a RuntimeException or sneaky checked exception. 1211 // 1212 // Don't call getMessage or toString() on the exception, in case the exception thrown by the 1213 // subclass is implemented with bugs similar to the subclass. 1214 pendingDescription = "Exception thrown from implementation: " + e.getClass(); 1215 } 1216 if (pendingDescription != null) { 1217 builder.append(", info=[").append(pendingDescription).append("]"); 1218 } 1219 } 1220 1221 // The future may complete while calculating the toString, so we check once more to see if the 1222 // future is done 1223 if (isDone()) { 1224 // Truncate anything that was appended before realizing this future is done 1225 builder.delete(truncateLength, builder.length()); 1226 addDoneString(builder); 1227 } 1228 } 1229 1230 @SuppressWarnings("CatchingUnchecked") // sneaky checked exception 1231 private void addDoneString(StringBuilder builder) { 1232 try { 1233 V value = getUninterruptibly(this); 1234 builder.append("SUCCESS, result=["); 1235 appendResultObject(builder, value); 1236 builder.append("]"); 1237 } catch (ExecutionException e) { 1238 builder.append("FAILURE, cause=[").append(e.getCause()).append("]"); 1239 } catch (CancellationException e) { 1240 builder.append("CANCELLED"); // shouldn't be reachable 1241 } catch (Exception e) { // sneaky checked exception 1242 builder.append("UNKNOWN, cause=[").append(e.getClass()).append(" thrown from get()]"); 1243 } 1244 } 1245 1246 /** 1247 * Any object can be the result of a Future, and not every object has a reasonable toString() 1248 * implementation. Using a reconstruction of the default Object.toString() prevents OOMs and stack 1249 * overflows, and helps avoid sensitive data inadvertently ending up in exception messages. 1250 */ 1251 private void appendResultObject(StringBuilder builder, @Nullable Object o) { 1252 if (o == null) { 1253 builder.append("null"); 1254 } else if (o == this) { 1255 builder.append("this future"); 1256 } else { 1257 builder 1258 .append(o.getClass().getName()) 1259 .append("@") 1260 .append(Integer.toHexString(System.identityHashCode(o))); 1261 } 1262 } 1263 1264 /** Helper for printing user supplied objects into our toString method. */ 1265 @SuppressWarnings("CatchingUnchecked") // sneaky checked exception 1266 private void appendUserObject(StringBuilder builder, @Nullable Object o) { 1267 // This is some basic recursion detection for when people create cycles via set/setFuture or 1268 // when deep chains of futures exist resulting in a StackOverflowException. We could detect 1269 // arbitrary cycles using a thread local but this should be a good enough solution (it is also 1270 // what jdk collections do in these cases) 1271 try { 1272 if (o == this) { 1273 builder.append("this future"); 1274 } else { 1275 builder.append(o); 1276 } 1277 } catch (Exception | StackOverflowError e) { 1278 // Any Exception is either a RuntimeException or sneaky checked exception. 1279 // 1280 // Don't call getMessage or toString() on the exception, in case the exception thrown by the 1281 // user object is implemented with bugs similar to the user object. 1282 builder.append("Exception thrown from implementation: ").append(e.getClass()); 1283 } 1284 } 1285 1286 /** 1287 * Submits the given runnable to the given {@link Executor} catching and logging all {@linkplain 1288 * RuntimeException runtime exceptions} thrown by the executor. 1289 */ 1290 @SuppressWarnings("CatchingUnchecked") // sneaky checked exception 1291 private static void executeListener(Runnable runnable, Executor executor) { 1292 try { 1293 executor.execute(runnable); 1294 } catch (Exception e) { // sneaky checked exception 1295 // Log it and keep going -- bad runnable and/or executor. Don't punish the other runnables if 1296 // we're given a bad one. We only catch Exception because we want Errors to propagate up. 1297 log.get() 1298 .log( 1299 Level.SEVERE, 1300 "RuntimeException while executing runnable " 1301 + runnable 1302 + " with executor " 1303 + executor, 1304 e); 1305 } 1306 } 1307 1308 private abstract static class AtomicHelper { 1309 /** Non-volatile write of the thread to the {@link Waiter#thread} field. */ 1310 abstract void putThread(Waiter waiter, Thread newValue); 1311 1312 /** Non-volatile write of the waiter to the {@link Waiter#next} field. */ 1313 abstract void putNext(Waiter waiter, @Nullable Waiter newValue); 1314 1315 /** Performs a CAS operation on the {@link AbstractFuture#waiters} field. */ 1316 abstract boolean casWaiters( 1317 AbstractFuture<?> future, @Nullable Waiter expect, @Nullable Waiter update); 1318 1319 /** Performs a CAS operation on the {@link AbstractFuture#listeners} field. */ 1320 abstract boolean casListeners( 1321 AbstractFuture<?> future, @Nullable Listener expect, Listener update); 1322 1323 /** Performs a GAS operation on the {@link AbstractFuture#waiters} field. */ 1324 abstract Waiter gasWaiters(AbstractFuture<?> future, Waiter update); 1325 1326 /** Performs a GAS operation on the {@link AbstractFuture#listeners} field. */ 1327 abstract Listener gasListeners(AbstractFuture<?> future, Listener update); 1328 1329 /** Performs a CAS operation on the {@link AbstractFuture#value} field. */ 1330 abstract boolean casValue(AbstractFuture<?> future, @Nullable Object expect, Object update); 1331 } 1332 1333 /** 1334 * {@link AtomicHelper} based on {@link sun.misc.Unsafe}. 1335 * 1336 * <p>Static initialization of this class will fail if the {@link sun.misc.Unsafe} object cannot 1337 * be accessed. 1338 */ 1339 @SuppressWarnings("SunApi") // b/345822163 1340 private static final class UnsafeAtomicHelper extends AtomicHelper { 1341 static final Unsafe UNSAFE; 1342 static final long LISTENERS_OFFSET; 1343 static final long WAITERS_OFFSET; 1344 static final long VALUE_OFFSET; 1345 static final long WAITER_THREAD_OFFSET; 1346 static final long WAITER_NEXT_OFFSET; 1347 1348 static { 1349 Unsafe unsafe = null; 1350 try { 1351 unsafe = Unsafe.getUnsafe(); 1352 } catch (SecurityException tryReflectionInstead) { 1353 try { 1354 unsafe = 1355 AccessController.doPrivileged( 1356 new PrivilegedExceptionAction<Unsafe>() { 1357 @Override 1358 public Unsafe run() throws Exception { 1359 Class<Unsafe> k = Unsafe.class; 1360 for (Field f : k.getDeclaredFields()) { 1361 f.setAccessible(true); 1362 Object x = f.get(null); 1363 if (k.isInstance(x)) { 1364 return k.cast(x); 1365 } 1366 } 1367 throw new NoSuchFieldError("the Unsafe"); 1368 } 1369 }); 1370 } catch (PrivilegedActionException e) { 1371 throw new RuntimeException("Could not initialize intrinsics", e.getCause()); 1372 } 1373 } 1374 try { 1375 Class<?> abstractFuture = AbstractFuture.class; 1376 WAITERS_OFFSET = unsafe.objectFieldOffset(abstractFuture.getDeclaredField("waiters")); 1377 LISTENERS_OFFSET = unsafe.objectFieldOffset(abstractFuture.getDeclaredField("listeners")); 1378 VALUE_OFFSET = unsafe.objectFieldOffset(abstractFuture.getDeclaredField("value")); 1379 WAITER_THREAD_OFFSET = unsafe.objectFieldOffset(Waiter.class.getDeclaredField("thread")); 1380 WAITER_NEXT_OFFSET = unsafe.objectFieldOffset(Waiter.class.getDeclaredField("next")); 1381 UNSAFE = unsafe; 1382 } catch (NoSuchFieldException e) { 1383 throw new RuntimeException(e); 1384 } 1385 } 1386 1387 @Override 1388 void putThread(Waiter waiter, Thread newValue) { 1389 UNSAFE.putObject(waiter, WAITER_THREAD_OFFSET, newValue); 1390 } 1391 1392 @Override 1393 void putNext(Waiter waiter, @Nullable Waiter newValue) { 1394 UNSAFE.putObject(waiter, WAITER_NEXT_OFFSET, newValue); 1395 } 1396 1397 @Override 1398 boolean casWaiters(AbstractFuture<?> future, @Nullable Waiter expect, @Nullable Waiter update) { 1399 return UNSAFE.compareAndSwapObject(future, WAITERS_OFFSET, expect, update); 1400 } 1401 1402 @Override 1403 boolean casListeners(AbstractFuture<?> future, @Nullable Listener expect, Listener update) { 1404 return UNSAFE.compareAndSwapObject(future, LISTENERS_OFFSET, expect, update); 1405 } 1406 1407 @Override 1408 Listener gasListeners(AbstractFuture<?> future, Listener update) { 1409 while (true) { 1410 Listener listener = future.listeners; 1411 if (update == listener) { 1412 return listener; 1413 } 1414 if (casListeners(future, listener, update)) { 1415 return listener; 1416 } 1417 } 1418 } 1419 1420 @Override 1421 Waiter gasWaiters(AbstractFuture<?> future, Waiter update) { 1422 while (true) { 1423 Waiter waiter = future.waiters; 1424 if (update == waiter) { 1425 return waiter; 1426 } 1427 if (casWaiters(future, waiter, update)) { 1428 return waiter; 1429 } 1430 } 1431 } 1432 1433 @Override 1434 boolean casValue(AbstractFuture<?> future, @Nullable Object expect, Object update) { 1435 return UNSAFE.compareAndSwapObject(future, VALUE_OFFSET, expect, update); 1436 } 1437 } 1438 1439 /** {@link AtomicHelper} based on {@link AtomicReferenceFieldUpdater}. */ 1440 private static final class AtomicReferenceFieldUpdaterAtomicHelper extends AtomicHelper { 1441 private static final AtomicReferenceFieldUpdater<Waiter, Thread> waiterThreadUpdater = 1442 newUpdater(Waiter.class, Thread.class, "thread"); 1443 private static final AtomicReferenceFieldUpdater<Waiter, Waiter> waiterNextUpdater = 1444 newUpdater(Waiter.class, Waiter.class, "next"); 1445 private static final AtomicReferenceFieldUpdater<? super AbstractFuture<?>, Waiter> 1446 waitersUpdater = waitersUpdaterFromWithinAbstractFuture(); 1447 private static final AtomicReferenceFieldUpdater<? super AbstractFuture<?>, Listener> 1448 listenersUpdater = listenersUpdaterFromWithinAbstractFuture(); 1449 private static final AtomicReferenceFieldUpdater<? super AbstractFuture<?>, Object> 1450 valueUpdater = valueUpdaterFromWithinAbstractFuture(); 1451 1452 @Override 1453 void putThread(Waiter waiter, Thread newValue) { 1454 waiterThreadUpdater.lazySet(waiter, newValue); 1455 } 1456 1457 @Override 1458 void putNext(Waiter waiter, @Nullable Waiter newValue) { 1459 waiterNextUpdater.lazySet(waiter, newValue); 1460 } 1461 1462 @Override 1463 boolean casWaiters(AbstractFuture<?> future, @Nullable Waiter expect, @Nullable Waiter update) { 1464 return waitersUpdater.compareAndSet(future, expect, update); 1465 } 1466 1467 @Override 1468 boolean casListeners(AbstractFuture<?> future, @Nullable Listener expect, Listener update) { 1469 return listenersUpdater.compareAndSet(future, expect, update); 1470 } 1471 1472 @Override 1473 Listener gasListeners(AbstractFuture<?> future, Listener update) { 1474 return listenersUpdater.getAndSet(future, update); 1475 } 1476 1477 @Override 1478 Waiter gasWaiters(AbstractFuture<?> future, Waiter update) { 1479 return waitersUpdater.getAndSet(future, update); 1480 } 1481 1482 @Override 1483 boolean casValue(AbstractFuture<?> future, @Nullable Object expect, Object update) { 1484 return valueUpdater.compareAndSet(future, expect, update); 1485 } 1486 } 1487 1488 /** Returns an {@link AtomicReferenceFieldUpdater} for {@link #waiters}. */ 1489 private static AtomicReferenceFieldUpdater<? super AbstractFuture<?>, Waiter> 1490 waitersUpdaterFromWithinAbstractFuture() { 1491 return newUpdater(AbstractFuture.class, Waiter.class, "waiters"); 1492 } 1493 1494 /** Returns an {@link AtomicReferenceFieldUpdater} for {@link #listeners}. */ 1495 private static AtomicReferenceFieldUpdater<? super AbstractFuture<?>, Listener> 1496 listenersUpdaterFromWithinAbstractFuture() { 1497 return newUpdater(AbstractFuture.class, Listener.class, "listeners"); 1498 } 1499 1500 /** Returns an {@link AtomicReferenceFieldUpdater} for {@link #value}. */ 1501 private static AtomicReferenceFieldUpdater<? super AbstractFuture<?>, Object> 1502 valueUpdaterFromWithinAbstractFuture() { 1503 return newUpdater(AbstractFuture.class, Object.class, "value"); 1504 } 1505 1506 /** 1507 * {@link AtomicHelper} based on {@code synchronized} and volatile writes. 1508 * 1509 * <p>This is an implementation of last resort for when certain basic VM features are broken (like 1510 * AtomicReferenceFieldUpdater). 1511 */ 1512 private static final class SynchronizedHelper extends AtomicHelper { 1513 @Override 1514 void putThread(Waiter waiter, Thread newValue) { 1515 waiter.thread = newValue; 1516 } 1517 1518 @Override 1519 void putNext(Waiter waiter, @Nullable Waiter newValue) { 1520 waiter.next = newValue; 1521 } 1522 1523 @Override 1524 boolean casWaiters(AbstractFuture<?> future, @Nullable Waiter expect, @Nullable Waiter update) { 1525 synchronized (future) { 1526 if (future.waiters == expect) { 1527 future.waiters = update; 1528 return true; 1529 } 1530 return false; 1531 } 1532 } 1533 1534 @Override 1535 boolean casListeners(AbstractFuture<?> future, @Nullable Listener expect, Listener update) { 1536 synchronized (future) { 1537 if (future.listeners == expect) { 1538 future.listeners = update; 1539 return true; 1540 } 1541 return false; 1542 } 1543 } 1544 1545 @Override 1546 Listener gasListeners(AbstractFuture<?> future, Listener update) { 1547 synchronized (future) { 1548 Listener old = future.listeners; 1549 if (old != update) { 1550 future.listeners = update; 1551 } 1552 return old; 1553 } 1554 } 1555 1556 @Override 1557 Waiter gasWaiters(AbstractFuture<?> future, Waiter update) { 1558 synchronized (future) { 1559 Waiter old = future.waiters; 1560 if (old != update) { 1561 future.waiters = update; 1562 } 1563 return old; 1564 } 1565 } 1566 1567 @Override 1568 boolean casValue(AbstractFuture<?> future, @Nullable Object expect, Object update) { 1569 synchronized (future) { 1570 if (future.value == expect) { 1571 future.value = update; 1572 return true; 1573 } 1574 return false; 1575 } 1576 } 1577 } 1578 1579 private static CancellationException cancellationExceptionWithCause( 1580 String message, @Nullable Throwable cause) { 1581 CancellationException exception = new CancellationException(message); 1582 exception.initCause(cause); 1583 return exception; 1584 } 1585}