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