This is an automated email from the ASF dual-hosted git repository. chrisdutz pushed a commit to branch develop in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 22ca9ed054d7cbdc51a781b3bfc3011f2bc2e084 Author: Christofer Dutz <[email protected]> AuthorDate: Fri Aug 14 17:36:46 2026 +0200 fix(plc4j/spi): Don't block the calling thread while waiting for a request permit RequestThrottle.execute() acquired its permit by blocking the caller. A driver that chains requests with thenCompose runs the follow-up on whichever thread completed the previous stage, which is typically the connection's receive thread. When all permits were held, that thread parked waiting for a permit and could no longer deliver the response that would have released one - the connection only recovered when an unrelated request hit its timeout. Negotiated max-amq is often 1-3 on S7-300, which makes this easy to reach. Requests are now queued and started in submission order as permits free up, so submitting never parks a thread. A single-drainer guard keeps a request that completes inline from recursing back into the queue, and raising the concurrency limit now starts queued work immediately instead of waiting for the next completion. The new tests hang against the previous implementation rather than failing, which is the deadlock itself. --- .../java/spi/drivers/throttle/RequestThrottle.java | 82 +++++++++++----- .../spi/drivers/throttle/RequestThrottleTest.java | 103 ++++++++++++++++++++- 2 files changed, 163 insertions(+), 22 deletions(-) diff --git a/plc4j/spi/drivers/src/main/java/org/apache/plc4x/java/spi/drivers/throttle/RequestThrottle.java b/plc4j/spi/drivers/src/main/java/org/apache/plc4x/java/spi/drivers/throttle/RequestThrottle.java index 141e792edc..0754cc16f4 100644 --- a/plc4j/spi/drivers/src/main/java/org/apache/plc4x/java/spi/drivers/throttle/RequestThrottle.java +++ b/plc4j/spi/drivers/src/main/java/org/apache/plc4x/java/spi/drivers/throttle/RequestThrottle.java @@ -21,8 +21,11 @@ package org.apache.plc4x.java.spi.drivers.throttle; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.Queue; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Supplier; /** @@ -35,6 +38,9 @@ public class RequestThrottle { private final Semaphore semaphore; private volatile int maxConcurrentRequests; + /** Requests waiting for a permit, started in the order they were submitted. */ + private final Queue<Runnable> pending = new ConcurrentLinkedQueue<>(); + private final AtomicBoolean draining = new AtomicBoolean(); public RequestThrottle(int maxConcurrentRequests) { if (maxConcurrentRequests < 1) { @@ -61,32 +67,65 @@ public class RequestThrottle { } public <T> CompletableFuture<T> execute(Supplier<CompletableFuture<T>> requestSupplier) { - try { - semaphore.acquire(); + CompletableFuture<T> result = new CompletableFuture<>(); + pending.add(() -> start(requestSupplier, result)); + drain(); + return result; + } - CompletableFuture<T> requestFuture; - try { - requestFuture = requestSupplier.get(); - } catch (Exception e) { - semaphore.release(); - return CompletableFuture.failedFuture(e); + /** + * Starts as many queued requests as there are permits. Only one thread drains at a time; any + * other thread hands its work to whoever holds the drain and returns immediately, so a request + * that completes inline cannot recurse back into the queue. + */ + private void drain() { + while (true) { + if (!draining.compareAndSet(false, true)) { + return; } - - // Use handle() to release the semaphore and propagate the result in a - // single stage, avoiding the extra CompletableFuture + whenComplete overhead. - return requestFuture.handle((result, error) -> { - semaphore.release(); - if (error != null) { - throw (error instanceof RuntimeException re) ? re : new RuntimeException(error); + try { + while (!pending.isEmpty() && semaphore.tryAcquire()) { + Runnable request = pending.poll(); + if (request == null) { + // Another thread took the entry between the check and the poll. + semaphore.release(); + break; + } + request.run(); } - return result; - }); + } finally { + draining.set(false); + } + // Something may have been queued by another thread while we were finishing the loop + // above; without this re-check it would sit there until the next permit is released. + if (pending.isEmpty() || semaphore.availablePermits() == 0) { + return; + } + } + } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return CompletableFuture.failedFuture( - new RuntimeException("Request throttling interrupted", e)); + private <T> void start(Supplier<CompletableFuture<T>> requestSupplier, CompletableFuture<T> result) { + CompletableFuture<T> requestFuture; + try { + requestFuture = requestSupplier.get(); + } catch (Exception e) { + releaseAndDrain(); + result.completeExceptionally(e); + return; } + requestFuture.whenComplete((value, error) -> { + releaseAndDrain(); + if (error != null) { + result.completeExceptionally(error); + } else { + result.complete(value); + } + }); + } + + private void releaseAndDrain() { + semaphore.release(); + drain(); } public synchronized void adjustMaxConcurrentRequests(int newMax) { @@ -100,6 +139,7 @@ public class RequestThrottle { int difference = newMax - this.maxConcurrentRequests; if (difference > 0) { semaphore.release(difference); + drain(); } else { int toDrain = Math.min(-difference, semaphore.availablePermits()); if (toDrain > 0) { diff --git a/plc4j/spi/drivers/src/test/java/org/apache/plc4x/java/spi/drivers/throttle/RequestThrottleTest.java b/plc4j/spi/drivers/src/test/java/org/apache/plc4x/java/spi/drivers/throttle/RequestThrottleTest.java index fac58417d6..2ed3af9140 100644 --- a/plc4j/spi/drivers/src/test/java/org/apache/plc4x/java/spi/drivers/throttle/RequestThrottleTest.java +++ b/plc4j/spi/drivers/src/test/java/org/apache/plc4x/java/spi/drivers/throttle/RequestThrottleTest.java @@ -21,7 +21,12 @@ package org.apache.plc4x.java.spi.drivers.throttle; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.*; @@ -217,4 +222,100 @@ class RequestThrottleTest { assertEquals(1, throttle.getAvailablePermits()); assertEquals(0, throttle.getInFlightRequests()); } -} + + /** + * Submitting a request must never park the calling thread. A driver that chains requests with + * thenCompose runs the follow-up on the thread that completed the previous stage - typically + * the connection's receive thread. If submitting blocked there while all permits were held, + * that thread could no longer deliver the response that would free a permit, and the + * connection stalled until an unrelated request timed out. + */ + @Test + void submittingDoesNotBlockWhenNoPermitIsAvailable() throws Exception { + RequestThrottle throttle = new RequestThrottle(1); + CompletableFuture<String> blocker = new CompletableFuture<>(); + throttle.execute(() -> blocker); + + // No permit left. This must return immediately rather than parking this thread. + AtomicBoolean started = new AtomicBoolean(); + CompletableFuture<String> queued = throttle.execute(() -> { + started.set(true); + return CompletableFuture.completedFuture("second"); + }); + + assertFalse(started.get(), "the queued request must not start before a permit is free"); + assertFalse(queued.isDone()); + + blocker.complete("first"); + + assertEquals("second", queued.get(5, TimeUnit.SECONDS)); + assertEquals(1, throttle.getAvailablePermits()); + } + + /** + * The deadlock itself: the queued request is released by the completion of the first one, from + * the very thread that is inside the first request's completion callback. + */ + @Test + void requestCompletingFromWithinAnotherRequestDoesNotDeadlock() throws Exception { + RequestThrottle throttle = new RequestThrottle(1); + CompletableFuture<String> first = new CompletableFuture<>(); + CompletableFuture<String> firstResult = throttle.execute(() -> first); + + // Chain a second request onto the first, the way a driver splits a large request. + CompletableFuture<String> chained = firstResult.thenCompose( + ignored -> throttle.execute(() -> CompletableFuture.completedFuture("chained"))); + + // Completing from this thread runs the continuation - and therefore the second submit - + // on this thread, while the permit is still held by the first request. + first.complete("first"); + + assertEquals("chained", chained.get(5, TimeUnit.SECONDS)); + assertEquals(1, throttle.getAvailablePermits()); + assertEquals(0, throttle.getInFlightRequests()); + } + + /** + * Queued requests start in submission order. + */ + @Test + void queuedRequestsRunInOrder() throws Exception { + RequestThrottle throttle = new RequestThrottle(1); + CompletableFuture<String> blocker = new CompletableFuture<>(); + throttle.execute(() -> blocker); + + List<Integer> order = Collections.synchronizedList(new ArrayList<>()); + List<CompletableFuture<String>> queued = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + int index = i; + queued.add(throttle.execute(() -> { + order.add(index); + return CompletableFuture.completedFuture("done" + index); + })); + } + + blocker.complete("first"); + for (CompletableFuture<String> future : queued) { + future.get(5, TimeUnit.SECONDS); + } + + assertEquals(List.of(0, 1, 2, 3, 4), order); + assertEquals(1, throttle.getAvailablePermits()); + } + + /** + * Raising the limit has to let already-queued requests start. + */ + @Test + void raisingTheLimitStartsQueuedRequests() throws Exception { + RequestThrottle throttle = new RequestThrottle(1); + throttle.execute(() -> new CompletableFuture<>()); + + CompletableFuture<String> queued = throttle.execute(() -> CompletableFuture.completedFuture("ok")); + assertFalse(queued.isDone()); + + throttle.adjustMaxConcurrentRequests(2); + + assertEquals("ok", queued.get(5, TimeUnit.SECONDS)); + } +} \ No newline at end of file
