This is an automated email from the ASF dual-hosted git repository.
markt-asf pushed a commit to branch 10.1.x
in repository https://gitbox.apache.org/repos/asf/tomcat.git
The following commit(s) were added to refs/heads/10.1.x by this push:
new 1f0ff30cf2 Review "Fix submitted task count drift in
ThreadPoolExecutor"
1f0ff30cf2 is described below
commit 1f0ff30cf2ef1a759c4811ee967507eba66b18d5
Author: Mark Thomas <[email protected]>
AuthorDate: Wed Nov 4 17:08:00 2026 +0000
Review "Fix submitted task count drift in ThreadPoolExecutor"
Fix a few more instances of task count drift
Co-authored by: Codex / GPT-6.1-Sol
---
.../tomcat/util/threads/ThreadPoolExecutor.java | 109 +++++++----
.../util/threads/TestThreadPoolExecutor.java | 216 +++++++++++++++++++--
2 files changed, 262 insertions(+), 63 deletions(-)
diff --git a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
index 2779f8e714..ce5076a77f 100644
--- a/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
+++ b/java/org/apache/tomcat/util/threads/ThreadPoolExecutor.java
@@ -25,7 +25,6 @@
package org.apache.tomcat.util.threads;
import java.util.ArrayList;
-import java.util.ConcurrentModificationException;
import java.util.HashSet;
import java.util.List;
import java.util.concurrent.AbstractExecutorService;
@@ -365,6 +364,7 @@ public class ThreadPoolExecutor extends
AbstractExecutorService {
* equal to {@link #getActiveCount()}.
*/
private final AtomicInteger submittedCount = new AtomicInteger(0);
+ private final ThreadLocal<RejectedTask> rejectedTask = new ThreadLocal<>();
private final AtomicLong lastContextStoppedTime = new AtomicLong(0L);
/**
@@ -687,20 +687,47 @@ public class ThreadPoolExecutor extends
AbstractExecutorService {
* ScheduledThreadPoolExecutor.
*/
final void reject(Runnable command) {
+ RejectedTask previous = rejectedTask.get();
+ RejectedTask current = new RejectedTask(command);
+ rejectedTask.set(current);
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;
+ try {
+ handler.rejectedExecution(command, this);
+ } catch (RejectedExecutionException rx) {
+ // execute() releases the count, or re-queues the task for a
retryable queue.
+ throw rx;
+ } catch (RuntimeException | Error e) {
+ submittedCount.decrementAndGet();
+ throw e;
+ }
+ // A custom handler may enqueue the task directly. Only release
its count if a policy reports that it
+ // was discarded, run directly or resubmitted through execute()
with a separate count.
+ if (current.handled) {
+ submittedCount.decrementAndGet();
+ }
+ } finally {
+ if (previous == null) {
+ rejectedTask.remove();
+ } else {
+ rejectedTask.set(previous);
+ }
+ }
+ }
+
+ private void markRejectedTaskHandled(Runnable command) {
+ RejectedTask current = rejectedTask.get();
+ if (current != null && current.command == command) {
+ current.handled = true;
+ }
+ }
+
+ private static class RejectedTask {
+ private final Runnable command;
+ private boolean handled;
+
+ RejectedTask(Runnable command) {
+ this.command = command;
}
- // 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();
}
/**
@@ -1209,6 +1236,7 @@ public class ThreadPoolExecutor extends
AbstractExecutorService {
// 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)) {
+ tryTerminate(); // In case SHUTDOWN and now empty
reject(command);
} else if (workerCountOf(recheck) == 0) {
addWorker(null, false);
@@ -1641,32 +1669,14 @@ 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 -> {
- if (r instanceof Future<?> && ((Future<?>) r).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). 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)) {
- purged.incrementAndGet();
- }
+ // A worker or another purge() may remove a task after it is selected.
Only release its count if this call
+ // actually removes it. A removeIf() predicate may run even when the
task is not subsequently removed.
+ for (Object r : q.toArray()) {
+ if (r instanceof Future<?> && ((Future<?>) r).isCancelled() &&
q.remove(r)) {
+ submittedCount.decrementAndGet();
}
}
- submittedCount.addAndGet(-purged.get());
-
tryTerminate(); // In case SHUTDOWN and now empty
}
@@ -1981,8 +1991,12 @@ public class ThreadPoolExecutor extends
AbstractExecutorService {
*/
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
- if (!e.isShutdown()) {
- r.run();
+ try {
+ if (!e.isShutdown()) {
+ r.run();
+ }
+ } finally {
+ e.markRejectedTaskHandled(r);
}
}
}
@@ -2030,6 +2044,7 @@ public class ThreadPoolExecutor extends
AbstractExecutorService {
*/
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
+ e.markRejectedTaskHandled(r);
}
}
@@ -2069,9 +2084,15 @@ public class ThreadPoolExecutor extends
AbstractExecutorService {
*/
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
- if (!e.isShutdown()) {
- e.getQueue().poll();
- e.execute(r);
+ try {
+ if (!e.isShutdown()) {
+ if (e.getQueue().poll() != null) {
+ e.submittedCount.decrementAndGet();
+ }
+ e.execute(r);
+ }
+ } finally {
+ e.markRejectedTaskHandled(r);
}
}
}
@@ -2096,6 +2117,10 @@ public class ThreadPoolExecutor extends
AbstractExecutorService {
* <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}.
+ * <p>
+ * Handlers that discard tasks or run them directly should delegate to
{@link DiscardPolicy} or
+ * {@link CallerRunsPolicy}, respectively, so the submitted task count
is adjusted. Handlers that enqueue tasks
+ * directly leave their counts to be released when the tasks complete.
*
* @param r the runnable task requested to be executed
* @param executor the executor attempting to execute this task
@@ -2104,4 +2129,4 @@ public class ThreadPoolExecutor extends
AbstractExecutorService {
*/
void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}
-}
\ No newline at end of file
+}
diff --git a/test/org/apache/tomcat/util/threads/TestThreadPoolExecutor.java
b/test/org/apache/tomcat/util/threads/TestThreadPoolExecutor.java
index 913ba79824..657a6b1be7 100644
--- a/test/org/apache/tomcat/util/threads/TestThreadPoolExecutor.java
+++ b/test/org/apache/tomcat/util/threads/TestThreadPoolExecutor.java
@@ -20,6 +20,9 @@ import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -83,36 +86,121 @@ public class TestThreadPoolExecutor {
}
/*
- * A rejected execution handler that does not throw discards the task or
runs it in the calling thread. Either way
+ * A built-in rejection policy 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.
+ // Test both policies, including inherited and delegated handling
methods.
for (ThreadPoolExecutor.RejectedExecutionHandler handler : new
ThreadPoolExecutor.RejectedExecutionHandler[] {
- new ThreadPoolExecutor.DiscardPolicy(), new
ThreadPoolExecutor.CallerRunsPolicy() }) {
- CountDownLatch release = new CountDownLatch(1);
+ new ThreadPoolExecutor.DiscardPolicy(), new
ThreadPoolExecutor.CallerRunsPolicy(),
+ new ThreadPoolExecutor.DiscardPolicy() {}, new
ThreadPoolExecutor.CallerRunsPolicy() {} }) {
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(),
Boolean.valueOf(expectRunInCaller),
- Boolean.valueOf(ran.get()));
- release.countDown();
- awaitSubmittedCount(pool, 0);
- } finally {
- pool.shutdownNow();
- }
+ doTestNonThrowingRejectionHandler(handler, expectRunInCaller);
+ doTestNonThrowingRejectionHandler((task, executor) ->
handler.rejectedExecution(task, executor),
+ expectRunInCaller);
+ }
+ }
+
+
+ private void
doTestNonThrowingRejectionHandler(ThreadPoolExecutor.RejectedExecutionHandler
handler,
+ boolean expectRunInCaller) throws Exception {
+ CountDownLatch release = new CountDownLatch(1);
+ 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(),
Boolean.valueOf(expectRunInCaller),
+ Boolean.valueOf(ran.get()));
+ release.countDown();
+ awaitSubmittedCount(pool, 0);
+ } finally {
+ pool.shutdownNow();
+ Assert.assertTrue(pool.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+
+ /*
+ * A rejection handler may enqueue the task instead of discarding it or
running it itself. Even a subclass of a
+ * built-in policy may do this, so returning from the handler must not
release the queued task's count.
+ */
+ @Test
+ public void testQueuingRejectionHandlerKeepsSubmittedCount() throws
Exception {
+ CountDownLatch firstStarted = new CountDownLatch(1);
+ CountDownLatch releaseFirst = new CountDownLatch(1);
+ CountDownLatch secondStarted = new CountDownLatch(1);
+ CountDownLatch releaseSecond = new CountDownLatch(1);
+ CountDownLatch thirdCompleted = new CountDownLatch(1);
+ TaskQueue queue = new TaskQueue(1);
+ ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 1, 60,
TimeUnit.SECONDS, queue,
+ Executors.defaultThreadFactory(), new
ThreadPoolExecutor.CallerRunsPolicy() {
+ @Override
+ public void rejectedExecution(Runnable task,
ThreadPoolExecutor executor) {
+ // Let the worker take the queued task so the handler
can enqueue the rejected task.
+ releaseFirst.countDown();
+ try {
+ awaitStarted(secondStarted);
+ executor.getQueue().put(task);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RejectedExecutionException(e);
+ }
+ }
+ });
+ queue.setParent(pool);
+ try {
+ pool.execute(blockingTask("first", firstStarted, releaseFirst));
+ awaitStarted(firstStarted);
+ pool.execute(blockingTask("second", secondStarted, releaseSecond));
+ pool.execute(thirdCompleted::countDown);
+
+ Assert.assertEquals("One running task and one task queued by the
handler", 2, pool.getSubmittedCount());
+ Assert.assertEquals(1, queue.size());
+ releaseSecond.countDown();
+ Assert.assertTrue(thirdCompleted.await(5, TimeUnit.SECONDS));
+ awaitSubmittedCount(pool, 0);
+ } finally {
+ releaseFirst.countDown();
+ releaseSecond.countDown();
+ pool.shutdownNow();
+ Assert.assertTrue(pool.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+
+ @Test
+ public void testDiscardOldestPolicyKeepsSubmittedCount() throws Exception {
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch completed = new CountDownLatch(1);
+ AtomicBoolean discardedTaskRan = new AtomicBoolean();
+ ThreadPoolExecutor pool = createPool(new
ThreadPoolExecutor.DiscardOldestPolicy());
+ try {
+ pool.execute(blockingTask("running", started, release));
+ awaitStarted(started);
+ pool.execute(() -> discardedTaskRan.set(true));
+ pool.execute(completed::countDown);
+ // The original queued task was discarded. Only the running task
and its replacement remain.
+ Assert.assertEquals(2, pool.getSubmittedCount());
+ release.countDown();
+ Assert.assertTrue(completed.await(5, TimeUnit.SECONDS));
+ awaitSubmittedCount(pool, 0);
+ Assert.assertFalse(discardedTaskRan.get());
+ } finally {
+ release.countDown();
+ pool.shutdownNow();
+ Assert.assertTrue(pool.awaitTermination(5, TimeUnit.SECONDS));
}
}
+
/*
* A null command is rejected with a NullPointerException before it can be
counted.
*/
@@ -181,6 +269,55 @@ public class TestThreadPoolExecutor {
}
}
+
+ /*
+ * A worker can dequeue a cancelled future after purge() has selected it
but before purge() removes it. In that
+ * case afterExecute() accounts for the future and purge() must not
decrement the count again.
+ */
+ @Test
+ public void testPurgeRacingWithWorkerKeepsSubmittedCount() throws
Exception {
+ CountDownLatch firstStarted = new CountDownLatch(1);
+ CountDownLatch releaseFirst = new CountDownLatch(1);
+ CountDownLatch secondStarted = new CountDownLatch(1);
+ CountDownLatch releaseSecond = new CountDownLatch(1);
+ TaskQueue queue = new TaskQueue();
+ ThreadPoolExecutor pool = new ThreadPoolExecutor(1, 1, 60,
TimeUnit.SECONDS, queue,
+ Executors.defaultThreadFactory());
+ queue.setParent(pool);
+ FutureTask<Void> cancelled = new FutureTask<>(() -> null) {
+ @Override
+ public boolean isCancelled() {
+ // The worker processes the cancelled future before starting
the second task.
+ releaseFirst.countDown();
+ try {
+ awaitStarted(secondStarted);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ return super.isCancelled();
+ }
+ };
+ try {
+ pool.execute(blockingTask("first", firstStarted, releaseFirst));
+ awaitStarted(firstStarted);
+ Assert.assertTrue(cancelled.cancel(false));
+ pool.execute(cancelled);
+ pool.execute(blockingTask("second", secondStarted, releaseSecond));
+ pool.purge();
+
+ Assert.assertEquals("Only the second task is outstanding", 1,
pool.getSubmittedCount());
+ releaseSecond.countDown();
+ awaitSubmittedCount(pool, 0);
+ } finally {
+ releaseFirst.countDown();
+ releaseSecond.countDown();
+ pool.shutdownNow();
+ Assert.assertTrue(pool.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+
/*
* 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.
@@ -217,6 +354,43 @@ public class TestThreadPoolExecutor {
}
}
+ /*
+ * If the pool is shut down after a task has been queued but before
execute() re-checks the pool state, execute()
+ * removes the task again and rejects it. The removal can empty the queue
of a shut down pool that has no workers,
+ * so it must trigger the transition to terminated since nothing else will.
+ */
+ @Test
+ public void testRollbackOfQueuedTaskDuringShutdownTerminatesPool() throws
Exception {
+ class ShutdownAfterOfferQueue extends LinkedBlockingQueue<Runnable> {
+ private static final long serialVersionUID = 1L;
+ private transient volatile ThreadPoolExecutor pool;
+
+ @Override
+ public boolean offer(Runnable r) {
+ boolean result = super.offer(r);
+ // Shut down while the task is queued, as a concurrent
shutdown() call could
+ pool.shutdown();
+ return result;
+ }
+ }
+
+ ShutdownAfterOfferQueue queue = new ShutdownAfterOfferQueue();
+ // No core threads so there is no worker to terminate the pool once
the queue has been emptied
+ ThreadPoolExecutor pool = new ThreadPoolExecutor(0, 1, 60,
TimeUnit.SECONDS, queue,
+ Executors.defaultThreadFactory(), new
ThreadPoolExecutor.AbortPolicy());
+ queue.pool = pool;
+ try {
+ Assert.assertThrows(RejectedExecutionException.class, () ->
pool.execute(() -> {
+ // Must not run
+ }));
+ Assert.assertTrue("Pool did not terminate",
pool.awaitTermination(5, TimeUnit.SECONDS));
+ Assert.assertEquals(0, pool.getSubmittedCount());
+ } finally {
+ pool.shutdownNow();
+ }
+ }
+
+
/*
* Tasks that complete normally or that throw must each decrement the
submitted count exactly once.
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
