* Thread pools address two different problems: they usually provide improved performance when executing large numbers * of asynchronous tasks, due to reduced per-task invocation overhead, and they provide a means of bounding and managing * the resources, including threads, consumed when executing a collection of tasks. Each {@code ThreadPoolExecutor} also * maintains some basic statistics, such as the number of completed tasks. *
* To be useful across a wide range of contexts, this class provides many adjustable parameters and extensibility hooks. * However, programmers are urged to use the more convenient {@link Executors} factory methods * {@link Executors#newCachedThreadPool} (unbounded thread pool, with automatic thread reclamation), * {@link Executors#newFixedThreadPool} (fixed size thread pool) and {@link Executors#newSingleThreadExecutor} (single * background thread), that preconfigure settings for the most common usage scenarios. Otherwise, use the following * guide when manually configuring and tuning this class: *
* If hook, callback, or BlockingQueue methods throw exceptions, internal worker threads may in turn fail, abruptly * terminate, and possibly be replaced.
* Extension example. Most extensions of this class override one or more of the protected hook methods. For * example, here is a subclass that adds a simple pause/resume feature: * *
{@code
* class PausableThreadPoolExecutor extends ThreadPoolExecutor {
* private boolean isPaused;
* private ReentrantLock pauseLock = new ReentrantLock();
* private Condition unpaused = pauseLock.newCondition();
*
* public PausableThreadPoolExecutor(...) { super(...); }
*
* protected void beforeExecute(Thread t, Runnable r) {
* super.beforeExecute(t, r);
* pauseLock.lock();
* try {
* while (isPaused) unpaused.await();
* } catch (InterruptedException ie) {
* t.interrupt();
* } finally {
* pauseLock.unlock();
* }
* }
*
* public void pause() {
* pauseLock.lock();
* try {
* isPaused = true;
* } finally {
* pauseLock.unlock();
* }
* }
*
* public void resume() {
* pauseLock.lock();
* try {
* isPaused = false;
* unpaused.signalAll();
* } finally {
* pauseLock.unlock();
* }
* }
* }}
*
* @since 1.5
*
* @author Doug Lea
*/
public class ThreadPoolExecutor extends AbstractExecutorService {
protected static final StringManager sm = StringManager.getManager(ThreadPoolExecutor.class);
/**
* The main pool control state, ctl, is an atomic integer packing two conceptual fields:
* * The workerCount is the number of workers that have been permitted to start and not permitted to stop. The value * may be transiently different from the actual number of live threads, for example when a ThreadFactory fails to * create a thread when asked, and when exiting threads are still performing bookkeeping before terminating. The * user-visible pool size is reported as the current size of the workers set. *
* The runState provides the main lifecycle control, taking on values: *
* Detecting the transition from SHUTDOWN to TIDYING is less straightforward than you'd like because the queue may
* become empty after non-empty and vice versa during SHUTDOWN state, but we can only terminate if, after seeing
* that it is empty, we see that workerCount is 0 (which sometimes entails a recheck -- see below).
*/
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;
// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
// Packing and unpacking ctl
private static int workerCountOf(int c) {
return c & COUNT_MASK;
}
private static int ctlOf(int rs, int wc) {
return rs | wc;
}
/*
* Bit field accessors that don't require unpacking ctl. These depend on the bit layout and on workerCount being
* never negative.
*/
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
/**
* Attempts to CAS-increment the workerCount field of ctl.
*/
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}
/**
* Attempts to CAS-decrement the workerCount field of ctl.
*/
private boolean compareAndDecrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect - 1);
}
/**
* Decrements the workerCount field of ctl. This is called only on abrupt termination of a thread (see
* processWorkerExit). Other decrements are performed within getTask.
*/
private void decrementWorkerCount() {
ctl.addAndGet(-1);
}
/**
* The queue used for holding tasks and handing off to worker threads. We do not require that workQueue.poll()
* returning null necessarily means that workQueue.isEmpty(), so rely solely on isEmpty to see if the queue is empty
* (which we must do for example when deciding whether to transition from SHUTDOWN to TIDYING). This accommodates
* special-purpose queues such as DelayQueues for which poll() is allowed to return null even if it may later return
* non-null when delays expire.
*/
private final BlockingQueue
* We go further and preserve pool invariants even in the face of errors such as OutOfMemoryError, that might be
* thrown while trying to create threads. Such errors are rather common due to the need to allocate a native stack
* in Thread.start, and users will want to perform clean pool shutdown to clean up. There will likely be enough
* memory available for the cleanup code to complete without encountering yet another OutOfMemoryError.
*/
private volatile ThreadFactory threadFactory;
/**
* Handler called when saturated or shutdown in execute.
*/
private volatile RejectedExecutionHandler handler;
/**
* Timeout in nanoseconds for idle threads waiting for work. Threads use this timeout when there are more than
* corePoolSize present or if allowCoreThreadTimeOut. Otherwise, they wait forever for new work.
*/
private volatile long keepAliveTime;
/**
* If false (default), core threads stay alive even when idle. If true, core threads use keepAliveTime to time out
* waiting for work.
*/
private volatile boolean allowCoreThreadTimeOut;
/**
* Core pool size is the minimum number of workers to keep alive (and not allow to time out etc) unless
* allowCoreThreadTimeOut is set, in which case the minimum is zero.
*
* Since the worker count is actually stored in COUNT_BITS bits, the effective limit is
* {@code corePoolSize & COUNT_MASK}.
*/
private volatile int corePoolSize;
/**
* Maximum pool size.
*
* Since the worker count is actually stored in COUNT_BITS bits, the effective limit is
* {@code maximumPoolSize & COUNT_MASK}.
*/
private volatile int maximumPoolSize;
/**
* The default rejected execution handler.
*/
private static final RejectedExecutionHandler defaultHandler = new RejectPolicy();
/**
* Class Worker mainly maintains interrupt control state for threads running tasks, along with other minor
* bookkeeping. This class opportunistically extends AbstractQueuedSynchronizer to simplify acquiring and releasing
* a lock surrounding each task execution. This protects against interrupts that are intended to wake up a worker
* thread waiting for a task from instead interrupting a task being run. We implement a simple non-reentrant mutual
* exclusion lock rather than use ReentrantLock because we do not want worker tasks to be able to reacquire the lock
* when they invoke pool control methods like setCorePoolSize. Additionally, to suppress interrupts until the thread
* actually starts running tasks, we initialize lock state to a negative value, and clear it upon start (in
* runWorker).
*/
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
/**
* This class will never be serialized, but we provide a serialVersionUID to suppress a javac warning.
*/
@Serial
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
// TODO: switch to AbstractQueuedLongSynchronizer and move
// completedTasks into the lock word.
/**
* Creates with given first task and thread from ThreadFactory.
*
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker. */
@Override
public void run() {
runWorker(this);
}
// Lock methods
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.
@Override
protected boolean isHeldExclusively() {
return getState() != 0;
}
@Override
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
@Override
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() {
acquire(1);
}
public boolean tryLock() {
return tryAcquire(1);
}
public void unlock() {
release(1);
}
public boolean isLocked() {
return isHeldExclusively();
}
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
/*
* Methods for setting control state
*/
/**
* Transitions runState to given target, or leaves it alone if already at least the given target.
*
* @param targetState the desired state, either SHUTDOWN or STOP (but not TIDYING or TERMINATED -- use tryTerminate
* for that)
*/
private void advanceRunState(int targetState) {
// assert targetState == SHUTDOWN || targetState == STOP;
for (;;) {
int c = ctl.get();
if (runStateAtLeast(c, targetState) || ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c)))) {
break;
}
}
}
/**
* Transitions to TERMINATED state if either (SHUTDOWN and pool and queue empty) or (STOP and pool empty). If
* otherwise eligible to terminate but workerCount is nonzero, interrupts an idle worker to ensure that shutdown
* signals propagate. This method must be called following any action that might make termination possible --
* reducing worker count or removing tasks from the queue during shutdown. The method is non-private to allow access
* from ScheduledThreadPoolExecutor.
*/
final void tryTerminate() {
for (;;) {
int c = ctl.get();
if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateLessThan(c, STOP) && !workQueue.isEmpty())) {
return;
}
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();
} finally {
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
/*
* Methods for controlling interrupts to worker threads.
*/
/**
* Interrupts all threads, even if active. Ignores SecurityExceptions (in which case some threads may remain
* uninterrupted).
*/
private void interruptWorkers() {
// assert mainLock.isHeldByCurrentThread();
for (Worker w : workers) {
w.interruptIfStarted();
}
}
/**
* Interrupts threads that might be waiting for tasks (as indicated by not being locked) so they can check for
* termination or configuration changes. Ignores SecurityExceptions (in which case some threads may remain
* uninterrupted).
*
* @param onlyOne If true, interrupt at most one worker. This is called only from tryTerminate when termination is
* otherwise enabled but there are still other workers. In this case, at most one waiting worker
* is interrupted to propagate shutdown signals in case all threads are currently waiting.
* Interrupting any arbitrary thread ensures that newly arriving workers since shutdown began
* will also eventually exit. To guarantee eventual termination, it suffices to always interrupt
* only one idle worker, but shutdown() interrupts all idle workers so that redundant workers
* exit promptly, not waiting for a straggler task to finish.
*/
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
for (Worker w : workers) {
Thread t = w.thread;
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne) {
break;
}
}
} finally {
mainLock.unlock();
}
}
/**
* Common form of interruptIdleWorkers, to avoid having to remember what the boolean argument means.
*/
private void interruptIdleWorkers() {
interruptIdleWorkers(false);
}
private static final boolean ONLY_ONE = true;
/*
* Misc utilities, most of which are also exported to ScheduledThreadPoolExecutor
*/
/**
* Invokes the rejected execution handler for the given command. Package-protected for use by
* ScheduledThreadPoolExecutor.
*/
final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}
/**
* Performs any further cleanup following run state transition on invocation of shutdown. A no-op here, but used by
* ScheduledThreadPoolExecutor to cancel delayed tasks.
*/
void onShutdown() {
}
/**
* Drains the task queue into a new list, normally using drainTo. But if the queue is a DelayQueue or any other kind
* of queue for which poll or drainTo may fail to remove some elements, it deletes them one by one.
*/
private List
* 1. We may start out with an initial task, in which case we don't need to get the first one. Otherwise, as long as
* pool is running, we get tasks from getTask. If it returns null then the worker exits due to changed pool state or
* configuration parameters. Other exits result from exception throws in external code, in which case
* completedAbruptly holds, which usually leads processWorkerExit to replace this thread.
*
* 2. Before running any task, the lock is acquired to prevent other pool interrupts while the task is executing,
* and then we ensure that unless pool is stopping, this thread does not have its interrupt set.
*
* 3. Each task run is preceded by a call to beforeExecute, which might throw an exception, in which case we cause
* thread to die (breaking loop with completedAbruptly true) without processing the task.
*
* 4. Assuming beforeExecute completes normally, we run the task, gathering any of its thrown exceptions to send to
* afterExecute. We separately handle RuntimeException, Error (both of which the specs guarantee that we trap) and
* arbitrary Throwables. Because we cannot rethrow Throwables within Runnable.run, we wrap them within Errors on the
* way out (to the thread's UncaughtExceptionHandler). Any thrown exception also conservatively causes thread to
* die.
*
* 5. After task.run completes, we call afterExecute, which may also throw an exception, which will also cause
* thread to die. According to JLS Sec 14.20, this exception is the one that will be in effect even if task.run
* throws.
*
* The net effect of the exception mechanics is that afterExecute and the thread's UncaughtExceptionHandler have as
* accurate information as we can provide about any problems encountered by user code.
*
* @param w the worker
*/
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted()) {
wt.interrupt();
}
try {
beforeExecute(wt, task);
try {
task.run();
afterExecute(task, null);
} catch (Throwable ex) {
afterExecute(task, ex);
throw ex;
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
// Public constructors and methods
/**
* Creates a new {@code ThreadPoolExecutor} with the given initial parameters, the
* {@linkplain Executors#defaultThreadFactory default thread factory} and the
* {@linkplain ThreadPoolExecutor.RejectPolicy default rejected execution handler}.
*
* It may be more convenient to use one of the {@link Executors} factory methods instead of this general purpose
* constructor.
*
* @param corePoolSize the number of threads to keep in the pool, even if they are idle, unless
* {@code allowCoreThreadTimeOut} is set
* @param maximumPoolSize the maximum number of threads to allow in the pool
* @param keepAliveTime when the number of threads is greater than the core, this is the maximum time that excess
* idle threads will wait for new tasks before terminating.
* @param unit the time unit for the {@code keepAliveTime} argument
* @param workQueue the queue to use for holding tasks before they are executed. This queue will hold only the
* {@code Runnable} tasks submitted by the {@code execute} method.
*
* @throws IllegalArgumentException if one of the following holds:
* This method does not wait for previously submitted tasks to complete execution. Use {@link #awaitTermination
* awaitTermination} to do that.
*
* @throws SecurityException {@inheritDoc}
*/
@Override
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
advanceRunState(SHUTDOWN);
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
/**
* Attempts to stop all actively executing tasks, halts the processing of waiting tasks, and returns a list of the
* tasks that were awaiting execution. These tasks are drained (removed) from the task queue upon return from this
* method.
*
* This method does not wait for actively executing tasks to terminate. Use {@link #awaitTermination
* awaitTermination} to do that.
*
* There are no guarantees beyond best-effort attempts to stop processing actively executing tasks. This
* implementation interrupts tasks via {@link Thread#interrupt}; any task that fails to respond to interrupts may
* never terminate.
*
* @throws SecurityException {@inheritDoc}
*/
@Override
public List
* This method may be useful as one part of a cancellation scheme. It may fail to remove tasks that have been
* converted into other forms before being placed on the internal queue. For example, a task entered using
* {@code submit} might be converted into a form that maintains {@code Future} status. However, in such cases,
* method {@link #purge} may be used to remove those Futures that have been cancelled.
*
* @param task the task to remove
*
* @return {@code true} if the task was removed
*/
public boolean remove(Runnable task) {
boolean removed = workQueue.remove(task);
tryTerminate(); // In case SHUTDOWN and now empty
return removed;
}
/**
* Tries to remove from the work queue all {@link Future} tasks that have been cancelled. This method can be useful
* as a storage reclamation operation, that has no other impact on functionality. Cancelled tasks are never
* executed, but may accumulate in work queues until worker threads can actively remove them. Invoking this method
* instead tries to remove them now. However, this method may fail to remove tasks in the presence of interference
* by other threads.
*/
public void purge() {
final BlockingQueue
* This implementation does nothing, but may be customized in subclasses. Note: To properly nest multiple
* overridings, subclasses should generally invoke {@code super.beforeExecute} at the end of this method.
*
* @param t the thread that will run task {@code r}
* @param r the task that will be executed
*/
protected void beforeExecute(Thread t, Runnable r) {
}
/**
* Method invoked upon completion of execution of the given Runnable. This method is invoked by the thread that
* executed the task. If non-null, the Throwable is the uncaught {@code RuntimeException} or {@code Error} that
* caused execution to terminate abruptly.
*
* This implementation does nothing, but may be customized in subclasses. Note: To properly nest multiple
* overridings, subclasses should generally invoke {@code super.afterExecute} at the beginning of this method.
*
* Note: When actions are enclosed in tasks (such as {@link java.util.concurrent.FutureTask}) either
* explicitly or via methods such as {@code submit}, these task objects catch and maintain computational exceptions,
* and so they do not cause abrupt termination, and the internal exceptions are not passed to this method.
* If you would like to trap both kinds of failures in this method, you can further probe for such cases, as in this
* sample subclass that prints either the direct cause or the underlying exception if a task has been aborted:
*
*
* In the absence of other alternatives, the method may throw an unchecked {@link RejectedExecutionException},
* which will be propagated to the caller of {@code execute}.
*
* @param r the runnable task requested to be executed
* @param executor the executor attempting to execute this task
*
* @throws RejectedExecutionException if there is no remedy
*/
void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}
}
*
*/
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (w != null) {
workers.remove(w);
}
decrementWorkerCount();
tryTerminate();
} finally {
mainLock.unlock();
}
}
/**
* Performs cleanup and bookkeeping for a dying worker. Called only from worker threads. Unless completedAbruptly is
* set, assumes that workerCount has already been adjusted to account for exit. This method removes thread from
* worker set, and possibly terminates the pool or replaces the worker if either it exited due to user task
* exception or if fewer than corePoolSize workers are running or queue is non-empty but there are no workers.
*
* @param w the worker
* @param completedAbruptly if the worker died due to user exception
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) {
decrementWorkerCount();
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && !workQueue.isEmpty()) {
min = 1;
}
// https://bz.apache.org/bugzilla/show_bug.cgi?id=65454
// If the work queue is not empty, it is likely that a task was
// added to the work queue between this thread timing out and
// the worker count being decremented a few lines above this
// comment. In this case, create a replacement worker so that
// the task isn't held in the queue waiting for one of the other
// workers to finish.
if (workerCountOf(c) >= min && workQueue.isEmpty()) {
return; // replacement not needed
}
}
addWorker(null, false);
}
}
/**
* Performs blocking or timed wait for a task, depending on current configuration settings, or returns null if this
* worker must exit because of any of:
*
*
*
* @return task, or null if the worker must exit, in which case workerCount is decremented
*/
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
// Check if queue empty only if necessary.
if (runStateAtLeast(c, SHUTDOWN) && (runStateAtLeast(c, STOP) || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c)) {
return null;
}
continue;
}
try {
Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take();
if (r != null) {
return r;
}
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
/**
* Main worker run loop. Repeatedly gets tasks from queue and executes them, while coping with a number of issues:
*
* {@code corePoolSize < 0}
* {@code keepAliveTime < 0}
* {@code maximumPoolSize <= 0}
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue} is null
*/
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
BlockingQueue
* {@code corePoolSize < 0}
* {@code keepAliveTime < 0}
* {@code maximumPoolSize <= 0}
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue} or {@code threadFactory} is null
*/
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
BlockingQueue
* {@code corePoolSize < 0}
* {@code keepAliveTime < 0}
* {@code maximumPoolSize <= 0}
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue} or {@code handler} is null
*/
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
BlockingQueue
* {@code corePoolSize < 0}
* {@code keepAliveTime < 0}
* {@code maximumPoolSize <= 0}
* {@code maximumPoolSize < corePoolSize}
* @throws NullPointerException if {@code workQueue} or {@code threadFactory} or {@code handler} is null
*/
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
BlockingQueue
* NOTE: this method only used in {@link TaskQueue#offer(Runnable)}, where operations are frequent, can
* greatly reduce unnecessary performance overhead by a lock-free way.
*
* @return the number of threads
*/
protected int getPoolSizeNoLock() {
return runStateAtLeast(ctl.get(), TIDYING) ? 0 : workers.size();
}
/**
* Returns the approximate number of threads that are actively executing tasks.
*
* @return the number of threads
*/
public int getActiveCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int n = 0;
for (Worker w : workers) {
if (w.isLocked()) {
++n;
}
}
return n;
} finally {
mainLock.unlock();
}
}
/**
* Returns the largest number of threads that have ever simultaneously been in the pool.
*
* @return the number of threads
*/
public int getLargestPoolSize() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
return largestPoolSize;
} finally {
mainLock.unlock();
}
}
/**
* Returns the approximate total number of tasks that have ever been scheduled for execution. Because the states of
* tasks and threads may change dynamically during computation, the returned value is only an approximation.
*
* @return the number of tasks
*/
public long getTaskCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
long n = completedTaskCount;
for (Worker w : workers) {
n += w.completedTasks;
if (w.isLocked()) {
++n;
}
}
return n + workQueue.size();
} finally {
mainLock.unlock();
}
}
/**
* Returns the approximate total number of tasks that have completed execution. Because the states of tasks and
* threads may change dynamically during computation, the returned value is only an approximation, but one that does
* not ever decrease across successive calls.
*
* @return the number of tasks
*/
public long getCompletedTaskCount() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
long n = completedTaskCount;
for (Worker w : workers) {
n += w.completedTasks;
}
return n;
} finally {
mainLock.unlock();
}
}
public int getSubmittedCount() {
return submittedCount.get();
}
/**
* Returns a string identifying this pool, as well as its state, including indications of run state and estimated
* worker and task counts.
*
* @return a string identifying this pool, as well as its state
*/
@Override
public String toString() {
long ncompleted;
int nworkers, nactive;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
ncompleted = completedTaskCount;
nactive = 0;
nworkers = workers.size();
for (Worker w : workers) {
ncompleted += w.completedTasks;
if (w.isLocked()) {
++nactive;
}
}
} finally {
mainLock.unlock();
}
int c = ctl.get();
String runState = isRunning(c) ? "Running" : runStateAtLeast(c, TERMINATED) ? "Terminated" : "Shutting down";
return super.toString() + "[" + runState + ", pool size = " + nworkers + ", active threads = " + nactive +
", queued tasks = " + workQueue.size() + ", completed tasks = " + ncompleted + "]";
}
/* Extension hooks */
/**
* Method invoked prior to executing the given Runnable in the given thread. This method is invoked by thread
* {@code t} that will execute task {@code r}, and may be used to re-initialize ThreadLocals, or to perform logging.
* {@code
* class ExtendedExecutor extends ThreadPoolExecutor {
* // ...
* protected void afterExecute(Runnable r, Throwable t) {
* super.afterExecute(r, t);
* if (t == null && r instanceof Future && ((Future) r).isDone()) {
* try {
* Object result = ((Future) r).get();
* } catch (CancellationException ce) {
* t = ce;
* } catch (ExecutionException ee) {
* t = ee.getCause();
* } catch (InterruptedException ie) {
* // ignore/reset
* Thread.currentThread().interrupt();
* }
* }
* if (t != null)
* System.out.println(t);
* }
* }
* }
*
* @param r the runnable that has completed
* @param t the exception that caused termination, or null if execution completed normally
*/
protected void afterExecute(Runnable r, Throwable t) {
// Throwing StopPooledThreadException is likely to cause this method to
// be called more than once for a given task based on the typical
// implementations of the parent class. This test ensures that
// decrementAndGet() is only called once after each task execution.
if (!(t instanceof StopPooledThreadException)) {
submittedCount.decrementAndGet();
}
if (t == null) {
stopCurrentThreadIfNeeded();
}
}
/**
* If the current thread was started before the last time when a context was stopped, an exception is thrown so that
* the current thread is stopped.
*/
protected void stopCurrentThreadIfNeeded() {
if (currentThreadShouldBeStopped()) {
long lastTime = lastTimeThreadKilledItself.longValue();
if (lastTime + threadRenewalDelay < System.currentTimeMillis()) {
if (lastTimeThreadKilledItself.compareAndSet(lastTime, System.currentTimeMillis() + 1)) {
// OK, it's really time to dispose of this thread
final String msg = sm.getString("threadPoolExecutor.threadStoppedToAvoidPotentialLeak",
Thread.currentThread().getName());
throw new StopPooledThreadException(msg);
}
}
}
}
protected boolean currentThreadShouldBeStopped() {
Thread currentThread = Thread.currentThread();
if (threadRenewalDelay >= 0 && currentThread instanceof TaskThread currentTaskThread) {
return currentTaskThread.getCreationTime() < this.lastContextStoppedTime.longValue();
}
return false;
}
/**
* Method invoked when the Executor has terminated. Default implementation does nothing. Note: To properly nest
* multiple overridings, subclasses should generally invoke {@code super.terminated} within this method.
*/
protected void terminated() {
}
/* Predefined RejectedExecutionHandlers */
/**
* A handler for rejected tasks that runs the rejected task directly in the calling thread of the {@code execute}
* method, unless the executor has been shut down, in which case the task is discarded.
*/
public static class CallerRunsPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code CallerRunsPolicy}.
*/
public CallerRunsPolicy() {
}
/**
* Executes task r in the caller's thread, unless the executor has been shut down, in which case the task is
* discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
/**
* A handler for rejected tasks that throws a {@link RejectedExecutionException}.
*/
public static class AbortPolicy implements RejectedExecutionHandler {
/**
* Creates an {@code AbortPolicy}.
*/
public AbortPolicy() {
}
/**
* Always throws RejectedExecutionException.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*
* @throws RejectedExecutionException always
*/
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException(
sm.getString("threadPoolExecutor.taskRejected", r.toString(), e.toString()));
}
}
/**
* A handler for rejected tasks that silently discards the rejected task.
*/
public static class DiscardPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardPolicy}.
*/
public DiscardPolicy() {
}
/**
* Does nothing, which has the effect of discarding task r.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
/**
* A handler for rejected tasks that discards the oldest unhandled request and then retries {@code execute}, unless
* the executor is shut down, in which case the task is discarded. This policy is rarely useful in cases where other
* threads may be waiting for tasks to terminate, or failures must be recorded. Instead, consider using a handler of
* the form:
*
* {@code
* new RejectedExecutionHandler() {
* public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
* Runnable dropped = e.getQueue().poll();
* if (dropped instanceof Future) {
* ((Future) dropped).cancel(false);
* // also consider logging the failure
* }
* e.execute(r); // retry
* }
* }
* }
*/
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
/**
* Creates a {@code DiscardOldestPolicy} for the given executor.
*/
public DiscardOldestPolicy() {
}
/**
* Obtains and ignores the next task that the executor would otherwise execute, if one is immediately available,
* and then retries execution of task r, unless the executor is shut down, in which case task r is instead
* discarded.
*
* @param r the runnable task requested to be executed
* @param e the executor attempting to execute this task
*/
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
private static class RejectPolicy implements RejectedExecutionHandler {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
throw new RejectedExecutionException();
}
}
public interface RejectedExecutionHandler {
/**
* Method that may be invoked by a {@link ThreadPoolExecutor} when {@link ThreadPoolExecutor#execute execute}
* cannot accept a task. This may occur when no more threads or queue slots are available because their bounds
* would be exceeded, or upon shutdown of the Executor.
*