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