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