util/concurrent/ScheduledThreadPoolExecutor.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/licenses/publicdomain
 */

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 scheduled for exactly 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);

    /** Base of nanosecond timings, to avoid wrapping */
    private static final long NANO_ORIGIN = System.nanoTime();

    /**
     * Returns nanosecond time offset by origin
     */
    final long now() {
        return System.nanoTime() - NANO_ORIGIN;
    }

    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;
        /**
         * Period in nanoseconds for repeating tasks.  A positive
         * value indicates fixed-rate execution.  A negative value
         * indicates fixed-delay execution.  A value of 0 indicates a
         * non-repeating task.
         */
        private final long period;

        /**
         * 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;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        /**
         * Creates a periodic action with given nano time and period
         */
        ScheduledFutureTask(Runnable r, V result, long ns,  long period) {
            super(r, result);
            this.time = ns;
            this.period = period;
            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;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

        public long getDelay(TimeUnit unit) {
            long d = unit.convert(time - now(), 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;
        }

        /**
         * Returns true if this is a periodic (not a one-shot) action.
         * @return true if periodic
         */
        boolean isPeriodic() {
            return period != 0;
        }

        /**
         * Run a periodic task
         */
        private void runPeriodic() {
            boolean ok = ScheduledFutureTask.super.runAndReset();
            boolean down = isShutdown();
            // Reschedule if not cancelled and not shutdown or policy allows
            if (ok && (!down ||
                       (getContinueExistingPeriodicTasksAfterShutdownPolicy() && 
                        !isTerminating()))) {
                long p = period;
                if (p > 0)
                    time += p;
                else
                    time = now() - p;
                ScheduledThreadPoolExecutor.super.getQueue().add(this);
            }
            // This might have been the final executed delayed
            // task.  Wake up threads to check.
            else if (down) 
                interruptIdleWorkers();
        }

        /**
         * Overrides FutureTask version so as to reset/requeue if periodic.
         */ 
        public void run() {
            if (isPeriodic())
                runPeriodic();
            else 
                ScheduledFutureTask.super.run();
        }
    }

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

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

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

    public ScheduledFuture<?> schedule(Runnable command, 
                                       long delay, 
                                       TimeUnit unit) {
        if (command == null || unit == null)
            throw new NullPointerException();
        long triggerTime = now() + 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();
        if (delay < 0) delay = 0;
        long triggerTime = now() + 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();
        if (initialDelay < 0) initialDelay = 0;
        long triggerTime = now() + unit.toNanos(initialDelay);
        ScheduledFutureTask<?> t = 
            new ScheduledFutureTask<Object>(command, 
                                            null,
                                            triggerTime,
                                            unit.toNanos(period));
        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();
        if (initialDelay < 0) initialDelay = 0;
        long triggerTime = now() + unit.toNanos(initialDelay);
        ScheduledFutureTask<?> t = 
            new ScheduledFutureTask<Boolean>(command, 
                                             null,
                                             triggerTime,
                                             unit.toNanos(-delay));
        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);
    }

    // Override AbstractExecutorService methods

    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.
     * @see #getExecuteExistingDelayedTasksAfterShutdownPolicy
     */
    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.
     * @see #setContinueExistingPeriodicTasksAfterShutdownPolicy
     */
    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.
     * @see #getExecuteExistingDelayedTasksAfterShutdownPolicy
     */
    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.
     * @see #setExecuteExistingDelayedTasksAfterShutdownPolicy
     */
    public boolean getExecuteExistingDelayedTasksAfterShutdownPolicy() {
        return executeExistingDelayedTasksAfterShutdown;
    }


    /**
     * 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 implementation
     * cancels tasks 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();
    }

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

    /**
     * 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(); }
            };
        }
    }
}
Revision:	1.18
Committed:	Mon Feb 9 13:28:48 2004 UTC (20 years, 4 months ago) by dl
Branch:	MAIN
Changes since 1.17:	+9 -8 lines
Log Message:	Wording fixes and improvements
#	User	Rev	Content
1	dl	1.1	/*
2			* Written by Doug Lea with assistance from members of JCP JSR-166
3	dl	1.11	* Expert Group and released to the public domain, as explained at
4			* http://creativecommons.org/licenses/publicdomain
5	dl	1.1	*/
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	dl	1.18	* without any real-time guarantees about when, after they are
21			* enabled, they will commence. Tasks scheduled for exactly the same
22			* execution time are enabled in first-in-first-out (FIFO) order of
23			* submission.
24	dl	1.1	*
25			* <p>While this class inherits from {@link ThreadPoolExecutor}, a few
26	dl	1.8	* of the inherited tuning methods are not useful for it. In
27			* particular, because it acts as a fixed-sized pool using
28			* <tt>corePoolSize</tt> threads and an unbounded queue, adjustments
29			* to <tt>maximumPoolSize</tt> have no useful effect.
30	dl	1.1	*
31			* @since 1.5
32			* @author Doug Lea
33			*/
34	tim	1.3	public class ScheduledThreadPoolExecutor
35			extends ThreadPoolExecutor
36			implements ScheduledExecutorService {
37	dl	1.1
38			/**
39			* False if should cancel/suppress periodic tasks on shutdown.
40			*/
41			private volatile boolean continueExistingPeriodicTasksAfterShutdown;
42
43			/**
44			* False if should cancel non-periodic tasks on shutdown.
45			*/
46			private volatile boolean executeExistingDelayedTasksAfterShutdown = true;
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	dl	1.14
54			/** Base of nanosecond timings, to avoid wrapping */
55			private static final long NANO_ORIGIN = System.nanoTime();
56
57			/**
58	dl	1.18	* Returns nanosecond time offset by origin
59	dl	1.14	*/
60	dl	1.17	final long now() {
61	dl	1.14	return System.nanoTime() - NANO_ORIGIN;
62			}
63
64	dl	1.5	private class ScheduledFutureTask<V>
65	dl	1.1	extends FutureTask<V> implements ScheduledFuture<V> {
66
67			/** Sequence number to break ties FIFO */
68			private final long sequenceNumber;
69			/** The time the task is enabled to execute in nanoTime units */
70			private long time;
71	dl	1.16	/**
72			* Period in nanoseconds for repeating tasks. A positive
73			* value indicates fixed-rate execution. A negative value
74			* indicates fixed-delay execution. A value of 0 indicates a
75			* non-repeating task.
76			*/
77	dl	1.1	private final long period;
78
79			/**
80			* Creates a one-shot action with given nanoTime-based trigger time
81			*/
82			ScheduledFutureTask(Runnable r, V result, long ns) {
83			super(r, result);
84			this.time = ns;
85			this.period = 0;
86			this.sequenceNumber = sequencer.getAndIncrement();
87			}
88
89			/**
90			* Creates a periodic action with given nano time and period
91			*/
92	dl	1.16	ScheduledFutureTask(Runnable r, V result, long ns, long period) {
93	dl	1.1	super(r, result);
94			this.time = ns;
95			this.period = period;
96			this.sequenceNumber = sequencer.getAndIncrement();
97			}
98
99			/**
100			* Creates a one-shot action with given nanoTime-based trigger
101			*/
102			ScheduledFutureTask(Callable<V> callable, long ns) {
103			super(callable);
104			this.time = ns;
105			this.period = 0;
106			this.sequenceNumber = sequencer.getAndIncrement();
107			}
108
109			public long getDelay(TimeUnit unit) {
110	dl	1.14	long d = unit.convert(time - now(), TimeUnit.NANOSECONDS);
111	dl	1.1	return d;
112			}
113
114			public int compareTo(Object other) {
115			if (other == this) // compare zero ONLY if same object
116			return 0;
117			ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
118			long diff = time - x.time;
119			if (diff < 0)
120			return -1;
121			else if (diff > 0)
122			return 1;
123			else if (sequenceNumber < x.sequenceNumber)
124			return -1;
125			else
126			return 1;
127			}
128
129			/**
130	dl	1.18	* Returns true if this is a periodic (not a one-shot) action.
131	dl	1.1	* @return true if periodic
132			*/
133	dl	1.10	boolean isPeriodic() {
134	dl	1.16	return period != 0;
135	dl	1.1	}
136
137			/**
138	dl	1.14	* Run a periodic task
139	dl	1.13	*/
140			private void runPeriodic() {
141			boolean ok = ScheduledFutureTask.super.runAndReset();
142			boolean down = isShutdown();
143			// Reschedule if not cancelled and not shutdown or policy allows
144			if (ok && (!down \|\|
145			(getContinueExistingPeriodicTasksAfterShutdownPolicy() &&
146			!isTerminating()))) {
147	dl	1.16	long p = period;
148			if (p > 0)
149			time += p;
150			else
151			time = now() - p;
152	dl	1.13	ScheduledThreadPoolExecutor.super.getQueue().add(this);
153			}
154			// This might have been the final executed delayed
155			// task. Wake up threads to check.
156			else if (down)
157			interruptIdleWorkers();
158			}
159
160			/**
161	dl	1.5	* Overrides FutureTask version so as to reset/requeue if periodic.
162	dl	1.1	*/
163			public void run() {
164	dl	1.13	if (isPeriodic())
165			runPeriodic();
166			else
167	dl	1.5	ScheduledFutureTask.super.run();
168	dl	1.1	}
169			}
170
171			/**
172	dl	1.13	* Specialized variant of ThreadPoolExecutor.execute for delayed tasks.
173			*/
174			private void delayedExecute(Runnable command) {
175			if (isShutdown()) {
176			reject(command);
177			return;
178	dl	1.1	}
179	dl	1.13	// Prestart a thread if necessary. We cannot prestart it
180			// running the task because the task (probably) shouldn't be
181			// run yet, so thread will just idle until delay elapses.
182			if (getPoolSize() < getCorePoolSize())
183			prestartCoreThread();
184
185			super.getQueue().add(command);
186			}
187	dl	1.1
188	dl	1.13	/**
189			* Cancel and clear the queue of all tasks that should not be run
190			* due to shutdown policy.
191			*/
192			private void cancelUnwantedTasks() {
193			boolean keepDelayed = getExecuteExistingDelayedTasksAfterShutdownPolicy();
194			boolean keepPeriodic = getContinueExistingPeriodicTasksAfterShutdownPolicy();
195			if (!keepDelayed && !keepPeriodic)
196			super.getQueue().clear();
197			else if (keepDelayed \|\| keepPeriodic) {
198			Object[] entries = super.getQueue().toArray();
199			for (int i = 0; i < entries.length; ++i) {
200			Object e = entries[i];
201			if (e instanceof ScheduledFutureTask) {
202			ScheduledFutureTask<?> t = (ScheduledFutureTask<?>)e;
203			if (t.isPeriodic()? !keepPeriodic : !keepDelayed)
204			t.cancel(false);
205			}
206			}
207			entries = null;
208			purge();
209	dl	1.1	}
210			}
211
212			/**
213	dl	1.13	* Creates a new ScheduledThreadPoolExecutor with the given core
214			* pool size.
215	dl	1.1	*
216			* @param corePoolSize the number of threads to keep in the pool,
217			* even if they are idle.
218			* @throws IllegalArgumentException if corePoolSize less than or
219			* equal to zero
220			*/
221			public ScheduledThreadPoolExecutor(int corePoolSize) {
222			super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
223			new DelayedWorkQueue());
224			}
225
226			/**
227	dl	1.13	* Creates a new ScheduledThreadPoolExecutor with the given
228			* initial parameters.
229	dl	1.1	*
230			* @param corePoolSize the number of threads to keep in the pool,
231			* even if they are idle.
232			* @param threadFactory the factory to use when the executor
233			* creates a new thread.
234			* @throws NullPointerException if threadFactory is null
235			*/
236			public ScheduledThreadPoolExecutor(int corePoolSize,
237			ThreadFactory threadFactory) {
238			super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
239			new DelayedWorkQueue(), threadFactory);
240			}
241
242			/**
243	dl	1.13	* Creates a new ScheduledThreadPoolExecutor with the given
244			* initial parameters.
245	dl	1.1	*
246			* @param corePoolSize the number of threads to keep in the pool,
247			* even if they are idle.
248			* @param handler the handler to use when execution is blocked
249			* because the thread bounds and queue capacities are reached.
250			* @throws NullPointerException if handler is null
251			*/
252			public ScheduledThreadPoolExecutor(int corePoolSize,
253			RejectedExecutionHandler handler) {
254			super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
255			new DelayedWorkQueue(), handler);
256			}
257
258			/**
259	dl	1.13	* Creates a new ScheduledThreadPoolExecutor with the given
260			* initial parameters.
261	dl	1.1	*
262			* @param corePoolSize the number of threads to keep in the pool,
263			* even if they are idle.
264			* @param threadFactory the factory to use when the executor
265			* creates a new thread.
266			* @param handler the handler to use when execution is blocked
267			* because the thread bounds and queue capacities are reached.
268			* @throws NullPointerException if threadFactory or handler is null
269			*/
270			public ScheduledThreadPoolExecutor(int corePoolSize,
271			ThreadFactory threadFactory,
272			RejectedExecutionHandler handler) {
273			super(corePoolSize, Integer.MAX_VALUE, 0, TimeUnit.NANOSECONDS,
274			new DelayedWorkQueue(), threadFactory, handler);
275			}
276
277	dl	1.13	public ScheduledFuture<?> schedule(Runnable command,
278			long delay,
279			TimeUnit unit) {
280	dl	1.9	if (command == null \|\| unit == null)
281	dl	1.1	throw new NullPointerException();
282	dl	1.14	long triggerTime = now() + unit.toNanos(delay);
283	dl	1.13	ScheduledFutureTask<?> t =
284			new ScheduledFutureTask<Boolean>(command, null, triggerTime);
285	dl	1.1	delayedExecute(t);
286			return t;
287			}
288
289	dl	1.13	public <V> ScheduledFuture<V> schedule(Callable<V> callable,
290			long delay,
291			TimeUnit unit) {
292	dl	1.9	if (callable == null \|\| unit == null)
293	dl	1.1	throw new NullPointerException();
294	dl	1.16	if (delay < 0) delay = 0;
295	dl	1.14	long triggerTime = now() + unit.toNanos(delay);
296	dl	1.13	ScheduledFutureTask<V> t =
297			new ScheduledFutureTask<V>(callable, triggerTime);
298	dl	1.1	delayedExecute(t);
299			return t;
300			}
301
302	dl	1.13	public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
303			long initialDelay,
304			long period,
305			TimeUnit unit) {
306	dl	1.9	if (command == null \|\| unit == null)
307	dl	1.1	throw new NullPointerException();
308			if (period <= 0)
309			throw new IllegalArgumentException();
310	dl	1.16	if (initialDelay < 0) initialDelay = 0;
311	dl	1.14	long triggerTime = now() + unit.toNanos(initialDelay);
312	dl	1.13	ScheduledFutureTask<?> t =
313			new ScheduledFutureTask<Object>(command,
314			null,
315			triggerTime,
316	dl	1.16	unit.toNanos(period));
317	dl	1.1	delayedExecute(t);
318			return t;
319			}
320
321	dl	1.13	public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
322			long initialDelay,
323			long delay,
324			TimeUnit unit) {
325	dl	1.9	if (command == null \|\| unit == null)
326	dl	1.1	throw new NullPointerException();
327			if (delay <= 0)
328			throw new IllegalArgumentException();
329	dl	1.16	if (initialDelay < 0) initialDelay = 0;
330	dl	1.14	long triggerTime = now() + unit.toNanos(initialDelay);
331	dl	1.13	ScheduledFutureTask<?> t =
332			new ScheduledFutureTask<Boolean>(command,
333			null,
334			triggerTime,
335	dl	1.16	unit.toNanos(-delay));
336	dl	1.1	delayedExecute(t);
337			return t;
338			}
339
340
341			/**
342			* Execute command with zero required delay. This has effect
343			* equivalent to <tt>schedule(command, 0, anyUnit)</tt>. Note
344			* that inspections of the queue and of the list returned by
345			* <tt>shutdownNow</tt> will access the zero-delayed
346			* {@link ScheduledFuture}, not the <tt>command</tt> itself.
347			*
348			* @param command the task to execute
349			* @throws RejectedExecutionException at discretion of
350			* <tt>RejectedExecutionHandler</tt>, if task cannot be accepted
351			* for execution because the executor has been shut down.
352			* @throws NullPointerException if command is null
353			*/
354			public void execute(Runnable command) {
355			if (command == null)
356			throw new NullPointerException();
357			schedule(command, 0, TimeUnit.NANOSECONDS);
358			}
359
360	dl	1.13	// Override AbstractExecutorService methods
361
362	dl	1.7	public Future<?> submit(Runnable task) {
363			return schedule(task, 0, TimeUnit.NANOSECONDS);
364			}
365
366			public <T> Future<T> submit(Runnable task, T result) {
367	dl	1.13	return schedule(Executors.callable(task, result),
368			0, TimeUnit.NANOSECONDS);
369	dl	1.7	}
370
371			public <T> Future<T> submit(Callable<T> task) {
372			return schedule(task, 0, TimeUnit.NANOSECONDS);
373			}
374	dl	1.1
375			/**
376			* Set policy on whether to continue executing existing periodic
377			* tasks even when this executor has been <tt>shutdown</tt>. In
378			* this case, these tasks will only terminate upon
379			* <tt>shutdownNow</tt>, or after setting the policy to
380			* <tt>false</tt> when already shutdown. This value is by default
381			* false.
382			* @param value if true, continue after shutdown, else don't.
383	dl	1.16	* @see #getExecuteExistingDelayedTasksAfterShutdownPolicy
384	dl	1.1	*/
385			public void setContinueExistingPeriodicTasksAfterShutdownPolicy(boolean value) {
386			continueExistingPeriodicTasksAfterShutdown = value;
387			if (!value && isShutdown())
388			cancelUnwantedTasks();
389			}
390
391			/**
392			* Get the policy on whether to continue executing existing
393			* periodic tasks even when this executor has been
394			* <tt>shutdown</tt>. In this case, these tasks will only
395			* terminate upon <tt>shutdownNow</tt> or after setting the policy
396			* to <tt>false</tt> when already shutdown. This value is by
397			* default false.
398			* @return true if will continue after shutdown.
399	dl	1.16	* @see #setContinueExistingPeriodicTasksAfterShutdownPolicy
400	dl	1.1	*/
401			public boolean getContinueExistingPeriodicTasksAfterShutdownPolicy() {
402			return continueExistingPeriodicTasksAfterShutdown;
403			}
404
405			/**
406			* Set policy on whether to execute existing delayed
407			* tasks even when this executor has been <tt>shutdown</tt>. In
408			* this case, these tasks will only terminate upon
409			* <tt>shutdownNow</tt>, or after setting the policy to
410			* <tt>false</tt> when already shutdown. This value is by default
411			* true.
412			* @param value if true, execute after shutdown, else don't.
413	dl	1.16	* @see #getExecuteExistingDelayedTasksAfterShutdownPolicy
414	dl	1.1	*/
415			public void setExecuteExistingDelayedTasksAfterShutdownPolicy(boolean value) {
416			executeExistingDelayedTasksAfterShutdown = value;
417			if (!value && isShutdown())
418			cancelUnwantedTasks();
419			}
420
421			/**
422			* Get policy on whether to execute existing delayed
423			* tasks even when this executor has been <tt>shutdown</tt>. In
424			* this case, these tasks will only terminate upon
425			* <tt>shutdownNow</tt>, or after setting the policy to
426			* <tt>false</tt> when already shutdown. This value is by default
427			* true.
428			* @return true if will execute after shutdown.
429	dl	1.16	* @see #setExecuteExistingDelayedTasksAfterShutdownPolicy
430	dl	1.1	*/
431			public boolean getExecuteExistingDelayedTasksAfterShutdownPolicy() {
432			return executeExistingDelayedTasksAfterShutdown;
433			}
434
435
436			/**
437			* Initiates an orderly shutdown in which previously submitted
438			* tasks are executed, but no new tasks will be accepted. If the
439			* <tt>ExecuteExistingDelayedTasksAfterShutdownPolicy</tt> has
440			* been set <tt>false</tt>, existing delayed tasks whose delays
441			* have not yet elapsed are cancelled. And unless the
442			* <tt>ContinueExistingPeriodicTasksAfterShutdownPolicy</tt> has
443			* been set <tt>true</tt>, future executions of existing periodic
444			* tasks will be cancelled.
445			*/
446			public void shutdown() {
447			cancelUnwantedTasks();
448			super.shutdown();
449			}
450
451			/**
452			* Attempts to stop all actively executing tasks, halts the
453			* processing of waiting tasks, and returns a list of the tasks that were
454			* awaiting execution.
455			*
456			* <p>There are no guarantees beyond best-effort attempts to stop
457	dl	1.18	* processing actively executing tasks. This implementation
458			* cancels tasks via {@link Thread#interrupt}, so if any tasks mask or
459	dl	1.1	* fail to respond to interrupts, they may never terminate.
460			*
461			* @return list of tasks that never commenced execution. Each
462			* element of this list is a {@link ScheduledFuture},
463	dl	1.18	* including those tasks submitted using <tt>execute</tt>, which
464	dl	1.1	* are for scheduling purposes used as the basis of a zero-delay
465			* <tt>ScheduledFuture</tt>.
466			*/
467	tim	1.4	public List<Runnable> shutdownNow() {
468	dl	1.1	return super.shutdownNow();
469			}
470
471			/**
472			* Returns the task queue used by this executor. Each element of
473			* this queue is a {@link ScheduledFuture}, including those
474			* tasks submitted using <tt>execute</tt> which are for scheduling
475			* purposes used as the basis of a zero-delay
476			* <tt>ScheduledFuture</tt>. Iteration over this queue is
477	dl	1.15	* <em>not</em> guaranteed to traverse tasks in the order in
478	dl	1.1	* which they will execute.
479			*
480			* @return the task queue
481			*/
482			public BlockingQueue<Runnable> getQueue() {
483			return super.getQueue();
484			}
485
486	dl	1.13	/**
487			* An annoying wrapper class to convince generics compiler to
488			* use a DelayQueue<ScheduledFutureTask> as a BlockingQueue<Runnable>
489			*/
490			private static class DelayedWorkQueue
491			extends AbstractCollection<Runnable>
492			implements BlockingQueue<Runnable> {
493
494			private final DelayQueue<ScheduledFutureTask> dq = new DelayQueue<ScheduledFutureTask>();
495			public Runnable poll() { return dq.poll(); }
496			public Runnable peek() { return dq.peek(); }
497			public Runnable take() throws InterruptedException { return dq.take(); }
498			public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
499			return dq.poll(timeout, unit);
500			}
501
502			public boolean add(Runnable x) { return dq.add((ScheduledFutureTask)x); }
503			public boolean offer(Runnable x) { return dq.offer((ScheduledFutureTask)x); }
504			public void put(Runnable x) {
505			dq.put((ScheduledFutureTask)x);
506			}
507			public boolean offer(Runnable x, long timeout, TimeUnit unit) {
508			return dq.offer((ScheduledFutureTask)x, timeout, unit);
509			}
510
511			public Runnable remove() { return dq.remove(); }
512			public Runnable element() { return dq.element(); }
513			public void clear() { dq.clear(); }
514			public int drainTo(Collection<? super Runnable> c) { return dq.drainTo(c); }
515			public int drainTo(Collection<? super Runnable> c, int maxElements) {
516			return dq.drainTo(c, maxElements);
517			}
518
519			public int remainingCapacity() { return dq.remainingCapacity(); }
520			public boolean remove(Object x) { return dq.remove(x); }
521			public boolean contains(Object x) { return dq.contains(x); }
522			public int size() { return dq.size(); }
523			public boolean isEmpty() { return dq.isEmpty(); }
524			public Object[] toArray() { return dq.toArray(); }
525			public <T> T[] toArray(T[] array) { return dq.toArray(array); }
526			public Iterator<Runnable> iterator() {
527			return new Iterator<Runnable>() {
528			private Iterator<ScheduledFutureTask> it = dq.iterator();
529			public boolean hasNext() { return it.hasNext(); }
530			public Runnable next() { return it.next(); }
531			public void remove() { it.remove(); }
532			};
533			}
534			}
535	dl	1.1	}