You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@tomcat.apache.org by re...@apache.org on 2026/10/16 16:16:31 UTC
(tomcat) 22/25: Fix submitted task count drift in ThreadPoolExecutor
This is an automated email from the ASF dual-hosted git repository.
rmaucher pushed a commit to branch 9.0.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 3f0bf8d596f380bd72e123a5d9408b7437bbe957
Author: opencode <re...@apache.org>
AuthorDate: Fri Oct 16 15:25:50 2026 +0200
Fix submitted task count drift in ThreadPoolExecutor
Adjust the submitted task counter on every path where a counted task
can no longer reach afterExecute(): a null command, a rejection
handler that does not throw, removal of a queued task via remove() or
purge(), and a beforeExecute override that throws. The counter feeds
TaskQueue's queue-or-create-thread decision, so a drifted value makes
idle pools create unnecessary threads.
---
.../tomcat/util/threads/ThreadPoolExecutor.java | 61 ++++-
.../util/threads/TestThreadPoolExecutor.java | 246 +++++++++++++++++++++
webapps/docs/changelog.xml | 9 +
3 files changed, 309 insertions(+), 7 deletions(-)
diff --git a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
index 6a66925368..7eab514ccc 100644
--- a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
+++ b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
@@ -687,7 +687,20 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
* ScheduledThreadPoolExecutor.
*/
final void reject(Runnable command) {
- handler.rejectedExecution(command, this);
+ try {
+ handler.rejectedExecution(command, this);
+ } catch (RejectedExecutionException rx) {
+ // The caller in execute() decrements the submitted count, or, when a retryable queue is used, re-queues
+ // the task which will then be counted by afterExecute() once it completes.
+ throw rx;
+ } catch (RuntimeException | Error e) {
+ // The task will not be executed by a worker, so afterExecute() will not decrement the submitted count.
+ submittedCount.decrementAndGet();
+ throw e;
+ }
+ // The handler completed without throwing. The task was discarded or run directly by the handler. It will not
+ // pass through afterExecute(), so decrement the submitted count here.
+ submittedCount.decrementAndGet();
}
/**
@@ -969,7 +982,15 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
wt.interrupt();
}
try {
- beforeExecute(wt, task);
+ try {
+ beforeExecute(wt, task);
+ } catch (Throwable ex) {
+ // beforeExecute() threw. The task will not be executed
+ // and afterExecute() will not be called, so decrement
+ // the submitted count here.
+ submittedCount.decrementAndGet();
+ throw ex;
+ }
try {
task.run();
afterExecute(task, null);
@@ -1138,6 +1159,11 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
*/
@Deprecated
public void execute(Runnable command, long timeout, TimeUnit unit) {
+ // Check for null before counting the task, so a rejected null
+ // argument does not leave the submitted count inflated.
+ if (command == null) {
+ throw new NullPointerException();
+ }
submittedCount.incrementAndGet();
try {
executeInternal(command);
@@ -1204,7 +1230,11 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
}
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
- if (!isRunning(recheck) && remove(command)) {
+ // Remove directly from the queue here. The public remove(Runnable)
+ // method also adjusts the submitted count, which would double the
+ // adjustment for this task, since the reject() call below and the
+ // catch block in execute() take care of that.
+ if (!isRunning(recheck) && workQueue.remove(command)) {
reject(command);
} else if (workerCountOf(recheck) == 0) {
addWorker(null, false);
@@ -1620,6 +1650,10 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
*/
public boolean remove(Runnable task) {
boolean removed = workQueue.remove(task);
+ if (removed) {
+ // The removed task will never be executed, so afterExecute() will not decrement the submitted count.
+ submittedCount.decrementAndGet();
+ }
tryTerminate(); // In case SHUTDOWN and now empty
return removed;
}
@@ -1633,19 +1667,32 @@ public class ThreadPoolExecutor extends AbstractExecutorService {
*/
public void purge() {
final BlockingQueue<Runnable> q = workQueue;
+ // Count the cancelled tasks that are removed. These will never be
+ // executed, so afterExecute() will not decrement the submitted count.
+ final AtomicInteger purged = new AtomicInteger(0);
try {
- q.removeIf(r -> r instanceof Future<?> && ((Future<?>) r).isCancelled());
+ q.removeIf(r -> {
+ if (r instanceof Future<?> future && future.isCancelled()) {
+ purged.incrementAndGet();
+ return true;
+ }
+ return false;
+ });
} 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).
+ // The slow path is more likely to be O(N*N). Any entries removed
+ // before the interference occurred are already accounted for in
+ // the purged count and will not appear in the copy taken below.
for (Object r : q.toArray()) {
- if (r instanceof Future<?> && ((Future<?>) r).isCancelled()) {
- q.remove(r);
+ if (r instanceof Future<?> future && future.isCancelled() && q.remove(r)) {
+ purged.incrementAndGet();
}
}
}
+ submittedCount.addAndGet(-purged.get());
+
tryTerminate(); // In case SHUTDOWN and now empty
}
diff --git a/test/org/apache/tomcat/util/threads/TestThreadPoolExecutor.java b/test/org/apache/tomcat/util/threads/TestThreadPoolExecutor.java
new file mode 100644
index 0000000000..0c299bf6fb
--- /dev/null
+++ b/test/org/apache/tomcat/util/threads/TestThreadPoolExecutor.java
@@ -0,0 +1,246 @@
+/*
+ * 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.
+ */
+package org.apache.tomcat.util.threads;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestThreadPoolExecutor {
+
+ /*
+ * The submitted task count is used by TaskQueue.offer() to decide if a task should be queued or a new thread
+ * should be created. A task that is counted when submitted but never reaches afterExecute() leaves the count
+ * permanently inflated.
+ */
+
+ private static ThreadPoolExecutor createPool(ThreadPoolExecutor.RejectedExecutionHandler handler) {
+ // Single thread with room for a single queued task. The task following
+ // the queued one is rejected.
+ return new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1),
+ Executors.defaultThreadFactory(), handler);
+ }
+
+ private static Runnable blockingTask(String name, CountDownLatch release) {
+ return blockingTask(name, null, release);
+ }
+
+ private static Runnable blockingTask(String name, CountDownLatch started, CountDownLatch release) {
+ return new Runnable() {
+ @Override
+ public void run() {
+ if (started != null) {
+ started.countDown();
+ }
+ try {
+ release.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ @Override
+ public String toString() {
+ return name;
+ }
+ };
+ }
+
+ /*
+ * Waits until the running task has started. getActiveCount() is not usable for this: it is approximate, and a
+ * worker that the pool constructor prestarted counts as active before it has taken its first task.
+ */
+ private static void awaitStarted(CountDownLatch started) throws InterruptedException {
+ Assert.assertTrue(started.await(5, TimeUnit.SECONDS));
+ }
+
+ private static void awaitSubmittedCount(ThreadPoolExecutor pool, int expected) throws InterruptedException {
+ long end = System.currentTimeMillis() + 5000;
+ while (pool.getSubmittedCount() != expected && System.currentTimeMillis() < end) {
+ Thread.sleep(20);
+ }
+ Assert.assertEquals(expected, pool.getSubmittedCount());
+ }
+
+ /*
+ * A rejected execution handler that does not throw discards the task or runs it in the calling thread. Either way
+ * the task never reaches afterExecute(), so the submitted count must be decremented when the handler returns.
+ */
+ @Test
+ public void testNonThrowingRejectionHandlerKeepsSubmittedCount() throws Exception {
+ // Test both handlers that do not throw.
+ for (ThreadPoolExecutor.RejectedExecutionHandler handler :
+ new ThreadPoolExecutor.RejectedExecutionHandler[] { new ThreadPoolExecutor.DiscardPolicy(),
+ new ThreadPoolExecutor.CallerRunsPolicy() }) {
+ CountDownLatch release = new CountDownLatch(1);
+ boolean expectRunInCaller = handler instanceof ThreadPoolExecutor.CallerRunsPolicy;
+ ThreadPoolExecutor pool = createPool(handler);
+ try {
+ CountDownLatch started = new CountDownLatch(1);
+ pool.execute(blockingTask("running", started, release));
+ awaitStarted(started);
+ pool.execute(blockingTask("queued", release));
+ AtomicBoolean ran = new AtomicBoolean();
+ pool.execute(() -> ran.set(true)); // rejected: discarded or run by this thread
+ // Only the running and the queued task are still outstanding.
+ Assert.assertEquals(handler.getClass().getName(), 2, pool.getSubmittedCount());
+ Assert.assertEquals(handler.getClass().getName(), expectRunInCaller, ran.get());
+ release.countDown();
+ awaitSubmittedCount(pool, 0);
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+ }
+
+ /*
+ * A null command is rejected with a NullPointerException before it can be counted.
+ */
+ @Test
+ public void testNullCommandKeepsSubmittedCount() throws Exception {
+ ThreadPoolExecutor pool = createPool(new ThreadPoolExecutor.AbortPolicy());
+ try {
+ try {
+ pool.execute(null);
+ Assert.fail("Expected NullPointerException");
+ } catch (NullPointerException e) {
+ // Expected.
+ }
+ Assert.assertEquals(0, pool.getSubmittedCount());
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+
+ /*
+ * A task removed from the queue is never executed and never reaches afterExecute(), so the removal must decrement
+ * the submitted count.
+ */
+ @Test
+ public void testRemoveKeepsSubmittedCount() throws Exception {
+ CountDownLatch release = new CountDownLatch(1);
+ ThreadPoolExecutor pool = createPool(new ThreadPoolExecutor.AbortPolicy());
+ try {
+ Runnable queued = blockingTask("queued", release);
+ CountDownLatch started = new CountDownLatch(1);
+ pool.execute(blockingTask("running", started, release));
+ awaitStarted(started);
+ pool.execute(queued);
+ Assert.assertTrue(pool.remove(queued));
+ // Only the running task is still outstanding.
+ Assert.assertEquals(1, pool.getSubmittedCount());
+ release.countDown();
+ awaitSubmittedCount(pool, 0);
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+
+ /*
+ * A cancelled future removed by purge() is never executed and never reaches afterExecute(), so the removal must
+ * decrement the submitted count.
+ */
+ @Test
+ public void testPurgeKeepsSubmittedCount() throws Exception {
+ CountDownLatch release = new CountDownLatch(1);
+ ThreadPoolExecutor pool = createPool(new ThreadPoolExecutor.AbortPolicy());
+ try {
+ CountDownLatch started = new CountDownLatch(1);
+ pool.execute(blockingTask("running", started, release));
+ awaitStarted(started);
+ Future<?> queued = pool.submit(blockingTask("queued", release));
+ // The submitted future is in the queue since the only worker is busy.
+ Assert.assertTrue(queued.cancel(false));
+ pool.purge();
+ // Only the running task is still outstanding.
+ Assert.assertEquals(1, pool.getSubmittedCount());
+ release.countDown();
+ awaitSubmittedCount(pool, 0);
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+
+ /*
+ * If beforeExecute() throws, the worker dies without processing the task and without calling afterExecute(), so
+ * the submitted count must be decremented for the lost task.
+ */
+ @Test
+ public void testBeforeExecuteThrowKeepsSubmittedCount() throws Exception {
+ CountDownLatch threw = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ class BeforeExecutorThrows extends ThreadPoolExecutor {
+ BeforeExecutorThrows() {
+ super(1, 1, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1), Executors.defaultThreadFactory(),
+ new AbortPolicy());
+ }
+
+ @Override
+ protected void beforeExecute(Thread t, Runnable r) {
+ if ("marker".equals(r.toString())) {
+ threw.countDown();
+ throw new IllegalStateException("Test beforeExecute failure");
+ }
+ }
+ }
+
+ BeforeExecutorThrows pool = new BeforeExecutorThrows();
+ try {
+ pool.execute(blockingTask("marker", release));
+ // The worker should die without executing the task.
+ Assert.assertTrue(threw.await(5, TimeUnit.SECONDS));
+ // The task is lost with the worker, so nothing is outstanding.
+ awaitSubmittedCount(pool, 0);
+ } finally {
+ release.countDown();
+ pool.shutdownNow();
+ }
+ }
+
+ /*
+ * Tasks that complete normally or that throw must each decrement the submitted count exactly once.
+ */
+ @Test
+ public void testTaskCompletionDecrementsOnce() throws Exception {
+ ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 2, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(20),
+ Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy());
+ try {
+ int taskCount = 10;
+ CountDownLatch done = new CountDownLatch(taskCount);
+ for (int i = 0; i < taskCount; i++) {
+ final int index = i;
+ pool.execute(() -> {
+ done.countDown();
+ if (index % 2 == 0) {
+ throw new IllegalStateException("Test task failure " + index);
+ }
+ });
+ }
+ Assert.assertTrue(done.await(5, TimeUnit.SECONDS));
+ awaitSubmittedCount(pool, 0);
+ Assert.assertTrue(pool.getSubmittedCount() >= 0);
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+}
diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml
index 7533dccd51..e377eb7e61 100644
--- a/webapps/docs/changelog.xml
+++ b/webapps/docs/changelog.xml
@@ -371,6 +371,15 @@
of the entry order, matching the behaviour of the stream based JAR
implementation. (remm)
</fix>
+ <fix>
+ Correct the accounting of the count of submitted tasks that
+ <code>ThreadPoolExecutor</code> maintains for <code>TaskQueue</code>.
+ The count was not always adjusted when a submitted task could no
+ longer complete, such as when a rejected execution handler did not
+ throw, or when a queued task was removed. A drifted value caused
+ thread pools with idle threads to create unnecessary threads.
+ (remm)
+ </fix>
</changelog>
</subsection>
<subsection name="Web applications">
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org