/*

 * 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.function.Supplier;

import java.util.function.Consumer;

import java.util.function.BiConsumer;

import java.util.function.Function;

import java.util.function.BiFunction;

import java.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.ForkJoinPool;

import java.util.concurrent.ForkJoinTask;

import java.util.concurrent.Executor;

import java.util.concurrent.ThreadLocalRandom;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.TimeoutException;

import java.util.concurrent.CancellationException;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.concurrent.locks.LockSupport;

/**

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

 * value and status), and may include 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>Methods are available for adding dependents based on

 * user-provided Functions, Consumers, or Runnables. The appropriate

 * form to use depends on whether actions require arguments and/or

 * produce results.  Completion of a dependent action will trigger the

 * completion of another CompletableFuture.  Actions may also be

 * triggered after either or both the current and another

 * CompletableFuture complete.  Multiple CompletableFutures may also

 * be grouped as one using {@link #anyOf(CompletableFuture...)} and

 * {@link #allOf(CompletableFuture...)}.

 *

 * <p>CompletableFutures themselves do not execute asynchronously.

 * However, actions supplied for dependent completions of another

 * CompletableFuture may do so, depending on whether they are provided

 * via one of the <em>async</em> methods (that is, methods with names

 * of the form <tt><var>xxx</var>Async</tt>).  The <em>async</em>

 * methods provide a way to commence asynchronous processing of an

 * action using either a given {@link Executor} or by default the

 * {@link ForkJoinPool#commonPool()}. To simplify monitoring,

 * debugging, and tracking, all generated asynchronous tasks are

 * instances of the marker interface {@link AsynchronousCompletionTask}.

 *

 * <p>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 these methods.  There

 * are no guarantees about the order of processing completions unless

 * constrained by these methods.

 *

 * <p>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())}.

 *

 * <p>Upon exceptional completion (including cancellation), or when a

 * completion entails an additional computation which terminates

 * abruptly with an (unchecked) exception or error, then all of their

 * dependent completions (and their dependents in turn) generally act

 * as {@code completeExceptionally} with a {@link CompletionException}

 * holding that exception as its cause.  However, the {@link

 * #exceptionally exceptionally} and {@link #handle handle}

 * completions <em>are</em> able to handle exceptional completions of

 * the CompletableFutures they depend on.

 *

 * <p>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.  However, in these cases,

 * methods {@link #join()} and {@link #getNow} throw the

 * CompletionException, which simplifies usage.

 *

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

 * of type {@code T}) may be null, but passing a null value for any other

 * parameter will result in a {@link NullPointerException} being thrown.

 *

 * @author Doug Lea

 * @since 1.8

 */

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

    /*

     * Overview:

     *

     * 1. 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 fast

     * and simple to detect and trigger, at the expense of a lot of

     * encoding and decoding that infiltrates many methods. One minor

     * simplification relies on the (static) NIL (to box null results)

     * being the only AltResult with a null exception field, so we

     * don't usually need explicit comparisons with NIL. The CF

     * exception propagation mechanics surrounding decoding rely on

     * unchecked casts of decoded results really being unchecked,

     * where user type errors are caught at point of use, as is

     * currently the case in Java. These are highlighted by using

     * SuppressWarnings-annotated temporaries.

     *

     * 2. Waiters are held in a Treiber stack similar to the one used

     * in FutureTask, Phaser, and SynchronousQueue. See their

     * internal documentation for algorithmic details.

     *

     * 3. Completions are also kept in a list/stack, and pulled off

     * and run when completion is triggered. (We could even use the

     * same stack as for waiters, but would give up the potential

     * parallelism obtained because woken waiters help release/run

     * others -- see method postComplete).  Because post-processing

     * may race with direct calls, class Completion opportunistically

     * extends AtomicInteger so callers can claim the action via

     * compareAndSet(0, 1).  The Completion.run methods are all

     * written a boringly similar uniform way (that sometimes includes

     * unnecessary-looking checks, kept to maintain uniformity).

     * There are enough dimensions upon which they differ that

     * attempts to factor commonalities while maintaining efficiency

     * require more lines of code than they would save.

     *

     * 4. The exported then/and/or methods do support a bit of

     * factoring (see doThenApply etc). They must cope with the

     * intrinsic races surrounding addition of a dependent action

     * versus performing the action directly because the task is

     * already complete.  For example, a CF may not be complete upon

     * entry, so a dependent completion is added, but by the time it

     * is added, the target CF is complete, so must be directly

     * executed. This is all done while avoiding unnecessary object

     * construction in safe-bypass cases.

     */

    // preliminaries

    static final class AltResult {

        final Throwable ex; // null only for NIL

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

    }

    static final AltResult NIL = new AltResult(null);

    // Fields

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

    volatile WaitNode waiters; // Treiber stack of threads blocked on get()

    volatile CompletionNode completions; // list (Treiber stack) of completions

    // Basic utilities for triggering and processing completions

    /**

     * Removes and signals all waiting threads and runs all completions.

     */

    final void postComplete() {

        WaitNode q; Thread t;

        while ((q = waiters) != null) {

            if (UNSAFE.compareAndSwapObject(this, WAITERS, q, q.next) &&

                (t = q.thread) != null) {

                q.thread = null;

                LockSupport.unpark(t);

            }

        }

        CompletionNode h; Completion c;

        while ((h = completions) != null) {

            if (UNSAFE.compareAndSwapObject(this, COMPLETIONS, h, h.next) &&

                (c = h.completion) != null)

                c.run();

        }

    }

    /**

     * Triggers completion with the encoding of the given arguments:

     * if the exception is non-null, encodes it as a wrapped

     * CompletionException unless it is one already.  Otherwise uses

     * the given result, boxed as NIL if null.

     */

    final void internalComplete(T v, Throwable ex) {

        if (result == null)

            UNSAFE.compareAndSwapObject

                (this, RESULT, null,

                 (ex == null) ? (v == null) ? NIL : v :

                 new AltResult((ex instanceof CompletionException) ? ex :

                               new CompletionException(ex)));

        postComplete(); // help out even if not triggered

    }

    /**

     * If triggered, helps release and/or process completions.

     */

    final void helpPostComplete() {

        if (result != null)

            postComplete();

    }

    /* ------------- waiting for completions -------------- */

    /** Number of processors, for spin control */

    static final int NCPU = Runtime.getRuntime().availableProcessors();

    /**

     * Heuristic spin value for waitingGet() before blocking on

     * multiprocessors

     */

    static final int SPINS = (NCPU > 1) ? 1 << 8 : 0;

    /**

     * Linked nodes to record waiting threads in a Treiber stack.  See

     * other classes such as Phaser and SynchronousQueue for more

     * detailed explanation. This class implements ManagedBlocker to

     * avoid starvation when blocking actions pile up in

     * ForkJoinPools.

     */

    static final class WaitNode implements ForkJoinPool.ManagedBlocker {

        long nanos;          // wait time if timed

        final long deadline; // non-zero if timed

        volatile int interruptControl; // > 0: interruptible, < 0: interrupted

        volatile Thread thread;

        volatile WaitNode next;

        WaitNode(boolean interruptible, long nanos, long deadline) {

            this.thread = Thread.currentThread();

            this.interruptControl = interruptible ? 1 : 0;

            this.nanos = nanos;

            this.deadline = deadline;

        }

        public boolean isReleasable() {

            if (thread == null)

                return true;

            if (Thread.interrupted()) {

                int i = interruptControl;

                interruptControl = -1;

                if (i > 0)

                    return true;

            }

            if (deadline != 0L &&

                (nanos <= 0L || (nanos = deadline - System.nanoTime()) <= 0L)) {

                thread = null;

                return true;

            }

            return false;

        }

        public boolean block() {

            if (isReleasable())

                return true;

            else if (deadline == 0L)

                LockSupport.park(this);

            else if (nanos > 0L)

                LockSupport.parkNanos(this, nanos);

            return isReleasable();

        }

    }

    /**

     * Returns raw result after waiting, or null if interruptible and

     * interrupted.

     */

    private Object waitingGet(boolean interruptible) {

        WaitNode q = null;

        boolean queued = false;

        int spins = SPINS;

        for (Object r;;) {

            if ((r = result) != null) {

                if (q != null) { // suppress unpark

                    q.thread = null;

                    if (q.interruptControl < 0) {

                        if (interruptible) {

                            removeWaiter(q);

                            return null;

                        }

                        Thread.currentThread().interrupt();

                    }

                }

                postComplete(); // help release others

                return r;

            }

            else if (spins > 0) {

                int rnd = ThreadLocalRandom.nextSecondarySeed();

                if (rnd == 0)

                    rnd = ThreadLocalRandom.current().nextInt();

                if (rnd >= 0)

                    --spins;

            }

            else if (q == null)

                q = new WaitNode(interruptible, 0L, 0L);

            else if (!queued)

                queued = UNSAFE.compareAndSwapObject(this, WAITERS,

                                                     q.next = waiters, q);

            else if (interruptible && q.interruptControl < 0) {

                removeWaiter(q);

                return null;

            }

            else if (q.thread != null && result == null) {

                try {

                    ForkJoinPool.managedBlock(q);

                } catch (InterruptedException ex) {

                    q.interruptControl = -1;

                }

            }

        }

    }

    /**

     * Awaits completion or aborts on interrupt or timeout.

     *

     * @param nanos time to wait

     * @return raw result

     */

    private Object timedAwaitDone(long nanos)

        throws InterruptedException, TimeoutException {

        WaitNode q = null;

        boolean queued = false;

        for (Object r;;) {

            if ((r = result) != null) {

                if (q != null) {

                    q.thread = null;

                    if (q.interruptControl < 0) {

                        removeWaiter(q);

                        throw new InterruptedException();

                    }

                }

                postComplete();

                return r;

            }

            else if (q == null) {

                if (nanos <= 0L)

                    throw new TimeoutException();

                long d = System.nanoTime() + nanos;

                q = new WaitNode(true, nanos, d == 0L ? 1L : d); // avoid 0

            }

            else if (!queued)

                queued = UNSAFE.compareAndSwapObject(this, WAITERS,

                                                     q.next = waiters, q);

            else if (q.interruptControl < 0) {

                removeWaiter(q);

                throw new InterruptedException();

            }

            else if (q.nanos <= 0L) {

                if (result == null) {

                    removeWaiter(q);

                    throw new TimeoutException();

                }

            }

            else if (q.thread != null && result == null) {

                try {

                    ForkJoinPool.managedBlock(q);

                } catch (InterruptedException ex) {

                    q.interruptControl = -1;

                }

            }

        }

    }

    /**

     * Tries to unlink a timed-out or interrupted wait node to avoid

     * accumulating garbage.  Internal nodes are simply unspliced

     * without CAS since it is harmless if they are traversed anyway

     * by releasers.  To avoid effects of unsplicing from already

     * removed nodes, the list is retraversed in case of an apparent

     * race.  This is slow when there are a lot of nodes, but we don't

     * expect lists to be long enough to outweigh higher-overhead

     * schemes.

     */

    private void removeWaiter(WaitNode node) {

        if (node != null) {

            node.thread = null;

            retry:

            for (;;) {          // restart on removeWaiter race

                for (WaitNode pred = null, q = waiters, s; q != null; q = s) {

                    s = q.next;

                    if (q.thread != null)

                        pred = q;

                    else if (pred != null) {

                        pred.next = s;

                        if (pred.thread == null) // check for race

                            continue retry;

                    }

                    else if (!UNSAFE.compareAndSwapObject(this, WAITERS, q, s))

                        continue retry;

                }

                break;

            }

        }

    }

    /* ------------- Async tasks -------------- */

    /**

     * 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 {

    }

    /** Base class can act as either FJ or plain Runnable */

    abstract static class Async extends ForkJoinTask<Void>

        implements Runnable, AsynchronousCompletionTask {

        public final Void getRawResult() { return null; }

        public final void setRawResult(Void v) { }

        public final void run() { exec(); }

    }

    static final class AsyncRun extends Async {

        final Runnable fn;

        final CompletableFuture<Void> dst;

        AsyncRun(Runnable fn, CompletableFuture<Void> dst) {

            this.fn = fn; this.dst = dst;

        }

        public final boolean exec() {

            CompletableFuture<Void> d; Throwable ex;

            if ((d = this.dst) != null && d.result == null) {

                try {

                    fn.run();

                    ex = null;

                } catch (Throwable rex) {

                    ex = rex;

                }

                d.internalComplete(null, ex);

            }

            return true;

        }

        private static final long serialVersionUID = 5232453952276885070L;

    }

    static final class AsyncSupply<U> extends Async {

        final Supplier<U> fn;

        final CompletableFuture<U> dst;

        AsyncSupply(Supplier<U> fn, CompletableFuture<U> dst) {

            this.fn = fn; this.dst = dst;

        }

        public final boolean exec() {

            CompletableFuture<U> d; U u; Throwable ex;

            if ((d = this.dst) != null && d.result == null) {

                try {

                    u = fn.get();

                    ex = null;

                } catch (Throwable rex) {

                    ex = rex;

                    u = null;

                }

                d.internalComplete(u, ex);

            }

            return true;

        }

        private static final long serialVersionUID = 5232453952276885070L;

    }

    static final class AsyncApply<T,U> extends Async {

        final T arg;

        final Function<? super T,? extends U> fn;

        final CompletableFuture<U> dst;

        AsyncApply(T arg, Function<? super T,? extends U> fn,

                   CompletableFuture<U> dst) {

            this.arg = arg; this.fn = fn; this.dst = dst;

        }

        public final boolean exec() {

            CompletableFuture<U> d; U u; Throwable ex;

            if ((d = this.dst) != null && d.result == null) {

                try {

                    u = fn.apply(arg);

                    ex = null;

                } catch (Throwable rex) {

                    ex = rex;

                    u = null;

                }

                d.internalComplete(u, ex);

            }

            return true;

        }

        private static final long serialVersionUID = 5232453952276885070L;

    }

    static final class AsyncCombine<T,U,V> extends Async {

        final T arg1;

        final U arg2;

        final BiFunction<? super T,? super U,? extends V> fn;

        final CompletableFuture<V> dst;

        AsyncCombine(T arg1, U arg2,

                     BiFunction<? super T,? super U,? extends V> fn,

                     CompletableFuture<V> dst) {

            this.arg1 = arg1; this.arg2 = arg2; this.fn = fn; this.dst = dst;

        }

        public final boolean exec() {

            CompletableFuture<V> d; V v; Throwable ex;

            if ((d = this.dst) != null && d.result == null) {

                try {

                    v = fn.apply(arg1, arg2);

                    ex = null;

                } catch (Throwable rex) {

                    ex = rex;

                    v = null;