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