util/concurrent/ScheduledThreadPoolExecutor.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain. Use, modify, and
 * redistribute this code in any way without acknowledgement.
 */

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);
    
    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;
        /** The delay following next time, or <= 0 if non-periodic */
        private final long period;
        /** true if at fixed rate; false if fixed delay */
        private final boolean rateBased; 

        /**
         * 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;
            rateBased = false;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        /**
         * Creates a periodic action with given nano time and period
         */
        ScheduledFutureTask(Runnable r, V result, long ns,  long period, boolean rateBased) {
            super(r, result);
            this.time = ns;
            this.period = period;
            this.rateBased = rateBased;
            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;
            rateBased = false;
            this.sequenceNumber = sequencer.getAndIncrement();
        }


        public long getDelay(TimeUnit unit) {
            long d =  unit.convert(time - System.nanoTime(), 
                                   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;
        }

        /**
         * Returns the period, or zero if non-periodic.
         *
         * @return the period
         */
        long getPeriod(TimeUnit unit) {
            return unit.convert(period, TimeUnit.NANOSECONDS);
        }

        /**
         * Overrides FutureTask version so as to reset/requeue if periodic.
         */ 
        public void run() {
            if (!isPeriodic())
                ScheduledFutureTask.super.run();
            else {
                if (!ScheduledFutureTask.super.runAndReset())
                    return;
                boolean down = isShutdown();
                if (!down ||
                    (getContinueExistingPeriodicTasksAfterShutdownPolicy() && 
                     !isTerminating())) {
                    time = period + (rateBased ? time : System.nanoTime());
                    ScheduledThreadPoolExecutor.super.getQueue().add(this);
                }
                // This might have been the final executed delayed
                // task.  Wake up threads to check.
                else if (down) 
                    interruptIdleWorkers();
            }
        }
    }

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

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

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

    public ScheduledFuture<?> schedule(Runnable command, long delay,  TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        long triggerTime = System.nanoTime() + 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();
        long triggerTime = System.nanoTime() + 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();
        long triggerTime = System.nanoTime() + unit.toNanos(initialDelay);
        ScheduledFutureTask<?> t = new ScheduledFutureTask<Object>
            (command, null,
             triggerTime,
             unit.toNanos(period), 
             true);
        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();
        long triggerTime = System.nanoTime() + unit.toNanos(initialDelay);
        ScheduledFutureTask<?> t = new ScheduledFutureTask<Boolean>
            (command, 
             null,
             triggerTime,
             unit.toNanos(delay), 
             false);
        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);
    }

    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.
     */
    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.
     */
    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.
     */
    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.
     */
    public boolean getExecuteExistingDelayedTasksAfterShutdownPolicy() {
        return executeExistingDelayedTasksAfterShutdown;
    }

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

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

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