Skip to content

Cap total number of concurrent requests per HTTP/2 connection #592

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
arturobernalg wants to merge 1 commit into
base: master
Choose a base branch
from
+976 −81
Open

Cap total number of concurrent requests per HTTP/2 connection #592

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions ...core5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2AsyncRequester.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,9 +77,10 @@ public H2AsyncRequester(
final TlsStrategy tlsStrategy,
final Timeout handshakeTimeout,
final IOReactorMetricsListener threadPoolListener,
final IOWorkerSelector workerSelector) {
final IOWorkerSelector workerSelector,
final int maxPendingCommandsPerConnection) {
super(ioReactorConfig, eventHandlerFactory, ioSessionDecorator, exceptionCallback, sessionListener, connPool,
tlsStrategy, handshakeTimeout, threadPoolListener, workerSelector);
tlsStrategy, handshakeTimeout, threadPoolListener, workerSelector, maxPendingCommandsPerConnection);
this.versionPolicy = versionPolicy != null ? versionPolicy : HttpVersionPolicy.NEGOTIATE;
}

Expand Down
139 changes: 64 additions & 75 deletions ...2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,10 +29,9 @@

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;

import org.apache.hc.core5.annotation.Internal;
import org.apache.hc.core5.concurrent.Cancellable;
Expand All @@ -44,21 +43,15 @@
import org.apache.hc.core5.function.Decorator;
import org.apache.hc.core5.function.Resolver;
import org.apache.hc.core5.http.ConnectionClosedException;
import org.apache.hc.core5.http.EntityDetails;
import org.apache.hc.core5.http.Header;
import org.apache.hc.core5.http.HttpException;
import org.apache.hc.core5.http.HttpHost;
import org.apache.hc.core5.http.HttpResponse;
import org.apache.hc.core5.http.impl.DefaultAddressResolver;
import org.apache.hc.core5.http.impl.bootstrap.AsyncRequester;
import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
import org.apache.hc.core5.http.nio.AsyncPushConsumer;
import org.apache.hc.core5.http.nio.AsyncRequestProducer;
import org.apache.hc.core5.http.nio.AsyncResponseConsumer;
import org.apache.hc.core5.http.nio.CapacityChannel;
import org.apache.hc.core5.http.nio.DataStreamChannel;
import org.apache.hc.core5.http.nio.HandlerFactory;
import org.apache.hc.core5.http.nio.RequestChannel;
import org.apache.hc.core5.http.nio.command.RequestExecutionCommand;
import org.apache.hc.core5.http.nio.command.ShutdownCommand;
import org.apache.hc.core5.http.nio.ssl.TlsStrategy;
Expand Down Expand Up @@ -87,6 +80,12 @@ public class H2MultiplexingRequester extends AsyncRequester {

private final H2ConnPool connPool;

/**
* Hard cap on per-connection queued / in-flight requests.
* {@code <= 0} disables the cap.
*/
private final int maxRequestsPerConnection;

/**
* Use {@link H2MultiplexingRequesterBootstrap} to create instances of this class.
*/
Expand All @@ -100,11 +99,13 @@ public H2MultiplexingRequester(
final Resolver<HttpHost, InetSocketAddress> addressResolver,
final TlsStrategy tlsStrategy,
final IOReactorMetricsListener threadPoolListener,
final IOWorkerSelector workerSelector) {
final IOWorkerSelector workerSelector,
final int maxRequestsPerConnection) {
super(eventHandlerFactory, ioReactorConfig, ioSessionDecorator, exceptionCallback, sessionListener,
ShutdownCommand.GRACEFUL_IMMEDIATE_CALLBACK, DefaultAddressResolver.INSTANCE,
threadPoolListener, workerSelector);
this.connPool = new H2ConnPool(this, addressResolver, tlsStrategy);
this.maxRequestsPerConnection = maxRequestsPerConnection;
}

public void closeIdle(final TimeValue idleTime) {
Expand Down Expand Up @@ -182,83 +183,42 @@ private void execute(
if (request.getAuthority() == null) {
request.setAuthority(new URIAuthority(host));
}
if (request.getScheme() == null) {
request.setScheme(host.getSchemeName());
}
connPool.getSession(host, timeout, new FutureCallback<IOSession>() {

@Override
public void completed(final IOSession ioSession) {
final AsyncClientExchangeHandler handlerProxy = new AsyncClientExchangeHandler() {

@Override
public void releaseResources() {
final int max = maxRequestsPerConnection;
Copy link
Member

@ok2c ok2c Jan 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@arturobernalg Why, why all this complexity? You are not going to impress no young ladies with it. You almost had it right last time. Restore the command count method from the previous revision and fail the request if the total number of pending commands is over the max limit. That should be all. The tricky bit is to come up with a reasonable config mechanism for the max pending command limit.

Copy link
Member Author

@arturobernalg arturobernalg Jan 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@arturobernalg Why, why all this complexity? You are not going to impress no young ladies with it. You almost had it right last time. Restore the command count method from the previous revision and fail the request if the total number of pending commands is over the max limit. That should be all. The tricky bit is to come up with a reasonable config mechanism for the max pending command limit.
Hi @ok2c , I’ve ditched the extra plumbing and went back to the simple approach: restored the command-count method and now H2MultiplexingRequester just fails fast when pending commands exceed the configured max.

if (max > 0) {
final int current = ioSession.getPendingCommandCount();
if (current >= 0 && current >= max) {
exchangeHandler.failed(new RejectedExecutionException(
"Maximum number of pending requests per connection reached (max=" + max + ")"));
exchangeHandler.releaseResources();
return;
}
}

@Override
public void produceRequest(final RequestChannel channel, final HttpContext httpContext) throws HttpException, IOException {
channel.sendRequest(request, entityDetails, httpContext);
}

@Override
public int available() {
return exchangeHandler.available();
}

@Override
public void produce(final DataStreamChannel channel) throws IOException {
exchangeHandler.produce(channel);
}

@Override
public void consumeInformation(final HttpResponse response, final HttpContext httpContext) throws HttpException, IOException {
exchangeHandler.consumeInformation(response, httpContext);
}

@Override
public void consumeResponse(
final HttpResponse response, final EntityDetails entityDetails, final HttpContext httpContext) throws HttpException, IOException {
exchangeHandler.consumeResponse(response, entityDetails, httpContext);
}

@Override
public void updateCapacity(final CapacityChannel capacityChannel) throws IOException {
exchangeHandler.updateCapacity(capacityChannel);
}

@Override
public void consume(final ByteBuffer src) throws IOException {
exchangeHandler.consume(src);
}

@Override
public void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException {
exchangeHandler.streamEnd(trailers);
}

@Override
public void cancel() {
exchangeHandler.cancel();
}

@Override
public void failed(final Exception cause) {
exchangeHandler.failed(cause);
}

};
final Timeout socketTimeout = ioSession.getSocketTimeout();
ioSession.enqueue(new RequestExecutionCommand(
handlerProxy,
pushHandlerFactory,
context,
streamControl -> {
cancellableDependency.setDependency(streamControl);
if (socketTimeout != null) {
streamControl.setTimeout(socketTimeout);
}
}),
Command.Priority.NORMAL);
final RequestExecutionCommand command = new RequestExecutionCommand(
exchangeHandler,
pushHandlerFactory,
context,
streamControl -> {
cancellableDependency.setDependency(streamControl);
if (socketTimeout != null) {
streamControl.setTimeout(socketTimeout);
}
});

ioSession.enqueue(command, Command.Priority.NORMAL);

if (!ioSession.isOpen()) {
exchangeHandler.failed(new ConnectionClosedException());
exchangeHandler.releaseResources();
}
}

Expand Down Expand Up @@ -350,4 +310,33 @@ public H2ConnPool getConnPool() {
return connPool;
}

/**
* Cancellable that can be wired to the stream control once it becomes available.
*/
private static final class CancellableExecution implements Cancellable, CancellableDependency {
Copy link
Member

@ok2c ok2c Jan 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@arturobernalg I am not sure why this is necessary. The original CancellableExecution is now unused. Is this an optimization of some sort?


private volatile Cancellable dependency;

@Override
public void setDependency(final Cancellable dependency) {
this.dependency = dependency;
}

@Override
public boolean isCancelled() {
return false;
}

@Override
public boolean cancel() {
final Cancellable local = this.dependency;
if (local != null) {
local.cancel();
return true;
}
return false;
}

}

}
23 changes: 22 additions & 1 deletion ...n/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequesterBootstrap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import java.util.ArrayList;
import java.util.List;

import org.apache.hc.core5.annotation.Experimental;
import org.apache.hc.core5.function.Callback;
import org.apache.hc.core5.function.Decorator;
import org.apache.hc.core5.function.Supplier;
Expand Down Expand Up @@ -76,6 +77,8 @@ public class H2MultiplexingRequesterBootstrap {

private IOReactorMetricsListener threadPoolListener;

private int maxRequestsPerConnection;

private H2MultiplexingRequesterBootstrap() {
this.routeEntries = new ArrayList<>();
}
Expand Down Expand Up @@ -180,6 +183,23 @@ public final H2MultiplexingRequesterBootstrap setIOReactorMetricsListener(final
return this;
}

/**
* Sets a hard limit on the number of pending request execution commands that can be queued per connection.
* When the limit is reached, new submissions fail fast with {@link java.util.concurrent.RejectedExecutionException}.
* A value {@code <= 0} disables the limit (default).
* Note: this limit applies to commands waiting in the connection's internal queue (backlog). HTTP/2 in-flight
* concurrency is governed separately by protocol settings (e.g. MAX_CONCURRENT_STREAMS).
*
* @param max maximum number of pending requests per connection; {@code <= 0} to disable the limit.
* @return this instance.
* @since 5.5
*/
@Experimental
public final H2MultiplexingRequesterBootstrap setMaxRequestsPerConnection(final int max) {
this.maxRequestsPerConnection = max;
return this;
}

/**
* Sets {@link H2StreamListener} instance.
*
Expand Down Expand Up @@ -274,7 +294,8 @@ public H2MultiplexingRequester create() {
DefaultAddressResolver.INSTANCE,
tlsStrategy != null ? tlsStrategy : new H2ClientTlsStrategy(),
threadPoolListener,
null);
null,
maxRequestsPerConnection);
}

}
9 changes: 8 additions & 1 deletion ...5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2RequesterBootstrap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,7 @@ public class H2RequesterBootstrap {
private ConnPoolListener<HttpHost> connPoolListener;
private IOReactorMetricsListener threadPoolListener;
private FrameFactory frameFactory;
private int maxPendingCommandsPerConnection;


private H2RequesterBootstrap() {
Expand Down Expand Up @@ -210,6 +211,11 @@ public final H2RequesterBootstrap setPoolConcurrencyPolicy(final PoolConcurrency
return this;
}

public final H2RequesterBootstrap setMaxPendingCommandsPerConnection(final int maxPendingCommandsPerConnection) {
this.maxPendingCommandsPerConnection = maxPendingCommandsPerConnection;
return this;
}

/**
* Sets {@link TlsStrategy} instance.
*
Expand Down Expand Up @@ -433,7 +439,8 @@ public H2AsyncRequester create() {
actualTlsStrategy,
handshakeTimeout,
threadPoolListener,
null);
null,
maxPendingCommandsPerConnection);
}

}
Loading
Loading