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