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
#	User	Rev	Content
1	tim	1.1	/*
2	dl	1.4	* 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	tim	1.1	*/
6
7			package java.util.concurrent;
8	dl	1.4	import java.util.concurrent.atomic.*;
9	dl	1.2	import java.util.*;
10	tim	1.1
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	dl	1.4	/**
28	dl	1.5	* 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	dl	1.4	* A delayed or periodic action.
35			*/
36	dl	1.5	public class DelayedTask extends CancellableTask implements Delayed {
37	dl	1.4	private final long sequenceNumber;
38			private final long time;
39			private final long period;
40
41			/**
42	dl	1.5	* Creates a one-shot action with given nanoTime-based trigger time
43	dl	1.4	*/
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	dl	1.5	* Creates a periodic action with given nano time and period
53	dl	1.4	*/
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	dl	1.5	public boolean cancel(boolean mayInterruptIfRunning) {
82			if (!isDone())
83			ScheduledExecutor.this.remove(this);
84			return super.cancel(mayInterruptIfRunning);
85			}
86
87	dl	1.4	/**
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	dl	1.5	DelayedTask nextTask() {
107	dl	1.4	if (period <= 0 \|\| isCancelled())
108			return null;
109			return new DelayedTask(getRunnable(), time+period, period);
110			}
111	dl	1.2
112	dl	1.4	}
113	dl	1.2
114	dl	1.4	/**
115			* A delayed result-bearing action.
116			*/
117	dl	1.5	public class DelayedFutureTask<V> extends DelayedTask implements Future<V> {
118	dl	1.4	/**
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	dl	1.2	}
126
127	dl	1.4	/**
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	dl	1.2	}
140
141	dl	1.4	public V get(long timeout, TimeUnit unit)
142			throws InterruptedException, ExecutionException, TimeoutException {
143			return ((InnerCancellableFuture<V>)getRunnable()).get(timeout, unit);
144	dl	1.2	}
145	tim	1.1
146	dl	1.4	protected void set(V v) {
147			((InnerCancellableFuture<V>)getRunnable()).set(v);
148	dl	1.2	}
149	tim	1.1
150	dl	1.4	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	dl	1.2	}
168	dl	1.4	public boolean offer(Runnable x) { return dq.offer((DelayedTask)x); }
169			public void put(Runnable x) throws InterruptedException {
170			dq.put((DelayedTask)x);
171	dl	1.2	}
172	dl	1.4	public boolean offer(Runnable x, long timeout, TimeUnit unit) throws InterruptedException {
173			return dq.offer((DelayedTask)x, timeout, unit);
174	dl	1.2	}
175	dl	1.4	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	tim	1.1	}
188	dl	1.4	}
189	tim	1.1
190	dl	1.4	/**
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	tim	1.1
201	dl	1.4	/**
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	tim	1.1	}
214
215	dl	1.4	/**
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	dl	1.2
229	tim	1.1	/**
230	dl	1.2	* 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	dl	1.4	* @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	tim	1.1	*/
239	dl	1.4	public ScheduledExecutor(int corePoolSize,
240			ThreadFactory threadFactory,
241			RejectedExecutionHandler handler) {
242	dl	1.2	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
243	dl	1.4	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	tim	1.1	}
295
296
297	dl	1.4	/**
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	tim	1.1	}
317
318	dl	1.4	/**
319			* Execute command with zero required delay
320			*/
321			public void execute(Runnable command) {
322			schedule(command, 0, TimeUnit.NANOSECONDS);
323	tim	1.1	}
324
325	dl	1.4	/**
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	tim	1.1	}
339	dl	1.4
340