This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new de836f4af365 CAMEL-26478: camel-util - BoundedExecutorService: a task
must not wait for a permit it holds itself (#28823)
de836f4af365 is described below
commit de836f4af365f97c310751b54fa381f48190e9c4
Author: Adriano Machado <[email protected]>
AuthorDate: Mon Nov 2 17:05:37 2026 -0500
CAMEL-26478: camel-util - BoundedExecutorService: a task must not wait for
a permit it holds itself (#28823)
* CAMEL-26478: camel-util - BoundedExecutorService: a task must not wait
for a permit it holds itself
A parallel split runs in the executor and submits its items to it, so with a
low maxQueueSize it waited for the permit it held, until keepAliveTime
expired,
or forever with the Block policy. With a split timeout the unfinished items
were dropped with no exception.
A task of the executor now runs the work itself when no permit is available,
unless the policy is Abort. A split submits itself again for each item, so
past a depth limit the work goes to a new task, without a permit, to not
overflow the stack. A submitter outside the executor waits as before.
_Claude Code on behalf of Adriano Machado (@ammachado)_
_This was generated by an AI agent and may contain inaccuracies.
Please verify before relying on it._
* CAMEL-26478: camel-util - BoundedExecutorService: a nested task that
throws restores its depth
Add a test for the failure path of runNested: a task that survives failures
of the work it submits must not reach the inline depth limit. Say in the
Javadoc of getCallerRunsCount that work past that limit is not counted.
_Claude Code on behalf of Adriano Machado (@ammachado)_
_This was generated by an AI agent and may contain inaccuracies.
Please verify before relying on it._
* CAMEL-26478: camel-util - BoundedExecutorService: address review comments
runTask restores the depth of the task that called it instead of removing
it,
document the inline behavior in the threading model page, and explain why
MAX_INLINE_DEPTH runs work outside the concurrency limit.
Co-authored-by: Claude Opus 5.5 <[email protected]>
Co-Authored-By: Claude Sonnet 5.5 <[email protected]>
---
.../util/concurrent/BoundedExecutorService.java | 63 ++++-
.../concurrent/BoundedExecutorServiceTest.java | 261 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 8 +
.../modules/ROOT/pages/threading-model.adoc | 9 +
4 files changed, 337 insertions(+), 4 deletions(-)
diff --git
a/core/camel-util/src/main/java/org/apache/camel/util/concurrent/BoundedExecutorService.java
b/core/camel-util/src/main/java/org/apache/camel/util/concurrent/BoundedExecutorService.java
index 6111377dcce5..73233e97e03a 100644
---
a/core/camel-util/src/main/java/org/apache/camel/util/concurrent/BoundedExecutorService.java
+++
b/core/camel-util/src/main/java/org/apache/camel/util/concurrent/BoundedExecutorService.java
@@ -44,6 +44,11 @@ import java.util.concurrent.atomic.LongAdder;
* <li><b>Block</b>: blocks indefinitely until a permit becomes available. No
timeout, no rejection.</li>
* </ul>
* <p>
+ * A task of this executor that submits more work, such as a parallel split
that submits its items, can hold the permit
+ * it would wait for. So when no permit is available it runs the work itself
at once, with the {@code CallerRuns} and
+ * {@code Block} policies, or in a new task when that gets too deep, which is
also outside the concurrency limit. With
+ * {@code Abort} it waits and is rejected as usual.
+ * <p>
* <b>Caller thread blocking:</b> while waiting for a permit, the calling
thread is blocked. When callers are virtual
* threads this is inexpensive (the carrier thread is released). When callers
are platform threads (e.g., HTTP server
* threads) the blocked thread is unavailable for other work — this is
standard backpressure behavior but worth noting
@@ -52,11 +57,17 @@ import java.util.concurrent.atomic.LongAdder;
*/
public class BoundedExecutorService extends AbstractExecutorService {
+ // Bounds the stack of tasks running inline. Past it, the work goes to a
new task of the delegate. That is the escape
+ // hatch against the deadlock, accepted at the cost of the concurrency
limit, as the inline runs are also outside it.
+ private static final int MAX_INLINE_DEPTH = 64;
+
private final ExecutorService delegate;
private final Semaphore semaphore;
private final int maxConcurrent;
private final long timeoutNanos;
private final ThreadPoolRejectedPolicy rejectedPolicy;
+ // how deep tasks are running inline in a task of this executor, and null
in any other thread
+ private final ThreadLocal<Integer> inlineDepth = new ThreadLocal<>();
private final LongAdder callerRunsCount = new LongAdder();
private final LongAdder rejectedCount = new LongAdder();
private final LongAdder delegatedTaskCount = new LongAdder();
@@ -86,9 +97,14 @@ public class BoundedExecutorService extends
AbstractExecutorService {
throw new RejectedExecutionException("Executor has been shut
down");
}
+ // a task of this executor can hold the permit it would wait for, so
it must not wait
+ Integer depth = rejectedPolicy != ThreadPoolRejectedPolicy.Abort ?
inlineDepth.get() : null;
+ boolean runInline = depth != null;
boolean acquired = false;
try {
- if (rejectedPolicy == ThreadPoolRejectedPolicy.Block) {
+ if (runInline) {
+ acquired = semaphore.tryAcquire();
+ } else if (rejectedPolicy == ThreadPoolRejectedPolicy.Block) {
semaphore.acquire();
acquired = true;
} else {
@@ -96,6 +112,10 @@ public class BoundedExecutorService extends
AbstractExecutorService {
}
if (!acquired) {
+ if (runInline) {
+ runNested(command, depth);
+ return;
+ }
if (rejectedPolicy == ThreadPoolRejectedPolicy.CallerRuns) {
callerRunsCount.increment();
command.run();
@@ -109,9 +129,8 @@ public class BoundedExecutorService extends
AbstractExecutorService {
try {
delegate.execute(() -> {
try {
- command.run();
+ runTask(command);
} finally {
- delegatedTaskCount.increment();
semaphore.release();
}
});
@@ -127,6 +146,41 @@ public class BoundedExecutorService extends
AbstractExecutorService {
}
}
+ private void runTask(Runnable command) {
+ // a delegate can run the task on the calling thread, which can be a
task of this executor itself
+ Integer previous = inlineDepth.get();
+ inlineDepth.set(0);
+ try {
+ command.run();
+ } finally {
+ if (previous == null) {
+ inlineDepth.remove();
+ } else {
+ inlineDepth.set(previous);
+ }
+ delegatedTaskCount.increment();
+ }
+ }
+
+ /**
+ * Runs a task submitted by a task of this executor when no permit is
available. It must not wait, as it can hold
+ * that permit itself, and it must not wait for a queue to run later
either, as it could be waiting for this task.
+ */
+ private void runNested(Runnable command, int depth) {
+ if (depth >= MAX_INLINE_DEPTH) {
+ // a task that submits itself again, as a parallel split does for
each item, must not overflow the stack
+ delegate.execute(() -> runTask(command));
+ return;
+ }
+ callerRunsCount.increment();
+ inlineDepth.set(depth + 1);
+ try {
+ command.run();
+ } finally {
+ inlineDepth.set(depth);
+ }
+ }
+
// -- Metrics --
/**
@@ -159,7 +213,8 @@ public class BoundedExecutorService extends
AbstractExecutorService {
}
/**
- * The number of times the timeout expired and a task fell back to running
on the caller's thread.
+ * The number of times a task ran on the caller's thread, because the
timeout expired or a task of this executor
+ * submitted it when no permit was available and the inline depth limit
was not reached.
*/
public long getCallerRunsCount() {
return callerRunsCount.sum();
diff --git
a/core/camel-util/src/test/java/org/apache/camel/util/concurrent/BoundedExecutorServiceTest.java
b/core/camel-util/src/test/java/org/apache/camel/util/concurrent/BoundedExecutorServiceTest.java
index 1b58012f35d9..ffe6b8c5f47b 100644
---
a/core/camel-util/src/test/java/org/apache/camel/util/concurrent/BoundedExecutorServiceTest.java
+++
b/core/camel-util/src/test/java/org/apache/camel/util/concurrent/BoundedExecutorServiceTest.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.util.concurrent;
+import java.util.List;
+import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -29,6 +31,7 @@ import org.junit.jupiter.api.Test;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -238,4 +241,262 @@ public class BoundedExecutorServiceTest {
delegate.shutdown();
}
}
+
+ @Test
+ public void testNestedTaskRunsInlineWhenItHoldsTheOnlyPermit() throws
Exception {
+ // a parallel split runs in the executor and submits its items to it,
so waiting cannot succeed
+ assertNestedTaskRunsInline(ThreadPoolRejectedPolicy.CallerRuns);
+ }
+
+ @Test
+ public void testNestedTaskRunsInlineWithBlockPolicy() throws Exception {
+ // this wait has no timeout, so the task would hang forever
+ assertNestedTaskRunsInline(ThreadPoolRejectedPolicy.Block);
+ }
+
+ private void assertNestedTaskRunsInline(ThreadPoolRejectedPolicy policy)
throws Exception {
+ var delegate = Executors.newCachedThreadPool();
+ var sized = new BoundedExecutorService(delegate, 1, 60,
TimeUnit.SECONDS, false, policy);
+ try {
+ CountDownLatch done = new CountDownLatch(1);
+ AtomicBoolean sameThread = new AtomicBoolean();
+
+ sized.execute(() -> {
+ Thread outer = Thread.currentThread();
+ sized.execute(() -> {
+ sameThread.set(Thread.currentThread() == outer);
+ done.countDown();
+ });
+ });
+
+ assertTrue(done.await(5, TimeUnit.SECONDS), "The nested task
waited for the permit its parent holds");
+ assertTrue(sameThread.get(), "The nested task should run in the
task that submitted it");
+ assertEquals(1, sized.getCallerRunsCount());
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testTaskThatResubmitsItselfDoesNotOverflowTheStack() throws
Exception {
+ // a parallel split submits itself again for each item, so a large
split must not run all of them inline
+ var delegate = Executors.newCachedThreadPool();
+ var sized = new BoundedExecutorService(
+ delegate, 1, 60, TimeUnit.SECONDS, false,
ThreadPoolRejectedPolicy.CallerRuns);
+ try {
+ int resubmits = 200_000;
+ AtomicInteger count = new AtomicInteger();
+ CountDownLatch done = new CountDownLatch(1);
+ Runnable[] task = new Runnable[1];
+ task[0] = () -> {
+ if (count.incrementAndGet() == resubmits) {
+ done.countDown();
+ } else {
+ sized.execute(task[0]);
+ }
+ };
+
+ sized.execute(task[0]);
+
+ assertTrue(done.await(30, TimeUnit.SECONDS), "Stopped after " +
count.get() + " resubmits");
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testNestedTaskThatThrowsDoesNotUseUpTheInlineDepth() throws
Exception {
+ // the depth must be restored on failure too, or a task that survives
failures of the work it submits would
+ // reach the depth limit and hand the work to new tasks, outside the
concurrency limit
+ var delegate = Executors.newCachedThreadPool();
+ var sized = new BoundedExecutorService(
+ delegate, 1, 60, TimeUnit.SECONDS, false,
ThreadPoolRejectedPolicy.CallerRuns);
+ try {
+ int failures = 100;
+ AtomicInteger inline = new AtomicInteger();
+ CountDownLatch done = new CountDownLatch(1);
+
+ sized.execute(() -> {
+ Thread outer = Thread.currentThread();
+ for (int i = 0; i < failures; i++) {
+ try {
+ sized.execute(() -> {
+ if (Thread.currentThread() == outer) {
+ inline.incrementAndGet();
+ }
+ throw new IllegalStateException("Forced");
+ });
+ } catch (IllegalStateException e) {
+ // an inline task fails in its caller, as a caller-run
task does
+ }
+ }
+ done.countDown();
+ });
+
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ assertEquals(failures, inline.get(), "Every nested task should
still run inline after a failure");
+ assertEquals(failures, sized.getCallerRunsCount());
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
+ @Test
+ public void
testDelegateRunningATaskOnTheCallingThreadKeepsTheDepthOfTheOuterTask() throws
Exception {
+ // a delegate that runs a task on the calling thread must not reset
the depth of the task that submitted it,
+ // or that task would wait for a permit it holds itself
+ var pool = Executors.newCachedThreadPool();
+ ThreadLocal<Boolean> direct = new ThreadLocal<>();
+ var delegate = new AbstractExecutorService() {
+ @Override
+ public void execute(Runnable command) {
+ if (direct.get() != null) {
+ command.run();
+ } else {
+ pool.execute(command);
+ }
+ }
+
+ @Override
+ public void shutdown() {
+ pool.shutdown();
+ }
+
+ @Override
+ public List<Runnable> shutdownNow() {
+ return pool.shutdownNow();
+ }
+
+ @Override
+ public boolean isShutdown() {
+ return pool.isShutdown();
+ }
+
+ @Override
+ public boolean isTerminated() {
+ return pool.isTerminated();
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException {
+ return pool.awaitTermination(timeout, unit);
+ }
+ };
+ var sized = new BoundedExecutorService(delegate, 2, 60,
TimeUnit.SECONDS, false, ThreadPoolRejectedPolicy.Block);
+ try {
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(1);
+ AtomicBoolean ranInline = new AtomicBoolean();
+
+ sized.execute(() -> {
+ Thread outer = Thread.currentThread();
+ // a free permit: the delegate runs this one on the calling
thread
+ direct.set(true);
+ sized.execute(() -> {
+ });
+ direct.remove();
+ // take the other permit, so that the next task finds none
+ sized.execute(() -> {
+ try {
+ release.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+ // with Block this waits forever if the depth of this task was
lost
+ sized.execute(() -> ranInline.set(Thread.currentThread() ==
outer));
+ done.countDown();
+ });
+
+ assertTrue(done.await(5, TimeUnit.SECONDS), "The nested task
should not wait for the permit");
+ assertTrue(ranInline.get(), "The nested task should run inline");
+ release.countDown();
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testNestedTaskUsesAFreePermit() throws Exception {
+ var delegate = Executors.newCachedThreadPool();
+ var sized = new BoundedExecutorService(
+ delegate, 2, 60, TimeUnit.SECONDS, false,
ThreadPoolRejectedPolicy.CallerRuns);
+ try {
+ CountDownLatch done = new CountDownLatch(1);
+ AtomicBoolean sameThread = new AtomicBoolean(true);
+
+ sized.execute(() -> {
+ Thread outer = Thread.currentThread();
+ sized.execute(() -> {
+ sameThread.set(Thread.currentThread() == outer);
+ done.countDown();
+ });
+ });
+
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ assertFalse(sameThread.get(), "The bound still allows this task to
run concurrently");
+ assertEquals(0, sized.getCallerRunsCount());
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testNestedTaskIsStillRejectedWithAbortPolicy() throws
Exception {
+ var delegate = Executors.newCachedThreadPool();
+ var sized = new BoundedExecutorService(
+ delegate, 1, 100, TimeUnit.MILLISECONDS, false,
ThreadPoolRejectedPolicy.Abort);
+ try {
+ CountDownLatch rejected = new CountDownLatch(1);
+
+ sized.execute(() -> {
+ try {
+ sized.execute(() -> {
+ // never runs
+ });
+ } catch (RejectedExecutionException e) {
+ rejected.countDown();
+ }
+ });
+
+ assertTrue(rejected.await(5, TimeUnit.SECONDS), "Abort must fail
the task and not run it");
+ assertEquals(0, sized.getCallerRunsCount());
+ } finally {
+ sized.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testTaskOfAnotherExecutorStillWaits() throws Exception {
+ // only a task of this executor can hold one of its permits, so others
wait for backpressure as usual
+ var full = new BoundedExecutorService(
+ Executors.newCachedThreadPool(), 1, 200,
TimeUnit.MILLISECONDS, false, ThreadPoolRejectedPolicy.CallerRuns);
+ var other = new BoundedExecutorService(
+ Executors.newCachedThreadPool(), 1, 60, TimeUnit.SECONDS,
false, ThreadPoolRejectedPolicy.CallerRuns);
+ try {
+ CountDownLatch release = new CountDownLatch(1);
+ CountDownLatch done = new CountDownLatch(1);
+ AtomicInteger waitedMillis = new AtomicInteger();
+ full.execute(() -> {
+ try {
+ release.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ });
+
+ other.execute(() -> {
+ long start = System.nanoTime();
+ full.execute(() -> waitedMillis.set((int)
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)));
+ done.countDown();
+ });
+
+ assertTrue(done.await(5, TimeUnit.SECONDS));
+ assertTrue(waitedMillis.get() >= 150, "It should wait for the
timeout, but waited " + waitedMillis.get());
+ release.countDown();
+ } finally {
+ full.shutdownNow();
+ other.shutdownNow();
+ }
+ }
}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 7affa39a05c1..8520a1dab8ba 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -141,6 +141,14 @@ processed items of the outer split. It counted its own
items when it had no thre
split when it had them. `SplitResult.getProcessedItems()` now reports the
items of that split only. A nested split with
`errorThreshold` can now stop earlier than before, as the ratio is computed
from the correct number.
+=== camel-util - BoundedExecutorService: a task of the executor does not wait
for a permit
+
+With virtual threads, a task that runs in a thread pool bounded by
`maxQueueSize` and submits more work to it, such as a
+parallel split, no longer waits for a permit when none is available. It runs
the work itself at once, with the
+`CallerRuns` and `Block` rejected policies. Before, it could wait for a permit
it held itself, until `keepAliveTime`
+expired, or forever with `Block`. A split with a `timeout` could lose items
that way. A submitter outside the pool waits
+as before.
+
=== camel-core - RedeliveryErrorHandler.defensiveCopyExchangeIfNeeded is
deprecated
When redelivery is enabled, the error handler (`DefaultErrorHandler`,
`DeadLetterChannel` and the transaction error
diff --git a/docs/user-manual/modules/ROOT/pages/threading-model.adoc
b/docs/user-manual/modules/ROOT/pages/threading-model.adoc
index 1452aecf2e16..2816849d2834 100644
--- a/docs/user-manual/modules/ROOT/pages/threading-model.adoc
+++ b/docs/user-manual/modules/ROOT/pages/threading-model.adoc
@@ -328,6 +328,9 @@ systems where failing fast is preferred over blocking.
message broker consumers and batch workloads where losing a task is
unacceptable and latency is less critical.
|===
+With virtual threads, a task of the pool that submits more work to the same
pool runs it at once when no permit is
+free, for `CallerRuns` and `Block`, instead of waiting (see
<<virtual-threads>>).
+
With platform threads, these policies apply when the `ThreadPoolExecutor` work
queue is full. With virtual threads,
the same policies apply when the concurrency semaphore has no available
permits (see <<virtual-threads>>).
@@ -364,6 +367,12 @@ the task runs on the caller's thread.
a `RejectedExecutionException` is thrown.
* *Block*: the caller blocks indefinitely until a permit becomes available.
+A task of the pool that submits more work to the same pool (for example a
parallel `split` that submits its items)
+can hold the permit it would wait for. So with `CallerRuns` and `Block`, when
no permit is free, that work runs at once
+on the submitting thread instead of waiting. For `Block` this applies even
when other tasks hold the permits. Such work
+runs outside the concurrency limit, as does the work handed to a new task when
it is nested too deeply. `Abort` is
+not affected: it still waits and is rejected as usual.
+
While waiting for a permit, the calling thread is blocked. When callers are
virtual threads this is inexpensive
(the carrier thread is released). When callers are platform threads (e.g.,
HTTP server threads) the blocked thread
is unavailable for other work.
