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