This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch backport/28823-to-camel-4.22.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit 9b5d2046f48179e42ce4a12febc9823b9db93db2 Author: Guillaume Nodet <[email protected]> AuthorDate: Mon Nov 2 22:13:50 2026 +0000 [backport camel-4.22.x] CAMEL-26478: camel-util - BoundedExecutorService: a task must not wait for a permit it holds itself --- .../util/concurrent/BoundedExecutorService.java | 63 ++++- .../concurrent/BoundedExecutorServiceTest.java | 261 +++++++++++++++++++++ .../ROOT/pages/camel-4x-upgrade-guide-4_22.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_22.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc index c4ec32ec16b0..05eb0aa86079 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc @@ -2277,3 +2277,11 @@ CamelContext of that provider. It now closes each of them only when no other `JC `JCacheManager` that was closed no longer opens its cache: `getCache()` then throws `IllegalStateException`; before, it returned the closed cache, whose operations throw `IllegalStateException`. This matters for code that calls a `JCacheManager` (for example one created by a custom `JCacheManagerFactory`) after it was closed. + +=== 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. diff --git a/docs/user-manual/modules/ROOT/pages/threading-model.adoc b/docs/user-manual/modules/ROOT/pages/threading-model.adoc index b419a3a1410a..8b2ea9065d3a 100644 --- a/docs/user-manual/modules/ROOT/pages/threading-model.adoc +++ b/docs/user-manual/modules/ROOT/pages/threading-model.adoc @@ -327,6 +327,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>>). @@ -363,6 +366,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.
