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