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

        /**
         * Overrides FutureTask version so as to reset/requeue if periodic.
         */ 
        public void run() {
            if (!isPeriodic())
                ScheduledFutureTask.super.run();
            else {
                ScheduledFutureTask.super.runAndReset();
                if (isCancelled())
                    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();
    }
            
    /**
     * 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 ScheduledFuture) 
            return super.remove(task);

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

        ScheduledFuture wrap = null;
        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;
                Object r = t.getTask();
                if (task.equals(r)) {
                    wrap = t;
                    break;
                }
            }
        }
        entries = null;
        return wrap != null && super.getQueue().remove(wrap);
    }


    /**
     * 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.9
Committed:	Mon Dec 22 00:48:14 2003 UTC (20 years, 5 months ago) by dl
Branch:	MAIN
Changes since 1.8:	+4 -4 lines
Log Message:	Streamline status settting
#	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	public 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	public 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	ScheduledFutureTask.super.runAndReset();
146	if (isCancelled())
147	return;
148	boolean down = isShutdown();
149	if (!down \|\|
150	(getContinueExistingPeriodicTasksAfterShutdownPolicy() &&
151	!isTerminating())) {
152	time = period + (rateBased ? time : System.nanoTime());
153	ScheduledThreadPoolExecutor.super.getQueue().add(this);
154	}
155	// This might have been the final executed delayed
156	// task. Wake up threads to check.
157	else if (down)
158	interruptIdleWorkers();
159	}
160	}
161	}
162
163	/**
164	* An annoying wrapper class to convince generics compiler to
165	* use a DelayQueue<ScheduledFutureTask> as a BlockingQueue<Runnable>
166	*/
167	private static class DelayedWorkQueue
168	extends AbstractCollection<Runnable> implements BlockingQueue<Runnable> {
169
170	private final DelayQueue<ScheduledFutureTask> dq = new DelayQueue<ScheduledFutureTask>();
171	public Runnable poll() { return dq.poll(); }
172	public Runnable peek() { return dq.peek(); }
173	public Runnable take() throws InterruptedException { return dq.take(); }
174	public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
175	return dq.poll(timeout, unit);
176	}
177
178	public boolean add(Runnable x) { return dq.add((ScheduledFutureTask)x); }
179	public boolean offer(Runnable x) { return dq.offer((ScheduledFutureTask)x); }
180	public void put(Runnable x) {
181	dq.put((ScheduledFutureTask)x);
182	}
183	public boolean offer(Runnable x, long timeout, TimeUnit unit) {
184	return dq.offer((ScheduledFutureTask)x, timeout, unit);
185	}
186
187	public Runnable remove() { return dq.remove(); }
188	public Runnable element() { return dq.element(); }
189	public void clear() { dq.clear(); }
190	public int drainTo(Collection<? super Runnable> c) { return dq.drainTo(c); }
191	public int drainTo(Collection<? super Runnable> c, int maxElements) {
192	return dq.drainTo(c, maxElements);
193	}
194
195	public int remainingCapacity() { return dq.remainingCapacity(); }
196	public boolean remove(Object x) { return dq.remove(x); }
197	public boolean contains(Object x) { return dq.contains(x); }
198	public int size() { return dq.size(); }
199	public boolean isEmpty() { return dq.isEmpty(); }
200	public Object[] toArray() { return dq.toArray(); }
201	public <T> T[] toArray(T[] array) { return dq.toArray(array); }
202	public Iterator<Runnable> iterator() {
203	return new Iterator<Runnable>() {
204	private Iterator<ScheduledFutureTask> it = dq.iterator();
205	public boolean hasNext() { return it.hasNext(); }
206	public Runnable next() { return it.next(); }
207	public void remove() { it.remove(); }
208	};
209	}
210	}
211
212	/**
213	* Creates a new ScheduledThreadPoolExecutor with the given core pool size.
214	*
215	* @param corePoolSize the number of threads to keep in the pool,
216	* even if they are idle.
217	* @throws IllegalArgumentException if corePoolSize less than or
218	* equal to zero
219	*/
220	public ScheduledThreadPoolExecutor(int corePoolSize) {
221	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
222	new DelayedWorkQueue());
223	}
224
225	/**
226	* Creates a new ScheduledThreadPoolExecutor with the given initial parameters.
227	*
228	* @param corePoolSize the number of threads to keep in the pool,
229	* even if they are idle.
230	* @param threadFactory the factory to use when the executor
231	* creates a new thread.
232	* @throws NullPointerException if threadFactory is null
233	*/
234	public ScheduledThreadPoolExecutor(int corePoolSize,
235	ThreadFactory threadFactory) {
236	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
237	new DelayedWorkQueue(), threadFactory);
238	}
239
240	/**
241	* Creates a new ScheduledThreadPoolExecutor 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 handler the handler to use when execution is blocked
246	* because the thread bounds and queue capacities are reached.
247	* @throws NullPointerException if handler is null
248	*/
249	public ScheduledThreadPoolExecutor(int corePoolSize,
250	RejectedExecutionHandler handler) {
251	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
252	new DelayedWorkQueue(), handler);
253	}
254
255	/**
256	* Creates a new ScheduledThreadPoolExecutor with the given initial parameters.
257	*
258	* @param corePoolSize the number of threads to keep in the pool,
259	* even if they are idle.
260	* @param threadFactory the factory to use when the executor
261	* creates a new thread.
262	* @param handler the handler to use when execution is blocked
263	* because the thread bounds and queue capacities are reached.
264	* @throws NullPointerException if threadFactory or handler is null
265	*/
266	public ScheduledThreadPoolExecutor(int corePoolSize,
267	ThreadFactory threadFactory,
268	RejectedExecutionHandler handler) {
269	super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
270	new DelayedWorkQueue(), threadFactory, handler);
271	}
272
273	/**
274	* Specialized variant of ThreadPoolExecutor.execute for delayed tasks.
275	*/
276	private void delayedExecute(Runnable command) {
277	if (isShutdown()) {
278	reject(command);
279	return;
280	}
281	// Prestart a thread if necessary. We cannot prestart it
282	// running the task because the task (probably) shouldn't be
283	// run yet, so thread will just idle until delay elapses.
284	if (getPoolSize() < getCorePoolSize())
285	prestartCoreThread();
286
287	super.getQueue().add(command);
288	}
289
290	public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
291	if (command == null \|\| unit == null)
292	throw new NullPointerException();
293	long triggerTime = System.nanoTime() + unit.toNanos(delay);
294	ScheduledFutureTask<?> t = new ScheduledFutureTask<Boolean>(command, null, triggerTime);
295	delayedExecute(t);
296	return t;
297	}
298
299	public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
300	if (callable == null \|\| unit == null)
301	throw new NullPointerException();
302	long triggerTime = System.nanoTime() + unit.toNanos(delay);
303	ScheduledFutureTask<V> t = new ScheduledFutureTask<V>(callable, triggerTime);
304	delayedExecute(t);
305	return t;
306	}
307
308	public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
309	if (command == null \|\| unit == null)
310	throw new NullPointerException();
311	if (period <= 0)
312	throw new IllegalArgumentException();
313	long triggerTime = System.nanoTime() + unit.toNanos(initialDelay);
314	ScheduledFutureTask<?> t = new ScheduledFutureTask<Object>
315	(command, null,
316	triggerTime,
317	unit.toNanos(period),
318	true);
319	delayedExecute(t);
320	return t;
321	}
322
323	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
324	if (command == null \|\| unit == null)
325	throw new NullPointerException();
326	if (delay <= 0)
327	throw new IllegalArgumentException();
328	long triggerTime = System.nanoTime() + unit.toNanos(initialDelay);
329	ScheduledFutureTask<?> t = new ScheduledFutureTask<Boolean>
330	(command,
331	null,
332	triggerTime,
333	unit.toNanos(delay),
334	false);
335	delayedExecute(t);
336	return t;
337	}
338
339
340	/**
341	* Execute command with zero required delay. This has effect
342	* equivalent to <tt>schedule(command, 0, anyUnit)</tt>. Note
343	* that inspections of the queue and of the list returned by
344	* <tt>shutdownNow</tt> will access the zero-delayed
345	* {@link ScheduledFuture}, not the <tt>command</tt> itself.
346	*
347	* @param command the task to execute
348	* @throws RejectedExecutionException at discretion of
349	* <tt>RejectedExecutionHandler</tt>, if task cannot be accepted
350	* for execution because the executor has been shut down.
351	* @throws NullPointerException if command is null
352	*/
353	public void execute(Runnable command) {
354	if (command == null)
355	throw new NullPointerException();
356	schedule(command, 0, TimeUnit.NANOSECONDS);
357	}
358
359	public Future<?> submit(Runnable task) {
360	return schedule(task, 0, TimeUnit.NANOSECONDS);
361	}
362
363	public <T> Future<T> submit(Runnable task, T result) {
364	return schedule(Executors.callable(task, result), 0, TimeUnit.NANOSECONDS);
365	}
366
367	public <T> Future<T> submit(Callable<T> task) {
368	return schedule(task, 0, TimeUnit.NANOSECONDS);
369	}
370
371	/**
372	* Set policy on whether to continue executing existing periodic
373	* tasks even when this executor has been <tt>shutdown</tt>. In
374	* this case, these tasks will only terminate upon
375	* <tt>shutdownNow</tt>, or after setting the policy to
376	* <tt>false</tt> when already shutdown. This value is by default
377	* false.
378	* @param value if true, continue after shutdown, else don't.
379	*/
380	public void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean value) {
381	continueExistingPeriodicTasksAfterShutdown = value;
382	if (!value && isShutdown())
383	cancelUnwantedTasks();
384	}
385
386	/**
387	* Get the policy on whether to continue executing existing
388	* periodic tasks even when this executor has been
389	* <tt>shutdown</tt>. In this case, these tasks will only
390	* terminate upon <tt>shutdownNow</tt> or after setting the policy
391	* to <tt>false</tt> when already shutdown. This value is by
392	* default false.
393	* @return true if will continue after shutdown.
394	*/
395	public boolean getContinueExistingPeriodicTasksAfterShutdownPolicy() {
396	return continueExistingPeriodicTasksAfterShutdown;
397	}
398
399	/**
400	* Set policy on whether to execute existing delayed
401	* tasks even when this executor has been <tt>shutdown</tt>. In
402	* this case, these tasks will only terminate upon
403	* <tt>shutdownNow</tt>, or after setting the policy to
404	* <tt>false</tt> when already shutdown. This value is by default
405	* true.
406	* @param value if true, execute after shutdown, else don't.
407	*/
408	public void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean value) {
409	executeExistingDelayedTasksAfterShutdown = value;
410	if (!value && isShutdown())
411	cancelUnwantedTasks();
412	}
413
414	/**
415	* Get policy on whether to execute existing delayed
416	* tasks even when this executor has been <tt>shutdown</tt>. In
417	* this case, these tasks will only terminate upon
418	* <tt>shutdownNow</tt>, or after setting the policy to
419	* <tt>false</tt> when already shutdown. This value is by default
420	* true.
421	* @return true if will execute after shutdown.
422	*/
423	public boolean getExecuteExistingDelayedTasksAfterShutdownPolicy() {
424	return executeExistingDelayedTasksAfterShutdown;
425	}
426
427	/**
428	* Cancel and clear the queue of all tasks that should not be run
429	* due to shutdown policy.
430	*/
431	private void cancelUnwantedTasks() {
432	boolean keepDelayed = getExecuteExistingDelayedTasksAfterShutdownPolicy();
433	boolean keepPeriodic = getContinueExistingPeriodicTasksAfterShutdownPolicy();
434	if (!keepDelayed && !keepPeriodic)
435	super.getQueue().clear();
436	else if (keepDelayed \|\| keepPeriodic) {
437	Object[] entries = super.getQueue().toArray();
438	for (int i = 0; i < entries.length; ++i) {
439	Object e = entries[i];
440	if (e instanceof ScheduledFutureTask) {
441	ScheduledFutureTask<?> t = (ScheduledFutureTask<?>)e;
442	if (t.isPeriodic()? !keepPeriodic : !keepDelayed)
443	t.cancel(false);
444	}
445	}
446	entries = null;
447	purge();
448	}
449	}
450
451	/**
452	* Initiates an orderly shutdown in which previously submitted
453	* tasks are executed, but no new tasks will be accepted. If the
454	* <tt>ExecuteExistingDelayedTasksAfterShutdownPolicy</tt> has
455	* been set <tt>false</tt>, existing delayed tasks whose delays
456	* have not yet elapsed are cancelled. And unless the
457	* <tt>ContinueExistingPeriodicTasksAfterShutdownPolicy</tt> has
458	* been set <tt>true</tt>, future executions of existing periodic
459	* tasks will be cancelled.
460	*/
461	public void shutdown() {
462	cancelUnwantedTasks();
463	super.shutdown();
464	}
465
466	/**
467	* Attempts to stop all actively executing tasks, halts the
468	* processing of waiting tasks, and returns a list of the tasks that were
469	* awaiting execution.
470	*
471	* <p>There are no guarantees beyond best-effort attempts to stop
472	* processing actively executing tasks. This implementations
473	* cancels via {@link Thread#interrupt}, so if any tasks mask or
474	* fail to respond to interrupts, they may never terminate.
475	*
476	* @return list of tasks that never commenced execution. Each
477	* element of this list is a {@link ScheduledFuture},
478	* including those tasks submitted using <tt>execute</tt> which
479	* are for scheduling purposes used as the basis of a zero-delay
480	* <tt>ScheduledFuture</tt>.
481	*/
482	public List<Runnable> shutdownNow() {
483	return super.shutdownNow();
484	}
485
486	/**
487	* Removes this task from internal queue if it is present, thus
488	* causing it not to be run if it has not already started. This
489	* method may be useful as one part of a cancellation scheme.
490	*
491	* @param task the task to remove
492	* @return true if the task was removed
493	*/
494	public boolean remove(Runnable task) {
495	if (task instanceof ScheduledFuture)
496	return super.remove(task);
497
498	// The task might actually have been wrapped as a ScheduledFuture
499	// in execute(), in which case we need to manually traverse
500	// looking for it.
501
502	ScheduledFuture wrap = null;
503	Object[] entries = super.getQueue().toArray();
504	for (int i = 0; i < entries.length; ++i) {
505	Object e = entries[i];
506	if (e instanceof ScheduledFutureTask) {
507	ScheduledFutureTask<?> t = (ScheduledFutureTask<?>)e;
508	Object r = t.getTask();
509	if (task.equals(r)) {
510	wrap = t;
511	break;
512	}
513	}
514	}
515	entries = null;
516	return wrap != null && super.getQueue().remove(wrap);
517	}
518
519
520	/**
521	* Returns the task queue used by this executor. Each element of
522	* this queue is a {@link ScheduledFuture}, including those
523	* tasks submitted using <tt>execute</tt> which are for scheduling
524	* purposes used as the basis of a zero-delay
525	* <tt>ScheduledFuture</tt>. Iteration over this queue is
526	* </em>not</em> guaranteed to traverse tasks in the order in
527	* which they will execute.
528	*
529	* @return the task queue
530	*/
531	public BlockingQueue<Runnable> getQueue() {
532	return super.getQueue();
533	}
534
535	}