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