/*

 * 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.  Sun designates this

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

 * by Sun 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 Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,

 * CA 95054 USA or visit www.sun.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/licenses/publicdomain

 */

package java.util.concurrent;

import java.io.Serializable;

import java.util.Collection;

import java.util.Collections;

import java.util.List;

import java.util.RandomAccess;

import java.util.Map;

import java.util.WeakHashMap;

/**

 * Abstract base class for tasks that run within a {@link ForkJoinPool}.

 * A {@code ForkJoinTask} is a thread-like entity that is much

 * lighter weight than a normal thread.  Huge numbers of tasks and

 * subtasks may be hosted by a small number of actual threads in a

 * ForkJoinPool, at the price of some usage limitations.

 *

 * <p>A "main" {@code ForkJoinTask} begins execution when submitted

 * to a {@link ForkJoinPool}.  Once started, it will usually in turn

 * start other subtasks.  As indicated by the name of this class,

 * many programs using {@code ForkJoinTask} employ only methods

 * {@link #fork} and {@link #join}, or derivatives such as {@link

 * #invokeAll}.  However, this class also provides a number of other

 * methods that can come into play in advanced usages, as well as

 * extension mechanics that allow support of new forms of fork/join

 * processing.

 *

 * <p>A {@code ForkJoinTask} is a lightweight form of {@link Future}.

 * The efficiency of {@code ForkJoinTask}s stems from a set of

 * restrictions (that are only partially statically enforceable)

 * reflecting their intended use as computational tasks calculating

 * pure functions or operating on purely isolated objects.  The

 * primary coordination mechanisms are {@link #fork}, that arranges

 * asynchronous execution, and {@link #join}, that doesn't proceed

 * until the task's result has been computed.  Computations should

 * avoid {@code synchronized} methods or blocks, and should minimize

 * other blocking synchronization apart from joining other tasks or

 * using synchronizers such as Phasers that are advertised to

 * cooperate with fork/join scheduling. Tasks should also not perform

 * blocking IO, and should ideally access variables that are

 * completely independent of those accessed by other running

 * tasks. Minor breaches of these restrictions, for example using

 * shared output streams, may be tolerable in practice, but frequent

 * use may result in poor performance, and the potential to

 * indefinitely stall if the number of threads not waiting for IO or

 * other external synchronization becomes exhausted. This usage

 * restriction is in part enforced by not permitting checked

 * exceptions such as {@code IOExceptions} to be thrown. However,

 * computations may still encounter unchecked exceptions, that are

 * rethrown to callers attempting to join them. These exceptions may

 * additionally include {@link RejectedExecutionException} stemming

 * from internal resource exhaustion, such as failure to allocate

 * internal task queues.

 *

 * <p>The primary method for awaiting completion and extracting

 * results of a task is {@link #join}, but there are several variants:

 * The {@link Future#get} methods support interruptible and/or timed

 * waits for completion and report results using {@code Future}

 * conventions. Method {@link #helpJoin} enables callers to actively

 * execute other tasks while awaiting joins, which is sometimes more

 * efficient but only applies when all subtasks are known to be

 * strictly tree-structured. Method {@link #invoke} is semantically

 * equivalent to {@code fork(); join()} but always attempts to begin

 * execution in the current thread. The "<em>quiet</em>" forms of

 * these methods do not extract results or report exceptions. These

 * may be useful when a set of tasks are being executed, and you need

 * to delay processing of results or exceptions until all complete.

 * Method {@code invokeAll} (available in multiple versions)

 * performs the most common form of parallel invocation: forking a set

 * of tasks and joining them all.

 *

 * <p>The execution status of tasks may be queried at several levels

 * of detail: {@link #isDone} is true if a task completed in any way

 * (including the case where a task was cancelled without executing);

 * {@link #isCompletedNormally} is true if a task completed without

 * cancellation or encountering an exception; {@link #isCancelled} is

 * true if the task was cancelled (in which case {@link #getException}

 * returns a {@link java.util.concurrent.CancellationException}); and

 * {@link #isCompletedAbnormally} is true if a task was either

 * cancelled or encountered an exception, in which case {@link

 * #getException} will return either the encountered exception or

 * {@link java.util.concurrent.CancellationException}.

 *

 * <p>The ForkJoinTask class is not usually directly subclassed.

 * Instead, you subclass one of the abstract classes that support a

 * particular style of fork/join processing, typically {@link

 * RecursiveAction} for computations that do not return results, or

 * {@link RecursiveTask} for those that do.  Normally, a concrete

 * ForkJoinTask subclass declares fields comprising its parameters,

 * established in a constructor, and then defines a {@code compute}

 * method that somehow uses the control methods supplied by this base

 * class. While these methods have {@code public} access (to allow

 * instances of different task subclasses to call each other's

 * methods), some of them may only be called from within other

 * ForkJoinTasks (as may be determined using method {@link

 * #inForkJoinPool}).  Attempts to invoke them in other contexts

 * result in exceptions or errors, possibly including

 * ClassCastException.

 *

 * <p>Most base support methods are {@code final}, to prevent

 * overriding of implementations that are intrinsically tied to the

 * underlying lightweight task scheduling framework.  Developers

 * creating new basic styles of fork/join processing should minimally

 * implement {@code protected} methods {@link #exec}, {@link

 * #setRawResult}, and {@link #getRawResult}, while also introducing

 * an abstract computational method that can be implemented in its

 * subclasses, possibly relying on other {@code protected} methods

 * provided by this class.

 *

 * <p>ForkJoinTasks should perform relatively small amounts of

 * computation. Large tasks should be split into smaller subtasks,

 * usually via recursive decomposition. As a very rough rule of thumb,

 * a task should perform more than 100 and less than 10000 basic

 * computational steps. If tasks are too big, then parallelism cannot

 * improve throughput. If too small, then memory and internal task

 * maintenance overhead may overwhelm processing.

 *

 * <p>This class provides {@code adapt} methods for {@link Runnable}

 * and {@link Callable}, that may be of use when mixing execution of

 * {@code ForkJoinTasks} with other kinds of tasks. When all tasks

 * are of this form, consider using a pool in

 * {@linkplain ForkJoinPool#setAsyncMode async mode}.

 *

 * <p>ForkJoinTasks are {@code Serializable}, which enables them to be

 * used in extensions such as remote execution frameworks. It is

 * sensible to serialize tasks only before or after, but not during,

 * execution. Serialization is not relied on during execution itself.

 *

 * @since 1.7

 * @author Doug Lea

 */

public abstract class ForkJoinTask<V> implements Future<V>, Serializable {

    /**

     * Run control status bits packed into a single int to minimize

     * footprint and to ensure atomicity (via CAS).  Status is

     * initially zero, and takes on nonnegative values until

     * completed, upon which status holds COMPLETED. CANCELLED, or

     * EXCEPTIONAL, which use the top 3 bits.  Tasks undergoing

     * blocking waits by other threads have SIGNAL_MASK bits set --

     * bit 15 for external (nonFJ) waits, and the rest a count of

     * waiting FJ threads.  (This representation relies on

     * ForkJoinPool max thread limits). Completion of a stolen task

     * with SIGNAL_MASK bits set awakens waiter via notifyAll. Even

     * though suboptimal for some purposes, we use basic builtin

     * wait/notify to take advantage of "monitor inflation" in JVMs

     * that we would otherwise need to emulate to avoid adding further

     * per-task bookkeeping overhead. Note that bits 16-28 are

     * currently unused. Also value 0x80000000 is available as spare

     * completion value.

     */

    volatile int status; // accessed directly by pool and workers

    static final int COMPLETION_MASK      = 0xe0000000;

    static final int NORMAL               = 0xe0000000; // == mask

    static final int CANCELLED            = 0xc0000000;

    static final int EXCEPTIONAL          = 0xa0000000;

    static final int SIGNAL_MASK          = 0x0000ffff;

    static final int INTERNAL_SIGNAL_MASK = 0x00007fff;

    static final int EXTERNAL_SIGNAL      = 0x00008000; // top bit of low word

    /**

     * Table of exceptions thrown by tasks, to enable reporting by

     * callers. Because exceptions are rare, we don't directly keep

     * them with task objects, but instead use a weak ref table.  Note

     * that cancellation exceptions don't appear in the table, but are

     * instead recorded as status values.

     * TODO: Use ConcurrentReferenceHashMap

     */

    static final Map<ForkJoinTask<?>, Throwable> exceptionMap =

        Collections.synchronizedMap

        (new WeakHashMap<ForkJoinTask<?>, Throwable>());

    // within-package utilities

    /**

     * Gets current worker thread, or null if not a worker thread.

     */

    static ForkJoinWorkerThread getWorker() {

        Thread t = Thread.currentThread();

        return ((t instanceof ForkJoinWorkerThread) ?

                (ForkJoinWorkerThread) t : null);

    }

    final boolean casStatus(int cmp, int val) {

        return UNSAFE.compareAndSwapInt(this, statusOffset, cmp, val);

    }

    /**

     * Workaround for not being able to rethrow unchecked exceptions.

     */

    static void rethrowException(Throwable ex) {

        if (ex != null)

            UNSAFE.throwException(ex);

    }

    // Setting completion status

    /**

     * Marks completion and wakes up threads waiting to join this task.

     *

     * @param completion one of NORMAL, CANCELLED, EXCEPTIONAL

     */

    final void setCompletion(int completion) {

        ForkJoinPool pool = getPool();

        if (pool != null) {

            int s; // Clear signal bits while setting completion status

            do {} while ((s = status) >= 0 && !casStatus(s, completion));

            if ((s & SIGNAL_MASK) != 0) {

                if ((s &= INTERNAL_SIGNAL_MASK) != 0)

                    pool.updateRunningCount(s);

                synchronized (this) { notifyAll(); }

            }

        }

        else

            externallySetCompletion(completion);

    }

    /**

     * Version of setCompletion for non-FJ threads.  Leaves signal

     * bits for unblocked threads to adjust, and always notifies.

     */

    private void externallySetCompletion(int completion) {

        int s;

        do {} while ((s = status) >= 0 &&

                     !casStatus(s, (s & SIGNAL_MASK) | completion));

        synchronized (this) { notifyAll(); }

    }

    /**

     * Sets status to indicate normal completion.

     */

    final void setNormalCompletion() {

        // Try typical fast case -- single CAS, no signal, not already done.

        // Manually expand casStatus to improve chances of inlining it

        if (!UNSAFE.compareAndSwapInt(this, statusOffset, 0, NORMAL))

            setCompletion(NORMAL);

    }

    // internal waiting and notification

    /**

     * Performs the actual monitor wait for awaitDone.

     */

    private void doAwaitDone() {

        // Minimize lock bias and in/de-flation effects by maximizing

        // chances of waiting inside sync

        try {

            while (status >= 0)

                synchronized (this) { if (status >= 0) wait(); }

        } catch (InterruptedException ie) {

            onInterruptedWait();

        }

    }

    /**

     * Performs the actual timed monitor wait for awaitDone.

     */

    private void doAwaitDone(long startTime, long nanos) {

        synchronized (this) {

            try {

                while (status >= 0) {

                    long nt = nanos - (System.nanoTime() - startTime);

                    if (nt <= 0)

                        break;

                    wait(nt / 1000000, (int) (nt % 1000000));

                }

            } catch (InterruptedException ie) {

                onInterruptedWait();

            }

        }

    }

    // Awaiting completion

    /**

     * Sets status to indicate there is joiner, then waits for join,

     * surrounded with pool notifications.

     *

     * @return status upon exit

     */

    private int awaitDone(ForkJoinWorkerThread w,

                          boolean maintainParallelism) {

        ForkJoinPool pool = (w == null) ? null : w.pool;

        int s;

        while ((s = status) >= 0) {

            if (casStatus(s, (pool == null) ? s|EXTERNAL_SIGNAL : s+1)) {

                if (pool == null || !pool.preJoin(this, maintainParallelism))

                    doAwaitDone();

                if (((s = status) & INTERNAL_SIGNAL_MASK) != 0)

                    adjustPoolCountsOnUnblock(pool);

                break;

            }

        }

        return s;

    }

    /**

     * Timed version of awaitDone

     *

     * @return status upon exit

     */

    private int awaitDone(ForkJoinWorkerThread w, long nanos) {

        ForkJoinPool pool = (w == null) ? null : w.pool;

        int s;

        while ((s = status) >= 0) {

            if (casStatus(s, (pool == null) ? s|EXTERNAL_SIGNAL : s+1)) {

                long startTime = System.nanoTime();

                if (pool == null || !pool.preJoin(this, false))

                    doAwaitDone(startTime, nanos);

                if ((s = status) >= 0) {

                    adjustPoolCountsOnCancelledWait(pool);

                    s = status;

                }

                if (s < 0 && (s & INTERNAL_SIGNAL_MASK) != 0)

                    adjustPoolCountsOnUnblock(pool);

                break;

            }

        }

        return s;

    }

    /**

     * Notifies pool that thread is unblocked. Called by signalled

     * threads when woken by non-FJ threads (which is atypical).

     */

    private void adjustPoolCountsOnUnblock(ForkJoinPool pool) {

        int s;

        do {} while ((s = status) < 0 && !casStatus(s, s & COMPLETION_MASK));

        if (pool != null && (s &= INTERNAL_SIGNAL_MASK) != 0)

            pool.updateRunningCount(s);

    }

    /**

     * Notifies pool to adjust counts on cancelled or timed out wait.

     */

    private void adjustPoolCountsOnCancelledWait(ForkJoinPool pool) {

        if (pool != null) {

            int s;

            while ((s = status) >= 0 && (s & INTERNAL_SIGNAL_MASK) != 0) {

                if (casStatus(s, s - 1)) {

                    pool.updateRunningCount(1);

                    break;

                }

            }

        }

    }

    /**

     * Handles interruptions during waits.

     */

    private void onInterruptedWait() {

        ForkJoinWorkerThread w = getWorker();

        if (w == null)

            Thread.currentThread().interrupt(); // re-interrupt

        else if (w.isTerminating())

            cancelIgnoringExceptions();

        // else if FJworker, ignore interrupt

    }

    // Recording and reporting exceptions

    private void setDoneExceptionally(Throwable rex) {

        exceptionMap.put(this, rex);

        setCompletion(EXCEPTIONAL);

    }

    /**

     * Throws the exception associated with status s.

     *

     * @throws the exception

     */

    private void reportException(int s) {

        if ((s &= COMPLETION_MASK) < NORMAL) {

            if (s == CANCELLED)

                throw new CancellationException();

            else

                rethrowException(exceptionMap.get(this));

        }

    }

    /**

     * Returns result or throws exception using j.u.c.Future conventions.

     * Only call when {@code isDone} known to be true or thread known

     * to be interrupted.

     */

    private V reportFutureResult()

        throws InterruptedException, ExecutionException {

        if (Thread.interrupted())

            throw new InterruptedException();

        int s = status & COMPLETION_MASK;

        if (s < NORMAL) {

            Throwable ex;

            if (s == CANCELLED)

                throw new CancellationException();

            if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)

                throw new ExecutionException(ex);

        }

        return getRawResult();

    }

    /**

     * Returns result or throws exception using j.u.c.Future conventions

     * with timeouts.

     */

    private V reportTimedFutureResult()

        throws InterruptedException, ExecutionException, TimeoutException {

        if (Thread.interrupted())

            throw new InterruptedException();

        Throwable ex;

        int s = status & COMPLETION_MASK;

        if (s == NORMAL)

            return getRawResult();

        else if (s == CANCELLED)

            throw new CancellationException();

        else if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)

            throw new ExecutionException(ex);

        else

            throw new TimeoutException();

    }

    // internal execution methods

    /**

     * Calls exec, recording completion, and rethrowing exception if

     * encountered. Caller should normally check status before calling.

     *

     * @return true if completed normally

     */

    private boolean tryExec() {

        try { // try block must contain only call to exec

            if (!exec())

                return false;

        } catch (Throwable rex) {

            setDoneExceptionally(rex);

            rethrowException(rex);

            return false; // not reached

        }

        setNormalCompletion();

        return true;

    }

    /**

     * Main execution method used by worker threads. Invokes

     * base computation unless already complete.

     */

    final void quietlyExec() {

        if (status >= 0) {

            try {

                if (!exec())

                    return;

            } catch (Throwable rex) {

                setDoneExceptionally(rex);

                return;

            }

            setNormalCompletion();

        }

    }

    /**

     * Calls exec(), recording but not rethrowing exception.

     * Caller should normally check status before calling.

     *

     * @return true if completed normally

     */

    private boolean tryQuietlyInvoke() {

        try {

            if (!exec())

                return false;

        } catch (Throwable rex) {

            setDoneExceptionally(rex);

            return false;

        }

        setNormalCompletion();

        return true;

    }

    /**

     * Cancels, ignoring any exceptions it throws.

     */

    final void cancelIgnoringExceptions() {

        try {

            cancel(false);

        } catch (Throwable ignore) {

        }

    }

    /**

     * Main implementation of helpJoin

     */

    private int busyJoin(ForkJoinWorkerThread w) {

        int s;

        ForkJoinTask<?> t;