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