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 tied for 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();

    /**
     * Return 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(Object 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;
        }

        /**
         * Return 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);
    }

    /**
     * Cancel and clear 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();
        }
    }

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

    /**
     * Execute 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);
    }

    /**
     * Set 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();
    }

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

    /**
     * Set 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();
    }

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