From 497da2da0aa77df0a5faf77c0ff4ad73ce68e206 Mon Sep 17 00:00:00 2001 From: Lars Krog-Jensen Date: Thu, 22 Aug 2024 10:32:29 +0200 Subject: [PATCH 1/7] Issue #12185 implementation/test of max suspended requests in QoSHandler --- .../jetty/server/handler/QoSHandler.java | 49 ++++++++++++++--- .../jetty/server/handler/QoSHandlerTest.java | 54 +++++++++++++++++++ 2 files changed, 97 insertions(+), 6 deletions(-) diff --git a/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/QoSHandler.java b/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/QoSHandler.java index eb7442c6229..2dee11d9923 100644 --- a/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/QoSHandler.java +++ b/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/QoSHandler.java @@ -52,7 +52,10 @@ import org.slf4j.LoggerFactory; * If more requests are received, they are suspended (that is, not * forwarded to the child {@code Handler}) and stored in a priority * queue. - * Priorities are determined via {@link #getPriority(Request)}, + * Maximum number of suspended request can be set {@link #setMaxSuspendedRequestCount(int)} to avoid + * out of memory error. When this limit is reached, the request will fail fast + * with status code {@code 503} (not available).

+ *

Priorities are determined via {@link #getPriority(Request)}, * that should return values between {@code 0} (the lowest priority) * and positive numbers, typically in the range {@code 0-10}.

*

When a request that is being processed completes, the suspended @@ -82,6 +85,7 @@ public class QoSHandler extends ConditionalHandler.Abstract private final Set priorities = new ConcurrentSkipListSet<>(Comparator.reverseOrder()); private CyclicTimeouts timeouts; private int maxRequests; + private int maxSuspendedRequests = Integer.MAX_VALUE; private Duration maxSuspend = Duration.ZERO; public QoSHandler() @@ -119,6 +123,29 @@ public class QoSHandler extends ConditionalHandler.Abstract this.maxRequests = maxRequests; } + /** + * @return the max number of suspended requests + */ + @ManagedAttribute(value = "The maximum number of suspended requests", readonly = true) + public int getMaxSuspendedRequestCount() + { + return maxSuspendedRequests; + } + + /** + *

Sets the max number of suspended requests.

+ *

Once the max suspended request limit is reached, the request is failed with a HTTP + * status of {@code 503 Service unavailable}.

+ * + * @param maxSuspendedRequests the max number of suspended requests + */ + public void setMaxSuspendedRequestCount(int maxSuspendedRequests) + { + if (isStarted()) + throw new IllegalStateException("Cannot change maxSuspendedRequests: " + this); + this.maxSuspendedRequests = maxSuspendedRequests; + } + /** * Get the max duration of time a request may stay suspended. * @return the max duration of time a request may stay suspended @@ -194,6 +221,7 @@ public class QoSHandler extends ConditionalHandler.Abstract LOG.debug("{} processing {}", this, request); boolean expired = false; + boolean tooManyRequests = false; // The read lock allows concurrency with resume(), // which is the common case, but not with expire(). @@ -203,7 +231,14 @@ public class QoSHandler extends ConditionalHandler.Abstract int permits = state.decrementAndGet(); if (permits < 0) { - if (request.getAttribute(EXPIRED_ATTRIBUTE_NAME) == null) + if (Math.abs(permits) > getMaxSuspendedRequestCount()) + { + // Reached the limit of number of suspended requests, + // complete request with 503, service unavailable + state.incrementAndGet(); + tooManyRequests = true; + } + else if (request.getAttribute(EXPIRED_ATTRIBUTE_NAME) == null) { // Cover this race condition: // T1 in this method may find no permits, so it will suspend the request. @@ -228,11 +263,13 @@ public class QoSHandler extends ConditionalHandler.Abstract lock.readLock().unlock(); } - if (!expired) - return handleWithPermit(request, response, callback); + if (expired || tooManyRequests) + { + notAvailable(response, callback); + return true; + } - notAvailable(response, callback); - return true; + return handleWithPermit(request, response, callback); } @Override diff --git a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java index 25c66f469aa..10d85ed0c35 100644 --- a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java +++ b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java @@ -16,6 +16,7 @@ package org.eclipse.jetty.server.handler; import java.time.Duration; import java.util.ArrayList; import java.util.List; +import java.util.Vector; import java.util.concurrent.TimeUnit; import java.util.stream.IntStream; @@ -407,4 +408,57 @@ public class QoSHandlerTest assertEquals(HttpStatus.OK_200, response.getStatus()); } + @Test + public void testMaxSuspendedRequests() throws Exception + { + int delay = 100; + QoSHandler qosHandler = new QoSHandler(); + qosHandler.setMaxRequestCount(2); + qosHandler.setMaxSuspendedRequestCount(2); + qosHandler.setHandler(new Handler.Abstract() + { + @Override + public boolean handle(Request request, Response response, Callback callback) + { + try + { + Thread.sleep(delay); + callback.succeeded(); + } + catch (Throwable x) { + callback.failed(x); + } + return true; + } + }); + start(qosHandler); + + int parallelism = 8; + Vector statusCodes = new Vector<>(); // due to synchronized List + IntStream.range(0, parallelism).parallel().forEach(i -> + { + try (LocalConnector.LocalEndPoint endPoint = connector.executeRequest(""" + GET /%d HTTP/1.1 + Host: localhost + + """.formatted(i))) { + String text = endPoint.getResponse(false, parallelism * delay * 5, TimeUnit.MILLISECONDS); + HttpTester.Response response = HttpTester.parseResponse(text); + statusCodes.add(response.getStatus()); + } + catch (Exception x) + { + fail(x); + } + }); + + await().atMost(5, TimeUnit.SECONDS).until(statusCodes::size, is(8)); + // expectation is that + // 2 requests will be handled straight away + // 2 will be suspended and eventually handled + // 4 will hit the max suspended request limit + assertEquals(4, statusCodes.stream().filter(sc -> sc == HttpStatus.OK_200).count()); + assertEquals(4, statusCodes.stream().filter(sc -> sc == HttpStatus.SERVICE_UNAVAILABLE_503).count()); + } + } From 7aa4c79ab75b520a4af03bf63970a3fe71e6426d Mon Sep 17 00:00:00 2001 From: Lars Krog-Jensen Date: Thu, 22 Aug 2024 10:58:51 +0200 Subject: [PATCH 2/7] Issue #12185 code formatting --- .../jetty/server/handler/QoSHandlerTest.java | 32 ++++++++++--------- 1 file changed, 17 insertions(+), 15 deletions(-) diff --git a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java index 10d85ed0c35..40243e33fdd 100644 --- a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java +++ b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java @@ -99,7 +99,7 @@ public class QoSHandlerTest LocalConnector.LocalEndPoint endPoint = connector.executeRequest(""" GET /%d HTTP/1.1 Host: localhost - + """.formatted(i)); endPoints.add(endPoint); // Wait that the request arrives at the server. @@ -110,7 +110,7 @@ public class QoSHandlerTest LocalConnector.LocalEndPoint endPoint = connector.executeRequest(""" GET /%d HTTP/1.1 Host: localhost - + """.formatted(maxRequests)); endPoints.add(endPoint); @@ -165,7 +165,7 @@ public class QoSHandlerTest LocalConnector.LocalEndPoint endPoint0 = connector.executeRequest(""" GET /0 HTTP/1.1 Host: localhost - + """); await().atMost(5, TimeUnit.SECONDS).until(callbacks::size, is(1)); @@ -173,7 +173,7 @@ public class QoSHandlerTest LocalConnector.LocalEndPoint endPoint1 = connector.executeRequest(""" GET /1 HTTP/1.1 Host: localhost - + """); await().atMost(5, TimeUnit.SECONDS).until(qosHandler::getSuspendedRequestCount, is(1L)); @@ -195,7 +195,7 @@ public class QoSHandlerTest LocalConnector.LocalEndPoint endPoint2 = connector.executeRequest(""" GET /2 HTTP/1.1 Host: localhost - + """); await().atMost(5, TimeUnit.SECONDS).until(callbacks::size, is(1)); callbacks.remove(0).succeeded(); @@ -234,7 +234,7 @@ public class QoSHandlerTest LocalConnector.LocalEndPoint endPoint0 = connector.executeRequest(""" GET /0 HTTP/1.1 Host: localhost - + """); await().atMost(5, TimeUnit.SECONDS).until(callbacks::size, is(1)); @@ -243,7 +243,7 @@ public class QoSHandlerTest GET /1 HTTP/1.1 Host: localhost Priority: 0 - + """); await().atMost(5, TimeUnit.SECONDS).until(callbacks::size, is(1)); @@ -252,7 +252,7 @@ public class QoSHandlerTest GET /2 HTTP/1.1 Host: localhost Priority: 1 - + """); await().atMost(5, TimeUnit.SECONDS).until(callbacks::size, is(1)); @@ -321,7 +321,7 @@ public class QoSHandlerTest try (LocalConnector.LocalEndPoint endPoint = connector.executeRequest(""" GET /%d/%d HTTP/1.1 Host: localhost - + """.formatted(i, j))) { String text = endPoint.getResponse(false, parallelism * iterations * delay * 5, TimeUnit.MILLISECONDS); @@ -361,7 +361,7 @@ public class QoSHandlerTest LocalConnector.LocalEndPoint normalEndPoint = connector.executeRequest(""" GET /normal/request HTTP/1.1 Host: localhost - + """); await().atMost(5, TimeUnit.SECONDS).until(callbacks::size, is(1)); @@ -369,7 +369,7 @@ public class QoSHandlerTest LocalConnector.LocalEndPoint anotherEndPoint = connector.executeRequest(""" GET /another/normal/request HTTP/1.1 Host: localhost - + """); await().atLeast(100, TimeUnit.MILLISECONDS).until(callbacks::size, is(1)); @@ -377,7 +377,7 @@ public class QoSHandlerTest LocalConnector.LocalEndPoint specialEndPoint = connector.executeRequest(""" GET /special/info HTTP/1.1 Host: localhost - + """); // Wait that the request arrives at the server. @@ -425,7 +425,8 @@ public class QoSHandlerTest Thread.sleep(delay); callback.succeeded(); } - catch (Throwable x) { + catch (Throwable x) + { callback.failed(x); } return true; @@ -440,8 +441,9 @@ public class QoSHandlerTest try (LocalConnector.LocalEndPoint endPoint = connector.executeRequest(""" GET /%d HTTP/1.1 Host: localhost - - """.formatted(i))) { + + """.formatted(i))) + { String text = endPoint.getResponse(false, parallelism * delay * 5, TimeUnit.MILLISECONDS); HttpTester.Response response = HttpTester.parseResponse(text); statusCodes.add(response.getStatus()); From 6868c34bdf80f4ca1b4f4e97c344e62a1de0aa47 Mon Sep 17 00:00:00 2001 From: Lars Krog-Jensen Date: Thu, 22 Aug 2024 11:28:34 +0200 Subject: [PATCH 3/7] Issue #12185 white space formatting and increase delay to avoid flaky tests --- .../java/org/eclipse/jetty/server/handler/QoSHandlerTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java index 40243e33fdd..f52761bfb9c 100644 --- a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java +++ b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java @@ -411,7 +411,7 @@ public class QoSHandlerTest @Test public void testMaxSuspendedRequests() throws Exception { - int delay = 100; + int delay = 500; QoSHandler qosHandler = new QoSHandler(); qosHandler.setMaxRequestCount(2); qosHandler.setMaxSuspendedRequestCount(2); From 7976e75b114701f7fa359c3490634f7c265fa30f Mon Sep 17 00:00:00 2001 From: Lars Krog-Jensen Date: Thu, 22 Aug 2024 12:07:32 +0200 Subject: [PATCH 4/7] Issue #12185 Custom FJP in unit test and more white space formatting --- .../jetty/server/handler/QoSHandlerTest.java | 51 ++++++++++--------- 1 file changed, 28 insertions(+), 23 deletions(-) diff --git a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java index f52761bfb9c..2e4392bbcb4 100644 --- a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java +++ b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java @@ -17,6 +17,7 @@ import java.time.Duration; import java.util.ArrayList; import java.util.List; import java.util.Vector; +import java.util.concurrent.ForkJoinPool; import java.util.concurrent.TimeUnit; import java.util.stream.IntStream; @@ -436,31 +437,35 @@ public class QoSHandlerTest int parallelism = 8; Vector statusCodes = new Vector<>(); // due to synchronized List - IntStream.range(0, parallelism).parallel().forEach(i -> + ForkJoinPool fjp = new ForkJoinPool(parallelism); + try { - try (LocalConnector.LocalEndPoint endPoint = connector.executeRequest(""" - GET /%d HTTP/1.1 - Host: localhost + fjp.submit(() -> IntStream.range(0, parallelism).parallel().forEach(i -> + { + try (LocalConnector.LocalEndPoint endPoint = connector.executeRequest(""" + GET /%d HTTP/1.1 + Host: localhost - """.formatted(i))) - { - String text = endPoint.getResponse(false, parallelism * delay * 5, TimeUnit.MILLISECONDS); - HttpTester.Response response = HttpTester.parseResponse(text); - statusCodes.add(response.getStatus()); - } - catch (Exception x) - { - fail(x); - } - }); - - await().atMost(5, TimeUnit.SECONDS).until(statusCodes::size, is(8)); - // expectation is that - // 2 requests will be handled straight away - // 2 will be suspended and eventually handled - // 4 will hit the max suspended request limit - assertEquals(4, statusCodes.stream().filter(sc -> sc == HttpStatus.OK_200).count()); - assertEquals(4, statusCodes.stream().filter(sc -> sc == HttpStatus.SERVICE_UNAVAILABLE_503).count()); + """.formatted(i))) { + String text = endPoint.getResponse(false, parallelism * delay * 5, TimeUnit.MILLISECONDS); + HttpTester.Response response = HttpTester.parseResponse(text); + statusCodes.add(response.getStatus()); + } catch (Exception x) { + fail(x); + } + })); + await().atMost(5, TimeUnit.SECONDS).until(statusCodes::size, is(8)); + // expectation is that + // 2 requests will be handled straight away + // 2 will be suspended and eventually handled + // 4 will hit the max suspended request limit + assertEquals(4, statusCodes.stream().filter(sc -> sc == HttpStatus.OK_200).count()); + assertEquals(4, statusCodes.stream().filter(sc -> sc == HttpStatus.SERVICE_UNAVAILABLE_503).count()); + } + finally + { + fjp.shutdown(); + } } } From 30791c16ee463655a205934bf702f7e62df6e38f Mon Sep 17 00:00:00 2001 From: Lars Krog-Jensen Date: Thu, 22 Aug 2024 12:18:02 +0200 Subject: [PATCH 5/7] Issue #12185 code formatting --- .../org/eclipse/jetty/server/handler/QoSHandlerTest.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java index 2e4392bbcb4..d4ad06b7af9 100644 --- a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java +++ b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java @@ -446,11 +446,14 @@ public class QoSHandlerTest GET /%d HTTP/1.1 Host: localhost - """.formatted(i))) { + """.formatted(i))) + { String text = endPoint.getResponse(false, parallelism * delay * 5, TimeUnit.MILLISECONDS); HttpTester.Response response = HttpTester.parseResponse(text); statusCodes.add(response.getStatus()); - } catch (Exception x) { + } + catch (Exception x) + { fail(x); } })); From dc41ad90ab508086e59f1e26a572d7463aa107c3 Mon Sep 17 00:00:00 2001 From: Jan Bartel Date: Thu, 22 Aug 2024 18:19:44 +1000 Subject: [PATCH 6/7] Fix order of jetty.http.port property for jetty maven plugin (#12183) --- .../jetty/maven/AbstractHomeForker.java | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/jetty-core/jetty-maven/src/main/java/org/eclipse/jetty/maven/AbstractHomeForker.java b/jetty-core/jetty-maven/src/main/java/org/eclipse/jetty/maven/AbstractHomeForker.java index 1e2ce6a45f3..8fed99ce6d2 100644 --- a/jetty-core/jetty-maven/src/main/java/org/eclipse/jetty/maven/AbstractHomeForker.java +++ b/jetty-core/jetty-maven/src/main/java/org/eclipse/jetty/maven/AbstractHomeForker.java @@ -198,6 +198,15 @@ public abstract class AbstractHomeForker extends AbstractForker if (stopKey != null) cmd.add("-DSTOP.KEY=" + stopKey); + //put any jetty properties onto the command line + if (jettyProperties != null) + { + for (Map.Entry e : jettyProperties.entrySet()) + { + cmd.add(e.getKey() + "=" + e.getValue()); + } + } + //set up enabled jetty modules StringBuilder tmp = new StringBuilder(); tmp.append("--module="); @@ -214,6 +223,7 @@ public abstract class AbstractHomeForker extends AbstractForker if (libExtJarFiles != null && !libExtJarFiles.isEmpty() && tmp.indexOf("ext") < 0) tmp.append(",ext"); tmp.append("," + environment + "-maven"); + cmd.add(tmp.toString()); //put any other jetty options onto the command line @@ -222,15 +232,6 @@ public abstract class AbstractHomeForker extends AbstractForker Arrays.stream(jettyOptions.split(" ")).filter(a -> StringUtil.isNotBlank(a)).forEach((a) -> cmd.add(a.trim())); } - //put any jetty properties onto the command line - if (jettyProperties != null) - { - for (Map.Entry e : jettyProperties.entrySet()) - { - cmd.add(e.getKey() + "=" + e.getValue()); - } - } - //existence of this file signals process started cmd.add("jetty.token.file=" + tokenFile.getAbsolutePath().toString()); From aae0a55104aa11155805acba6d70be9ab4767430 Mon Sep 17 00:00:00 2001 From: Simone Bordet Date: Fri, 23 Aug 2024 11:31:11 +0200 Subject: [PATCH 7/7] Limited max suspended requests to 1024 by default. Javadocs, comment and logging improvements. Simplified test case. Signed-off-by: Simone Bordet --- .../jetty/server/handler/QoSHandler.java | 29 ++++--- .../jetty/server/handler/QoSHandlerTest.java | 80 +++++++++++-------- 2 files changed, 64 insertions(+), 45 deletions(-) diff --git a/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/QoSHandler.java b/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/QoSHandler.java index 2dee11d9923..69722c13d40 100644 --- a/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/QoSHandler.java +++ b/jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/QoSHandler.java @@ -51,9 +51,10 @@ import org.slf4j.LoggerFactory; * to the number configured via {@link #setMaxRequestCount(int)}. * If more requests are received, they are suspended (that is, not * forwarded to the child {@code Handler}) and stored in a priority - * queue. - * Maximum number of suspended request can be set {@link #setMaxSuspendedRequestCount(int)} to avoid - * out of memory error. When this limit is reached, the request will fail fast + * queue.

+ *

The maximum number of suspended request can be set with + * {@link #setMaxSuspendedRequestCount(int)} to avoid out of memory errors. + * When this limit is reached, the request will fail fast * with status code {@code 503} (not available).

*

Priorities are determined via {@link #getPriority(Request)}, * that should return values between {@code 0} (the lowest priority) @@ -85,7 +86,7 @@ public class QoSHandler extends ConditionalHandler.Abstract private final Set priorities = new ConcurrentSkipListSet<>(Comparator.reverseOrder()); private CyclicTimeouts timeouts; private int maxRequests; - private int maxSuspendedRequests = Integer.MAX_VALUE; + private int maxSuspendedRequests = 1024; private Duration maxSuspend = Duration.ZERO; public QoSHandler() @@ -134,8 +135,11 @@ public class QoSHandler extends ConditionalHandler.Abstract /** *

Sets the max number of suspended requests.

- *

Once the max suspended request limit is reached, the request is failed with a HTTP - * status of {@code 503 Service unavailable}.

+ *

Once the max suspended request limit is reached, + * the request is failed with a HTTP status of + * {@code 503 Service unavailable}.

+ *

A negative value indicate an unlimited number + * of suspended requests.

* * @param maxSuspendedRequests the max number of suspended requests */ @@ -171,7 +175,7 @@ public class QoSHandler extends ConditionalHandler.Abstract } @ManagedAttribute("The number of suspended requests") - public long getSuspendedRequestCount() + public int getSuspendedRequestCount() { int permits = state.get(); return Math.max(0, -permits); @@ -231,10 +235,11 @@ public class QoSHandler extends ConditionalHandler.Abstract int permits = state.decrementAndGet(); if (permits < 0) { - if (Math.abs(permits) > getMaxSuspendedRequestCount()) + int maxSuspended = getMaxSuspendedRequestCount(); + if (maxSuspended >= 0 && Math.abs(permits) > maxSuspended) { - // Reached the limit of number of suspended requests, - // complete request with 503, service unavailable + // Reached the limit of suspended requests, + // complete the request with 503 unavailable. state.incrementAndGet(); tooManyRequests = true; } @@ -278,8 +283,10 @@ public class QoSHandler extends ConditionalHandler.Abstract return nextHandler(request, response, callback); } - private static void notAvailable(Response response, Callback callback) + private void notAvailable(Response response, Callback callback) { + if (LOG.isDebugEnabled()) + LOG.debug("{} rejecting {}", this, response.getRequest()); response.setStatus(HttpStatus.SERVICE_UNAVAILABLE_503); if (response.isCommitted()) callback.failed(new IllegalStateException("Response already committed")); diff --git a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java index d4ad06b7af9..76f67fd2095 100644 --- a/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java +++ b/jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java @@ -16,9 +16,8 @@ package org.eclipse.jetty.server.handler; import java.time.Duration; import java.util.ArrayList; import java.util.List; -import java.util.Vector; -import java.util.concurrent.ForkJoinPool; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.IntStream; import org.eclipse.jetty.http.HttpStatus; @@ -412,10 +411,11 @@ public class QoSHandlerTest @Test public void testMaxSuspendedRequests() throws Exception { - int delay = 500; + int delay = 1000; QoSHandler qosHandler = new QoSHandler(); qosHandler.setMaxRequestCount(2); qosHandler.setMaxSuspendedRequestCount(2); + AtomicInteger handling = new AtomicInteger(); qosHandler.setHandler(new Handler.Abstract() { @Override @@ -423,6 +423,7 @@ public class QoSHandlerTest { try { + handling.incrementAndGet(); Thread.sleep(delay); callback.succeeded(); } @@ -435,40 +436,51 @@ public class QoSHandlerTest }); start(qosHandler); - int parallelism = 8; - Vector statusCodes = new Vector<>(); // due to synchronized List - ForkJoinPool fjp = new ForkJoinPool(parallelism); - try + List endPoints = new ArrayList<>(); + // Send 2 requests that should pass through QoSHandler. + for (int i = 0; i < 2; i++) { - fjp.submit(() -> IntStream.range(0, parallelism).parallel().forEach(i -> - { - try (LocalConnector.LocalEndPoint endPoint = connector.executeRequest(""" - GET /%d HTTP/1.1 - Host: localhost - - """.formatted(i))) - { - String text = endPoint.getResponse(false, parallelism * delay * 5, TimeUnit.MILLISECONDS); - HttpTester.Response response = HttpTester.parseResponse(text); - statusCodes.add(response.getStatus()); - } - catch (Exception x) - { - fail(x); - } - })); - await().atMost(5, TimeUnit.SECONDS).until(statusCodes::size, is(8)); - // expectation is that - // 2 requests will be handled straight away - // 2 will be suspended and eventually handled - // 4 will hit the max suspended request limit - assertEquals(4, statusCodes.stream().filter(sc -> sc == HttpStatus.OK_200).count()); - assertEquals(4, statusCodes.stream().filter(sc -> sc == HttpStatus.SERVICE_UNAVAILABLE_503).count()); + LocalConnector.LocalEndPoint endPoint = connector.executeRequest(""" + GET /pass/%d HTTP/1.1 + Host: localhost + + """.formatted(i)); + endPoints.add(endPoint); } - finally + await().atMost(5, TimeUnit.SECONDS).until(handling::get, is(2)); + // Send 2 requests that should be suspended by QoSHandler. + for (int i = 0; i < 2; i++) { - fjp.shutdown(); + LocalConnector.LocalEndPoint endPoint = connector.executeRequest(""" + GET /suspend/%d HTTP/1.1 + Host: localhost + + """.formatted(i)); + endPoints.add(endPoint); } + await().atMost(5, TimeUnit.SECONDS).until(qosHandler::getSuspendedRequestCount, is(2)); + // Send 2 requests that should be failed immediately by QoSHandler. + for (int i = 0; i < 2; i++) + { + HttpTester.Response response = HttpTester.parseResponse(connector.getResponse(""" + GET /rejected/%d HTTP/1.1 + Host: localhost + + """.formatted(i))); + assertEquals(HttpStatus.SERVICE_UNAVAILABLE_503, response.getStatus()); + } + // Wait for the other requests to finish normally. + endPoints.forEach(endPoint -> + { + try + { + HttpTester.Response response = HttpTester.parseResponse(endPoint.getResponse(false, 2 * delay, TimeUnit.MILLISECONDS)); + assertEquals(HttpStatus.OK_200, response.getStatus()); + } + catch (Exception x) + { + fail(x); + } + }); } - }