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