You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@tomcat.apache.org by ma...@apache.org on 2021/07/21 17:34:11 UTC
[tomcat] branch main updated (6d02df2 -> 0f2e084)
This is an automated email from the ASF dual-hosted git repository.
markt pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/tomcat.git.
from 6d02df2 Deprecate unused method. Improve Javadoc.
new 4289f47 Remove deprecated code
new c03c7c8 Replace Tomcat's ThreadPoolExecutor with the current JSR 166 version
new 8bd316a Clean up imported code
new 4c31464 Merge in changes from Tomcat's extended ThreadPoolExecutor
new 0f2e084 Fix BZ 65454. Correct a timing issue that could delay a request
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
java/org/apache/tomcat/util/threads/TaskQueue.java | 24 -
.../tomcat/util/threads/ThreadPoolExecutor.java | 2335 +++++++++++++++++++-
webapps/docs/changelog.xml | 6 +
3 files changed, 2249 insertions(+), 116 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
[tomcat] 02/05: Replace Tomcat's ThreadPoolExecutor with the
current JSR 166 version
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
markt pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit c03c7c82d61d3b6e5f42e022db4435d0bfce04f5
Author: Mark Thomas <ma...@apache.org>
AuthorDate: Wed Jul 21 15:59:30 2021 +0100
Replace Tomcat's ThreadPoolExecutor with the current JSR 166 version
Note: This commit will not compile.
---
.../tomcat/util/threads/ThreadPoolExecutor.java | 2213 ++++++++++++++++++--
1 file changed, 2076 insertions(+), 137 deletions(-)
diff --git a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
index b6d5105..de9e1a8 100644
--- a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
+++ b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
@@ -1,211 +1,2150 @@
+
/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * 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/publicdomain/zero/1.0/
*/
-package org.apache.tomcat.util.threads;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.RejectedExecutionHandler;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
+package java.util.concurrent;
-import org.apache.tomcat.util.res.StringManager;
+import java.util.ArrayList;
+import java.util.ConcurrentModificationException;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.AbstractQueuedSynchronizer;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
/**
- * Same as a java.util.concurrent.ThreadPoolExecutor but implements a much more efficient
- * {@link #getSubmittedCount()} method, to be used to properly handle the work queue.
- * If a RejectedExecutionHandler is not specified a default one will be configured
- * and that one will always throw a RejectedExecutionException
+ * An {@link ExecutorService} that executes each submitted task using
+ * one of possibly several pooled threads, normally configured
+ * using {@link Executors} factory methods.
+ *
+ * <p>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.
+ *
+ * <p>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:
+ *
+ * <dl>
+ *
+ * <dt>Core and maximum pool sizes</dt>
+ *
+ * <dd>A {@code ThreadPoolExecutor} will automatically adjust the
+ * pool size (see {@link #getPoolSize})
+ * according to the bounds set by
+ * corePoolSize (see {@link #getCorePoolSize}) and
+ * maximumPoolSize (see {@link #getMaximumPoolSize}).
+ *
+ * When a new task is submitted in method {@link #execute(Runnable)},
+ * if fewer than corePoolSize threads are running, a new thread is
+ * created to handle the request, even if other worker threads are
+ * idle. Else if fewer than maximumPoolSize threads are running, a
+ * new thread will be created to handle the request only if the queue
+ * is full. By setting corePoolSize and maximumPoolSize the same, you
+ * create a fixed-size thread pool. By setting maximumPoolSize to an
+ * essentially unbounded value such as {@code Integer.MAX_VALUE}, you
+ * allow the pool to accommodate an arbitrary number of concurrent
+ * tasks. Most typically, core and maximum pool sizes are set only
+ * upon construction, but they may also be changed dynamically using
+ * {@link #setCorePoolSize} and {@link #setMaximumPoolSize}. </dd>
+ *
+ * <dt>On-demand construction</dt>
+ *
+ * <dd>By default, even core threads are initially created and
+ * started only when new tasks arrive, but this can be overridden
+ * dynamically using method {@link #prestartCoreThread} or {@link
+ * #prestartAllCoreThreads}. You probably want to prestart threads if
+ * you construct the pool with a non-empty queue. </dd>
+ *
+ * <dt>Creating new threads</dt>
+ *
+ * <dd>New threads are created using a {@link ThreadFactory}. If not
+ * otherwise specified, a {@link Executors#defaultThreadFactory} is
+ * used, that creates threads to all be in the same {@link
+ * ThreadGroup} and with the same {@code NORM_PRIORITY} priority and
+ * non-daemon status. By supplying a different ThreadFactory, you can
+ * alter the thread's name, thread group, priority, daemon status,
+ * etc. If a {@code ThreadFactory} fails to create a thread when asked
+ * by returning null from {@code newThread}, the executor will
+ * continue, but might not be able to execute any tasks. Threads
+ * should possess the "modifyThread" {@code RuntimePermission}. If
+ * worker threads or other threads using the pool do not possess this
+ * permission, service may be degraded: configuration changes may not
+ * take effect in a timely manner, and a shutdown pool may remain in a
+ * state in which termination is possible but not completed.</dd>
+ *
+ * <dt>Keep-alive times</dt>
+ *
+ * <dd>If the pool currently has more than corePoolSize threads,
+ * excess threads will be terminated if they have been idle for more
+ * than the keepAliveTime (see {@link #getKeepAliveTime(TimeUnit)}).
+ * This provides a means of reducing resource consumption when the
+ * pool is not being actively used. If the pool becomes more active
+ * later, new threads will be constructed. This parameter can also be
+ * changed dynamically using method {@link #setKeepAliveTime(long,
+ * TimeUnit)}. Using a value of {@code Long.MAX_VALUE} {@link
+ * TimeUnit#NANOSECONDS} effectively disables idle threads from ever
+ * terminating prior to shut down. By default, the keep-alive policy
+ * applies only when there are more than corePoolSize threads, but
+ * method {@link #allowCoreThreadTimeOut(boolean)} can be used to
+ * apply this time-out policy to core threads as well, so long as the
+ * keepAliveTime value is non-zero. </dd>
+ *
+ * <dt>Queuing</dt>
+ *
+ * <dd>Any {@link BlockingQueue} may be used to transfer and hold
+ * submitted tasks. The use of this queue interacts with pool sizing:
+ *
+ * <ul>
+ *
+ * <li>If fewer than corePoolSize threads are running, the Executor
+ * always prefers adding a new thread
+ * rather than queuing.
+ *
+ * <li>If corePoolSize or more threads are running, the Executor
+ * always prefers queuing a request rather than adding a new
+ * thread.
+ *
+ * <li>If a request cannot be queued, a new thread is created unless
+ * this would exceed maximumPoolSize, in which case, the task will be
+ * rejected.
+ *
+ * </ul>
+ *
+ * There are three general strategies for queuing:
+ * <ol>
+ *
+ * <li><em> Direct handoffs.</em> A good default choice for a work
+ * queue is a {@link SynchronousQueue} that hands off tasks to threads
+ * without otherwise holding them. Here, an attempt to queue a task
+ * will fail if no threads are immediately available to run it, so a
+ * new thread will be constructed. This policy avoids lockups when
+ * handling sets of requests that might have internal dependencies.
+ * Direct handoffs generally require unbounded maximumPoolSizes to
+ * avoid rejection of new submitted tasks. This in turn admits the
+ * possibility of unbounded thread growth when commands continue to
+ * arrive on average faster than they can be processed.
+ *
+ * <li><em> Unbounded queues.</em> Using an unbounded queue (for
+ * example a {@link LinkedBlockingQueue} without a predefined
+ * capacity) will cause new tasks to wait in the queue when all
+ * corePoolSize threads are busy. Thus, no more than corePoolSize
+ * threads will ever be created. (And the value of the maximumPoolSize
+ * therefore doesn't have any effect.) This may be appropriate when
+ * each task is completely independent of others, so tasks cannot
+ * affect each others execution; for example, in a web page server.
+ * While this style of queuing can be useful in smoothing out
+ * transient bursts of requests, it admits the possibility of
+ * unbounded work queue growth when commands continue to arrive on
+ * average faster than they can be processed.
+ *
+ * <li><em>Bounded queues.</em> A bounded queue (for example, an
+ * {@link ArrayBlockingQueue}) helps prevent resource exhaustion when
+ * used with finite maximumPoolSizes, but can be more difficult to
+ * tune and control. Queue sizes and maximum pool sizes may be traded
+ * off for each other: Using large queues and small pools minimizes
+ * CPU usage, OS resources, and context-switching overhead, but can
+ * lead to artificially low throughput. If tasks frequently block (for
+ * example if they are I/O bound), a system may be able to schedule
+ * time for more threads than you otherwise allow. Use of small queues
+ * generally requires larger pool sizes, which keeps CPUs busier but
+ * may encounter unacceptable scheduling overhead, which also
+ * decreases throughput.
+ *
+ * </ol>
+ *
+ * </dd>
*
+ * <dt>Rejected tasks</dt>
+ *
+ * <dd>New tasks submitted in method {@link #execute(Runnable)} will be
+ * <em>rejected</em> when the Executor has been shut down, and also when
+ * the Executor uses finite bounds for both maximum threads and work queue
+ * capacity, and is saturated. In either case, the {@code execute} method
+ * invokes the {@link
+ * RejectedExecutionHandler#rejectedExecution(Runnable, ThreadPoolExecutor)}
+ * method of its {@link RejectedExecutionHandler}. Four predefined handler
+ * policies are provided:
+ *
+ * <ol>
+ *
+ * <li>In the default {@link ThreadPoolExecutor.AbortPolicy}, the handler
+ * throws a runtime {@link RejectedExecutionException} upon rejection.
+ *
+ * <li>In {@link ThreadPoolExecutor.CallerRunsPolicy}, the thread
+ * that invokes {@code execute} itself runs the task. This provides a
+ * simple feedback control mechanism that will slow down the rate that
+ * new tasks are submitted.
+ *
+ * <li>In {@link ThreadPoolExecutor.DiscardPolicy}, a task that cannot
+ * be executed is simply dropped. This policy is designed only for
+ * those rare cases in which task completion is never relied upon.
+ *
+ * <li>In {@link ThreadPoolExecutor.DiscardOldestPolicy}, if the
+ * executor is not shut down, the task at the head of the work queue
+ * is dropped, and then execution is retried (which can fail again,
+ * causing this to be repeated.) This policy is rarely acceptable. In
+ * nearly all cases, you should also cancel the task to cause an
+ * exception in any component waiting for its completion, and/or log
+ * the failure, as illustrated in {@link
+ * ThreadPoolExecutor.DiscardOldestPolicy} documentation.
+ *
+ * </ol>
+ *
+ * It is possible to define and use other kinds of {@link
+ * RejectedExecutionHandler} classes. Doing so requires some care
+ * especially when policies are designed to work only under particular
+ * capacity or queuing policies. </dd>
+ *
+ * <dt>Hook methods</dt>
+ *
+ * <dd>This class provides {@code protected} overridable
+ * {@link #beforeExecute(Thread, Runnable)} and
+ * {@link #afterExecute(Runnable, Throwable)} methods that are called
+ * before and after execution of each task. These can be used to
+ * manipulate the execution environment; for example, reinitializing
+ * ThreadLocals, gathering statistics, or adding log entries.
+ * Additionally, method {@link #terminated} can be overridden to perform
+ * any special processing that needs to be done once the Executor has
+ * fully terminated.
+ *
+ * <p>If hook, callback, or BlockingQueue methods throw exceptions,
+ * internal worker threads may in turn fail, abruptly terminate, and
+ * possibly be replaced.</dd>
+ *
+ * <dt>Queue maintenance</dt>
+ *
+ * <dd>Method {@link #getQueue()} allows access to the work queue
+ * for purposes of monitoring and debugging. Use of this method for
+ * any other purpose is strongly discouraged. Two supplied methods,
+ * {@link #remove(Runnable)} and {@link #purge} are available to
+ * assist in storage reclamation when large numbers of queued tasks
+ * become cancelled.</dd>
+ *
+ * <dt>Reclamation</dt>
+ *
+ * <dd>A pool that is no longer referenced in a program <em>AND</em>
+ * has no remaining threads may be reclaimed (garbage collected)
+ * without being explicitly shutdown. You can configure a pool to
+ * allow all unused threads to eventually die by setting appropriate
+ * keep-alive times, using a lower bound of zero core threads and/or
+ * setting {@link #allowCoreThreadTimeOut(boolean)}. </dd>
+ *
+ * </dl>
+ *
+ * <p><b>Extension example.</b> 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:
+ *
+ * <pre> {@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();
+ * }
+ * }
+ * }}</pre>
+ *
+ * @since 1.5
+ * @author Doug Lea
*/
-public class ThreadPoolExecutor extends java.util.concurrent.ThreadPoolExecutor {
+public class ThreadPoolExecutor extends AbstractExecutorService {
+ /**
+ * The main pool control state, ctl, is an atomic integer packing
+ * two conceptual fields
+ * workerCount, indicating the effective number of threads
+ * runState, indicating whether running, shutting down etc
+ *
+ * In order to pack them into one int, we limit workerCount to
+ * (2^29)-1 (about 500 million) threads rather than (2^31)-1 (2
+ * billion) otherwise representable. If this is ever an issue in
+ * the future, the variable can be changed to be an AtomicLong,
+ * and the shift/mask constants below adjusted. But until the need
+ * arises, this code is a bit faster and simpler using an int.
+ *
+ * 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:
+ *
+ * RUNNING: Accept new tasks and process queued tasks
+ * SHUTDOWN: Don't accept new tasks, but process queued tasks
+ * STOP: Don't accept new tasks, don't process queued tasks,
+ * and interrupt in-progress tasks
+ * TIDYING: All tasks have terminated, workerCount is zero,
+ * the thread transitioning to state TIDYING
+ * will run the terminated() hook method
+ * TERMINATED: terminated() has completed
+ *
+ * The numerical order among these values matters, to allow
+ * ordered comparisons. The runState monotonically increases over
+ * time, but need not hit each state. The transitions are:
+ *
+ * RUNNING -> SHUTDOWN
+ * On invocation of shutdown()
+ * (RUNNING or SHUTDOWN) -> STOP
+ * On invocation of shutdownNow()
+ * SHUTDOWN -> TIDYING
+ * When both queue and pool are empty
+ * STOP -> TIDYING
+ * When pool is empty
+ * TIDYING -> TERMINATED
+ * When the terminated() hook method has completed
+ *
+ * Threads waiting in awaitTermination() will return when the
+ * state reaches TERMINATED.
+ *
+ * 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 << COUNT_BITS;
+ 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 runStateOf(int c) { return c & ~COUNT_MASK; }
+ 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<Runnable> workQueue;
+
+ /**
+ * Lock held on access to workers set and related bookkeeping.
+ * While we could use a concurrent set of some sort, it turns out
+ * to be generally preferable to use a lock. Among the reasons is
+ * that this serializes interruptIdleWorkers, which avoids
+ * unnecessary interrupt storms, especially during shutdown.
+ * Otherwise exiting threads would concurrently interrupt those
+ * that have not yet interrupted. It also simplifies some of the
+ * associated statistics bookkeeping of largestPoolSize etc. We
+ * also hold mainLock on shutdown and shutdownNow, for the sake of
+ * ensuring workers set is stable while separately checking
+ * permission to interrupt and actually interrupting.
+ */
+ private final ReentrantLock mainLock = new ReentrantLock();
+
+ /**
+ * Set containing all worker threads in pool. Accessed only when
+ * holding mainLock.
+ */
+ private final HashSet<Worker> workers = new HashSet<>();
+
+ /**
+ * Wait condition to support awaitTermination.
+ */
+ private final Condition termination = mainLock.newCondition();
+
+ /**
+ * Tracks largest attained pool size. Accessed only under
+ * mainLock.
+ */
+ private int largestPoolSize;
+
+ /**
+ * Counter for completed tasks. Updated only on termination of
+ * worker threads. Accessed only under mainLock.
+ */
+ private long completedTaskCount;
+
+ /*
+ * All user control parameters are declared as volatiles so that
+ * ongoing actions are based on freshest values, but without need
+ * for locking, since no internal invariants depend on them
+ * changing synchronously with respect to other actions.
+ */
+
+ /**
+ * Factory for new threads. All threads are created using this
+ * factory (via method addWorker). All callers must be prepared
+ * for addWorker to fail, which may reflect a system or user's
+ * policy limiting the number of threads. Even though it is not
+ * treated as an error, failure to create threads may result in
+ * new tasks being rejected or existing ones remaining stuck in
+ * the queue.
+ *
+ * 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;
+
/**
- * The string manager for this package.
+ * If false (default), core threads stay alive even when idle.
+ * If true, core threads use keepAliveTime to time out waiting
+ * for work.
*/
- protected static final StringManager sm = StringManager
- .getManager("org.apache.tomcat.util.threads.res");
+ private volatile boolean allowCoreThreadTimeOut;
/**
- * The number of tasks submitted but not yet finished. This includes tasks
- * in the queue and tasks that have been handed to a worker thread but the
- * latter did not start executing the task yet.
- * This number is always greater or equal to {@link #getActiveCount()}.
+ * 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 final AtomicInteger submittedCount = new AtomicInteger(0);
- private final AtomicLong lastContextStoppedTime = new AtomicLong(0L);
+ private volatile int corePoolSize;
/**
- * Most recent time in ms when a thread decided to kill itself to avoid
- * potential memory leaks. Useful to throttle the rate of renewals of
- * threads.
+ * Maximum pool size.
+ *
+ * Since the worker count is actually stored in COUNT_BITS bits,
+ * the effective limit is {@code maximumPoolSize & COUNT_MASK}.
*/
- private final AtomicLong lastTimeThreadKilledItself = new AtomicLong(0L);
+ private volatile int maximumPoolSize;
/**
- * Delay in ms between 2 threads being renewed. If negative, do not renew threads.
+ * The default rejected execution handler.
*/
- private long threadRenewalDelay = Constants.DEFAULT_THREAD_RENEWAL_DELAY;
+ private static final RejectedExecutionHandler defaultHandler =
+ new AbortPolicy();
+
+ /**
+ * Permission required for callers of shutdown and shutdownNow.
+ * We additionally require (see checkShutdownAccess) that callers
+ * have permission to actually interrupt threads in the worker set
+ * (as governed by Thread.interrupt, which relies on
+ * ThreadGroup.checkAccess, which in turn relies on
+ * SecurityManager.checkAccess). Shutdowns are attempted only if
+ * these checks pass.
+ *
+ * All actual invocations of Thread.interrupt (see
+ * interruptIdleWorkers and interruptWorkers) ignore
+ * SecurityExceptions, meaning that the attempted interrupts
+ * silently fail. In the case of shutdown, they should not fail
+ * unless the SecurityManager has inconsistent policies, sometimes
+ * allowing access to a thread and sometimes not. In such cases,
+ * failure to actually interrupt threads may disable or delay full
+ * termination. Other uses of interruptIdleWorkers are advisory,
+ * and failure to actually interrupt will merely delay response to
+ * configuration changes so is not handled exceptionally.
+ */
+ private static final RuntimePermission shutdownPerm =
+ new RuntimePermission("modifyThread");
+
+ /**
+ * 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.
+ */
+ private static final long serialVersionUID = 6138294804551838833L;
+
+ /** Thread this worker is running in. Null if factory fails. */
+ @SuppressWarnings("serial") // Unlikely to be serializable
+ final Thread thread;
+ /** Initial task to run. Possibly null. */
+ @SuppressWarnings("serial") // Not statically typed as Serializable
+ Runnable firstTask;
+ /** Per-thread task counter */
+ volatile long completedTasks;
- public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) {
- super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
- prestartAllCoreThreads();
+ // 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. */
+ public void run() {
+ runWorker(this);
+ }
+
+ // Lock methods
+ //
+ // The value 0 represents the unlocked state.
+ // The value 1 represents the locked state.
+
+ protected boolean isHeldExclusively() {
+ return getState() != 0;
+ }
+
+ protected boolean tryAcquire(int unused) {
+ if (compareAndSetState(0, 1)) {
+ setExclusiveOwnerThread(Thread.currentThread());
+ return true;
+ }
+ return false;
+ }
+
+ 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;
+ }
+ }
}
- public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
- RejectedExecutionHandler handler) {
- super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
- prestartAllCoreThreads();
+ /**
+ * 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
+ }
}
- public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
- super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, new RejectHandler());
- prestartAllCoreThreads();
+ /*
+ * Methods for controlling interrupts to worker threads.
+ */
+
+ /**
+ * If there is a security manager, makes sure caller has
+ * permission to shut down threads in general (see shutdownPerm).
+ * If this passes, additionally makes sure the caller is allowed
+ * to interrupt each worker thread. This might not be true even if
+ * first check passed, if the SecurityManager treats some threads
+ * specially.
+ */
+ private void checkShutdownAccess() {
+ // assert mainLock.isHeldByCurrentThread();
+ SecurityManager security = System.getSecurityManager();
+ if (security != null) {
+ security.checkPermission(shutdownPerm);
+ for (Worker w : workers) {
+ security.checkAccess(w.thread);
+ }
+ }
}
- public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
- super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new RejectHandler());
- prestartAllCoreThreads();
+ /**
+ * 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();
+ }
}
- public long getThreadRenewalDelay() {
- return threadRenewalDelay;
+ /**
+ * 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();
+ }
}
- public void setThreadRenewalDelay(long threadRenewalDelay) {
- this.threadRenewalDelay = threadRenewalDelay;
+ /**
+ * Common form of interruptIdleWorkers, to avoid having to
+ * remember what the boolean argument means.
+ */
+ private void interruptIdleWorkers() {
+ interruptIdleWorkers(false);
}
- @Override
- 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();
+ 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<Runnable> drainQueue() {
+ BlockingQueue<Runnable> q = workQueue;
+ ArrayList<Runnable> taskList = new ArrayList<>();
+ q.drainTo(taskList);
+ if (!q.isEmpty()) {
+ for (Runnable r : q.toArray(new Runnable[0])) {
+ if (q.remove(r)) {
+ taskList.add(r);
+ }
+ }
}
+ return taskList;
+ }
+
+ /*
+ * Methods for creating, running and cleaning up after workers
+ */
- if (t == null) {
- stopCurrentThreadIfNeeded();
+ /**
+ * Checks if a new worker can be added with respect to current
+ * pool state and the given bound (either core or maximum). If so,
+ * the worker count is adjusted accordingly, and, if possible, a
+ * new worker is created and started, running firstTask as its
+ * first task. This method returns false if the pool is stopped or
+ * eligible to shut down. It also returns false if the thread
+ * factory fails to create a thread when asked. If the thread
+ * creation fails, either due to the thread factory returning
+ * null, or due to an exception (typically OutOfMemoryError in
+ * Thread.start()), we roll back cleanly.
+ *
+ * @param firstTask the task the new thread should run first (or
+ * null if none). Workers are created with an initial first task
+ * (in method execute()) to bypass queuing when there are fewer
+ * than corePoolSize threads (in which case we always start one),
+ * or when the queue is full (in which case we must bypass queue).
+ * Initially idle threads are usually created via
+ * prestartCoreThread or to replace other dying workers.
+ *
+ * @param core if true use corePoolSize as bound, else
+ * maximumPoolSize. (A boolean indicator is used here rather than a
+ * value to ensure reads of fresh values after checking other pool
+ * state).
+ * @return true if successful
+ */
+ private boolean addWorker(Runnable firstTask, boolean core) {
+ retry:
+ for (int c = ctl.get();;) {
+ // Check if queue empty only if necessary.
+ if (runStateAtLeast(c, SHUTDOWN)
+ && (runStateAtLeast(c, STOP)
+ || firstTask != null
+ || workQueue.isEmpty())) {
+ return false;
+ }
+
+ for (;;) {
+ if (workerCountOf(c)
+ >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK)) {
+ return false;
+ }
+ if (compareAndIncrementWorkerCount(c)) {
+ break retry;
+ }
+ c = ctl.get(); // Re-read ctl
+ if (runStateAtLeast(c, SHUTDOWN))
+ {
+ continue retry;
+ // else CAS failed due to workerCount change; retry inner loop
+ }
+ }
+ }
+
+ boolean workerStarted = false;
+ boolean workerAdded = false;
+ Worker w = null;
+ try {
+ w = new Worker(firstTask);
+ final Thread t = w.thread;
+ if (t != null) {
+ final ReentrantLock mainLock = this.mainLock;
+ mainLock.lock();
+ try {
+ // Recheck while holding lock.
+ // Back out on ThreadFactory failure or if
+ // shut down before lock acquired.
+ int c = ctl.get();
+
+ if (isRunning(c) ||
+ (runStateLessThan(c, STOP) && firstTask == null)) {
+ if (t.getState() != Thread.State.NEW) {
+ throw new IllegalThreadStateException();
+ }
+ workers.add(w);
+ workerAdded = true;
+ int s = workers.size();
+ if (s > largestPoolSize) {
+ largestPoolSize = s;
+ }
+ }
+ } finally {
+ mainLock.unlock();
+ }
+ if (workerAdded) {
+ t.start();
+ workerStarted = true;
+ }
+ }
+ } finally {
+ if (! workerStarted) {
+ addWorkerFailed(w);
+ }
+ }
+ return workerStarted;
+ }
+
+ /**
+ * Rolls back the worker thread creation.
+ * - removes worker from workers, if present
+ * - decrements worker count
+ * - rechecks for termination, in case the existence of this
+ * worker was holding up termination
+ */
+ private void addWorkerFailed(Worker w) {
+ final ReentrantLock mainLock = this.mainLock;
+ mainLock.lock();
+ try {
+ if (w != null) {
+ workers.remove(w);
+ }
+ decrementWorkerCount();
+ tryTerminate();
+ } finally {
+ mainLock.unlock();
}
}
/**
- * 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.
+ * 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
*/
- 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
+ private void processWorkerExit(Worker w, boolean completedAbruptly) {
+ if (completedAbruptly) {
+ decrementWorkerCount();
+ }
- final String msg = sm.getString(
- "threadPoolExecutor.threadStoppedToAvoidPotentialLeak",
- Thread.currentThread().getName());
+ final ReentrantLock mainLock = this.mainLock;
+ mainLock.lock();
+ try {
+ completedTaskCount += w.completedTasks;
+ workers.remove(w);
+ } finally {
+ mainLock.unlock();
+ }
+
+ tryTerminate();
- throw new StopPooledThreadException(msg);
+ int c = ctl.get();
+ if (runStateLessThan(c, STOP)) {
+ if (!completedAbruptly) {
+ int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
+ if (min == 0 && ! workQueue.isEmpty()) {
+ min = 1;
+ }
+ if (workerCountOf(c) >= min)
+ {
+ return; // replacement not needed
}
}
+ addWorker(null, false);
}
}
- protected boolean currentThreadShouldBeStopped() {
- if (threadRenewalDelay >= 0
- && Thread.currentThread() instanceof TaskThread) {
- TaskThread currentTaskThread = (TaskThread) Thread.currentThread();
- if (currentTaskThread.getCreationTime() <
- this.lastContextStoppedTime.longValue()) {
- return true;
+ /**
+ * 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:
+ * 1. There are more than maximumPoolSize workers (due to
+ * a call to setMaximumPoolSize).
+ * 2. The pool is stopped.
+ * 3. The pool is shutdown and the queue is empty.
+ * 4. This worker timed out waiting for a task, and timed-out
+ * workers are subject to termination (that is,
+ * {@code allowCoreThreadTimeOut || workerCount > corePoolSize})
+ * both before and after the timed wait, and if the queue is
+ * non-empty, this worker is not the last thread in the pool.
+ *
+ * @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;
}
}
- return false;
}
- public int getSubmittedCount() {
- return submittedCount.get();
+ /**
+ * Main worker run loop. Repeatedly gets tasks from queue and
+ * executes them, while coping with a number of issues:
+ *
+ * 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.AbortPolicy
+ * default rejected execution handler}.
+ *
+ * <p>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:<br>
+ * {@code corePoolSize < 0}<br>
+ * {@code keepAliveTime < 0}<br>
+ * {@code maximumPoolSize <= 0}<br>
+ * {@code maximumPoolSize < corePoolSize}
+ * @throws NullPointerException if {@code workQueue} is null
+ */
+ public ThreadPoolExecutor(int corePoolSize,
+ int maximumPoolSize,
+ long keepAliveTime,
+ TimeUnit unit,
+ BlockingQueue<Runnable> workQueue) {
+ this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
+ Executors.defaultThreadFactory(), defaultHandler);
+ }
+
+ /**
+ * Creates a new {@code ThreadPoolExecutor} with the given initial
+ * parameters and the {@linkplain ThreadPoolExecutor.AbortPolicy
+ * default rejected execution handler}.
+ *
+ * @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.
+ * @param threadFactory the factory to use when the executor
+ * creates a new thread
+ * @throws IllegalArgumentException if one of the following holds:<br>
+ * {@code corePoolSize < 0}<br>
+ * {@code keepAliveTime < 0}<br>
+ * {@code maximumPoolSize <= 0}<br>
+ * {@code maximumPoolSize < corePoolSize}
+ * @throws NullPointerException if {@code workQueue}
+ * or {@code threadFactory} is null
+ */
+ public ThreadPoolExecutor(int corePoolSize,
+ int maximumPoolSize,
+ long keepAliveTime,
+ TimeUnit unit,
+ BlockingQueue<Runnable> workQueue,
+ ThreadFactory threadFactory) {
+ this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
+ threadFactory, defaultHandler);
+ }
+
+ /**
+ * Creates a new {@code ThreadPoolExecutor} with the given initial
+ * parameters and the
+ * {@linkplain Executors#defaultThreadFactory default thread factory}.
+ *
+ * @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.
+ * @param handler the handler to use when execution is blocked
+ * because the thread bounds and queue capacities are reached
+ * @throws IllegalArgumentException if one of the following holds:<br>
+ * {@code corePoolSize < 0}<br>
+ * {@code keepAliveTime < 0}<br>
+ * {@code maximumPoolSize <= 0}<br>
+ * {@code maximumPoolSize < corePoolSize}
+ * @throws NullPointerException if {@code workQueue}
+ * or {@code handler} is null
+ */
+ public ThreadPoolExecutor(int corePoolSize,
+ int maximumPoolSize,
+ long keepAliveTime,
+ TimeUnit unit,
+ BlockingQueue<Runnable> workQueue,
+ RejectedExecutionHandler handler) {
+ this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
+ Executors.defaultThreadFactory(), handler);
+ }
+
+ /**
+ * Creates a new {@code ThreadPoolExecutor} with the given initial
+ * parameters.
+ *
+ * @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.
+ * @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 IllegalArgumentException if one of the following holds:<br>
+ * {@code corePoolSize < 0}<br>
+ * {@code keepAliveTime < 0}<br>
+ * {@code maximumPoolSize <= 0}<br>
+ * {@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<Runnable> workQueue,
+ ThreadFactory threadFactory,
+ RejectedExecutionHandler handler) {
+ if (corePoolSize < 0 ||
+ maximumPoolSize <= 0 ||
+ maximumPoolSize < corePoolSize ||
+ keepAliveTime < 0) {
+ throw new IllegalArgumentException();
+ }
+ if (workQueue == null || threadFactory == null || handler == null) {
+ throw new NullPointerException();
+ }
+ this.corePoolSize = corePoolSize;
+ this.maximumPoolSize = maximumPoolSize;
+ this.workQueue = workQueue;
+ this.keepAliveTime = unit.toNanos(keepAliveTime);
+ this.threadFactory = threadFactory;
+ this.handler = handler;
}
/**
- * {@inheritDoc}
+ * Executes the given task sometime in the future. The task
+ * may execute in a new thread or in an existing pooled thread.
+ *
+ * If the task cannot be submitted for execution, either because this
+ * executor has been shutdown or because its capacity has been reached,
+ * the task is handled by the current {@link RejectedExecutionHandler}.
+ *
+ * @param command the task to execute
+ * @throws RejectedExecutionException at discretion of
+ * {@code RejectedExecutionHandler}, if the task
+ * cannot be accepted for execution
+ * @throws NullPointerException if {@code command} is null
*/
- @Override
public void execute(Runnable command) {
- submittedCount.incrementAndGet();
+ if (command == null) {
+ throw new NullPointerException();
+ }
+ /*
+ * Proceed in 3 steps:
+ *
+ * 1. If fewer than corePoolSize threads are running, try to
+ * start a new thread with the given command as its first
+ * task. The call to addWorker atomically checks runState and
+ * workerCount, and so prevents false alarms that would add
+ * threads when it shouldn't, by returning false.
+ *
+ * 2. If a task can be successfully queued, then we still need
+ * to double-check whether we should have added a thread
+ * (because existing ones died since last checking) or that
+ * the pool shut down since entry into this method. So we
+ * recheck state and if necessary roll back the enqueuing if
+ * stopped, or start a new thread if there are none.
+ *
+ * 3. If we cannot queue task, then we try to add a new
+ * thread. If it fails, we know we are shut down or saturated
+ * and so reject the task.
+ */
+ int c = ctl.get();
+ if (workerCountOf(c) < corePoolSize) {
+ if (addWorker(command, true)) {
+ return;
+ }
+ c = ctl.get();
+ }
+ if (isRunning(c) && workQueue.offer(command)) {
+ int recheck = ctl.get();
+ if (! isRunning(recheck) && remove(command)) {
+ reject(command);
+ } else if (workerCountOf(recheck) == 0) {
+ addWorker(null, false);
+ }
+ }
+ else if (!addWorker(command, false)) {
+ reject(command);
+ }
+ }
+
+ /**
+ * Initiates an orderly shutdown in which previously submitted
+ * tasks are executed, but no new tasks will be accepted.
+ * Invocation has no additional effect if already shut down.
+ *
+ * <p>This method does not wait for previously submitted tasks to
+ * complete execution. Use {@link #awaitTermination awaitTermination}
+ * to do that.
+ *
+ * @throws SecurityException {@inheritDoc}
+ */
+ public void shutdown() {
+ final ReentrantLock mainLock = this.mainLock;
+ mainLock.lock();
+ try {
+ checkShutdownAccess();
+ 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.
+ *
+ * <p>This method does not wait for actively executing tasks to
+ * terminate. Use {@link #awaitTermination awaitTermination} to
+ * do that.
+ *
+ * <p>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}
+ */
+ public List<Runnable> shutdownNow() {
+ List<Runnable> tasks;
+ final ReentrantLock mainLock = this.mainLock;
+ mainLock.lock();
+ try {
+ checkShutdownAccess();
+ advanceRunState(STOP);
+ interruptWorkers();
+ tasks = drainQueue();
+ } finally {
+ mainLock.unlock();
+ }
+ tryTerminate();
+ return tasks;
+ }
+
+ public boolean isShutdown() {
+ return runStateAtLeast(ctl.get(), SHUTDOWN);
+ }
+
+ /** Used by ScheduledThreadPoolExecutor. */
+ boolean isStopped() {
+ return runStateAtLeast(ctl.get(), STOP);
+ }
+
+ /**
+ * Returns true if this executor is in the process of terminating
+ * after {@link #shutdown} or {@link #shutdownNow} but has not
+ * completely terminated. This method may be useful for
+ * debugging. A return of {@code true} reported a sufficient
+ * period after shutdown may indicate that submitted tasks have
+ * ignored or suppressed interruption, causing this executor not
+ * to properly terminate.
+ *
+ * @return {@code true} if terminating but not yet terminated
+ */
+ public boolean isTerminating() {
+ int c = ctl.get();
+ return runStateAtLeast(c, SHUTDOWN) && runStateLessThan(c, TERMINATED);
+ }
+
+ public boolean isTerminated() {
+ return runStateAtLeast(ctl.get(), TERMINATED);
+ }
+
+ public boolean awaitTermination(long timeout, TimeUnit unit)
+ throws InterruptedException {
+ long nanos = unit.toNanos(timeout);
+ final ReentrantLock mainLock = this.mainLock;
+ mainLock.lock();
try {
- super.execute(command);
- } catch (RejectedExecutionException rx) {
- if (super.getQueue() instanceof TaskQueue) {
- // If the Executor is close to maximum pool size, concurrent
- // calls to execute() may result (due to Tomcat's use of
- // TaskQueue) in some tasks being rejected rather than queued.
- // If this happens, add them to the queue.
- final TaskQueue queue = (TaskQueue)super.getQueue();
- if (!queue.force(command)) {
- submittedCount.decrementAndGet();
- throw new RejectedExecutionException(sm.getString("threadPoolExecutor.queueFull"));
+ while (runStateLessThan(ctl.get(), TERMINATED)) {
+ if (nanos <= 0L) {
+ return false;
}
- } else {
- submittedCount.decrementAndGet();
- throw rx;
+ nanos = termination.awaitNanos(nanos);
}
+ return true;
+ } finally {
+ mainLock.unlock();
}
}
- public void contextStopping() {
- this.lastContextStoppedTime.set(System.currentTimeMillis());
+ // Override without "throws Throwable" for compatibility with subclasses
+ // whose finalize method invokes super.finalize() (as is recommended).
+ // Before JDK 11, finalize() had a non-empty method body.
- // save the current pool parameters to restore them later
- int savedCorePoolSize = this.getCorePoolSize();
- TaskQueue taskQueue =
- getQueue() instanceof TaskQueue ? (TaskQueue) getQueue() : null;
- if (taskQueue != null) {
- // note by slaurent : quite oddly threadPoolExecutor.setCorePoolSize
- // checks that queue.remainingCapacity()==0. I did not understand
- // why, but to get the intended effect of waking up idle threads, I
- // temporarily fake this condition.
- taskQueue.setForcedRemainingCapacity(0);
+ /**
+ * @implNote Previous versions of this class had a finalize method
+ * that shut down this executor, but in this version, finalize
+ * does nothing.
+ */
+ @Deprecated(since="9")
+ protected void finalize() {}
+
+ /**
+ * Sets the thread factory used to create new threads.
+ *
+ * @param threadFactory the new thread factory
+ * @throws NullPointerException if threadFactory is null
+ * @see #getThreadFactory
+ */
+ public void setThreadFactory(ThreadFactory threadFactory) {
+ if (threadFactory == null) {
+ throw new NullPointerException();
+ }
+ this.threadFactory = threadFactory;
+ }
+
+ /**
+ * Returns the thread factory used to create new threads.
+ *
+ * @return the current thread factory
+ * @see #setThreadFactory(ThreadFactory)
+ */
+ public ThreadFactory getThreadFactory() {
+ return threadFactory;
+ }
+
+ /**
+ * Sets a new handler for unexecutable tasks.
+ *
+ * @param handler the new handler
+ * @throws NullPointerException if handler is null
+ * @see #getRejectedExecutionHandler
+ */
+ public void setRejectedExecutionHandler(RejectedExecutionHandler handler) {
+ if (handler == null) {
+ throw new NullPointerException();
}
+ this.handler = handler;
+ }
+
+ /**
+ * Returns the current handler for unexecutable tasks.
+ *
+ * @return the current handler
+ * @see #setRejectedExecutionHandler(RejectedExecutionHandler)
+ */
+ public RejectedExecutionHandler getRejectedExecutionHandler() {
+ return handler;
+ }
+
+ /**
+ * Sets the core number of threads. This overrides any value set
+ * in the constructor. If the new value is smaller than the
+ * current value, excess existing threads will be terminated when
+ * they next become idle. If larger, new threads will, if needed,
+ * be started to execute any queued tasks.
+ *
+ * @param corePoolSize the new core size
+ * @throws IllegalArgumentException if {@code corePoolSize < 0}
+ * or {@code corePoolSize} is greater than the {@linkplain
+ * #getMaximumPoolSize() maximum pool size}
+ * @see #getCorePoolSize
+ */
+ public void setCorePoolSize(int corePoolSize) {
+ if (corePoolSize < 0 || maximumPoolSize < corePoolSize) {
+ throw new IllegalArgumentException();
+ }
+ int delta = corePoolSize - this.corePoolSize;
+ this.corePoolSize = corePoolSize;
+ if (workerCountOf(ctl.get()) > corePoolSize) {
+ interruptIdleWorkers();
+ } else if (delta > 0) {
+ // We don't really know how many new threads are "needed".
+ // As a heuristic, prestart enough new workers (up to new
+ // core size) to handle the current number of tasks in
+ // queue, but stop if queue becomes empty while doing so.
+ int k = Math.min(delta, workQueue.size());
+ while (k-- > 0 && addWorker(null, true)) {
+ if (workQueue.isEmpty()) {
+ break;
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns the core number of threads.
+ *
+ * @return the core number of threads
+ * @see #setCorePoolSize
+ */
+ public int getCorePoolSize() {
+ return corePoolSize;
+ }
- // setCorePoolSize(0) wakes idle threads
- this.setCorePoolSize(0);
+ /**
+ * Starts a core thread, causing it to idly wait for work. This
+ * overrides the default policy of starting core threads only when
+ * new tasks are executed. This method will return {@code false}
+ * if all core threads have already been started.
+ *
+ * @return {@code true} if a thread was started
+ */
+ public boolean prestartCoreThread() {
+ return workerCountOf(ctl.get()) < corePoolSize &&
+ addWorker(null, true);
+ }
- // TaskQueue.take() takes care of timing out, so that we are sure that
- // all threads of the pool are renewed in a limited time, something like
- // (threadKeepAlive + longest request time)
+ /**
+ * Same as prestartCoreThread except arranges that at least one
+ * thread is started even if corePoolSize is 0.
+ */
+ void ensurePrestart() {
+ int wc = workerCountOf(ctl.get());
+ if (wc < corePoolSize) {
+ addWorker(null, true);
+ } else if (wc == 0) {
+ addWorker(null, false);
+ }
+ }
- if (taskQueue != null) {
- // ok, restore the state of the queue and pool
- taskQueue.resetForcedRemainingCapacity();
+ /**
+ * Starts all core threads, causing them to idly wait for work. This
+ * overrides the default policy of starting core threads only when
+ * new tasks are executed.
+ *
+ * @return the number of threads started
+ */
+ public int prestartAllCoreThreads() {
+ int n = 0;
+ while (addWorker(null, true)) {
+ ++n;
}
- this.setCorePoolSize(savedCorePoolSize);
+ return n;
+ }
+
+ /**
+ * Returns true if this pool allows core threads to time out and
+ * terminate if no tasks arrive within the keepAlive time, being
+ * replaced if needed when new tasks arrive. When true, the same
+ * keep-alive policy applying to non-core threads applies also to
+ * core threads. When false (the default), core threads are never
+ * terminated due to lack of incoming tasks.
+ *
+ * @return {@code true} if core threads are allowed to time out,
+ * else {@code false}
+ *
+ * @since 1.6
+ */
+ public boolean allowsCoreThreadTimeOut() {
+ return allowCoreThreadTimeOut;
}
- private static class RejectHandler implements RejectedExecutionHandler {
- @Override
- public void rejectedExecution(Runnable r,
- java.util.concurrent.ThreadPoolExecutor executor) {
- throw new RejectedExecutionException();
+ /**
+ * Sets the policy governing whether core threads may time out and
+ * terminate if no tasks arrive within the keep-alive time, being
+ * replaced if needed when new tasks arrive. When false, core
+ * threads are never terminated due to lack of incoming
+ * tasks. When true, the same keep-alive policy applying to
+ * non-core threads applies also to core threads. To avoid
+ * continual thread replacement, the keep-alive time must be
+ * greater than zero when setting {@code true}. This method
+ * should in general be called before the pool is actively used.
+ *
+ * @param value {@code true} if should time out, else {@code false}
+ * @throws IllegalArgumentException if value is {@code true}
+ * and the current keep-alive time is not greater than zero
+ *
+ * @since 1.6
+ */
+ public void allowCoreThreadTimeOut(boolean value) {
+ if (value && keepAliveTime <= 0) {
+ throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
}
+ if (value != allowCoreThreadTimeOut) {
+ allowCoreThreadTimeOut = value;
+ if (value) {
+ interruptIdleWorkers();
+ }
+ }
+ }
+ /**
+ * Sets the maximum allowed number of threads. This overrides any
+ * value set in the constructor. If the new value is smaller than
+ * the current value, excess existing threads will be
+ * terminated when they next become idle.
+ *
+ * @param maximumPoolSize the new maximum
+ * @throws IllegalArgumentException if the new maximum is
+ * less than or equal to zero, or
+ * less than the {@linkplain #getCorePoolSize core pool size}
+ * @see #getMaximumPoolSize
+ */
+ public void setMaximumPoolSize(int maximumPoolSize) {
+ if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize) {
+ throw new IllegalArgumentException();
+ }
+ this.maximumPoolSize = maximumPoolSize;
+ if (workerCountOf(ctl.get()) > maximumPoolSize) {
+ interruptIdleWorkers();
+ }
}
+ /**
+ * Returns the maximum allowed number of threads.
+ *
+ * @return the maximum allowed number of threads
+ * @see #setMaximumPoolSize
+ */
+ public int getMaximumPoolSize() {
+ return maximumPoolSize;
+ }
-}
+ /**
+ * Sets the thread keep-alive time, which is the amount of time
+ * that threads may remain idle before being terminated.
+ * Threads that wait this amount of time without processing a
+ * task will be terminated if there are more than the core
+ * number of threads currently in the pool, or if this pool
+ * {@linkplain #allowsCoreThreadTimeOut() allows core thread timeout}.
+ * This overrides any value set in the constructor.
+ *
+ * @param time the time to wait. A time value of zero will cause
+ * excess threads to terminate immediately after executing tasks.
+ * @param unit the time unit of the {@code time} argument
+ * @throws IllegalArgumentException if {@code time} less than zero or
+ * if {@code time} is zero and {@code allowsCoreThreadTimeOut}
+ * @see #getKeepAliveTime(TimeUnit)
+ */
+ public void setKeepAliveTime(long time, TimeUnit unit) {
+ if (time < 0) {
+ throw new IllegalArgumentException();
+ }
+ if (time == 0 && allowsCoreThreadTimeOut()) {
+ throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
+ }
+ long keepAliveTime = unit.toNanos(time);
+ long delta = keepAliveTime - this.keepAliveTime;
+ this.keepAliveTime = keepAliveTime;
+ if (delta < 0) {
+ interruptIdleWorkers();
+ }
+ }
+
+ /**
+ * Returns the thread keep-alive time, which is the amount of time
+ * that threads may remain idle before being terminated.
+ * Threads that wait this amount of time without processing a
+ * task will be terminated if there are more than the core
+ * number of threads currently in the pool, or if this pool
+ * {@linkplain #allowsCoreThreadTimeOut() allows core thread timeout}.
+ *
+ * @param unit the desired time unit of the result
+ * @return the time limit
+ * @see #setKeepAliveTime(long, TimeUnit)
+ */
+ public long getKeepAliveTime(TimeUnit unit) {
+ return unit.convert(keepAliveTime, TimeUnit.NANOSECONDS);
+ }
+
+ /* User-level queue utilities */
+
+ /**
+ * Returns the task queue used by this executor. Access to the
+ * task queue is intended primarily for debugging and monitoring.
+ * This queue may be in active use. Retrieving the task queue
+ * does not prevent queued tasks from executing.
+ *
+ * @return the task queue
+ */
+ public BlockingQueue<Runnable> getQueue() {
+ return workQueue;
+ }
+
+ /**
+ * Removes this task from the executor's internal queue if it is
+ * present, thus causing it not to be run if it has not already
+ * started.
+ *
+ * <p>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<Runnable> q = workQueue;
+ try {
+ Iterator<Runnable> it = q.iterator();
+ while (it.hasNext()) {
+ Runnable r = it.next();
+ if (r instanceof Future<?> && ((Future<?>)r).isCancelled()) {
+ it.remove();
+ }
+ }
+ } catch (ConcurrentModificationException fallThrough) {
+ // Take slow path if we encounter interference during traversal.
+ // Make copy for traversal and call remove for cancelled entries.
+ // The slow path is more likely to be O(N*N).
+ for (Object r : q.toArray()) {
+ if (r instanceof Future<?> && ((Future<?>)r).isCancelled()) {
+ q.remove(r);
+ }
+ }
+ }
+
+ tryTerminate(); // In case SHUTDOWN and now empty
+ }
+
+ /* Statistics */
+
+ /**
+ * Returns the current number of threads in the pool.
+ *
+ * @return the number of threads
+ */
+ public int getPoolSize() {
+ final ReentrantLock mainLock = this.mainLock;
+ mainLock.lock();
+ try {
+ // Remove rare and surprising possibility of
+ // isTerminated() && getPoolSize() > 0
+ return runStateAtLeast(ctl.get(), TIDYING) ? 0
+ : workers.size();
+ } finally {
+ mainLock.unlock();
+ }
+ }
+
+ /**
+ * 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();
+ }
+ }
+
+ /**
+ * 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
+ */
+ 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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p><b>Note:</b> When actions are enclosed in tasks (such as
+ * {@link 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 <em>not</em>
+ * 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:
+ *
+ * <pre> {@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);
+ * }
+ * }}</pre>
+ *
+ * @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) { }
+
+ /**
+ * 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
+ */
+ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
+ if (!e.isShutdown()) {
+ r.run();
+ }
+ }
+ }
+
+ /**
+ * A handler for rejected tasks that throws a
+ * {@link RejectedExecutionException}.
+ *
+ * This is the default handler for {@link ThreadPoolExecutor} and
+ * {@link ScheduledThreadPoolExecutor}.
+ */
+ 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
+ */
+ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
+ throw new RejectedExecutionException("Task " + r.toString() +
+ " rejected from " +
+ 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
+ */
+ 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:
+ * <pre> {@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
+ * }}}</pre>
+ */
+ 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
+ */
+ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
+ if (!e.isShutdown()) {
+ e.getQueue().poll();
+ e.execute(r);
+ }
+ }
+ }
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
[tomcat] 05/05: Fix BZ 65454. Correct a timing issue that could
delay a request
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
markt pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 0f2e084d1583cfa4c00cec1958ec30113fb4f41e
Author: Mark Thomas <ma...@apache.org>
AuthorDate: Wed Jul 21 17:45:01 2021 +0100
Fix BZ 65454. Correct a timing issue that could delay a request
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.
---
java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java | 10 ++++++++--
webapps/docs/changelog.xml | 6 ++++++
2 files changed, 14 insertions(+), 2 deletions(-)
diff --git a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
index 7b1ec55..8771c22 100644
--- a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
+++ b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
@@ -1050,8 +1050,14 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
if (min == 0 && ! workQueue.isEmpty()) {
min = 1;
}
- if (workerCountOf(c) >= min)
- {
+ // 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
}
}
diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml
index 3838cf3..cafd371 100644
--- a/webapps/docs/changelog.xml
+++ b/webapps/docs/changelog.xml
@@ -146,6 +146,12 @@
associated with the response until the connection timed out at which
point the final packet would be sent and the connection closed. (markt)
</fix>
+ <fix>
+ <bug>65454</bug>: Fix a race condition that could result in a delay to
+ a new request. The new request could be queued to wait for an existing
+ request to finish processing rather than the thread pool creating a new
+ thread to process the new request. (markt)
+ </fix>
</changelog>
</subsection>
<subsection name="Jasper">
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
[tomcat] 01/05: Remove deprecated code
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
markt pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 4289f4745cd831775abe0c5d1b08c9936889c47f
Author: Mark Thomas <ma...@apache.org>
AuthorDate: Wed Jul 21 17:27:12 2021 +0100
Remove deprecated code
---
java/org/apache/tomcat/util/threads/TaskQueue.java | 24 ----------------------
1 file changed, 24 deletions(-)
diff --git a/java/org/apache/tomcat/util/threads/TaskQueue.java b/java/org/apache/tomcat/util/threads/TaskQueue.java
index 3aad107..24c0ad6 100644
--- a/java/org/apache/tomcat/util/threads/TaskQueue.java
+++ b/java/org/apache/tomcat/util/threads/TaskQueue.java
@@ -76,30 +76,6 @@ public class TaskQueue extends LinkedBlockingQueue<Runnable> {
}
- /**
- * Used to add a task to the queue if the task has been rejected by the Executor.
- *
- * @param o The task to add to the queue
- * @param timeout The timeout to use when adding the task
- * @param unit The units in which the timeout is expressed
- *
- * @return {@code true} if the task was added to the queue,
- * otherwise {@code false}
- *
- * @throws InterruptedException If the call is interrupted before the
- * timeout expires
- *
- * @deprecated Unused. Will be removed in Tomcat 10.1.x.
- */
- @Deprecated
- public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
- if (parent == null || parent.isShutdown()) {
- throw new RejectedExecutionException(sm.getString("taskQueue.notRunning"));
- }
- return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task is rejected
- }
-
-
@Override
public boolean offer(Runnable o) {
//we can't do any checks
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
[tomcat] 03/05: Clean up imported code
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
markt pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 8bd316a226a7ca267fce12de3a6c9553a42b2d04
Author: Mark Thomas <ma...@apache.org>
AuthorDate: Wed Jul 21 16:15:00 2021 +0100
Clean up imported code
Compilation will still fail
Added ALv2 header
Fixed imports
Remove deprecated code
Add RejectedExecutionHandler interface
Fixed IDE warnings
Remove unused code
---
.../tomcat/util/threads/ThreadPoolExecutor.java | 95 +++++++++++++++++-----
1 file changed, 73 insertions(+), 22 deletions(-)
diff --git a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
index de9e1a8..5193209 100644
--- a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
+++ b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
@@ -1,24 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
/*
+ * The original version of this file carried the following notice:
+ *
* 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/publicdomain/zero/1.0/
*/
-
-package java.util.concurrent;
+package org.apache.tomcat.util.threads;
import java.util.ArrayList;
import java.util.ConcurrentModificationException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
/**
- * An {@link ExecutorService} that executes each submitted task using
+ * An {@link java.util.concurrent.ExecutorService}
+ * that executes each submitted task using
* one of possibly several pooled threads, normally configured
* using {@link Executors} factory methods.
*
@@ -132,7 +157,8 @@ import java.util.concurrent.locks.ReentrantLock;
* <ol>
*
* <li><em> Direct handoffs.</em> A good default choice for a work
- * queue is a {@link SynchronousQueue} that hands off tasks to threads
+ * queue is a {@link java.util.concurrent.SynchronousQueue}
+ * that hands off tasks to threads
* without otherwise holding them. Here, an attempt to queue a task
* will fail if no threads are immediately available to run it, so a
* new thread will be constructed. This policy avoids lockups when
@@ -143,7 +169,8 @@ import java.util.concurrent.locks.ReentrantLock;
* arrive on average faster than they can be processed.
*
* <li><em> Unbounded queues.</em> Using an unbounded queue (for
- * example a {@link LinkedBlockingQueue} without a predefined
+ * example a {@link java.util.concurrent.LinkedBlockingQueue}
+ * without a predefined
* capacity) will cause new tasks to wait in the queue when all
* corePoolSize threads are busy. Thus, no more than corePoolSize
* threads will ever be created. (And the value of the maximumPoolSize
@@ -156,7 +183,8 @@ import java.util.concurrent.locks.ReentrantLock;
* average faster than they can be processed.
*
* <li><em>Bounded queues.</em> A bounded queue (for example, an
- * {@link ArrayBlockingQueue}) helps prevent resource exhaustion when
+ * {@link java.util.concurrent.ArrayBlockingQueue})
+ * helps prevent resource exhaustion when
* used with finite maximumPoolSizes, but can be more difficult to
* tune and control. Queue sizes and maximum pool sizes may be traded
* off for each other: Using large queues and small pools minimizes
@@ -366,7 +394,6 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
private static final int TERMINATED = 3 << COUNT_BITS;
// Packing and unpacking ctl
- private static int runStateOf(int c) { return c & ~COUNT_MASK; }
private static int workerCountOf(int c) { return c & COUNT_MASK; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
@@ -581,10 +608,8 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
- @SuppressWarnings("serial") // Unlikely to be serializable
final Thread thread;
/** Initial task to run. Possibly null. */
- @SuppressWarnings("serial") // Not statically typed as Serializable
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
@@ -603,6 +628,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
}
/** Delegates main run loop to outer runWorker. */
+ @Override
public void run() {
runWorker(this);
}
@@ -612,10 +638,12 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
// 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());
@@ -624,6 +652,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
return false;
}
+ @Override
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
@@ -1104,6 +1133,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
*
* @param w the worker
*/
+ @SuppressWarnings("null") // task cannot be null
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
@@ -1320,6 +1350,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
* cannot be accepted for execution
* @throws NullPointerException if {@code command} is null
*/
+ @Override
public void execute(Runnable command) {
if (command == null) {
throw new NullPointerException();
@@ -1375,6 +1406,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
*
* @throws SecurityException {@inheritDoc}
*/
+ @Override
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
@@ -1406,6 +1438,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
*
* @throws SecurityException {@inheritDoc}
*/
+ @Override
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
@@ -1422,6 +1455,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
return tasks;
}
+ @Override
public boolean isShutdown() {
return runStateAtLeast(ctl.get(), SHUTDOWN);
}
@@ -1447,10 +1481,12 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
return runStateAtLeast(c, SHUTDOWN) && runStateLessThan(c, TERMINATED);
}
+ @Override
public boolean isTerminated() {
return runStateAtLeast(ctl.get(), TERMINATED);
}
+ @Override
public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
long nanos = unit.toNanos(timeout);
@@ -1469,18 +1505,6 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
}
}
- // Override without "throws Throwable" for compatibility with subclasses
- // whose finalize method invokes super.finalize() (as is recommended).
- // Before JDK 11, finalize() had a non-empty method body.
-
- /**
- * @implNote Previous versions of this class had a finalize method
- * that shut down this executor, but in this version, finalize
- * does nothing.
- */
- @Deprecated(since="9")
- protected void finalize() {}
-
/**
* Sets the thread factory used to create new threads.
*
@@ -1923,6 +1947,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
*
* @return a string identifying this pool, as well as its state
*/
+ @Override
public String toString() {
long ncompleted;
int nworkers, nactive;
@@ -1985,7 +2010,8 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
* beginning of this method.
*
* <p><b>Note:</b> When actions are enclosed in tasks (such as
- * {@link FutureTask}) either explicitly or via methods 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 <em>not</em>
@@ -2053,6 +2079,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
* @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();
@@ -2080,6 +2107,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
* @param e the executor attempting to execute this task
* @throws RejectedExecutionException always
*/
+ @Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
@@ -2103,6 +2131,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
* @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) {
}
}
@@ -2140,6 +2169,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
* @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();
@@ -2147,4 +2177,25 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
}
}
}
+
+
+ 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.
+ *
+ * <p>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);
+ }
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org
[tomcat] 04/05: Merge in changes from Tomcat's extended
ThreadPoolExecutor
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
markt pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 4c314648eccc75c0f8108b3ed394dea69bf75593
Author: Mark Thomas <ma...@apache.org>
AuthorDate: Wed Jul 21 16:36:32 2021 +0100
Merge in changes from Tomcat's extended ThreadPoolExecutor
---
.../tomcat/util/threads/ThreadPoolExecutor.java | 165 ++++++++++++++++++++-
1 file changed, 160 insertions(+), 5 deletions(-)
diff --git a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
index 5193209..7b1ec55 100644
--- a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
+++ b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
@@ -37,10 +37,13 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
+import org.apache.tomcat.util.res.StringManager;
+
/**
* An {@link java.util.concurrent.ExecutorService}
* that executes each submitted task using
@@ -325,6 +328,9 @@ import java.util.concurrent.locks.ReentrantLock;
* @author Doug Lea
*/
public class ThreadPoolExecutor extends AbstractExecutorService {
+
+ protected static final StringManager sm = StringManager.getManager("org.apache.tomcat.util.threads.res");
+
/**
* The main pool control state, ctl, is an atomic integer packing
* two conceptual fields
@@ -488,6 +494,22 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
*/
private long completedTaskCount;
+ /**
+ * The number of tasks submitted but not yet finished. This includes tasks
+ * in the queue and tasks that have been handed to a worker thread but the
+ * latter did not start executing the task yet.
+ * This number is always greater or equal to {@link #getActiveCount()}.
+ */
+ private final AtomicInteger submittedCount = new AtomicInteger(0);
+ private final AtomicLong lastContextStoppedTime = new AtomicLong(0L);
+
+ /**
+ * Most recent time in ms when a thread decided to kill itself to avoid
+ * potential memory leaks. Useful to throttle the rate of renewals of
+ * threads.
+ */
+ private final AtomicLong lastTimeThreadKilledItself = new AtomicLong(0L);
+
/*
* All user control parameters are declared as volatiles so that
* ongoing actions are based on freshest values, but without need
@@ -496,6 +518,11 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
*/
/**
+ * Delay in ms between 2 threads being renewed. If negative, do not renew threads.
+ */
+ private volatile long threadRenewalDelay = Constants.DEFAULT_THREAD_RENEWAL_DELAY;
+
+ /**
* Factory for new threads. All threads are created using this
* factory (via method addWorker). All callers must be prepared
* for addWorker to fail, which may reflect a system or user's
@@ -555,8 +582,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
/**
* The default rejected execution handler.
*/
- private static final RejectedExecutionHandler defaultHandler =
- new AbortPolicy();
+ private static final RejectedExecutionHandler defaultHandler = new RejectHandler();
/**
* Permission required for callers of shutdown and shutdownNow.
@@ -1336,6 +1362,31 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
this.handler = handler;
}
+
+ @Override
+ public void execute(Runnable command) {
+ submittedCount.incrementAndGet();
+ try {
+ executeInternal(command);
+ } catch (RejectedExecutionException rx) {
+ if (getQueue() instanceof TaskQueue) {
+ // If the Executor is close to maximum pool size, concurrent
+ // calls to execute() may result (due to Tomcat's use of
+ // TaskQueue) in some tasks being rejected rather than queued.
+ // If this happens, add them to the queue.
+ final TaskQueue queue = (TaskQueue) getQueue();
+ if (!queue.force(command)) {
+ submittedCount.decrementAndGet();
+ throw new RejectedExecutionException(sm.getString("threadPoolExecutor.queueFull"));
+ }
+ } else {
+ submittedCount.decrementAndGet();
+ throw rx;
+ }
+ }
+ }
+
+
/**
* Executes the given task sometime in the future. The task
* may execute in a new thread or in an existing pooled thread.
@@ -1350,8 +1401,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
* cannot be accepted for execution
* @throws NullPointerException if {@code command} is null
*/
- @Override
- public void execute(Runnable command) {
+ private void executeInternal(Runnable command) {
if (command == null) {
throw new NullPointerException();
}
@@ -1764,6 +1814,17 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
return unit.convert(keepAliveTime, TimeUnit.NANOSECONDS);
}
+
+ public long getThreadRenewalDelay() {
+ return threadRenewalDelay;
+ }
+
+
+ public void setThreadRenewalDelay(long threadRenewalDelay) {
+ this.threadRenewalDelay = threadRenewalDelay;
+ }
+
+
/* User-level queue utilities */
/**
@@ -1834,6 +1895,37 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
tryTerminate(); // In case SHUTDOWN and now empty
}
+
+ public void contextStopping() {
+ this.lastContextStoppedTime.set(System.currentTimeMillis());
+
+ // save the current pool parameters to restore them later
+ int savedCorePoolSize = this.getCorePoolSize();
+ TaskQueue taskQueue =
+ getQueue() instanceof TaskQueue ? (TaskQueue) getQueue() : null;
+ if (taskQueue != null) {
+ // note by slaurent : quite oddly threadPoolExecutor.setCorePoolSize
+ // checks that queue.remainingCapacity()==0. I did not understand
+ // why, but to get the intended effect of waking up idle threads, I
+ // temporarily fake this condition.
+ taskQueue.setForcedRemainingCapacity(0);
+ }
+
+ // setCorePoolSize(0) wakes idle threads
+ this.setCorePoolSize(0);
+
+ // TaskQueue.take() takes care of timing out, so that we are sure that
+ // all threads of the pool are renewed in a limited time, something like
+ // (threadKeepAlive + longest request time)
+
+ if (taskQueue != null) {
+ // ok, restore the state of the queue and pool
+ taskQueue.resetForcedRemainingCapacity();
+ }
+ this.setCorePoolSize(savedCorePoolSize);
+ }
+
+
/* Statistics */
/**
@@ -1940,6 +2032,12 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
}
}
+
+ 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
@@ -1998,6 +2096,7 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
*/
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
@@ -2048,7 +2147,56 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
* @param t the exception that caused termination, or null if
* execution completed normally
*/
- protected void afterExecute(Runnable r, Throwable t) { }
+ 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() {
+ if (threadRenewalDelay >= 0
+ && Thread.currentThread() instanceof TaskThread) {
+ TaskThread currentTaskThread = (TaskThread) Thread.currentThread();
+ if (currentTaskThread.getCreationTime() <
+ this.lastContextStoppedTime.longValue()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/**
* Method invoked when the Executor has terminated. Default
@@ -2178,6 +2326,13 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
}
}
+ private static class RejectHandler implements RejectedExecutionHandler {
+ @Override
+ public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
+ throw new RejectedExecutionException();
+ }
+ }
+
public interface RejectedExecutionHandler {
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org