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 tasks 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>ThreadExecutor</tt> are required.  Tasks submitted using the
 * <tt>execute</tt> method are scheduled as if they had a requested
 * delay of zero. 
 *
 * @since 1.5
 * @see Executors
 *
 * @spec JSR-166
 */
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 class DelayedTask extends CancellableTask implements Delayed {
        private final long sequenceNumber;
        private final long time;
        private final long period;

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

        /**
         * Creates a periodic action with given nano time and period
         */
        DelayedTask(Runnable r, long ns,  long period) {
            super(r);
            if (period <= 0)
                throw new IllegalArgumentException();
            this.time = ns;
            this.period = period;
            this.sequenceNumber = sequencer.getAndIncrement();
        }


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

        public int compareTo(Object other) {
            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;
        }

        public boolean cancel(boolean mayInterruptIfRunning) {
            if (!isDone()) 
                ScheduledExecutor.this.remove(this);
            return super.cancel(mayInterruptIfRunning);
        }

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

        /**
         * Returns the period, or zero if non-periodic
         */
        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;
            return new DelayedTask(getRunnable(), time+period, period);
        }

    }

    /**
     * A delayed result-bearing action.
     */
    public 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, TimeUnit.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 AbstractQueue<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 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 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);
    }

    /**
     * Creates and executes a one-shot action that may trigger after
     * the given delay.
     */

    public DelayedTask schedule(Runnable r, long delay,  TimeUnit unit) {
        DelayedTask t = new DelayedTask(r, TimeUnit.nanoTime() + unit.toNanos(delay));
        super.execute(t);
        return t;
    }

    /**
     * Creates and executes a one-shot action that may trigger after the given date.
     */
    public DelayedTask schedule(Runnable r, Date date) {
        DelayedTask t = new DelayedTask
            (r, 
             TimeUnit.MILLISECONDS.toNanos(date.getTime() - 
                                           System.currentTimeMillis()));
        super.execute(t);
        return t;
    }
    
    /**
     * Creates and executes a periodic action that may trigger first
     * after the given initial delay, and subsequently with the given
     * period.
     */
    public DelayedTask schedule (Runnable r, long initialDelay,  long period, TimeUnit unit) {
        DelayedTask t = new DelayedTask
            (r, TimeUnit.nanoTime() + unit.toNanos(initialDelay),
             unit.toNanos(period));
        super.execute(t);
        return t;
    }
    
    /**
     * Creates a periodic action that may trigger first after the
     * given date, and subsequently with the given period.
     */
    public DelayedTask schedule(Runnable r, Date firstDate, long period, TimeUnit unit) {
        DelayedTask t = new DelayedTask
            (r, 
             TimeUnit.MILLISECONDS.toNanos(firstDate.getTime() - 
                                           System.currentTimeMillis()),
             unit.toNanos(period));
        super.execute(t);
        return t;
    }


    /**
     * Creates and executes a Future that may trigger after the given delay.
     */
    public <V> DelayedFutureTask<V> schedule(Callable<V> callable, long delay,  TimeUnit unit) {
        DelayedFutureTask<V> t = new DelayedFutureTask<V>
            (callable, delay, unit);
        super.execute(t);
        return t;
    }

    /**
     * Creates and executes a one-shot action that may trigger after
     * the given date.
     */
    public <V> DelayedFutureTask<V> schedule(Callable<V> callable, Date date) {
        DelayedFutureTask<V> t = new DelayedFutureTask<V>
            (callable, date);
        super.execute(t);
        return t;
    }

    /**
     * Execute command with zero required delay
     */
    public void execute(Runnable command) {
        schedule(command, 0, TimeUnit.NANOSECONDS);
    }

    /**
     * If executed task was periodic, cause the task for the next
     * period to execute.
     */
    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)
            super.execute(next);
    }
}


Revision:	1.5
Committed:	Wed Jun 4 11:34:20 2003 UTC (21 years ago) by dl
Branch:	MAIN
Changes since 1.4:	+17 -11 lines
Log Message:	Minor documentation updates Remove remove(task) from ExecutorService, Add removablity to ScheduledExecutor tasks, Revert Executors.execute to return FutureTask
#	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 tasks to run after a given delay,
13	* 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>ThreadExecutor</tt> are required. Tasks submitted using the
17	* <tt>execute</tt> method are scheduled as if they had a requested
18	* delay of zero.
19	*
20	* @since 1.5
21	* @see Executors
22	*
23	* @spec JSR-166
24	*/
25	public class ScheduledExecutor extends ThreadPoolExecutor {
26
27	/**
28	* Sequence number to break scheduling ties, and in turn to
29	* guarantee FIFO order among tied entries.
30	*/
31	private static final AtomicLong sequencer = new AtomicLong(0);
32
33	/**
34	* A delayed or periodic action.
35	*/
36	public class DelayedTask extends CancellableTask implements Delayed {
37	private final long sequenceNumber;
38	private final long time;
39	private final long period;
40
41	/**
42	* Creates a one-shot action with given nanoTime-based trigger time
43	*/
44	DelayedTask(Runnable r, long ns) {
45	super(r);
46	this.time = ns;
47	this.period = 0;
48	this.sequenceNumber = sequencer.getAndIncrement();
49	}
50
51	/**
52	* Creates a periodic action with given nano time and period
53	*/
54	DelayedTask(Runnable r, long ns, long period) {
55	super(r);
56	if (period <= 0)
57	throw new IllegalArgumentException();
58	this.time = ns;
59	this.period = period;
60	this.sequenceNumber = sequencer.getAndIncrement();
61	}
62
63
64	public long getDelay(TimeUnit unit) {
65	return unit.convert(time - TimeUnit.nanoTime(), TimeUnit.NANOSECONDS);
66	}
67
68	public int compareTo(Object other) {
69	DelayedTask x = (DelayedTask)other;
70	long diff = time - x.time;
71	if (diff < 0)
72	return -1;
73	else if (diff > 0)
74	return 1;
75	else if (sequenceNumber < x.sequenceNumber)
76	return -1;
77	else
78	return 1;
79	}
80
81	public boolean cancel(boolean mayInterruptIfRunning) {
82	if (!isDone())
83	ScheduledExecutor.this.remove(this);
84	return super.cancel(mayInterruptIfRunning);
85	}
86
87	/**
88	* Return true if this is a periodic (not a one-shot) action.
89	*/
90	public boolean isPeriodic() {
91	return period > 0;
92	}
93
94	/**
95	* Returns the period, or zero if non-periodic
96	*/
97	public long getPeriod(TimeUnit unit) {
98	return unit.convert(period, TimeUnit.NANOSECONDS);
99	}
100
101	/**
102	* Return a new DelayedTask that will trigger in the period
103	* subsequent to current task, or null if non-periodic
104	* or canceled.
105	*/
106	DelayedTask nextTask() {
107	if (period <= 0 \|\| isCancelled())
108	return null;
109	return new DelayedTask(getRunnable(), time+period, period);
110	}
111
112	}
113
114	/**
115	* A delayed result-bearing action.
116	*/
117	public class DelayedFutureTask<V> extends DelayedTask implements Future<V> {
118	/**
119	* Creates a Future that may trigger after the given delay.
120	*/
121	DelayedFutureTask(Callable<V> callable, long delay, TimeUnit unit) {
122	// must set after super ctor call to use inner class
123	super(null, TimeUnit.nanoTime() + unit.toNanos(delay));
124	setRunnable(new InnerCancellableFuture<V>(callable));
125	}
126
127	/**
128	* Creates a one-shot action that may trigger after the given date.
129	*/
130	DelayedFutureTask(Callable<V> callable, Date date) {
131	super(null,
132	TimeUnit.MILLISECONDS.toNanos(date.getTime() -
133	System.currentTimeMillis()));
134	setRunnable(new InnerCancellableFuture<V>(callable));
135	}
136
137	public V get() throws InterruptedException, ExecutionException {
138	return ((InnerCancellableFuture<V>)getRunnable()).get();
139	}
140
141	public V get(long timeout, TimeUnit unit)
142	throws InterruptedException, ExecutionException, TimeoutException {
143	return ((InnerCancellableFuture<V>)getRunnable()).get(timeout, unit);
144	}
145
146	protected void set(V v) {
147	((InnerCancellableFuture<V>)getRunnable()).set(v);
148	}
149
150	protected void setException(Throwable t) {
151	((InnerCancellableFuture<V>)getRunnable()).setException(t);
152	}
153	}
154
155
156	/**
157	* An annoying wrapper class to convince generics compiler to
158	* use a DelayQueue<DelayedTask> as a BlockingQUeue<Runnable>
159	*/
160	private static class DelayedWorkQueue extends AbstractQueue<Runnable> implements BlockingQueue<Runnable> {
161	private final DelayQueue<DelayedTask> dq = new DelayQueue<DelayedTask>();
162	public Runnable poll() { return dq.poll(); }
163	public Runnable peek() { return dq.peek(); }
164	public Runnable take() throws InterruptedException { return dq.take(); }
165	public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
166	return dq.poll(timeout, unit);
167	}
168	public boolean offer(Runnable x) { return dq.offer((DelayedTask)x); }
169	public void put(Runnable x) throws InterruptedException {
170	dq.put((DelayedTask)x);
171	}
172	public boolean offer(Runnable x, long timeout, TimeUnit unit) throws InterruptedException {
173	return dq.offer((DelayedTask)x, timeout, unit);
174	}
175	public int remainingCapacity() { return dq.remainingCapacity(); }
176	public boolean remove(Object x) { return dq.remove(x); }
177	public boolean contains(Object x) { return dq.contains(x); }
178	public int size() { return dq.size(); }
179	public boolean isEmpty() { return dq.isEmpty(); }
180	public Iterator<Runnable> iterator() {
181	return new Iterator<Runnable>() {
182	private Iterator<DelayedTask> it = dq.iterator();
183	public boolean hasNext() { return it.hasNext(); }
184	public Runnable next() { return it.next(); }
185	public void remove() { it.remove(); }
186	};
187	}
188	}
189
190	/**
191	* Creates a new ScheduledExecutor with the given initial parameters.
192	*
193	* @param corePoolSize the number of threads to keep in the pool,
194	* even if they are idle.
195	*/
196	public ScheduledExecutor(int corePoolSize) {
197	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
198	new DelayedWorkQueue());
199	}
200
201	/**
202	* Creates a new ScheduledExecutor with the given initial parameters.
203	*
204	* @param corePoolSize the number of threads to keep in the pool,
205	* even if they are idle.
206	* @param threadFactory the factory to use when the executor
207	* creates a new thread.
208	*/
209	public ScheduledExecutor(int corePoolSize,
210	ThreadFactory threadFactory) {
211	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
212	new DelayedWorkQueue(), threadFactory);
213	}
214
215	/**
216	* Creates a new ScheduledExecutor with the given initial parameters.
217	*
218	* @param corePoolSize the number of threads to keep in the pool,
219	* even if they are idle.
220	* @param handler the handler to use when execution is blocked
221	* because the thread bounds and queue capacities are reached.
222	*/
223	public ScheduledExecutor(int corePoolSize,
224	RejectedExecutionHandler handler) {
225	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
226	new DelayedWorkQueue(), handler);
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	* @param threadFactory the factory to use when the executor
235	* creates a new thread.
236	* @param handler the handler to use when execution is blocked
237	* because the thread bounds and queue capacities are reached.
238	*/
239	public ScheduledExecutor(int corePoolSize,
240	ThreadFactory threadFactory,
241	RejectedExecutionHandler handler) {
242	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
243	new DelayedWorkQueue(), threadFactory, handler);
244	}
245
246	/**
247	* Creates and executes a one-shot action that may trigger after
248	* the given delay.
249	*/
250
251	public DelayedTask schedule(Runnable r, long delay, TimeUnit unit) {
252	DelayedTask t = new DelayedTask(r, TimeUnit.nanoTime() + unit.toNanos(delay));
253	super.execute(t);
254	return t;
255	}
256
257	/**
258	* Creates and executes a one-shot action that may trigger after the given date.
259	*/
260	public DelayedTask schedule(Runnable r, Date date) {
261	DelayedTask t = new DelayedTask
262	(r,
263	TimeUnit.MILLISECONDS.toNanos(date.getTime() -
264	System.currentTimeMillis()));
265	super.execute(t);
266	return t;
267	}
268
269	/**
270	* Creates and executes a periodic action that may trigger first
271	* after the given initial delay, and subsequently with the given
272	* period.
273	*/
274	public DelayedTask schedule (Runnable r, long initialDelay, long period, TimeUnit unit) {
275	DelayedTask t = new DelayedTask
276	(r, TimeUnit.nanoTime() + unit.toNanos(initialDelay),
277	unit.toNanos(period));
278	super.execute(t);
279	return t;
280	}
281
282	/**
283	* Creates a periodic action that may trigger first after the
284	* given date, and subsequently with the given period.
285	*/
286	public DelayedTask schedule(Runnable r, Date firstDate, long period, TimeUnit unit) {
287	DelayedTask t = new DelayedTask
288	(r,
289	TimeUnit.MILLISECONDS.toNanos(firstDate.getTime() -
290	System.currentTimeMillis()),
291	unit.toNanos(period));
292	super.execute(t);
293	return t;
294	}
295
296
297	/**
298	* Creates and executes a Future that may trigger after the given delay.
299	*/
300	public <V> DelayedFutureTask<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
301	DelayedFutureTask<V> t = new DelayedFutureTask<V>
302	(callable, delay, unit);
303	super.execute(t);
304	return t;
305	}
306
307	/**
308	* Creates and executes a one-shot action that may trigger after
309	* the given date.
310	*/
311	public <V> DelayedFutureTask<V> schedule(Callable<V> callable, Date date) {
312	DelayedFutureTask<V> t = new DelayedFutureTask<V>
313	(callable, date);
314	super.execute(t);
315	return t;
316	}
317
318	/**
319	* Execute command with zero required delay
320	*/
321	public void execute(Runnable command) {
322	schedule(command, 0, TimeUnit.NANOSECONDS);
323	}
324
325	/**
326	* If executed task was periodic, cause the task for the next
327	* period to execute.
328	*/
329	protected void afterExecute(Runnable r, Throwable t) {
330	if (isShutdown())
331	return;
332	super.afterExecute(r, t);
333	DelayedTask d = (DelayedTask)r;
334	DelayedTask next = d.nextTask();
335	if (next != null)
336	super.execute(next);
337	}
338	}
339
340