util/concurrent/ScheduledExecutor.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.*;

/**
 * An <tt>Executor</tt> that can schedule command to run after a given
 * delay, or to execute periodically. This class is preferable to
 * <tt>java.util.Timer</tt> when multiple worker threads are needed,
 * or when the additional flexibility or capabilities of
 * <tt>ThreadPoolExecutor</tt> (which this class extends) are
 * required.
 *
 * <p> The <tt>schedule</tt> methods create tasks with various delays
 * and return a task object that can be used to cancel or check
 * execution. The <tt>scheduleAtFixedRate</tt> and
 * <tt>scheduleWithFixedDelay</tt> methods create and execute tasks
 * that run periodically until cancelled.  Commands submitted using
 * the <tt>execute</tt> method are scheduled with a requested delay of
 * zero.
 *
 * <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. An
 * internal {@link DelayQueue} used for scheduling relies on relative
 * delays, which may drift from absolute times (as returned by
 * <tt>System.currentTimeMillis</tt>) over sufficiently long periods.
 *
 * <p>While this class inherits from {@link ThreadPoolExecutor}, a few
 * of the inherited tuning methods are not especially useful for
 * it. In particular, because a <tt>ScheduledExecutor</tt> always 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
 * @see Executors
 *
 * @spec JSR-166
 * @author Doug Lea
 */
public class ScheduledExecutor extends ThreadPoolExecutor {

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

    /**
     * A delayed or periodic action.
     */
    public static class DelayedTask extends CancellableTask implements Delayed {
        /** Sequence number to break ties FIFO */
        private final long sequenceNumber;
        /** The time the task is enabled to execute in nanoTime units */
        private final long time;
        /** The delay forllowing 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
         */
        DelayedTask(Runnable r, long ns) {
            super(r);
            this.time = ns;
            this.period = 0;
            rateBased = false;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        /**
         * Creates a periodic action with given nano time and period
         */
        DelayedTask(Runnable r, long ns,  long period, boolean rateBased) {
            super(r);
            if (period <= 0)
                throw new IllegalArgumentException();
            this.time = ns;
            this.period = period;
            this.rateBased = rateBased;
            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)
                return 0;
            DelayedTask x = (DelayedTask)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
         */
        public boolean isPeriodic() {
            return period > 0;
        }

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

        /**
         * Return a new DelayedTask that will trigger in the period
         * subsequent to current task, or null if non-periodic
         * or canceled.
         */
        DelayedTask nextTask() {
            if (period <= 0 || isCancelled())
                return null;
            long nextTime = period + (rateBased ? time : System.nanoTime());
            return new DelayedTask(getRunnable(), nextTime, period, rateBased);
        }

    }

    /**
     * A delayed result-bearing action.
     */
    public static class DelayedFutureTask<V> extends DelayedTask implements Future<V> {
        /**
         * Creates a Future that may trigger after the given delay.
         */
        DelayedFutureTask(Callable<V> callable, long delay,  TimeUnit unit) {
            // must set after super ctor call to use inner class
            super(null, System.nanoTime() + unit.toNanos(delay));
            setRunnable(new InnerCancellableFuture<V>(callable));
        }

        /**
         * Creates a one-shot action that may trigger after the given date.
         */
        DelayedFutureTask(Callable<V> callable, Date date) {
            super(null, 
                  TimeUnit.MILLISECONDS.toNanos(date.getTime() - 
                                                System.currentTimeMillis()));
            setRunnable(new InnerCancellableFuture<V>(callable));
        }
    
        public V get() throws InterruptedException, ExecutionException {
            return ((InnerCancellableFuture<V>)getRunnable()).get();
        }

        public V get(long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
            return ((InnerCancellableFuture<V>)getRunnable()).get(timeout, unit);
        }

        protected void set(V v) {
            ((InnerCancellableFuture<V>)getRunnable()).set(v);
        }

        protected void setException(Throwable t) {
            ((InnerCancellableFuture<V>)getRunnable()).setException(t);
        }
    }


    /**
     * An annoying wrapper class to convince generics compiler to
     * use a DelayQueue<DelayedTask> as a BlockingQueue<Runnable>
     */ 
    private static class DelayedWorkQueue extends AbstractCollection<Runnable> implements BlockingQueue<Runnable> {
        private final DelayQueue<DelayedTask> dq = new DelayQueue<DelayedTask>();
        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((DelayedTask)x); }
        public boolean offer(Runnable x) { return dq.offer((DelayedTask)x); }
        public void put(Runnable x) throws InterruptedException {
            dq.put((DelayedTask)x); 
        }
        public boolean offer(Runnable x, long timeout, TimeUnit unit) throws InterruptedException {
            return dq.offer((DelayedTask)x, timeout, unit);
        }

        public Runnable remove() { return dq.remove(); }
        public Runnable element() { return dq.element(); }
        public void clear() { dq.clear(); }

        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 Iterator<Runnable> iterator() { 
            return new Iterator<Runnable>() {
                private Iterator<DelayedTask> it = dq.iterator();
                public boolean hasNext() { return it.hasNext(); }
                public Runnable next() { return it.next(); }
                public void remove() {  it.remove(); }
            };
        }
    }

    /**
     * Creates a new ScheduledExecutor with the given initial parameters.
     * 
     * @param corePoolSize the number of threads to keep in the pool,
     * even if they are idle.
     */
    public ScheduledExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue());
    }

    /**
     * Creates a new ScheduledExecutor 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. 
     */
    public ScheduledExecutor(int corePoolSize,
                             ThreadFactory threadFactory) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue(), threadFactory);
    }

    /**
     * Creates a new ScheduledExecutor 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.
     */
    public ScheduledExecutor(int corePoolSize,
                              RejectedExecutionHandler handler) {
        super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
              new DelayedWorkQueue(), handler);
    }

    /**
     * Creates a new ScheduledExecutor 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.
     */
    public ScheduledExecutor(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.
     */
    void delayedExecute(Runnable command) {
        if (isShutdown()) {
            reject(command);
            return;
        }
        // Prestart 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())
            addIfUnderCorePoolSize(null);
            
        getQueue().offer(command);
    }

    /**
     * Creates and executes a one-shot action that becomes enabled after
     * the given delay.
     * @param command the task to execute.
     * @param delay the time from now to delay execution.
     * @param unit the time unit of the delay parameter.
     * @return a handle that can be used to cancel the task.
     */

    public DelayedTask schedule(Runnable command, long delay,  TimeUnit unit) {
        DelayedTask t = new DelayedTask(command, System.nanoTime() + unit.toNanos(delay));
        delayedExecute(t);
        return t;
    }

    /**
     * Creates and executes a one-shot action that becomes enabled
     * after the given date.
     * @param command the task to execute.
     * @param date the time to commence excution.
     * @return a handle that can be used to cancel the task.
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution because the executor has been shut down.
     */
    public DelayedTask schedule(Runnable command, Date date) {
        DelayedTask t = new DelayedTask
            (command, 
             TimeUnit.MILLISECONDS.toNanos(date.getTime() - 
                                           System.currentTimeMillis()));
        delayedExecute(t);
        return t;
    }
    
    /**
     * Creates and executes a periodic action that becomes enabled first
     * after the given initial delay, and subsequently with the given
     * period; that is executions will commence after
     * <tt>initialDelay</tt> then <tt>initialDelay+period</tt>, then
     * <tt>initialDelay + 2 * period</tt>, and so on.
     * @param command the task to execute.
     * @param initialDelay the time to delay first execution.
     * @param period the period between successive executions.
     * @param unit the time unit of the delay and period parameters
     * @return a handle that can be used to cancel the task.
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution because the executor has been shut down.
     */
    public DelayedTask scheduleAtFixedRate(Runnable command, long initialDelay,  long period, TimeUnit unit) {
        DelayedTask t = new DelayedTask
            (command, System.nanoTime() + unit.toNanos(initialDelay),
             unit.toNanos(period), true);
        delayedExecute(t);
        return t;
    }
    
    /**
     * Creates a periodic action that becomes enabled first after the
     * given date, and subsequently with the given period
     * period; that is executions will commence after
     * <tt>initialDate</tt> then <tt>initialDate+period</tt>, then
     * <tt>initialDate + 2 * period</tt>, and so on.
     * @param command the task to execute.
     * @param initialDate the time to delay first execution.
     * @param period the period between commencement of successive
     * executions.
     * @param unit the time unit of the  period parameter.
     * @return a handle that can be used to cancel the task.
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution because the executor has been shut down.
     */
    public DelayedTask scheduleAtFixedRate(Runnable command, Date initialDate, long period, TimeUnit unit) {
        DelayedTask t = new DelayedTask
            (command, 
             TimeUnit.MILLISECONDS.toNanos(initialDate.getTime() - 
                                           System.currentTimeMillis()),
             unit.toNanos(period), true);
        delayedExecute(t);
        return t;
    }

    /**
     * Creates and executes a periodic action that becomes enabled first
     * after the given initial delay, and and subsequently with the
     * given delay between the termination of one execution and the
     * commencement of the next.
     * @param command the task to execute.
     * @param initialDelay the time to delay first execution.
     * @param delay the delay between the termination of one
     * execution and the commencement of the next.
     * @param unit the time unit of the delay and delay parameters
     * @return a handle that can be used to cancel the task.
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution because the executor has been shut down.
     */
    public DelayedTask scheduleWithFixedDelay(Runnable command, long initialDelay,  long delay, TimeUnit unit) {
        DelayedTask t = new DelayedTask
            (command, System.nanoTime() + unit.toNanos(initialDelay),
             unit.toNanos(delay), false);
        delayedExecute(t);
        return t;
    }
    
    /**
     * Creates a periodic action that becomes enabled first after the
     * given date, and subsequently with the given delay between
     * the termination of one execution and the commencement of the
     * next.
     * @param command the task to execute.
     * @param initialDate the time to delay first execution.
     * @param delay the delay between the termination of one
     * execution and the commencement of the next.
     * @param unit the time unit of the  delay parameter.
     * @return a handle that can be used to cancel the task.
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution because the executor has been shut down.
     */
    public DelayedTask scheduleWithFixedDelay(Runnable command, Date initialDate, long delay, TimeUnit unit) {
        DelayedTask t = new DelayedTask
            (command, 
             TimeUnit.MILLISECONDS.toNanos(initialDate.getTime() - 
                                           System.currentTimeMillis()),
             unit.toNanos(delay), false);
        delayedExecute(t);
        return t;
    }


    /**
     * Creates and executes a Future that becomes enabled after the
     * given delay.
     * @param callable the function to execute.
     * @param delay the time from now to delay execution.
     * @param unit the time unit of the delay parameter.
     * @return a Future that can be used to extract result or cancel.
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution because the executor has been shut down.
     */
    public <V> DelayedFutureTask<V> schedule(Callable<V> callable, long delay,  TimeUnit unit) {
        DelayedFutureTask<V> t = new DelayedFutureTask<V>
            (callable, delay, unit);
        delayedExecute(t);
        return t;
    }

    /**
     * Creates and executes a one-shot action that becomes enabled after
     * the given date.
     * @param callable the function to execute.
     * @param date the time to commence excution.
     * @return a Future that can be used to extract result or cancel.
     * @throws RejectedExecutionException if task cannot be scheduled
     * for execution because the executor has been shut down.
     */
    public <V> DelayedFutureTask<V> schedule(Callable<V> callable, Date date) {
        DelayedFutureTask<V> t = new DelayedFutureTask<V>
            (callable, date);
        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
     * <tt>DelayedTask</tt>, 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.
     */
    public void execute(Runnable command) {
        schedule(command, 0, TimeUnit.NANOSECONDS);
    }

    /**
     * Removes this task from internal queue if it is present, thus
     * causing it not to be run if it has not already started.  This
     * method may be useful as one part of a cancellation scheme.
     *
     * @param task the task to remove
     * @return true if the task was removed
     */
    public boolean remove(Runnable task) {
        if (task instanceof DelayedTask && getQueue().remove(task))
            return true;

        // The task might actually have been wrapped as a DelayedTask
        // in execute(), in which case we need to maually traverse
        // looking for it.

        DelayedTask wrap = null;
        Object[] entries = getQueue().toArray();
        for (int i = 0; i < entries.length; ++i) {
            DelayedTask t = (DelayedTask)entries[i];
            Runnable r = t.getRunnable();
            if (task.equals(r)) {
                wrap = t;
                break;
            }
        }
        entries = null;
        return wrap != null && getQueue().remove(wrap);
    }

    /**
     * If executed task was periodic, cause the task for the next
     * period to execute.
     * @param r the task (assumed to be a DelayedTask)
     * @param t the exception
     */
    protected void afterExecute(Runnable r, Throwable t) { 
        if (isShutdown()) 
            return;
        super.afterExecute(r, t);
        DelayedTask d = (DelayedTask)r;
        DelayedTask next = d.nextTask();
        if (next == null) 
            return;
        try {
            delayedExecute(next);
        } catch(RejectedExecutionException ex) {
            // lost race to detect shutdown; ignore
        }
    }
}


Revision:	1.16
Committed:	Mon Aug 11 19:41:03 2003 UTC (20 years, 9 months ago) by dl
Branch:	MAIN
Changes since 1.15:	+5 -1 lines
Log Message:	Clarified execute javadoc
#	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	* An <tt>Executor</tt> that can schedule command to run after a given
13	* delay, or to execute periodically. This class is preferable to
14	* <tt>java.util.Timer</tt> when multiple worker threads are needed,
15	* or when the additional flexibility or capabilities of
16	* <tt>ThreadPoolExecutor</tt> (which this class extends) are
17	* required.
18	*
19	* <p> The <tt>schedule</tt> methods create tasks with various delays
20	* and return a task object that can be used to cancel or check
21	* execution. The <tt>scheduleAtFixedRate</tt> and
22	* <tt>scheduleWithFixedDelay</tt> methods create and execute tasks
23	* that run periodically until cancelled. Commands submitted using
24	* the <tt>execute</tt> method are scheduled with a requested delay of
25	* zero.
26	*
27	* <p> Delayed tasks execute no sooner than they are enabled, but
28	* without any real-time guarantees about when, after they are enabled
29	* they will commence. Tasks tied for the same execution time are
30	* enabled in first-in-first-out (FIFO) order of submission. An
31	* internal {@link DelayQueue} used for scheduling relies on relative
32	* delays, which may drift from absolute times (as returned by
33	* <tt>System.currentTimeMillis</tt>) over sufficiently long periods.
34	*
35	* <p>While this class inherits from {@link ThreadPoolExecutor}, a few
36	* of the inherited tuning methods are not especially useful for
37	* it. In particular, because a <tt>ScheduledExecutor</tt> always acts
38	* as a fixed-sized pool using <tt>corePoolSize</tt> threads and an
39	* unbounded queue, adjustments to <tt>maximumPoolSize</tt> have no
40	* useful effect.
41	*
42	* @since 1.5
43	* @see Executors
44	*
45	* @spec JSR-166
46	* @author Doug Lea
47	*/
48	public class ScheduledExecutor extends ThreadPoolExecutor {
49
50	/**
51	* Sequence number to break scheduling ties, and in turn to
52	* guarantee FIFO order among tied entries.
53	*/
54	private static final AtomicLong sequencer = new AtomicLong(0);
55
56	/**
57	* A delayed or periodic action.
58	*/
59	public static class DelayedTask extends CancellableTask implements Delayed {
60	/** Sequence number to break ties FIFO */
61	private final long sequenceNumber;
62	/** The time the task is enabled to execute in nanoTime units */
63	private final long time;
64	/** The delay forllowing next time, or <= 0 if non-periodic */
65	private final long period;
66	/** true if at fixed rate; false if fixed delay */
67	private final boolean rateBased;
68
69	/**
70	* Creates a one-shot action with given nanoTime-based trigger time
71	*/
72	DelayedTask(Runnable r, long ns) {
73	super(r);
74	this.time = ns;
75	this.period = 0;
76	rateBased = false;
77	this.sequenceNumber = sequencer.getAndIncrement();
78	}
79
80	/**
81	* Creates a periodic action with given nano time and period
82	*/
83	DelayedTask(Runnable r, long ns, long period, boolean rateBased) {
84	super(r);
85	if (period <= 0)
86	throw new IllegalArgumentException();
87	this.time = ns;
88	this.period = period;
89	this.rateBased = rateBased;
90	this.sequenceNumber = sequencer.getAndIncrement();
91	}
92
93
94	public long getDelay(TimeUnit unit) {
95	long d = unit.convert(time - System.nanoTime(),
96	TimeUnit.NANOSECONDS);
97	return d;
98	}
99
100	public int compareTo(Object other) {
101	if (other == this)
102	return 0;
103	DelayedTask x = (DelayedTask)other;
104	long diff = time - x.time;
105	if (diff < 0)
106	return -1;
107	else if (diff > 0)
108	return 1;
109	else if (sequenceNumber < x.sequenceNumber)
110	return -1;
111	else
112	return 1;
113	}
114
115	/**
116	* Return true if this is a periodic (not a one-shot) action.
117	* @return true if periodic
118	*/
119	public boolean isPeriodic() {
120	return period > 0;
121	}
122
123	/**
124	* Returns the period, or zero if non-periodic.
125	*
126	* @return the period
127	*/
128	public long getPeriod(TimeUnit unit) {
129	return unit.convert(period, TimeUnit.NANOSECONDS);
130	}
131
132	/**
133	* Return a new DelayedTask that will trigger in the period
134	* subsequent to current task, or null if non-periodic
135	* or canceled.
136	*/
137	DelayedTask nextTask() {
138	if (period <= 0 \|\| isCancelled())
139	return null;
140	long nextTime = period + (rateBased ? time : System.nanoTime());
141	return new DelayedTask(getRunnable(), nextTime, period, rateBased);
142	}
143
144	}
145
146	/**
147	* A delayed result-bearing action.
148	*/
149	public static class DelayedFutureTask<V> extends DelayedTask implements Future<V> {
150	/**
151	* Creates a Future that may trigger after the given delay.
152	*/
153	DelayedFutureTask(Callable<V> callable, long delay, TimeUnit unit) {
154	// must set after super ctor call to use inner class
155	super(null, System.nanoTime() + unit.toNanos(delay));
156	setRunnable(new InnerCancellableFuture<V>(callable));
157	}
158
159	/**
160	* Creates a one-shot action that may trigger after the given date.
161	*/
162	DelayedFutureTask(Callable<V> callable, Date date) {
163	super(null,
164	TimeUnit.MILLISECONDS.toNanos(date.getTime() -
165	System.currentTimeMillis()));
166	setRunnable(new InnerCancellableFuture<V>(callable));
167	}
168
169	public V get() throws InterruptedException, ExecutionException {
170	return ((InnerCancellableFuture<V>)getRunnable()).get();
171	}
172
173	public V get(long timeout, TimeUnit unit)
174	throws InterruptedException, ExecutionException, TimeoutException {
175	return ((InnerCancellableFuture<V>)getRunnable()).get(timeout, unit);
176	}
177
178	protected void set(V v) {
179	((InnerCancellableFuture<V>)getRunnable()).set(v);
180	}
181
182	protected void setException(Throwable t) {
183	((InnerCancellableFuture<V>)getRunnable()).setException(t);
184	}
185	}
186
187
188	/**
189	* An annoying wrapper class to convince generics compiler to
190	* use a DelayQueue<DelayedTask> as a BlockingQueue<Runnable>
191	*/
192	private static class DelayedWorkQueue extends AbstractCollection<Runnable> implements BlockingQueue<Runnable> {
193	private final DelayQueue<DelayedTask> dq = new DelayQueue<DelayedTask>();
194	public Runnable poll() { return dq.poll(); }
195	public Runnable peek() { return dq.peek(); }
196	public Runnable take() throws InterruptedException { return dq.take(); }
197	public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
198	return dq.poll(timeout, unit);
199	}
200
201	public boolean add(Runnable x) { return dq.add((DelayedTask)x); }
202	public boolean offer(Runnable x) { return dq.offer((DelayedTask)x); }
203	public void put(Runnable x) throws InterruptedException {
204	dq.put((DelayedTask)x);
205	}
206	public boolean offer(Runnable x, long timeout, TimeUnit unit) throws InterruptedException {
207	return dq.offer((DelayedTask)x, timeout, unit);
208	}
209
210	public Runnable remove() { return dq.remove(); }
211	public Runnable element() { return dq.element(); }
212	public void clear() { dq.clear(); }
213
214	public int remainingCapacity() { return dq.remainingCapacity(); }
215	public boolean remove(Object x) { return dq.remove(x); }
216	public boolean contains(Object x) { return dq.contains(x); }
217	public int size() { return dq.size(); }
218	public boolean isEmpty() { return dq.isEmpty(); }
219	public Iterator<Runnable> iterator() {
220	return new Iterator<Runnable>() {
221	private Iterator<DelayedTask> it = dq.iterator();
222	public boolean hasNext() { return it.hasNext(); }
223	public Runnable next() { return it.next(); }
224	public void remove() { it.remove(); }
225	};
226	}
227	}
228
229	/**
230	* Creates a new ScheduledExecutor with the given initial parameters.
231	*
232	* @param corePoolSize the number of threads to keep in the pool,
233	* even if they are idle.
234	*/
235	public ScheduledExecutor(int corePoolSize) {
236	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
237	new DelayedWorkQueue());
238	}
239
240	/**
241	* Creates a new ScheduledExecutor with the given initial parameters.
242	*
243	* @param corePoolSize the number of threads to keep in the pool,
244	* even if they are idle.
245	* @param threadFactory the factory to use when the executor
246	* creates a new thread.
247	*/
248	public ScheduledExecutor(int corePoolSize,
249	ThreadFactory threadFactory) {
250	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
251	new DelayedWorkQueue(), threadFactory);
252	}
253
254	/**
255	* Creates a new ScheduledExecutor 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 handler the handler to use when execution is blocked
260	* because the thread bounds and queue capacities are reached.
261	*/
262	public ScheduledExecutor(int corePoolSize,
263	RejectedExecutionHandler handler) {
264	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
265	new DelayedWorkQueue(), handler);
266	}
267
268	/**
269	* Creates a new ScheduledExecutor with the given initial parameters.
270	*
271	* @param corePoolSize the number of threads to keep in the pool,
272	* even if they are idle.
273	* @param threadFactory the factory to use when the executor
274	* creates a new thread.
275	* @param handler the handler to use when execution is blocked
276	* because the thread bounds and queue capacities are reached.
277	*/
278	public ScheduledExecutor(int corePoolSize,
279	ThreadFactory threadFactory,
280	RejectedExecutionHandler handler) {
281	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
282	new DelayedWorkQueue(), threadFactory, handler);
283	}
284
285	/**
286	* Specialized variant of ThreadPoolExecutor.execute for delayed tasks.
287	*/
288	void delayedExecute(Runnable command) {
289	if (isShutdown()) {
290	reject(command);
291	return;
292	}
293	// Prestart thread if necessary. We cannot prestart it running
294	// the task because the task (probably) shouldn't be run yet,
295	// so thread will just idle until delay elapses.
296	if (getPoolSize() < getCorePoolSize())
297	addIfUnderCorePoolSize(null);
298
299	getQueue().offer(command);
300	}
301
302	/**
303	* Creates and executes a one-shot action that becomes enabled after
304	* the given delay.
305	* @param command the task to execute.
306	* @param delay the time from now to delay execution.
307	* @param unit the time unit of the delay parameter.
308	* @return a handle that can be used to cancel the task.
309	*/
310
311	public DelayedTask schedule(Runnable command, long delay, TimeUnit unit) {
312	DelayedTask t = new DelayedTask(command, System.nanoTime() + unit.toNanos(delay));
313	delayedExecute(t);
314	return t;
315	}
316
317	/**
318	* Creates and executes a one-shot action that becomes enabled
319	* after the given date.
320	* @param command the task to execute.
321	* @param date the time to commence excution.
322	* @return a handle that can be used to cancel the task.
323	* @throws RejectedExecutionException if task cannot be scheduled
324	* for execution because the executor has been shut down.
325	*/
326	public DelayedTask schedule(Runnable command, Date date) {
327	DelayedTask t = new DelayedTask
328	(command,
329	TimeUnit.MILLISECONDS.toNanos(date.getTime() -
330	System.currentTimeMillis()));
331	delayedExecute(t);
332	return t;
333	}
334
335	/**
336	* Creates and executes a periodic action that becomes enabled first
337	* after the given initial delay, and subsequently with the given
338	* period; that is executions will commence after
339	* <tt>initialDelay</tt> then <tt>initialDelay+period</tt>, then
340	* <tt>initialDelay + 2 * period</tt>, and so on.
341	* @param command the task to execute.
342	* @param initialDelay the time to delay first execution.
343	* @param period the period between successive executions.
344	* @param unit the time unit of the delay and period parameters
345	* @return a handle that can be used to cancel the task.
346	* @throws RejectedExecutionException if task cannot be scheduled
347	* for execution because the executor has been shut down.
348	*/
349	public DelayedTask scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
350	DelayedTask t = new DelayedTask
351	(command, System.nanoTime() + unit.toNanos(initialDelay),
352	unit.toNanos(period), true);
353	delayedExecute(t);
354	return t;
355	}
356
357	/**
358	* Creates a periodic action that becomes enabled first after the
359	* given date, and subsequently with the given period
360	* period; that is executions will commence after
361	* <tt>initialDate</tt> then <tt>initialDate+period</tt>, then
362	* <tt>initialDate + 2 * period</tt>, and so on.
363	* @param command the task to execute.
364	* @param initialDate the time to delay first execution.
365	* @param period the period between commencement of successive
366	* executions.
367	* @param unit the time unit of the period parameter.
368	* @return a handle that can be used to cancel the task.
369	* @throws RejectedExecutionException if task cannot be scheduled
370	* for execution because the executor has been shut down.
371	*/
372	public DelayedTask scheduleAtFixedRate(Runnable command, Date initialDate, long period, TimeUnit unit) {
373	DelayedTask t = new DelayedTask
374	(command,
375	TimeUnit.MILLISECONDS.toNanos(initialDate.getTime() -
376	System.currentTimeMillis()),
377	unit.toNanos(period), true);
378	delayedExecute(t);
379	return t;
380	}
381
382	/**
383	* Creates and executes a periodic action that becomes enabled first
384	* after the given initial delay, and and subsequently with the
385	* given delay between the termination of one execution and the
386	* commencement of the next.
387	* @param command the task to execute.
388	* @param initialDelay the time to delay first execution.
389	* @param delay the delay between the termination of one
390	* execution and the commencement of the next.
391	* @param unit the time unit of the delay and delay parameters
392	* @return a handle that can be used to cancel the task.
393	* @throws RejectedExecutionException if task cannot be scheduled
394	* for execution because the executor has been shut down.
395	*/
396	public DelayedTask scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
397	DelayedTask t = new DelayedTask
398	(command, System.nanoTime() + unit.toNanos(initialDelay),
399	unit.toNanos(delay), false);
400	delayedExecute(t);
401	return t;
402	}
403
404	/**
405	* Creates a periodic action that becomes enabled first after the
406	* given date, and subsequently with the given delay between
407	* the termination of one execution and the commencement of the
408	* next.
409	* @param command the task to execute.
410	* @param initialDate the time to delay first execution.
411	* @param delay the delay between the termination of one
412	* execution and the commencement of the next.
413	* @param unit the time unit of the delay parameter.
414	* @return a handle that can be used to cancel the task.
415	* @throws RejectedExecutionException if task cannot be scheduled
416	* for execution because the executor has been shut down.
417	*/
418	public DelayedTask scheduleWithFixedDelay(Runnable command, Date initialDate, long delay, TimeUnit unit) {
419	DelayedTask t = new DelayedTask
420	(command,
421	TimeUnit.MILLISECONDS.toNanos(initialDate.getTime() -
422	System.currentTimeMillis()),
423	unit.toNanos(delay), false);
424	delayedExecute(t);
425	return t;
426	}
427
428
429	/**
430	* Creates and executes a Future that becomes enabled after the
431	* given delay.
432	* @param callable the function to execute.
433	* @param delay the time from now to delay execution.
434	* @param unit the time unit of the delay parameter.
435	* @return a Future that can be used to extract result or cancel.
436	* @throws RejectedExecutionException if task cannot be scheduled
437	* for execution because the executor has been shut down.
438	*/
439	public <V> DelayedFutureTask<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
440	DelayedFutureTask<V> t = new DelayedFutureTask<V>
441	(callable, delay, unit);
442	delayedExecute(t);
443	return t;
444	}
445
446	/**
447	* Creates and executes a one-shot action that becomes enabled after
448	* the given date.
449	* @param callable the function to execute.
450	* @param date the time to commence excution.
451	* @return a Future that can be used to extract result or cancel.
452	* @throws RejectedExecutionException if task cannot be scheduled
453	* for execution because the executor has been shut down.
454	*/
455	public <V> DelayedFutureTask<V> schedule(Callable<V> callable, Date date) {
456	DelayedFutureTask<V> t = new DelayedFutureTask<V>
457	(callable, date);
458	delayedExecute(t);
459	return t;
460	}
461
462	/**
463	* Execute command with zero required delay. This has effect
464	* equivalent to <tt>schedule(command, 0, anyUnit)</tt>. Note
465	* that inspections of the queue and of the list returned by
466	* <tt>shutdownNow</tt> will access the zero-delayed
467	* <tt>DelayedTask</tt>, not the <tt>command</tt> itself.
468	*
469	* @param command the task to execute
470	* @throws RejectedExecutionException at discretion of
471	* <tt>RejectedExecutionHandler</tt>, if task cannot be accepted
472	* for execution because the executor has been shut down.
473	*/
474	public void execute(Runnable command) {
475	schedule(command, 0, TimeUnit.NANOSECONDS);
476	}
477
478	/**
479	* Removes this task from internal queue if it is present, thus
480	* causing it not to be run if it has not already started. This
481	* method may be useful as one part of a cancellation scheme.
482	*
483	* @param task the task to remove
484	* @return true if the task was removed
485	*/
486	public boolean remove(Runnable task) {
487	if (task instanceof DelayedTask && getQueue().remove(task))
488	return true;
489
490	// The task might actually have been wrapped as a DelayedTask
491	// in execute(), in which case we need to maually traverse
492	// looking for it.
493
494	DelayedTask wrap = null;
495	Object[] entries = getQueue().toArray();
496	for (int i = 0; i < entries.length; ++i) {
497	DelayedTask t = (DelayedTask)entries[i];
498	Runnable r = t.getRunnable();
499	if (task.equals(r)) {
500	wrap = t;
501	break;
502	}
503	}
504	entries = null;
505	return wrap != null && getQueue().remove(wrap);
506	}
507
508	/**
509	* If executed task was periodic, cause the task for the next
510	* period to execute.
511	* @param r the task (assumed to be a DelayedTask)
512	* @param t the exception
513	*/
514	protected void afterExecute(Runnable r, Throwable t) {
515	if (isShutdown())
516	return;
517	super.afterExecute(r, t);
518	DelayedTask d = (DelayedTask)r;
519	DelayedTask next = d.nextTask();
520	if (next == null)
521	return;
522	try {
523	delayedExecute(next);
524	} catch(RejectedExecutionException ex) {
525	// lost race to detect shutdown; ignore
526	}
527	}
528	}
529
530