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