util/concurrent/ScheduledThreadPoolExecutor.java

/*
 * 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.util.concurrent.atomic.*;
import java.util.*;

/**
 * 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>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
 * <tt>corePoolSize</tt> threads and an unbounded queue, adjustments
 * to <tt>maximumPoolSize</tt> have no useful effect.
 *
 * <p>This class supports protected extension method
 * <tt>decorateTask</tt> (one version each for <tt>Runnable</tt> and
 * <tt>Callable</tt>) that can be used to customize the concrete task
 * types used to execute commands. By default,  
 * a <tt>ScheduledThreadPoolExecutor</tt> uses
 * a task type extending {@link FutureTask}. However, this may
 * be modified or replaced using subclasses of the form:
 * <pre>
 * public class CustomScheduledExecutor extends ScheduledThreadPoolExecutor {
 *
 *    static class CustomTask&lt;V&gt; implements RunnableScheduledFuture&lt;V&gt; { ... }
 *
 *    &lt;V&gt; protected RunnableScheduledFuture&lt;V&gt; decorateTask(
 *                 Runnable r, RunnableScheduledFuture&lt;V&gt; task) {
 *        return new CustomTask&lt;V&gt;(r, task);
 *     }
 *
 *    &lt;V&gt; protected RunnableScheduledFuture&lt;V&gt; decorateTask(
 *                 Callable c, RunnableScheduledFuture&lt;V&gt; task) {
 *        return new CustomTask&lt;V&gt;(c, task);
 *     }
 *     // ... add constructors, etc.
 * }
 * </pre>
 * @since 1.5
 * @author Doug Lea
 */
public class ScheduledThreadPoolExecutor
        extends ThreadPoolExecutor
        implements ScheduledExecutorService {

    /**
     * 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;

    /**
     * Sequence number to break scheduling ties, and in turn to
     * guarantee FIFO order among tied entries.
     */
    private static final AtomicLong sequencer = new AtomicLong(0);

    /** Base of nanosecond timings, to avoid wrapping */
    private static final long NANO_ORIGIN = System.nanoTime();

    /**
     * Returns nanosecond time offset by origin
     */
    final long now() {
        return System.nanoTime() - NANO_ORIGIN;
    }

    private class ScheduledFutureTask<V>
            extends FutureTask<V> implements RunnableScheduledFuture<V> {

        /** Sequence number to break ties FIFO */
        private final long sequenceNumber;
        /** The time the task is enabled to execute in nanoTime units */
        private long time;
        /**
         * Period in nanoseconds for repeating tasks.  A positive
         * value indicates fixed-rate execution.  A negative value
         * indicates fixed-delay execution.  A value of 0 indicates a
         * non-repeating task.
         */
        private final long period;

        /**
         * Creates a one-shot action with given nanoTime-based trigger time
         */
        ScheduledFutureTask(Runnable r, V result, long ns) {
            super(r, result);
            this.time = ns;
            this.period = 0;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        /**
         * Creates a periodic action with given nano time and period
         */
        ScheduledFutureTask(Runnable r, V result, long ns,  long period) {
            super(r, result);
            this.time = ns;
            this.period = period;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        /**
         * Creates a one-shot action with given nanoTime-based trigger
         */
        ScheduledFutureTask(Callable<V> callable, long ns) {
            super(callable);
            this.time = ns;
            this.period = 0;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        public long getDelay(TimeUnit unit) {
            long d = unit.convert(time - now(), TimeUnit.NANOSECONDS);
            return d;
        }

        public int compareTo(Delayed other) {
            if (other == this) // compare zero ONLY if same object
                return 0;
            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;
        }

        /**
         * Returns true if this is a periodic (not a one-shot) action.
         * @return true if periodic
         */
        public boolean isPeriodic() {
            return period != 0;
        }

        /**
         * Run a periodic task
         */
        private void runPeriodic() {
            boolean ok = ScheduledFutureTask.super.runAndReset();
            boolean down = isShutdown();
            // Reschedule if not cancelled and not shutdown or policy allows
            if (ok && (!down ||
                       (getContinueExistingPeriodicTasksAfterShutdownPolicy() &&
                        !isTerminating()))) {
                long p = period;
                if (p > 0)
                    time += p;
                else
                    time = now() - p;
                ScheduledThreadPoolExecutor.super.getQueue().add(this);
            }
            // This might have been the final executed delayed
            // task.  Wake up threads to check.
            else if (down)
                interruptIdleWorkers();
        }

        /**
         * Overrides FutureTask version so as to reset/requeue if periodic.
         */
        public void run() {
            if (isPeriodic())
                runPeriodic();
            else
                ScheduledFutureTask.super.run();
        }
    }

    /**
     * Specialized variant of ThreadPoolExecutor.execute for delayed tasks.
     */
    private void delayedExecute(Runnable command) {
        if (isShutdown()) {
            reject(command);
            return;
        }
        // Prestart a thread if necessary. We cannot prestart it
        // running the task because the task (probably) shouldn't be
        // run yet, so thread will just idle until delay elapses.
        if (getPoolSize() < getCorePoolSize())
            prestartCoreThread();

        super.getQueue().add(command);
    }

    /**
     * Cancels and clears the queue of all tasks that should not be run
     * due to shutdown policy.
     */
    private void cancelUnwantedTasks() {
        boolean keepDelayed = getExecuteExistingDelayedTasksAfterShutdownPolicy();
        boolean keepPeriodic = getContinueExistingPeriodicTasksAfterShutdownPolicy();
        if (!keepDelayed && !keepPeriodic)
            super.getQueue().clear();
        else if (keepDelayed || keepPeriodic) {
            Object[] entries = super.getQueue().toArray();
            for (int i = 0; i < entries.length; ++i) {
                Object e = entries[i];
                if (e instanceof RunnableScheduledFuture) {
                    RunnableScheduledFuture<?> t = (RunnableScheduledFuture<?>)e;
                    if (t.isPeriodic()? !keepPeriodic : !keepDelayed)
                        t.cancel(false);
                }
            }
            entries = null;
            purge();
        }
    }

    public boolean remove(Runnable task) {
        if (!(task instanceof RunnableScheduledFuture))
            return false;
        return getQueue().remove(task);
    }

    /**
     * Modify or replace 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
     * @return a task that can execute the runnable
     * @since 1.6
     */
    protected <V> RunnableScheduledFuture<V> decorateTask(
        Runnable runnable, RunnableScheduledFuture<V> task) {
        return task;
    }

    /**
     * Modify or replace 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
     * @return a task that can execute the callable
     * @since 1.6
     */
    protected <V> RunnableScheduledFuture<V> decorateTask(
        Callable<V> callable, RunnableScheduledFuture<V> task) {
        return task;
    }

    /**
     * Creates a new ScheduledThreadPoolExecutor with the given core
     * pool size.
     *
     * @param corePoolSize the number of threads to keep in the pool,
     * even if they are idle.
     * @throws IllegalArgumentException if corePoolSize less than or
     * equal to zero
     */
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue());
    }

    /**
     * Creates a new ScheduledThreadPoolExecutor with the given
     * initial parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool,
     * even if they are idle.
     * @param threadFactory the factory to use when the executor
     * creates a new thread.
     * @throws NullPointerException if threadFactory is null
     */
    public ScheduledThreadPoolExecutor(int corePoolSize,
                             ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue(), threadFactory);
    }

    /**
     * Creates a new ScheduledThreadPoolExecutor with the given
     * initial parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool,
     * even if they are idle.
     * @param handler the handler to use when execution is blocked
     * because the thread bounds and queue capacities are reached.
     * @throws NullPointerException if handler is null
     */
    public ScheduledThreadPoolExecutor(int corePoolSize,
                              RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue(), handler);
    }

    /**
     * Creates a new ScheduledThreadPoolExecutor with the given
     * initial parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool,
     * even if they are idle.
     * @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 NullPointerException if threadFactory or handler is null
     */
    public ScheduledThreadPoolExecutor(int corePoolSize,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue(), threadFactory, handler);
    }

    public ScheduledFuture<?> schedule(Runnable command,
                                       long delay,
                                       TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        long triggerTime = now() + unit.toNanos(delay);
        RunnableScheduledFuture<?> t = decorateTask(command,
            new ScheduledFutureTask<Boolean>(command, null, triggerTime));
        delayedExecute(t);
        return t;
    }

    public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay,
                                           TimeUnit unit) {
        if (callable == null || unit == null)
            throw new NullPointerException();
        if (delay < 0) delay = 0;
        long triggerTime = now() + unit.toNanos(delay);
        RunnableScheduledFuture<V> t = decorateTask(callable,
            new ScheduledFutureTask<V>(callable, triggerTime));
        delayedExecute(t);
        return t;
    }

    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (period <= 0)
            throw new IllegalArgumentException();
        if (initialDelay < 0) initialDelay = 0;
        long triggerTime = now() + unit.toNanos(initialDelay);
        RunnableScheduledFuture<?> t = decorateTask(command,
            new ScheduledFutureTask<Object>(command,
                                            null,
                                            triggerTime,
                                            unit.toNanos(period)));
        delayedExecute(t);
        return t;
    }

    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        if (delay <= 0)
            throw new IllegalArgumentException();
        if (initialDelay < 0) initialDelay = 0;
        long triggerTime = now() + unit.toNanos(initialDelay);
        RunnableScheduledFuture<?> t = decorateTask(command,
            new ScheduledFutureTask<Boolean>(command,
                                             null,
                                             triggerTime,
                                             unit.toNanos(-delay)));
        delayedExecute(t);
        return t;
    }


    /**
     * Executes command with zero required delay. This has effect
     * equivalent to <tt>schedule(command, 0, anyUnit)</tt>.  Note
     * that inspections of the queue and of the list returned by
     * <tt>shutdownNow</tt> will access the zero-delayed
     * {@link ScheduledFuture}, not the <tt>command</tt> itself.
     *
     * @param command the task to execute
     * @throws RejectedExecutionException at discretion of
     * <tt>RejectedExecutionHandler</tt>, if task cannot be accepted
     * for execution because the executor has been shut down.
     * @throws NullPointerException if command is null
     */
    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        schedule(command, 0, TimeUnit.NANOSECONDS);
    }

    // Override AbstractExecutorService methods

    public Future<?> submit(Runnable task) {
        return schedule(task, 0, TimeUnit.NANOSECONDS);
    }

    public <T> Future<T> submit(Runnable task, T result) {
        return schedule(Executors.callable(task, result),
                        0, TimeUnit.NANOSECONDS);
    }

    public <T> Future<T> submit(Callable<T> task) {
        return schedule(task, 0, TimeUnit.NANOSECONDS);
    }

    /**
     * Sets the policy on whether to continue executing existing periodic
     * tasks even when this executor has been <tt>shutdown</tt>. In
     * this case, these tasks will only terminate upon
     * <tt>shutdownNow</tt>, or after setting the policy to
     * <tt>false</tt> when already shutdown. This value is by default
     * false.
     * @param value if true, continue after shutdown, else don't.
     * @see #getExecuteExistingDelayedTasksAfterShutdownPolicy
     */
    public void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean value) {
        continueExistingPeriodicTasksAfterShutdown = value;
        if (!value && isShutdown())
            cancelUnwantedTasks();
    }

    /**
     * Gets the policy on whether to continue executing existing
     * periodic tasks even when this executor has been
     * <tt>shutdown</tt>. In this case, these tasks will only
     * terminate upon <tt>shutdownNow</tt> or after setting the policy
     * to <tt>false</tt> when already shutdown. This value is by
     * default false.
     * @return true if will continue after shutdown.
     * @see #setContinueExistingPeriodicTasksAfterShutdownPolicy
     */
    public boolean getContinueExistingPeriodicTasksAfterShutdownPolicy() {
        return continueExistingPeriodicTasksAfterShutdown;
    }

    /**
     * Sets the policy on whether to execute existing delayed
     * tasks even when this executor has been <tt>shutdown</tt>. In
     * this case, these tasks will only terminate upon
     * <tt>shutdownNow</tt>, or after setting the policy to
     * <tt>false</tt> when already shutdown. This value is by default
     * true.
     * @param value if true, execute after shutdown, else don't.
     * @see #getExecuteExistingDelayedTasksAfterShutdownPolicy
     */
    public void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean value) {
        executeExistingDelayedTasksAfterShutdown = value;
        if (!value && isShutdown())
            cancelUnwantedTasks();
    }

    /**
     * Gets the policy on whether to execute existing delayed
     * tasks even when this executor has been <tt>shutdown</tt>. In
     * this case, these tasks will only terminate upon
     * <tt>shutdownNow</tt>, or after setting the policy to
     * <tt>false</tt> when already shutdown. This value is by default
     * true.
     * @return true if will execute after shutdown.
     * @see #setExecuteExistingDelayedTasksAfterShutdownPolicy
     */
    public boolean getExecuteExistingDelayedTasksAfterShutdownPolicy() {
        return executeExistingDelayedTasksAfterShutdown;
    }


    /**
     * Initiates an orderly shutdown in which previously submitted
     * tasks are executed, but no new tasks will be accepted. If the
     * <tt>ExecuteExistingDelayedTasksAfterShutdownPolicy</tt> has
     * been set <tt>false</tt>, existing delayed tasks whose delays
     * have not yet elapsed are cancelled. And unless the
     * <tt>ContinueExistingPeriodicTasksAfterShutdownPolicy</tt> has
     * been set <tt>true</tt>, future executions of existing periodic
     * tasks will be cancelled.
     */
    public void shutdown() {
        cancelUnwantedTasks();
        super.shutdown();
    }

    /**
     * Attempts to stop all actively executing tasks, halts the
     * processing of waiting tasks, and returns a list of the tasks that were
     * awaiting execution.
     *
     * <p>There are no guarantees beyond best-effort attempts to stop
     * processing actively executing tasks.  This implementation
     * cancels tasks via {@link Thread#interrupt}, so if any tasks mask or
     * fail to respond to interrupts, they may never terminate.
     *
     * @return list of tasks that never commenced execution.  Each
     * element of this list is a {@link ScheduledFuture},
     * including those tasks submitted using <tt>execute</tt>, which
     * are for scheduling purposes used as the basis of a zero-delay
     * <tt>ScheduledFuture</tt>.
     */
    public List<Runnable> shutdownNow() {
        return super.shutdownNow();
    }

    /**
     * Returns the task queue used by this executor.  Each element of
     * this queue is a {@link ScheduledFuture}, including those
     * tasks submitted using <tt>execute</tt> which are for scheduling
     * purposes used as the basis of a zero-delay
     * <tt>ScheduledFuture</tt>. Iteration over this queue is
     * <em>not</em> guaranteed to traverse tasks in the order in
     * which they will execute.
     *
     * @return the task queue
     */
    public BlockingQueue<Runnable> getQueue() {
        return super.getQueue();
    }

    /**
     * An annoying wrapper class to convince generics compiler to
     * use a DelayQueue<RunnableScheduledFuture> as a BlockingQueue<Runnable>
     */
    private static class DelayedWorkQueue
        extends AbstractCollection<Runnable>
        implements BlockingQueue<Runnable> {

        private final DelayQueue<RunnableScheduledFuture> dq = new DelayQueue<RunnableScheduledFuture>();
        public Runnable poll() { return dq.poll(); }
        public Runnable peek() { return dq.peek(); }
        public Runnable take() throws InterruptedException { return dq.take(); }
        public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
            return dq.poll(timeout, unit);
        }

        public boolean add(Runnable x) { return dq.add((RunnableScheduledFuture)x); }
        public boolean offer(Runnable x) { return dq.offer((RunnableScheduledFuture)x); }
        public void put(Runnable x)  {
            dq.put((RunnableScheduledFuture)x);
        }
        public boolean offer(Runnable x, long timeout, TimeUnit unit) {
            return dq.offer((RunnableScheduledFuture)x, timeout, unit);
        }

        public Runnable remove() { return dq.remove(); }
        public Runnable element() { return dq.element(); }
        public void clear() { dq.clear(); }
        public int drainTo(Collection<? super Runnable> c) { return dq.drainTo(c); }
        public int drainTo(Collection<? super Runnable> c, int maxElements) {
            return dq.drainTo(c, maxElements);
        }

        public int remainingCapacity() { return dq.remainingCapacity(); }
        public boolean remove(Object x) { return dq.remove(x); }
        public boolean contains(Object x) { return dq.contains(x); }
        public int size() { return dq.size(); }
        public boolean isEmpty() { return dq.isEmpty(); }
        public Object[] toArray() { return dq.toArray(); }
        public <T> T[] toArray(T[] array) { return dq.toArray(array); }
        public Iterator<Runnable> iterator() {
            return new Iterator<Runnable>() {
                private Iterator<RunnableScheduledFuture> it = dq.iterator();
                public boolean hasNext() { return it.hasNext(); }
                public Runnable next() { return it.next(); }
                public void remove() {  it.remove(); }
            };
        }
    }
}
Revision:	1.23
Committed:	Tue May 17 13:08:08 2005 UTC (19 years ago) by dl
Branch:	MAIN
Changes since 1.22:	+63 -16 lines
Log Message:	Add documentation; adjust STPE internal types
#	Content
1	/*
2	* Written by Doug Lea with assistance from members of JCP JSR-166
3	* Expert Group and released to the public domain, as explained at
4	* http://creativecommons.org/licenses/publicdomain
5	*/
6
7	package java.util.concurrent;
8	import java.util.concurrent.atomic.*;
9	import java.util.*;
10
11	/**
12	* A {@link ThreadPoolExecutor} that can additionally schedule
13	* commands to run after a given delay, or to execute
14	* periodically. This class is preferable to {@link java.util.Timer}
15	* when multiple worker threads are needed, or when the additional
16	* flexibility or capabilities of {@link ThreadPoolExecutor} (which
17	* this class extends) are required.
18	*
19	* <p> Delayed tasks execute no sooner than they are enabled, but
20	* without any real-time guarantees about when, after they are
21	* enabled, they will commence. Tasks scheduled for exactly the same
22	* execution time are enabled in first-in-first-out (FIFO) order of
23	* submission.
24	*
25	* <p>While this class inherits from {@link ThreadPoolExecutor}, a few
26	* of the inherited tuning methods are not useful for it. In
27	* particular, because it acts as a fixed-sized pool using
28	* <tt>corePoolSize</tt> threads and an unbounded queue, adjustments
29	* to <tt>maximumPoolSize</tt> have no useful effect.
30	*
31	* <p>This class supports protected extension method
32	* <tt>decorateTask</tt> (one version each for <tt>Runnable</tt> and
33	* <tt>Callable</tt>) that can be used to customize the concrete task
34	* types used to execute commands. By default,
35	* a <tt>ScheduledThreadPoolExecutor</tt> uses
36	* a task type extending {@link FutureTask}. However, this may
37	* be modified or replaced using subclasses of the form:
38	* <pre>
39	* public class CustomScheduledExecutor extends ScheduledThreadPoolExecutor {
40	*
41	* static class CustomTask<V> implements RunnableScheduledFuture<V> { ... }
42	*
43	* <V> protected RunnableScheduledFuture<V> decorateTask(
44	* Runnable r, RunnableScheduledFuture<V> task) {
45	* return new CustomTask<V>(r, task);
46	* }
47	*
48	* <V> protected RunnableScheduledFuture<V> decorateTask(
49	* Callable c, RunnableScheduledFuture<V> task) {
50	* return new CustomTask<V>(c, task);
51	* }
52	* // ... add constructors, etc.
53	* }
54	* </pre>
55	* @since 1.5
56	* @author Doug Lea
57	*/
58	public class ScheduledThreadPoolExecutor
59	extends ThreadPoolExecutor
60	implements ScheduledExecutorService {
61
62	/**
63	* False if should cancel/suppress periodic tasks on shutdown.
64	*/
65	private volatile boolean continueExistingPeriodicTasksAfterShutdown;
66
67	/**
68	* False if should cancel non-periodic tasks on shutdown.
69	*/
70	private volatile boolean executeExistingDelayedTasksAfterShutdown = true;
71
72	/**
73	* Sequence number to break scheduling ties, and in turn to
74	* guarantee FIFO order among tied entries.
75	*/
76	private static final AtomicLong sequencer = new AtomicLong(0);
77
78	/** Base of nanosecond timings, to avoid wrapping */
79	private static final long NANO_ORIGIN = System.nanoTime();
80
81	/**
82	* Returns nanosecond time offset by origin
83	*/
84	final long now() {
85	return System.nanoTime() - NANO_ORIGIN;
86	}
87
88	private class ScheduledFutureTask<V>
89	extends FutureTask<V> implements RunnableScheduledFuture<V> {
90
91	/** Sequence number to break ties FIFO */
92	private final long sequenceNumber;
93	/** The time the task is enabled to execute in nanoTime units */
94	private long time;
95	/**
96	* Period in nanoseconds for repeating tasks. A positive
97	* value indicates fixed-rate execution. A negative value
98	* indicates fixed-delay execution. A value of 0 indicates a
99	* non-repeating task.
100	*/
101	private final long period;
102
103	/**
104	* Creates a one-shot action with given nanoTime-based trigger time
105	*/
106	ScheduledFutureTask(Runnable r, V result, long ns) {
107	super(r, result);
108	this.time = ns;
109	this.period = 0;
110	this.sequenceNumber = sequencer.getAndIncrement();
111	}
112
113	/**
114	* Creates a periodic action with given nano time and period
115	*/
116	ScheduledFutureTask(Runnable r, V result, long ns, long period) {
117	super(r, result);
118	this.time = ns;
119	this.period = period;
120	this.sequenceNumber = sequencer.getAndIncrement();
121	}
122
123	/**
124	* Creates a one-shot action with given nanoTime-based trigger
125	*/
126	ScheduledFutureTask(Callable<V> callable, long ns) {
127	super(callable);
128	this.time = ns;
129	this.period = 0;
130	this.sequenceNumber = sequencer.getAndIncrement();
131	}
132
133	public long getDelay(TimeUnit unit) {
134	long d = unit.convert(time - now(), TimeUnit.NANOSECONDS);
135	return d;
136	}
137
138	public int compareTo(Delayed other) {
139	if (other == this) // compare zero ONLY if same object
140	return 0;
141	ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
142	long diff = time - x.time;
143	if (diff < 0)
144	return -1;
145	else if (diff > 0)
146	return 1;
147	else if (sequenceNumber < x.sequenceNumber)
148	return -1;
149	else
150	return 1;
151	}
152
153	/**
154	* Returns true if this is a periodic (not a one-shot) action.
155	* @return true if periodic
156	*/
157	public boolean isPeriodic() {
158	return period != 0;
159	}
160
161	/**
162	* Run a periodic task
163	*/
164	private void runPeriodic() {
165	boolean ok = ScheduledFutureTask.super.runAndReset();
166	boolean down = isShutdown();
167	// Reschedule if not cancelled and not shutdown or policy allows
168	if (ok && (!down \|\|
169	(getContinueExistingPeriodicTasksAfterShutdownPolicy() &&
170	!isTerminating()))) {
171	long p = period;
172	if (p > 0)
173	time += p;
174	else
175	time = now() - p;
176	ScheduledThreadPoolExecutor.super.getQueue().add(this);
177	}
178	// This might have been the final executed delayed
179	// task. Wake up threads to check.
180	else if (down)
181	interruptIdleWorkers();
182	}
183
184	/**
185	* Overrides FutureTask version so as to reset/requeue if periodic.
186	*/
187	public void run() {
188	if (isPeriodic())
189	runPeriodic();
190	else
191	ScheduledFutureTask.super.run();
192	}
193	}
194
195	/**
196	* Specialized variant of ThreadPoolExecutor.execute for delayed tasks.
197	*/
198	private void delayedExecute(Runnable command) {
199	if (isShutdown()) {
200	reject(command);
201	return;
202	}
203	// Prestart a thread if necessary. We cannot prestart it
204	// running the task because the task (probably) shouldn't be
205	// run yet, so thread will just idle until delay elapses.
206	if (getPoolSize() < getCorePoolSize())
207	prestartCoreThread();
208
209	super.getQueue().add(command);
210	}
211
212	/**
213	* Cancels and clears the queue of all tasks that should not be run
214	* due to shutdown policy.
215	*/
216	private void cancelUnwantedTasks() {
217	boolean keepDelayed = getExecuteExistingDelayedTasksAfterShutdownPolicy();
218	boolean keepPeriodic = getContinueExistingPeriodicTasksAfterShutdownPolicy();
219	if (!keepDelayed && !keepPeriodic)
220	super.getQueue().clear();
221	else if (keepDelayed \|\| keepPeriodic) {
222	Object[] entries = super.getQueue().toArray();
223	for (int i = 0; i < entries.length; ++i) {
224	Object e = entries[i];
225	if (e instanceof RunnableScheduledFuture) {
226	RunnableScheduledFuture<?> t = (RunnableScheduledFuture<?>)e;
227	if (t.isPeriodic()? !keepPeriodic : !keepDelayed)
228	t.cancel(false);
229	}
230	}
231	entries = null;
232	purge();
233	}
234	}
235
236	public boolean remove(Runnable task) {
237	if (!(task instanceof RunnableScheduledFuture))
238	return false;
239	return getQueue().remove(task);
240	}
241
242	/**
243	* Modify or replace the task used to execute a runnable.
244	* This method can be used to override the concrete
245	* class used for managing internal tasks.
246	* The default implementation simply returns the given
247	* task.
248	*
249	* @param runnable the submitted Runnable
250	* @param task the task created to execute the runnable
251	* @return a task that can execute the runnable
252	* @since 1.6
253	*/
254	protected <V> RunnableScheduledFuture<V> decorateTask(
255	Runnable runnable, RunnableScheduledFuture<V> task) {
256	return task;
257	}
258
259	/**
260	* Modify or replace the task used to execute a callable.
261	* This method can be used to override the concrete
262	* class used for managing internal tasks.
263	* The default implementation simply returns the given
264	* task.
265	*
266	* @param callable the submitted Callable
267	* @param task the task created to execute the callable
268	* @return a task that can execute the callable
269	* @since 1.6
270	*/
271	protected <V> RunnableScheduledFuture<V> decorateTask(
272	Callable<V> callable, RunnableScheduledFuture<V> task) {
273	return task;
274	}
275
276	/**
277	* Creates a new ScheduledThreadPoolExecutor with the given core
278	* pool size.
279	*
280	* @param corePoolSize the number of threads to keep in the pool,
281	* even if they are idle.
282	* @throws IllegalArgumentException if corePoolSize less than or
283	* equal to zero
284	*/
285	public ScheduledThreadPoolExecutor(int corePoolSize) {
286	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
287	new DelayedWorkQueue());
288	}
289
290	/**
291	* Creates a new ScheduledThreadPoolExecutor with the given
292	* initial parameters.
293	*
294	* @param corePoolSize the number of threads to keep in the pool,
295	* even if they are idle.
296	* @param threadFactory the factory to use when the executor
297	* creates a new thread.
298	* @throws NullPointerException if threadFactory is null
299	*/
300	public ScheduledThreadPoolExecutor(int corePoolSize,
301	ThreadFactory threadFactory) {
302	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
303	new DelayedWorkQueue(), threadFactory);
304	}
305
306	/**
307	* Creates a new ScheduledThreadPoolExecutor with the given
308	* initial parameters.
309	*
310	* @param corePoolSize the number of threads to keep in the pool,
311	* even if they are idle.
312	* @param handler the handler to use when execution is blocked
313	* because the thread bounds and queue capacities are reached.
314	* @throws NullPointerException if handler is null
315	*/
316	public ScheduledThreadPoolExecutor(int corePoolSize,
317	RejectedExecutionHandler handler) {
318	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
319	new DelayedWorkQueue(), handler);
320	}
321
322	/**
323	* Creates a new ScheduledThreadPoolExecutor with the given
324	* initial parameters.
325	*
326	* @param corePoolSize the number of threads to keep in the pool,
327	* even if they are idle.
328	* @param threadFactory the factory to use when the executor
329	* creates a new thread.
330	* @param handler the handler to use when execution is blocked
331	* because the thread bounds and queue capacities are reached.
332	* @throws NullPointerException if threadFactory or handler is null
333	*/
334	public ScheduledThreadPoolExecutor(int corePoolSize,
335	ThreadFactory threadFactory,
336	RejectedExecutionHandler handler) {
337	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
338	new DelayedWorkQueue(), threadFactory, handler);
339	}
340
341	public ScheduledFuture<?> schedule(Runnable command,
342	long delay,
343	TimeUnit unit) {
344	if (command == null \|\| unit == null)
345	throw new NullPointerException();
346	long triggerTime = now() + unit.toNanos(delay);
347	RunnableScheduledFuture<?> t = decorateTask(command,
348	new ScheduledFutureTask<Boolean>(command, null, triggerTime));
349	delayedExecute(t);
350	return t;
351	}
352
353	public <V> ScheduledFuture<V> schedule(Callable<V> callable,
354	long delay,
355	TimeUnit unit) {
356	if (callable == null \|\| unit == null)
357	throw new NullPointerException();
358	if (delay < 0) delay = 0;
359	long triggerTime = now() + unit.toNanos(delay);
360	RunnableScheduledFuture<V> t = decorateTask(callable,
361	new ScheduledFutureTask<V>(callable, triggerTime));
362	delayedExecute(t);
363	return t;
364	}
365
366	public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
367	long initialDelay,
368	long period,
369	TimeUnit unit) {
370	if (command == null \|\| unit == null)
371	throw new NullPointerException();
372	if (period <= 0)
373	throw new IllegalArgumentException();
374	if (initialDelay < 0) initialDelay = 0;
375	long triggerTime = now() + unit.toNanos(initialDelay);
376	RunnableScheduledFuture<?> t = decorateTask(command,
377	new ScheduledFutureTask<Object>(command,
378	null,
379	triggerTime,
380	unit.toNanos(period)));
381	delayedExecute(t);
382	return t;
383	}
384
385	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
386	long initialDelay,
387	long delay,
388	TimeUnit unit) {
389	if (command == null \|\| unit == null)
390	throw new NullPointerException();
391	if (delay <= 0)
392	throw new IllegalArgumentException();
393	if (initialDelay < 0) initialDelay = 0;
394	long triggerTime = now() + unit.toNanos(initialDelay);
395	RunnableScheduledFuture<?> t = decorateTask(command,
396	new ScheduledFutureTask<Boolean>(command,
397	null,
398	triggerTime,
399	unit.toNanos(-delay)));
400	delayedExecute(t);
401	return t;
402	}
403
404
405	/**
406	* Executes command with zero required delay. This has effect
407	* equivalent to <tt>schedule(command, 0, anyUnit)</tt>. Note
408	* that inspections of the queue and of the list returned by
409	* <tt>shutdownNow</tt> will access the zero-delayed
410	* {@link ScheduledFuture}, not the <tt>command</tt> itself.
411	*
412	* @param command the task to execute
413	* @throws RejectedExecutionException at discretion of
414	* <tt>RejectedExecutionHandler</tt>, if task cannot be accepted
415	* for execution because the executor has been shut down.
416	* @throws NullPointerException if command is null
417	*/
418	public void execute(Runnable command) {
419	if (command == null)
420	throw new NullPointerException();
421	schedule(command, 0, TimeUnit.NANOSECONDS);
422	}
423
424	// Override AbstractExecutorService methods
425
426	public Future<?> submit(Runnable task) {
427	return schedule(task, 0, TimeUnit.NANOSECONDS);
428	}
429
430	public <T> Future<T> submit(Runnable task, T result) {
431	return schedule(Executors.callable(task, result),
432	0, TimeUnit.NANOSECONDS);
433	}
434
435	public <T> Future<T> submit(Callable<T> task) {
436	return schedule(task, 0, TimeUnit.NANOSECONDS);
437	}
438
439	/**
440	* Sets the policy on whether to continue executing existing periodic
441	* tasks even when this executor has been <tt>shutdown</tt>. In
442	* this case, these tasks will only terminate upon
443	* <tt>shutdownNow</tt>, or after setting the policy to
444	* <tt>false</tt> when already shutdown. This value is by default
445	* false.
446	* @param value if true, continue after shutdown, else don't.
447	* @see #getExecuteExistingDelayedTasksAfterShutdownPolicy
448	*/
449	public void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean value) {
450	continueExistingPeriodicTasksAfterShutdown = value;
451	if (!value && isShutdown())
452	cancelUnwantedTasks();
453	}
454
455	/**
456	* Gets the policy on whether to continue executing existing
457	* periodic tasks even when this executor has been
458	* <tt>shutdown</tt>. In this case, these tasks will only
459	* terminate upon <tt>shutdownNow</tt> or after setting the policy
460	* to <tt>false</tt> when already shutdown. This value is by
461	* default false.
462	* @return true if will continue after shutdown.
463	* @see #setContinueExistingPeriodicTasksAfterShutdownPolicy
464	*/
465	public boolean getContinueExistingPeriodicTasksAfterShutdownPolicy() {
466	return continueExistingPeriodicTasksAfterShutdown;
467	}
468
469	/**
470	* Sets the policy on whether to execute existing delayed
471	* tasks even when this executor has been <tt>shutdown</tt>. In
472	* this case, these tasks will only terminate upon
473	* <tt>shutdownNow</tt>, or after setting the policy to
474	* <tt>false</tt> when already shutdown. This value is by default
475	* true.
476	* @param value if true, execute after shutdown, else don't.
477	* @see #getExecuteExistingDelayedTasksAfterShutdownPolicy
478	*/
479	public void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean value) {
480	executeExistingDelayedTasksAfterShutdown = value;
481	if (!value && isShutdown())
482	cancelUnwantedTasks();
483	}
484
485	/**
486	* Gets the policy on whether to execute existing delayed
487	* tasks even when this executor has been <tt>shutdown</tt>. In
488	* this case, these tasks will only terminate upon
489	* <tt>shutdownNow</tt>, or after setting the policy to
490	* <tt>false</tt> when already shutdown. This value is by default
491	* true.
492	* @return true if will execute after shutdown.
493	* @see #setExecuteExistingDelayedTasksAfterShutdownPolicy
494	*/
495	public boolean getExecuteExistingDelayedTasksAfterShutdownPolicy() {
496	return executeExistingDelayedTasksAfterShutdown;
497	}
498
499
500	/**
501	* Initiates an orderly shutdown in which previously submitted
502	* tasks are executed, but no new tasks will be accepted. If the
503	* <tt>ExecuteExistingDelayedTasksAfterShutdownPolicy</tt> has
504	* been set <tt>false</tt>, existing delayed tasks whose delays
505	* have not yet elapsed are cancelled. And unless the
506	* <tt>ContinueExistingPeriodicTasksAfterShutdownPolicy</tt> has
507	* been set <tt>true</tt>, future executions of existing periodic
508	* tasks will be cancelled.
509	*/
510	public void shutdown() {
511	cancelUnwantedTasks();
512	super.shutdown();
513	}
514
515	/**
516	* Attempts to stop all actively executing tasks, halts the
517	* processing of waiting tasks, and returns a list of the tasks that were
518	* awaiting execution.
519	*
520	* <p>There are no guarantees beyond best-effort attempts to stop
521	* processing actively executing tasks. This implementation
522	* cancels tasks via {@link Thread#interrupt}, so if any tasks mask or
523	* fail to respond to interrupts, they may never terminate.
524	*
525	* @return list of tasks that never commenced execution. Each
526	* element of this list is a {@link ScheduledFuture},
527	* including those tasks submitted using <tt>execute</tt>, which
528	* are for scheduling purposes used as the basis of a zero-delay
529	* <tt>ScheduledFuture</tt>.
530	*/
531	public List<Runnable> shutdownNow() {
532	return super.shutdownNow();
533	}
534
535	/**
536	* Returns the task queue used by this executor. Each element of
537	* this queue is a {@link ScheduledFuture}, including those
538	* tasks submitted using <tt>execute</tt> which are for scheduling
539	* purposes used as the basis of a zero-delay
540	* <tt>ScheduledFuture</tt>. Iteration over this queue is
541	* <em>not</em> guaranteed to traverse tasks in the order in
542	* which they will execute.
543	*
544	* @return the task queue
545	*/
546	public BlockingQueue<Runnable> getQueue() {
547	return super.getQueue();
548	}
549
550	/**
551	* An annoying wrapper class to convince generics compiler to
552	* use a DelayQueue<RunnableScheduledFuture> as a BlockingQueue<Runnable>
553	*/
554	private static class DelayedWorkQueue
555	extends AbstractCollection<Runnable>
556	implements BlockingQueue<Runnable> {
557
558	private final DelayQueue<RunnableScheduledFuture> dq = new DelayQueue<RunnableScheduledFuture>();
559	public Runnable poll() { return dq.poll(); }
560	public Runnable peek() { return dq.peek(); }
561	public Runnable take() throws InterruptedException { return dq.take(); }
562	public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
563	return dq.poll(timeout, unit);
564	}
565
566	public boolean add(Runnable x) { return dq.add((RunnableScheduledFuture)x); }
567	public boolean offer(Runnable x) { return dq.offer((RunnableScheduledFuture)x); }
568	public void put(Runnable x) {
569	dq.put((RunnableScheduledFuture)x);
570	}
571	public boolean offer(Runnable x, long timeout, TimeUnit unit) {
572	return dq.offer((RunnableScheduledFuture)x, timeout, unit);
573	}
574
575	public Runnable remove() { return dq.remove(); }
576	public Runnable element() { return dq.element(); }
577	public void clear() { dq.clear(); }
578	public int drainTo(Collection<? super Runnable> c) { return dq.drainTo(c); }
579	public int drainTo(Collection<? super Runnable> c, int maxElements) {
580	return dq.drainTo(c, maxElements);
581	}
582
583	public int remainingCapacity() { return dq.remainingCapacity(); }
584	public boolean remove(Object x) { return dq.remove(x); }
585	public boolean contains(Object x) { return dq.contains(x); }
586	public int size() { return dq.size(); }
587	public boolean isEmpty() { return dq.isEmpty(); }
588	public Object[] toArray() { return dq.toArray(); }
589	public <T> T[] toArray(T[] array) { return dq.toArray(array); }
590	public Iterator<Runnable> iterator() {
591	return new Iterator<Runnable>() {
592	private Iterator<RunnableScheduledFuture> it = dq.iterator();
593	public boolean hasNext() { return it.hasNext(); }
594	public Runnable next() { return it.next(); }
595	public void remove() { it.remove(); }
596	};
597	}
598	}
599	}