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