Eclipse
/
jetty.project
mirror of https://github.com/jetty/jetty.project.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Limited max suspended requests to 1024 by default. 

Javadocs, comment and logging improvements.
Simplified test case.

Signed-off-by: Simone Bordet <simone.bordet@gmail.com>
This commit is contained in:
Simone Bordet 2024-08-23 11:31:11 +02:00
parent dc41ad90ab
commit aae0a55104
No known key found for this signature in database
GPG Key ID: 1677D141BCF3584D
2 changed files with 64 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/handler/QoSHandler.java
View File

@ -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.</p>
* <p>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).</p>
* <p>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<Integer> priorities = new ConcurrentSkipListSet<>(Comparator.reverseOrder());
private CyclicTimeouts<Entry> 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
/**
* <p>Sets the max number of suspended requests.</p>
* <p>Once the max suspended request limit is reached, the request is failed with a HTTP
* status of {@code 503 Service unavailable}.</p>
* <p>Once the max suspended request limit is reached,
* the request is failed with a HTTP status of
* {@code 503 Service unavailable}.</p>
* <p>A negative value indicate an unlimited number
* of suspended requests.</p>
*
* @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"));

80
jetty-core/jetty-server/src/test/java/org/eclipse/jetty/server/handler/QoSHandlerTest.java
View File

@ -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<Integer> statusCodes = new Vector<>(); // due to synchronized List
ForkJoinPool fjp = new ForkJoinPool(parallelism);
try
List<LocalConnector.LocalEndPoint> 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);
}
});
}
}