/*

 * 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 static java.util.concurrent.TimeUnit.MILLISECONDS;

import static java.util.concurrent.TimeUnit.NANOSECONDS;

import java.util.AbstractQueue;

import java.util.Arrays;

import java.util.Collection;

import java.util.Iterator;

import java.util.List;

import java.util.NoSuchElementException;

import java.util.concurrent.atomic.AtomicLong;

import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.ReentrantLock;

/**

 * A {@link ThreadPoolExecutor} that can additionally schedule

 * commands to run after a given delay, or to execute periodically.

 * This class is preferable to {@link java.util.Timer} when multiple

 * worker threads are needed, or when the additional flexibility or

 * capabilities of {@link ThreadPoolExecutor} (which this class

 * extends) are required.

 *

 * <p>Delayed tasks execute no sooner than they are enabled, but

 * without any real-time guarantees about when, after they are

 * enabled, they will commence. Tasks scheduled for exactly the same

 * execution time are enabled in first-in-first-out (FIFO) order of

 * submission.

 *

 * <p>When a submitted task is cancelled before it is run, execution

 * is suppressed.  By default, such a cancelled task is not

 * automatically removed from the work queue until its delay elapses.

 * While this enables further inspection and monitoring, it may also

 * cause unbounded retention of cancelled tasks.  To avoid this, use

 * {@link #setRemoveOnCancelPolicy} to cause tasks to be immediately

 * removed from the work queue at time of cancellation.

 *

 * <p>Successive executions of a periodic task scheduled via

 * {@link #scheduleAtFixedRate scheduleAtFixedRate} or

 * {@link #scheduleWithFixedDelay scheduleWithFixedDelay}

 * do not overlap. While different executions may be performed by

 * different threads, the effects of prior executions

 * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>

 * those of subsequent ones.

 *

 * <p>While this class inherits from {@link ThreadPoolExecutor}, a few

 * of the inherited tuning methods are not useful for it. In

 * particular, because it acts as a fixed-sized pool using

 * {@code corePoolSize} threads and an unbounded queue, adjustments

 * to {@code maximumPoolSize} have no useful effect. Additionally, it

 * is almost never a good idea to set {@code corePoolSize} to zero or

 * use {@code allowCoreThreadTimeOut} because this may leave the pool

 * without threads to handle tasks once they become eligible to run.

 *

 * <p><b>Extension notes:</b> This class overrides the

 * {@link ThreadPoolExecutor#execute(Runnable) execute} and

 * {@link AbstractExecutorService#submit(Runnable) submit}

 * methods to generate internal {@link ScheduledFuture} objects to

 * control per-task delays and scheduling.  To preserve

 * functionality, any further overrides of these methods in

 * subclasses must invoke superclass versions, which effectively

 * disables additional task customization.  However, this class

 * provides alternative protected extension method

 * {@code decorateTask} (one version each for {@code Runnable} and

 * {@code Callable}) that can be used to customize the concrete task

 * types used to execute commands entered via {@code execute},

 * {@code submit}, {@code schedule}, {@code scheduleAtFixedRate},

 * and {@code scheduleWithFixedDelay}.  By default, a

 * {@code ScheduledThreadPoolExecutor} uses a task type extending

 * {@link FutureTask}. However, this may be modified or replaced using

 * subclasses of the form:

 *

 * <pre> {@code

 * public class CustomScheduledExecutor extends ScheduledThreadPoolExecutor {

 *

 *   static class CustomTask<V> implements RunnableScheduledFuture<V> { ... }

 *

 *   protected <V> RunnableScheduledFuture<V> decorateTask(

 *                Runnable r, RunnableScheduledFuture<V> task) {

 *       return new CustomTask<V>(r, task);

 *   }

 *

 *   protected <V> RunnableScheduledFuture<V> decorateTask(

 *                Callable<V> c, RunnableScheduledFuture<V> task) {

 *       return new CustomTask<V>(c, task);

 *   }

 *   // ... add constructors, etc.

 * }}</pre>

 *

 * @since 1.5

 * @author Doug Lea

 */

public class ScheduledThreadPoolExecutor

        extends ThreadPoolExecutor

        implements ScheduledExecutorService {

    /*

     * This class specializes ThreadPoolExecutor implementation by

     *

     * 1. Using a custom task type ScheduledFutureTask, even for tasks

     *    that don't require scheduling because they are submitted

     *    using ExecutorService rather than ScheduledExecutorService

     *    methods, which are treated as tasks with a delay of zero.

     *

     * 2. Using a custom queue (DelayedWorkQueue), a variant of

     *    unbounded DelayQueue. The lack of capacity constraint and

     *    the fact that corePoolSize and maximumPoolSize are

     *    effectively identical simplifies some execution mechanics

     *    (see delayedExecute) compared to ThreadPoolExecutor.

     *

     * 3. Supporting optional run-after-shutdown parameters, which

     *    leads to overrides of shutdown methods to remove and cancel

     *    tasks that should NOT be run after shutdown, as well as

     *    different recheck logic when task (re)submission overlaps

     *    with a shutdown.

     *

     * 4. Task decoration methods to allow interception and

     *    instrumentation, which are needed because subclasses cannot

     *    otherwise override submit methods to get this effect. These

     *    don't have any impact on pool control logic though.

     */

    /**

     * False if should cancel/suppress periodic tasks on shutdown.

     */

    private volatile boolean continueExistingPeriodicTasksAfterShutdown;

    /**

     * False if should cancel non-periodic tasks on shutdown.

     */

    private volatile boolean executeExistingDelayedTasksAfterShutdown = true;

    /**

     * True if ScheduledFutureTask.cancel should remove from queue.

     */

    volatile boolean removeOnCancel;

    /**

     * Sequence number to break scheduling ties, and in turn to

     * guarantee FIFO order among tied entries.

     */

    private static final AtomicLong sequencer = new AtomicLong();

    private class ScheduledFutureTask<V>

            extends FutureTask<V> implements RunnableScheduledFuture<V> {

        /** Sequence number to break ties FIFO */

        private final long sequenceNumber;

        /** The nanoTime-based time when the task is enabled to execute. */

        private volatile long time;

        /**

         * Period for repeating tasks, in nanoseconds.

         * A positive value indicates fixed-rate execution.

         * A negative value indicates fixed-delay execution.

         * A value of 0 indicates a non-repeating (one-shot) task.

         */

        private final long period;

        /** The actual task to be re-enqueued by reExecutePeriodic */

        RunnableScheduledFuture<V> outerTask = this;

        /**

         * Index into delay queue, to support faster cancellation.

         */

        int heapIndex;

        /**

         * Creates a one-shot action with given nanoTime-based trigger time.

         */

        ScheduledFutureTask(Runnable r, V result, long triggerTime,

                            long sequenceNumber) {

            super(r, result);

            this.time = triggerTime;

            this.period = 0;

            this.sequenceNumber = sequenceNumber;

        }

        /**

         * Creates a periodic action with given nanoTime-based initial

         * trigger time and period.

         */

        ScheduledFutureTask(Runnable r, V result, long triggerTime,

                            long period, long sequenceNumber) {

            super(r, result);

            this.time = triggerTime;

            this.period = period;

            this.sequenceNumber = sequenceNumber;

        }

        /**

         * Creates a one-shot action with given nanoTime-based trigger time.

         */

        ScheduledFutureTask(Callable<V> callable, long triggerTime,

                            long sequenceNumber) {

            super(callable);

            this.time = triggerTime;

            this.period = 0;

            this.sequenceNumber = sequenceNumber;

        }

        public long getDelay(TimeUnit unit) {

            return unit.convert(time - System.nanoTime(), NANOSECONDS);

        }

        public int compareTo(Delayed other) {

            if (other == this) // compare zero if same object

                return 0;

            if (other instanceof ScheduledFutureTask) {

                ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;

                long diff = time - x.time;

                if (diff < 0)

                    return -1;

                else if (diff > 0)

                    return 1;

                else if (sequenceNumber < x.sequenceNumber)

                    return -1;

                else

                    return 1;

            }

            long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);

            return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;

        }

        /**

         * Returns {@code true} if this is a periodic (not a one-shot) action.

         *

         * @return {@code true} if periodic

         */

        public boolean isPeriodic() {

            return period != 0;

        }

        /**

         * Sets the next time to run for a periodic task.

         */

        private void setNextRunTime() {

            long p = period;

            if (p > 0)

                time += p;

            else

                time = triggerTime(-p);

        }

        public boolean cancel(boolean mayInterruptIfRunning) {

            // The racy read of heapIndex below is benign:

            // if heapIndex < 0, then OOTA guarantees that we have surely

            // been removed; else we recheck under lock in remove()

            boolean cancelled = super.cancel(mayInterruptIfRunning);

            if (cancelled && removeOnCancel && heapIndex >= 0)

                remove(this);

            return cancelled;

        }

        /**

         * Overrides FutureTask version so as to reset/requeue if periodic.

         */

        public void run() {

            boolean periodic = isPeriodic();

            if (!canRunInCurrentRunState(periodic))

                cancel(false);

            else if (!periodic)

                super.run();

            else if (super.runAndReset()) {

                setNextRunTime();

                reExecutePeriodic(outerTask);

            }

        }

    }

    /**

     * Returns true if can run a task given current run state

     * and run-after-shutdown parameters.

     *

     * @param periodic true if this task periodic, false if delayed

     */

    boolean canRunInCurrentRunState(boolean periodic) {

        return isRunningOrShutdown(periodic ?

                                   continueExistingPeriodicTasksAfterShutdown :

                                   executeExistingDelayedTasksAfterShutdown);

    }

    /**

     * Main execution method for delayed or periodic tasks.  If pool

     * is shut down, rejects the task. Otherwise adds task to queue

     * and starts a thread, if necessary, to run it.  (We cannot

     * prestart the thread to run the task because the task (probably)

     * shouldn't be run yet.)  If the pool is shut down while the task

     * is being added, cancel and remove it if required by state and

     * run-after-shutdown parameters.

     *

     * @param task the task

     */

    private void delayedExecute(RunnableScheduledFuture<?> task) {

        if (isShutdown())

            reject(task);

        else {

            super.getQueue().add(task);

            if (isShutdown() &&

                !canRunInCurrentRunState(task.isPeriodic()) &&

                remove(task))

                task.cancel(false);

            else

                ensurePrestart();

        }

    }

    /**

     * Requeues a periodic task unless current run state precludes it.

     * Same idea as delayedExecute except drops task rather than rejecting.

     *

     * @param task the task

     */

    void reExecutePeriodic(RunnableScheduledFuture<?> task) {

        if (canRunInCurrentRunState(true)) {

            super.getQueue().add(task);

            if (!canRunInCurrentRunState(true) && remove(task))

                task.cancel(false);

            else

                ensurePrestart();

        }

    }

    /**

     * Cancels and clears the queue of all tasks that should not be run

     * due to shutdown policy.  Invoked within super.shutdown.

     */

    @Override void onShutdown() {

        BlockingQueue<Runnable> q = super.getQueue();

        boolean keepDelayed =

            getExecuteExistingDelayedTasksAfterShutdownPolicy();

        boolean keepPeriodic =

            getContinueExistingPeriodicTasksAfterShutdownPolicy();

        if (!keepDelayed && !keepPeriodic) {

            for (Object e : q.toArray())

                if (e instanceof RunnableScheduledFuture<?>)

                    ((RunnableScheduledFuture<?>) e).cancel(false);

            q.clear();

        }

        else {

            // Traverse snapshot to avoid iterator exceptions

            for (Object e : q.toArray()) {

                if (e instanceof RunnableScheduledFuture) {

                    RunnableScheduledFuture<?> t =

                        (RunnableScheduledFuture<?>)e;

                    if ((t.isPeriodic() ? !keepPeriodic : !keepDelayed) ||

                        t.isCancelled()) { // also remove if already cancelled

                        if (q.remove(t))

                            t.cancel(false);

                    }

                }

            }

        }

        tryTerminate();

    }

    /**

     * Modifies or replaces the task used to execute a runnable.

     * This method can be used to override the concrete

     * class used for managing internal tasks.

     * The default implementation simply returns the given task.

     *

     * @param runnable the submitted Runnable

     * @param task the task created to execute the runnable

     * @param <V> the type of the task's result

     * @return a task that can execute the runnable

     * @since 1.6

     */

    protected <V> RunnableScheduledFuture<V> decorateTask(

        Runnable runnable, RunnableScheduledFuture<V> task) {

        return task;

    }

    /**

     * Modifies or replaces the task used to execute a callable.

     * This method can be used to override the concrete

     * class used for managing internal tasks.

     * The default implementation simply returns the given task.

     *

     * @param callable the submitted Callable

     * @param task the task created to execute the callable

     * @param <V> the type of the task's result

     * @return a task that can execute the callable

     * @since 1.6

     */

    protected <V> RunnableScheduledFuture<V> decorateTask(

        Callable<V> callable, RunnableScheduledFuture<V> task) {

        return task;

    }

    /**

     * The default keep-alive time for pool threads.

     *

     * Normally, this value is unused because all pool threads will be

     * core threads, but if a user creates a pool with a corePoolSize

     * of zero (against our advice), we keep a thread alive as long as

     * there are queued tasks.  If the keep alive time is zero (the

     * historic value), we end up hot-spinning in getTask, wasting a

     * CPU.  But on the other hand, if we set the value too high, and

     * users create a one-shot pool which they don't cleanly shutdown,

     * the pool's non-daemon threads will prevent JVM termination.  A

     * small but non-zero value (relative to a JVM's lifetime) seems

     * best.

     */

    private static final long DEFAULT_KEEPALIVE_MILLIS = 10L;

    /**

     * Creates a new {@code ScheduledThreadPoolExecutor} with the

     * given core pool size.

     *

     * @param corePoolSize the number of threads to keep in the pool, even

     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set

     * @throws IllegalArgumentException if {@code corePoolSize < 0}

     */

    public ScheduledThreadPoolExecutor(int corePoolSize) {

        super(corePoolSize, Integer.MAX_VALUE,

              DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,

              new DelayedWorkQueue());

    }

    /**

     * Creates a new {@code ScheduledThreadPoolExecutor} with the

     * given initial parameters.

     *

     * @param corePoolSize the number of threads to keep in the pool, even

     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set

     * @param threadFactory the factory to use when the executor

     *        creates a new thread

     * @throws IllegalArgumentException if {@code corePoolSize < 0}

     * @throws NullPointerException if {@code threadFactory} is null

     */

    public ScheduledThreadPoolExecutor(int corePoolSize,

                                       ThreadFactory threadFactory) {

        super(corePoolSize, Integer.MAX_VALUE,

              DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,

              new DelayedWorkQueue(), threadFactory);

    }

    /**

     * Creates a new {@code ScheduledThreadPoolExecutor} with the

     * given initial parameters.

     *

     * @param corePoolSize the number of threads to keep in the pool, even

     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set

     * @param handler the handler to use when execution is blocked

     *        because the thread bounds and queue capacities are reached

     * @throws IllegalArgumentException if {@code corePoolSize < 0}

     * @throws NullPointerException if {@code handler} is null

     */

    public ScheduledThreadPoolExecutor(int corePoolSize,

                                       RejectedExecutionHandler handler) {

        super(corePoolSize, Integer.MAX_VALUE,

              DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,

              new DelayedWorkQueue(), handler);

    }

    /**

     * Creates a new {@code ScheduledThreadPoolExecutor} with the

     * given initial parameters.

     *

     * @param corePoolSize the number of threads to keep in the pool, even

     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set

     * @param threadFactory the factory to use when the executor

     *        creates a new thread

     * @param handler the handler to use when execution is blocked

     *        because the thread bounds and queue capacities are reached

     * @throws IllegalArgumentException if {@code corePoolSize < 0}

     * @throws NullPointerException if {@code threadFactory} or

     *         {@code handler} is null

     */

    public ScheduledThreadPoolExecutor(int corePoolSize,

                                       ThreadFactory threadFactory,

                                       RejectedExecutionHandler handler) {

        super(corePoolSize, Integer.MAX_VALUE,

              DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,