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)
            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)
            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)
            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)
            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.8
Committed:	Sun Dec 21 15:14:36 2003 UTC (20 years, 5 months ago) by dl
Branch:	MAIN
Changes since 1.7:	+4 -5 lines
Log Message:	Documentation improvements
#	User	Rev	Content
1	dl	1.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	dl	1.7	* 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	dl	1.1	*
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	dl	1.8	* 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	dl	1.1	*
30			* @since 1.5
31			* @author Doug Lea
32			*/
33	tim	1.3	public class ScheduledThreadPoolExecutor
34			extends ThreadPoolExecutor
35			implements ScheduledExecutorService {
36	dl	1.1
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	dl	1.5	private class ScheduledFutureTask<V>
55	dl	1.1	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	dl	1.5	* Overrides FutureTask version so as to reset/requeue if periodic.
140	dl	1.1	*/
141			public void run() {
142	dl	1.5	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	dl	1.1	}
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	dl	1.6	public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
291	dl	1.1	if (command == null)
292			throw new NullPointerException();
293			long triggerTime = System.nanoTime() + unit.toNanos(delay);
294	dl	1.6	ScheduledFutureTask<?> t = new ScheduledFutureTask<Boolean>(command, null, triggerTime);
295	dl	1.1	delayedExecute(t);
296			return t;
297			}
298
299			public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
300			if (callable == 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	dl	1.6	public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
309	dl	1.1	if (command == null)
310			throw new NullPointerException();
311			if (period <= 0)
312			throw new IllegalArgumentException();
313			long triggerTime = System.nanoTime() + unit.toNanos(initialDelay);
314	dl	1.6	ScheduledFutureTask<?> t = new ScheduledFutureTask<Object>
315			(command, null,
316	dl	1.1	triggerTime,
317			unit.toNanos(period),
318			true);
319			delayedExecute(t);
320			return t;
321			}
322
323	dl	1.6	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
324	dl	1.1	if (command == null)
325			throw new NullPointerException();
326			if (delay <= 0)
327			throw new IllegalArgumentException();
328			long triggerTime = System.nanoTime() + unit.toNanos(initialDelay);
329	dl	1.6	ScheduledFutureTask<?> t = new ScheduledFutureTask<Boolean>
330	dl	1.1	(command,
331	dl	1.6	null,
332	dl	1.1	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	dl	1.7	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	dl	1.1
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	dl	1.7	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	dl	1.1	}
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	tim	1.4	public List<Runnable> shutdownNow() {
483	dl	1.1	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	dl	1.7	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	dl	1.1	}
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			}