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