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.
 *
 * @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 ScheduledFuture<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
         */
        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 ScheduledFutureTask) {
                    ScheduledFutureTask<?> t = (ScheduledFutureTask<?>)e;
                    if (t.isPeriodic()? !keepPeriodic : !keepDelayed)
                        t.cancel(false);
                }
            }
            entries = null;
            purge();
        }
    }

    public boolean remove(Runnable task) {
        if (!(task instanceof ScheduledFutureTask))
            return false;
        return getQueue().remove(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);
        ScheduledFutureTask<?> t =
            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);
        ScheduledFutureTask<V> t =
            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);
        ScheduledFutureTask<?> t =
            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);
        ScheduledFutureTask<?> t =
            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<ScheduledFutureTask> as a BlockingQueue<Runnable>
     */
    private static class DelayedWorkQueue
        extends AbstractCollection<Runnable>
        implements BlockingQueue<Runnable> {

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