/*

 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

 *

 * This code is free software; you can redistribute it and/or modify it

 * under the terms of the GNU General Public License version 2 only, as

 * published by the Free Software Foundation.  Oracle designates this

 * particular file as subject to the "Classpath" exception as provided

 * by Oracle in the LICENSE file that accompanied this code.

 *

 * This code is distributed in the hope that it will be useful, but WITHOUT

 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

 * version 2 for more details (a copy is included in the LICENSE file that

 * accompanied this code).

 *

 * You should have received a copy of the GNU General Public License version

 * 2 along with this work; if not, write to the Free Software Foundation,

 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

 *

 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

 * or visit www.oracle.com if you need additional information or have any

 * questions.

 */

/*

 * This file is available under and governed by the GNU General Public

 * License version 2 only, as published by the Free Software Foundation.

 * However, the following notice accompanied the original version of this

 * file:

 *

 * Written by Doug Lea with assistance from members of JCP JSR-166

 * Expert Group and released to the public domain, as explained at

 * http://creativecommons.org/publicdomain/zero/1.0/

 */

package java.util.concurrent;

import java.util.concurrent.locks.LockSupport;

import java.util.function.BiConsumer;

import java.util.function.BiFunction;

import java.util.function.Consumer;

import java.util.function.Function;

import java.util.function.Supplier;

/**

 * A {@link Future} that may be explicitly completed (setting its

 * value and status), and may be used as a {@link CompletionStage},

 * supporting dependent functions and actions that trigger upon its

 * completion.

 *

 * <p>When two or more threads attempt to

 * {@link #complete complete},

 * {@link #completeExceptionally completeExceptionally}, or

 * {@link #cancel cancel}

 * a CompletableFuture, only one of them succeeds.

 *

 * <p>In addition to these and related methods for directly

 * manipulating status and results, CompletableFuture implements

 * interface {@link CompletionStage} with the following policies: <ul>

 *

 * <li>Actions supplied for dependent completions of

 * <em>non-async</em> methods may be performed by the thread that

 * completes the current CompletableFuture, or by any other caller of

 * a completion method.

 *

 * <li>All <em>async</em> methods without an explicit Executor

 * argument are performed using the {@link ForkJoinPool#commonPool()}

 * (unless it does not support a parallelism level of at least two, in

 * which case, a new Thread is created to run each task).  This may be

 * overridden for non-static methods in subclasses by defining method

 * {@link #defaultExecutor()}. To simplify monitoring, debugging,

 * and tracking, all generated asynchronous tasks are instances of the

 * marker interface {@link AsynchronousCompletionTask}.  Operations

 * with time-delays can use adapter methods defined in this class, for

 * example: {@code supplyAsync(supplier, delayedExecutor(timeout,

 * timeUnit))}.  To support methods with delays and timeouts, this

 * class maintains at most one daemon thread for triggering and

 * cancelling actions, not for running them.

 *

 * <li>All CompletionStage methods are implemented independently of

 * other public methods, so the behavior of one method is not impacted

 * by overrides of others in subclasses.

 *

 * <li>All CompletionStage methods return CompletableFutures.  To

 * restrict usages to only those methods defined in interface

 * CompletionStage, use method {@link #minimalCompletionStage}. Or to

 * ensure only that clients do not themselves modify a future, use

 * method {@link #copy}.

 * </ul>

 *

 * <p>CompletableFuture also implements {@link Future} with the following

 * policies: <ul>

 *

 * <li>Since (unlike {@link FutureTask}) this class has no direct

 * control over the computation that causes it to be completed,

 * cancellation is treated as just another form of exceptional

 * completion.  Method {@link #cancel cancel} has the same effect as

 * {@code completeExceptionally(new CancellationException())}. Method

 * {@link #isCompletedExceptionally} can be used to determine if a

 * CompletableFuture completed in any exceptional fashion.

 *

 * <li>In case of exceptional completion with a CompletionException,

 * methods {@link #get()} and {@link #get(long, TimeUnit)} throw an

 * {@link ExecutionException} with the same cause as held in the

 * corresponding CompletionException.  To simplify usage in most

 * contexts, this class also defines methods {@link #join()} and

 * {@link #getNow} that instead throw the CompletionException directly

 * in these cases.

 * </ul>

 *

 * <p>Arguments used to pass a completion result (that is, for

 * parameters of type {@code T}) for methods accepting them may be

 * null, but passing a null value for any other parameter will result

 * in a {@link NullPointerException} being thrown.

 *

 * <p>Subclasses of this class should normally override the "virtual

 * constructor" method {@link #newIncompleteFuture}, which establishes

 * the concrete type returned by CompletionStage methods. For example,

 * here is a class that substitutes a different default Executor and

 * disables the {@code obtrude} methods:

 *

 * <pre> {@code

 * class MyCompletableFuture<T> extends CompletableFuture<T> {

 *   static final Executor myExecutor = ...;

 *   public MyCompletableFuture() { }

 *   public <U> CompletableFuture<U> newIncompleteFuture() {

 *     return new MyCompletableFuture<U>(); }

 *   public Executor defaultExecutor() {

 *     return myExecutor; }

 *   public void obtrudeValue(T value) {

 *     throw new UnsupportedOperationException(); }

 *   public void obtrudeException(Throwable ex) {

 *     throw new UnsupportedOperationException(); }

 * }}</pre>

 *

 * @author Doug Lea

 * @since 1.8

 * @param <T> The result type returned by this future's {@code join}

 * and {@code get} methods

 */

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {

    /*

     * Overview:

     *

     * A CompletableFuture may have dependent completion actions,

     * collected in a linked stack. It atomically completes by CASing

     * a result field, and then pops off and runs those actions. This

     * applies across normal vs exceptional outcomes, sync vs async

     * actions, binary triggers, and various forms of completions.

     *

     * Non-nullness of field result (set via CAS) indicates done.  An

     * AltResult is used to box null as a result, as well as to hold

     * exceptions.  Using a single field makes completion simple to

     * detect and trigger.  Encoding and decoding is straightforward

     * but adds to the sprawl of trapping and associating exceptions

     * with targets.  Minor simplifications rely on (static) NIL (to

     * box null results) being the only AltResult with a null

     * exception field, so we don't usually need explicit comparisons.

     * Even though some of the generics casts are unchecked (see

     * SuppressWarnings annotations), they are placed to be

     * appropriate even if checked.

     *

     * Dependent actions are represented by Completion objects linked

     * as Treiber stacks headed by field "stack". There are Completion

     * classes for each kind of action, grouped into single-input

     * (UniCompletion), two-input (BiCompletion), projected

     * (BiCompletions using either (not both) of two inputs), shared

     * (CoCompletion, used by the second of two sources), zero-input

     * source actions, and Signallers that unblock waiters. Class

     * Completion extends ForkJoinTask to enable async execution

     * (adding no space overhead because we exploit its "tag" methods

     * to maintain claims). It is also declared as Runnable to allow

     * usage with arbitrary executors.

     *

     * Support for each kind of CompletionStage relies on a separate

     * class, along with two CompletableFuture methods:

     *

     * * A Completion class with name X corresponding to function,

     *   prefaced with "Uni", "Bi", or "Or". Each class contains

     *   fields for source(s), actions, and dependent. They are

     *   boringly similar, differing from others only with respect to

     *   underlying functional forms. We do this so that users don't

     *   encounter layers of adapters in common usages.

     *

     * * Boolean CompletableFuture method x(...) (for example

     *   uniApply) takes all of the arguments needed to check that an

     *   action is triggerable, and then either runs the action or

     *   arranges its async execution by executing its Completion

     *   argument, if present. The method returns true if known to be

     *   complete.

     *

     * * Completion method tryFire(int mode) invokes the associated x

     *   method with its held arguments, and on success cleans up.

     *   The mode argument allows tryFire to be called twice (SYNC,

     *   then ASYNC); the first to screen and trap exceptions while

     *   arranging to execute, and the second when called from a

     *   task. (A few classes are not used async so take slightly

     *   different forms.)  The claim() callback suppresses function

     *   invocation if already claimed by another thread.

     *

     * * CompletableFuture method xStage(...) is called from a public

     *   stage method of CompletableFuture x. It screens user

     *   arguments and invokes and/or creates the stage object.  If

     *   not async and x is already complete, the action is run

     *   immediately.  Otherwise a Completion c is created, pushed to

     *   x's stack (unless done), and started or triggered via

     *   c.tryFire.  This also covers races possible if x completes

     *   while pushing.  Classes with two inputs (for example BiApply)

     *   deal with races across both while pushing actions.  The

     *   second completion is a CoCompletion pointing to the first,

     *   shared so that at most one performs the action.  The

     *   multiple-arity methods allOf and anyOf do this pairwise to

     *   form trees of completions.

     *

     * Note that the generic type parameters of methods vary according

     * to whether "this" is a source, dependent, or completion.

     *

     * Method postComplete is called upon completion unless the target

     * is guaranteed not to be observable (i.e., not yet returned or

     * linked). Multiple threads can call postComplete, which

     * atomically pops each dependent action, and tries to trigger it

     * via method tryFire, in NESTED mode.  Triggering can propagate

     * recursively, so NESTED mode returns its completed dependent (if

     * one exists) for further processing by its caller (see method

     * postFire).

     *

     * Blocking methods get() and join() rely on Signaller Completions

     * that wake up waiting threads.  The mechanics are similar to

     * Treiber stack wait-nodes used in FutureTask, Phaser, and

     * SynchronousQueue. See their internal documentation for

     * algorithmic details.

     *

     * Without precautions, CompletableFutures would be prone to

     * garbage accumulation as chains of Completions build up, each

     * pointing back to its sources. So we null out fields as soon as

     * possible (see especially method Completion.detach). The

     * screening checks needed anyway harmlessly ignore null arguments

     * that may have been obtained during races with threads nulling

     * out fields.  We also try to unlink fired Completions from

     * stacks that might never be popped (see method postFire).

     * Completion fields need not be declared as final or volatile

     * because they are only visible to other threads upon safe

     * publication.

     */

    volatile Object result;       // Either the result or boxed AltResult

    volatile Completion stack;    // Top of Treiber stack of dependent actions

    final boolean internalComplete(Object r) { // CAS from null to r

        return U.compareAndSwapObject(this, RESULT, null, r);

    }

    final boolean casStack(Completion cmp, Completion val) {

        return U.compareAndSwapObject(this, STACK, cmp, val);

    }

    /** Returns true if successfully pushed c onto stack. */

    final boolean tryPushStack(Completion c) {

        Completion h = stack;

        lazySetNext(c, h);

        return U.compareAndSwapObject(this, STACK, h, c);

    }

    /** Unconditionally pushes c onto stack, retrying if necessary. */

    final void pushStack(Completion c) {

        do {} while (!tryPushStack(c));

    }

    /* ------------- Encoding and decoding outcomes -------------- */

    static final class AltResult { // See above

        final Throwable ex;        // null only for NIL

        AltResult(Throwable x) { this.ex = x; }

    }

    /** The encoding of the null value. */

    static final AltResult NIL = new AltResult(null);

    /** Completes with the null value, unless already completed. */

    final boolean completeNull() {

        return U.compareAndSwapObject(this, RESULT, null,

                                      NIL);

    }

    /** Returns the encoding of the given non-exceptional value. */

    final Object encodeValue(T t) {

        return (t == null) ? NIL : t;

    }

    /** Completes with a non-exceptional result, unless already completed. */

    final boolean completeValue(T t) {

        return U.compareAndSwapObject(this, RESULT, null,

                                      (t == null) ? NIL : t);

    }

    /**

     * Returns the encoding of the given (non-null) exception as a

     * wrapped CompletionException unless it is one already.

     */

    static AltResult encodeThrowable(Throwable x) {

        return new AltResult((x instanceof CompletionException) ? x :

                             new CompletionException(x));

    }

    /** Completes with an exceptional result, unless already completed. */

    final boolean completeThrowable(Throwable x) {

        return U.compareAndSwapObject(this, RESULT, null,

                                      encodeThrowable(x));

    }

    /**

     * Returns the encoding of the given (non-null) exception as a

     * wrapped CompletionException unless it is one already.  May

     * return the given Object r (which must have been the result of a

     * source future) if it is equivalent, i.e. if this is a simple

     * relay of an existing CompletionException.

     */

    static Object encodeThrowable(Throwable x, Object r) {

        if (!(x instanceof CompletionException))

            x = new CompletionException(x);

        else if (r instanceof AltResult && x == ((AltResult)r).ex)

            return r;

        return new AltResult(x);

    }

    /**

     * Completes with the given (non-null) exceptional result as a

     * wrapped CompletionException unless it is one already, unless

     * already completed.  May complete with the given Object r

     * (which must have been the result of a source future) if it is

     * equivalent, i.e. if this is a simple propagation of an

     * existing CompletionException.

     */

    final boolean completeThrowable(Throwable x, Object r) {

        return U.compareAndSwapObject(this, RESULT, null,

                                      encodeThrowable(x, r));

    }

    /**

     * Returns the encoding of the given arguments: if the exception

     * is non-null, encodes as AltResult.  Otherwise uses the given

     * value, boxed as NIL if null.

     */

    Object encodeOutcome(T t, Throwable x) {

        return (x == null) ? (t == null) ? NIL : t : encodeThrowable(x);

    }

    /**

     * Returns the encoding of a copied outcome; if exceptional,

     * rewraps as a CompletionException, else returns argument.

     */

    static Object encodeRelay(Object r) {

        Throwable x;

        return (((r instanceof AltResult) &&

                 (x = ((AltResult)r).ex) != null &&

                 !(x instanceof CompletionException)) ?

                new AltResult(new CompletionException(x)) : r);

    }

    /**

     * Completes with r or a copy of r, unless already completed.

     * If exceptional, r is first coerced to a CompletionException.

     */

    final boolean completeRelay(Object r) {

        return U.compareAndSwapObject(this, RESULT, null,

                                      encodeRelay(r));

    }

    /**

     * Reports result using Future.get conventions.

     */

    private static <T> T reportGet(Object r)

        throws InterruptedException, ExecutionException {

        if (r == null) // by convention below, null means interrupted

            throw new InterruptedException();

        if (r instanceof AltResult) {

            Throwable x, cause;

            if ((x = ((AltResult)r).ex) == null)

                return null;

            if (x instanceof CancellationException)

                throw (CancellationException)x;

            if ((x instanceof CompletionException) &&

                (cause = x.getCause()) != null)

                x = cause;

            throw new ExecutionException(x);

        }

        @SuppressWarnings("unchecked") T t = (T) r;

        return t;

    }

    /**

     * Decodes outcome to return result or throw unchecked exception.

     */

    private static <T> T reportJoin(Object r) {

        if (r instanceof AltResult) {

            Throwable x;

            if ((x = ((AltResult)r).ex) == null)

                return null;

            if (x instanceof CancellationException)

                throw (CancellationException)x;

            if (x instanceof CompletionException)

                throw (CompletionException)x;

            throw new CompletionException(x);

        }

        @SuppressWarnings("unchecked") T t = (T) r;

        return t;

    }

    /* ------------- Async task preliminaries -------------- */

    /**

     * A marker interface identifying asynchronous tasks produced by

     * {@code async} methods. This may be useful for monitoring,

     * debugging, and tracking asynchronous activities.

     *

     * @since 1.8

     */

    public static interface AsynchronousCompletionTask {

    }

    private static final boolean USE_COMMON_POOL =

        (ForkJoinPool.getCommonPoolParallelism() > 1);

    /**

     * Default executor -- ForkJoinPool.commonPool() unless it cannot

     * support parallelism.

     */

    private static final Executor ASYNC_POOL = USE_COMMON_POOL ?

        ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

    /** Fallback if ForkJoinPool.commonPool() cannot support parallelism */

    static final class ThreadPerTaskExecutor implements Executor {

        public void execute(Runnable r) { new Thread(r).start(); }

    }

    /**

     * Null-checks user executor argument, and translates uses of

     * commonPool to ASYNC_POOL in case parallelism disabled.

     */

    static Executor screenExecutor(Executor e) {

        if (!USE_COMMON_POOL && e == ForkJoinPool.commonPool())

            return ASYNC_POOL;

        if (e == null) throw new NullPointerException();

        return e;

    }

    // Modes for Completion.tryFire. Signedness matters.

    static final int SYNC   =  0;

    static final int ASYNC  =  1;

    static final int NESTED = -1;

    /**

     * Spins before blocking in waitingGet

     */

    static final int SPINS = (Runtime.getRuntime().availableProcessors() > 1 ?

                              1 << 8 : 0);

    /* ------------- Base Completion classes and operations -------------- */

    @SuppressWarnings("serial")

    abstract static class Completion extends ForkJoinTask<Void>

        implements Runnable, AsynchronousCompletionTask {

        volatile Completion next;      // Treiber stack link

        /**

         * Performs completion action if triggered, returning a

         * dependent that may need propagation, if one exists.

         *

         * @param mode SYNC, ASYNC, or NESTED

         */

        abstract CompletableFuture<?> tryFire(int mode);

        /** Returns true if possibly still triggerable. Used by cleanStack. */

        abstract boolean isLive();

        public final void run()                { tryFire(ASYNC); }

        public final boolean exec()            { tryFire(ASYNC); return false; }

        public final Void getRawResult()       { return null; }

        public final void setRawResult(Void v) {}

    }

    static void lazySetNext(Completion c, Completion next) {

        U.putOrderedObject(c, NEXT, next);

    }

    /**

     * Pops and tries to trigger all reachable dependents.  Call only

     * when known to be done.

     */

    final void postComplete() {

        /*

         * On each step, variable f holds current dependents to pop

         * and run.  It is extended along only one path at a time,

         * pushing others to avoid unbounded recursion.