/*
* 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.
*
*
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.
*
*
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
* corePoolSize threads and an unbounded queue, adjustments
* to maximumPoolSize 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
extends FutureTask implements ScheduledFuture {
/** 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 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
*/
boolean isPeriodic() {
return period > 0;
}
/**
* Returns the period, or zero if non-periodic.
*
* @return the period
*/
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 {
if (!ScheduledFutureTask.super.runAndReset())
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 as a BlockingQueue
*/
private static class DelayedWorkQueue
extends AbstractCollection implements BlockingQueue {
private final DelayQueue dq = new DelayQueue();
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[] toArray(T[] array) { return dq.toArray(array); }
public Iterator iterator() {
return new Iterator() {
private Iterator 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(command, null, triggerTime);
delayedExecute(t);
return t;
}
public ScheduledFuture schedule(Callable callable, long delay, TimeUnit unit) {
if (callable == null || unit == null)
throw new NullPointerException();
long triggerTime = System.nanoTime() + unit.toNanos(delay);
ScheduledFutureTask t = new ScheduledFutureTask(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