This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 9.0.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit caa6c7eec18614b5e8446b9062d827655a3054ef Author: opencode <[email protected]> AuthorDate: Mon Oct 12 12:46:59 2026 +0200 Hold the SemaphoreValve permit across asynchronous processing When a request entering the valve was already in async dispatch (state DISPATCHING or STARTED), SemaphoreValve acquired a second permit for the dispatch cycle although the permit from the initial request was still held. With a saturated semaphore the dispatch was rejected with a 503 response, or, for a non blocking valve, completing the failed dispatch threw an IllegalStateException because asynchronous processing had already ended. With block set to true a saturated semaphore additionally deadlocked the dispatch forever, since the timeout mechanism is not available before the dispatch starts processing. Adopt the asynchronous handling used by PersistentValve: a dispatch of an asynchronous request neither acquires nor releases a permit, and a listener attached to the async context releases the permit acquired when asynchronous processing started exactly once, when the asynchronous processing completes or fails. The permit is held across an asynchronous timeout dispatch, and it is reacquired when a new asynchronous cycle is started from a dispatch, so a request always holds exactly one permit while it occupies the pipeline. Add regression tests covering the permit being held across an asynchronous timeout dispatch and a full start, dispatch and complete cycle under a saturated semaphore. --- .../org/apache/catalina/valves/SemaphoreValve.java | 102 +++++++++++--- .../apache/catalina/valves/TestSemaphoreValve.java | 148 ++++++++++++++++++++- webapps/docs/changelog.xml | 9 ++ 3 files changed, 238 insertions(+), 21 deletions(-) diff --git a/java/org/apache/catalina/valves/SemaphoreValve.java b/java/org/apache/catalina/valves/SemaphoreValve.java index 096a473069..37f715be15 100644 --- a/java/org/apache/catalina/valves/SemaphoreValve.java +++ b/java/org/apache/catalina/valves/SemaphoreValve.java @@ -18,7 +18,10 @@ package org.apache.catalina.valves; import java.io.IOException; import java.util.concurrent.Semaphore; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.servlet.AsyncEvent; +import javax.servlet.AsyncListener; import javax.servlet.ServletException; import org.apache.catalina.LifecycleException; @@ -33,7 +36,9 @@ import org.apache.catalina.connector.Response; * <p> * This Valve may be attached to any Container, depending on the granularity of the concurrency control you wish to * perform. Note that internally, some async requests may require multiple serial requests to complete what - to the - * user - appears as a single request. + * user - appears as a single request. The permit acquired for a request that enters asynchronous mode is held until + * the asynchronous processing is complete. Subsequent dispatch cycles of the same asynchronous exchange are not + * separately controlled, since the exchange already holds a permit. * </p> */ public class SemaphoreValve extends ValveBase { @@ -230,32 +235,47 @@ public class SemaphoreValve extends ValveBase { public void invoke(Request request, Response response) throws IOException, ServletException { if (controlConcurrency(request, response)) { - boolean shouldRelease = true; - try { - if (block) { - if (interruptible) { - try { - semaphore.acquire(); - } catch (InterruptedException e) { + if (request.isAsync()) { + // Asynchronous dispatch of an exchange that is already holding a permit from the cycle that started + // the asynchronous processing. Acquiring a second permit for the same exchange could block this + // dispatch waiting for a permit held by the exchange itself, so concurrency control for the exchange + // is handled entirely by the cycle that started the asynchronous processing. + getNext().invoke(request, response); + } else { + boolean shouldRelease = true; + try { + if (block) { + if (interruptible) { + try { + semaphore.acquire(); + } catch (InterruptedException e) { + shouldRelease = false; + permitDenied(request, response); + return; + } + } else { + semaphore.acquireUninterruptibly(); + } + } else { + if (!semaphore.tryAcquire()) { shouldRelease = false; permitDenied(request, response); return; } - } else { - semaphore.acquireUninterruptibly(); } - } else { - if (!semaphore.tryAcquire()) { - shouldRelease = false; - permitDenied(request, response); - return; + getNext().invoke(request, response); + } finally { + if (shouldRelease) { + if (request.isAsync()) { + // The request entered asynchronous mode during this call. Asynchronous processing is + // still in progress when invoke returns. Hold the permit and register a listener that + // releases it once the asynchronous processing is complete. + request.getAsyncContext().addListener(new ReleasePermitAsyncListener(semaphore)); + } else { + semaphore.release(); + } } } - getNext().invoke(request, response); - } finally { - if (shouldRelease) { - semaphore.release(); - } } } else { getNext().invoke(request, response); @@ -293,4 +313,46 @@ public class SemaphoreValve extends ValveBase { } + /** + * Async listener that releases the permit held for a request that entered asynchronous mode, exactly once, when + * the asynchronous processing completes or fails. + */ + private class ReleasePermitAsyncListener implements AsyncListener { + + private final Semaphore semaphore; + private final AtomicBoolean released = new AtomicBoolean(false); + + ReleasePermitAsyncListener(Semaphore semaphore) { + this.semaphore = semaphore; + } + + @Override + public void onComplete(AsyncEvent event) throws IOException { + releaseOnce(); + } + + @Override + public void onTimeout(AsyncEvent event) throws IOException { + // NO-OP. Asynchronous processing continues via a dispatch and the + // request will eventually complete, triggering onComplete. + } + + @Override + public void onError(AsyncEvent event) throws IOException { + // Asynchronous processing terminated with an error and will not complete. + releaseOnce(); + } + + @Override + public void onStartAsync(AsyncEvent event) throws IOException { + // Keep holding the permit across subsequent asynchronous cycles. + event.getAsyncContext().addListener(this); + } + + private void releaseOnce() { + if (released.compareAndSet(false, true)) { + semaphore.release(); + } + } + } } diff --git a/test/org/apache/catalina/valves/TestSemaphoreValve.java b/test/org/apache/catalina/valves/TestSemaphoreValve.java index d041c7e81d..5254f97846 100644 --- a/test/org/apache/catalina/valves/TestSemaphoreValve.java +++ b/test/org/apache/catalina/valves/TestSemaphoreValve.java @@ -24,6 +24,8 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import javax.servlet.AsyncContext; +import javax.servlet.DispatcherType; import javax.servlet.http.HttpServlet; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; @@ -32,6 +34,7 @@ import org.junit.Assert; import org.junit.Test; import org.apache.catalina.Context; +import org.apache.catalina.Wrapper; import org.apache.catalina.connector.Request; import org.apache.catalina.connector.Response; import org.apache.catalina.startup.Tomcat; @@ -177,6 +180,95 @@ public class TestSemaphoreValve extends TomcatBaseTest { } + @Test + public void testPermitHeldDuringAsyncProcessing() throws Exception { + Tomcat tomcat = getTomcatInstance(); + + Context ctx = getProgrammaticRootContext(); + + CountDownLatch asyncStarted = new CountDownLatch(1); + AtomicReference<AsyncContext> asyncContextRef = new AtomicReference<>(); + Wrapper asyncWrapper = Tomcat.addServlet(ctx, "async", + new StartAsyncServlet(asyncStarted, asyncContextRef)); + asyncWrapper.setAsyncSupported(true); + ctx.addServletMapping("/async", "async"); + + Tomcat.addServlet(ctx, "hello", new HelloWorldServlet()); + ctx.addServletMapping("/hello", "hello"); + + SemaphoreValve valve = new SemaphoreValve(); + valve.setConcurrency(1); + valve.setBlock(false); + valve.setHighConcurrencyStatus(HttpServletResponse.SC_SERVICE_UNAVAILABLE); + ctx.getPipeline().addValve(valve); + + tomcat.start(); + + // First request starts asynchronous processing and keeps the permit + AtomicInteger firstRc = new AtomicInteger(); + Thread firstThread = new Thread(() -> { + try { + firstRc.set(getUrl("http://localhost:"; + getPort() + "/async", new ByteChunk(), null)); + } catch (IOException e) { + // Ignore + } + }); + firstThread.start(); + + Assert.assertTrue("First request should start async processing", + asyncStarted.await(10, TimeUnit.SECONDS)); + + // Give the first request time to return from the synchronous phase of + // the asynchronous exchange + Thread.sleep(500); + + // The permit must still be held while async processing is in progress + int rc2 = getUrl("http://localhost:"; + getPort() + "/hello", new ByteChunk(), null); + + Assert.assertEquals(HttpServletResponse.SC_SERVICE_UNAVAILABLE, rc2); + + // Complete the async processing. The permit should be released. + asyncContextRef.get().complete(); + firstThread.join(10000); + Assert.assertFalse(firstThread.isAlive()); + Assert.assertEquals(HttpServletResponse.SC_OK, firstRc.get()); + + // The permit is available again once async processing has completed + ByteChunk res = new ByteChunk(); + int rc3 = getUrl("http://localhost:"; + getPort() + "/hello", res, null); + + Assert.assertEquals(HttpServletResponse.SC_OK, rc3); + Assert.assertEquals(HelloWorldServlet.RESPONSE_TEXT, res.toString()); + } + + + @Test + public void testAsyncDispatchCycleDoesNotAcquireSecondPermit() throws Exception { + Tomcat tomcat = getTomcatInstance(); + + Context ctx = getProgrammaticRootContext(); + + Wrapper dispatchWrapper = Tomcat.addServlet(ctx, "dispatch", new DispatchServlet()); + dispatchWrapper.setAsyncSupported(true); + ctx.addServletMapping("/dispatch", "dispatch"); + + SemaphoreValve valve = new SemaphoreValve(); + valve.setConcurrency(1); + // Default block=true. If the dispatch cycle attempted to acquire a + // second permit, the valve would deadlock since the exchange itself + // holds the only permit. + ctx.getPipeline().addValve(valve); + + tomcat.start(); + + ByteChunk res = new ByteChunk(); + int rc = getUrl("http://localhost:"; + getPort() + "/dispatch", res, null); + + Assert.assertEquals(HttpServletResponse.SC_OK, rc); + Assert.assertEquals("DISPATCHED", res.toString()); + } + + @Test public void testGetSetProperties() { SemaphoreValve valve = new SemaphoreValve(); @@ -368,7 +460,7 @@ public class TestSemaphoreValve extends TomcatBaseTest { Thread blocked = new Thread(() -> { invokeStarted.countDown(); try { - semaphoreValve.invoke(null, response); + semaphoreValve.invoke(new MockRequest(), response); } catch (Throwable t) { // Ignored } @@ -416,6 +508,60 @@ public class TestSemaphoreValve extends TomcatBaseTest { } } + private static final class StartAsyncServlet extends HttpServlet { + + @Serial + private static final long serialVersionUID = 1L; + private final CountDownLatch asyncStarted; + private final AtomicReference<AsyncContext> asyncContextRef; + + private StartAsyncServlet(CountDownLatch asyncStarted, + AtomicReference<AsyncContext> asyncContextRef) { + this.asyncStarted = asyncStarted; + this.asyncContextRef = asyncContextRef; + } + + @Override + protected void doGet(HttpServletRequest req, HttpServletResponse resp) + throws IOException { + AsyncContext asyncContext = req.startAsync(); + // Completion is driven by the test via the captured context + asyncContext.setTimeout(0); + asyncContextRef.set(asyncContext); + asyncStarted.countDown(); + } + } + + private static final class DispatchServlet extends HttpServlet { + + @Serial + private static final long serialVersionUID = 1L; + + @Override + protected void doGet(HttpServletRequest req, HttpServletResponse resp) + throws IOException { + if (req.getDispatcherType() == DispatcherType.ASYNC) { + // Async dispatch cycle. The pipeline is re-entered while the + // exchange holds the permit acquired by the initial cycle. + // The container completes the exchange when this cycle + // returns without restarting the asynchronous processing. + resp.getWriter().print("DISPATCHED"); + } else { + AsyncContext asyncContext = req.startAsync(); + asyncContext.setTimeout(0); + // Dispatch back through the pipeline + asyncContext.dispatch(); + } + } + } + + public static class MockRequest extends Request { + + public MockRequest() { + super(null); + } + } + public static class MockResponse extends Response { public MockResponse() { diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index cea15efa49..0e03224af3 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -2566,6 +2566,15 @@ configured address, while the unresolvable host was advertised to the other cluster members. (remm) </fix> + <fix> + When the Servlet asynchronous processing mechanism is used, + <code>SemaphoreValve</code> no longer attempts to acquire a second + permit when the request is dispatched back to the container. A + saturated semaphore previously rejected such a dispatch with a 503 + response or an <code>IllegalStateException</code>. The permit + obtained when asynchronous processing starts is now released when + that asynchronous processing completes or fails. (remm) + </fix> </changelog> </subsection> <subsection name="Other"> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
