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

Merged branch 'LarsKrogJensen-issue-12185' into 'jetty-12.0.x'.

This commit is contained in:
Simone Bordet 2024-08-23 11:33:16 +02:00
parent 36ec04f92e aae0a55104
commit 185b423e1a
No known key found for this signature in database
GPG Key ID: 1677D141BCF3584D
2 changed files with 141 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -51,8 +51,12 @@ 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.
* Priorities are determined via {@link #getPriority(Request)},
* 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)
* and positive numbers, typically in the range {@code 0-10}.</p>
* <p>When a request that is being processed completes, the suspended
@ -82,6 +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 = 1024;
private Duration maxSuspend = Duration.ZERO;
public QoSHandler()
@ -119,6 +124,32 @@ 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;
}
/**
* <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>A negative value indicate an unlimited number
* of suspended requests.</p>
*
* @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
@ -144,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);
@ -194,6 +225,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 +235,15 @@ public class QoSHandler extends ConditionalHandler.Abstract
int permits = state.decrementAndGet();
if (permits < 0)
{
if (request.getAttribute(EXPIRED_ATTRIBUTE_NAME) == null)
int maxSuspended = getMaxSuspendedRequestCount();
if (maxSuspended >= 0 && Math.abs(permits) > maxSuspended)
{
// Reached the limit of suspended requests,
// complete the request with 503 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 +268,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
@ -241,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"));

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

@ -17,6 +17,7 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
import org.eclipse.jetty.http.HttpStatus;
@ -98,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.
@ -109,7 +110,7 @@ public class QoSHandlerTest
LocalConnector.LocalEndPoint endPoint = connector.executeRequest("""
GET /%d HTTP/1.1
Host: localhost
""".formatted(maxRequests));
endPoints.add(endPoint);
@ -164,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));
@ -172,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));
@ -194,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();
@ -233,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));
@ -242,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));
@ -251,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));
@ -320,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);
@ -360,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));
@ -368,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));
@ -376,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.
@ -407,4 +408,79 @@ public class QoSHandlerTest
assertEquals(HttpStatus.OK_200, response.getStatus());
}
@Test
public void testMaxSuspendedRequests() throws Exception
{
int delay = 1000;
QoSHandler qosHandler = new QoSHandler();
qosHandler.setMaxRequestCount(2);
qosHandler.setMaxSuspendedRequestCount(2);
AtomicInteger handling = new AtomicInteger();
qosHandler.setHandler(new Handler.Abstract()
{
@Override
public boolean handle(Request request, Response response, Callback callback)
{
try
{
handling.incrementAndGet();
Thread.sleep(delay);
callback.succeeded();
}
catch (Throwable x)
{
callback.failed(x);
}
return true;
}
});
start(qosHandler);
List<LocalConnector.LocalEndPoint> endPoints = new ArrayList<>();
// Send 2 requests that should pass through QoSHandler.
for (int i = 0; i < 2; i++)
{
LocalConnector.LocalEndPoint endPoint = connector.executeRequest("""
GET /pass/%d HTTP/1.1
Host: localhost
""".formatted(i));
endPoints.add(endPoint);
}
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++)
{
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);
}
});
}
}