This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 11.0.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 3d92ff34401048496384e38390a5f0aac7cd406c Author: opencode <[email protected]> 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 3ce64d281f..5c31dfa737 100644 --- a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java +++ b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java @@ -660,7 +660,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(); } /** @@ -942,7 +955,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); @@ -1090,6 +1111,11 @@ public class ThreadPoolExecutor extends AbstractExecutorService { @Override public void execute(Runnable command) { + // 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); @@ -1150,7 +1176,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); @@ -1564,6 +1594,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; } @@ -1577,19 +1611,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 519a4a3ae0..83101c9a1d 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -382,6 +382,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: [email protected] For additional commands, e-mail: [email protected]
