Cap total number of concurrent commands per connection

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2AsyncRequester.java

@@ -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;
     }
 

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java

@@ -33,6 +33,7 @@
 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;
@@ -87,6 +88,12 @@ public class H2MultiplexingRequester extends AsyncRequester {
 
     private final H2ConnPool connPool;
 
+    /**
+     * Hard cap on per-connection queued / in-flight commands.
+     * {@code <= 0} disables the cap.
+     */
+    private final int maxCommandsPerConnection;
+
     /**
      * Use {@link H2MultiplexingRequesterBootstrap} to create instances of this class.
      */
@@ -100,11 +107,13 @@ public H2MultiplexingRequester(
             final Resolver<HttpHost, InetSocketAddress> addressResolver,
             final TlsStrategy tlsStrategy,
             final IOReactorMetricsListener threadPoolListener,
-            final IOWorkerSelector workerSelector) {
+            final IOWorkerSelector workerSelector,
+            final int maxCommandsPerConnection) {
         super(eventHandlerFactory, ioReactorConfig, ioSessionDecorator, exceptionCallback, sessionListener,
                 ShutdownCommand.GRACEFUL_IMMEDIATE_CALLBACK, DefaultAddressResolver.INSTANCE,
                 threadPoolListener, workerSelector);
         this.connPool = new H2ConnPool(this, addressResolver, tlsStrategy);
+        this.maxCommandsPerConnection = maxCommandsPerConnection;
     }
 
     public void closeIdle(final TimeValue idleTime) {
@@ -245,6 +254,16 @@ public void failed(final Exception cause) {
                             }
 
                         };
+                        final int max = maxCommandsPerConnection;
+                        if (max > 0) {
+                            final int current = ioSession.getPendingCommandCount();
+                            if (current >= 0 && current >= max) {
+                                exchangeHandler.failed(new RejectedExecutionException(
+                                        "Maximum number of pending commands per connection reached (max=" + max + ")"));
+                                exchangeHandler.releaseResources();
+                                return;
+                            }
+                        }
                         final Timeout socketTimeout = ioSession.getSocketTimeout();
                         ioSession.enqueue(new RequestExecutionCommand(
                                         handlerProxy,
@@ -349,5 +368,4 @@ public final <T> Future<T> execute(
     public H2ConnPool getConnPool() {
         return connPool;
     }
-
 }

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequesterBootstrap.java

@@ -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;
@@ -76,6 +77,8 @@ public class H2MultiplexingRequesterBootstrap {
 
     private IOReactorMetricsListener threadPoolListener;
 
+    private int maxCommandsPerConnection;
+
     private H2MultiplexingRequesterBootstrap() {
         this.routeEntries = new ArrayList<>();
     }
@@ -180,6 +183,23 @@ public final H2MultiplexingRequesterBootstrap setIOReactorMetricsListener(final
         return this;
     }
 
+    /**
+     * Sets a hard limit on the number of pending commands 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 commands per connection; {@code <= 0} to disable the limit.
+     * @return this instance.
+     * @since 5.5
+     */
+    @Experimental
+    public final H2MultiplexingRequesterBootstrap setMaxCommandsPerConnection(final int max) {
+        this.maxCommandsPerConnection = max;
+        return this;
+    }
+
     /**
      * Sets {@link H2StreamListener} instance.
      *
@@ -274,7 +294,8 @@ public H2MultiplexingRequester create() {
                 DefaultAddressResolver.INSTANCE,
                 tlsStrategy != null ? tlsStrategy : new H2ClientTlsStrategy(),
                 threadPoolListener,
-                null);
+                null,
+                maxCommandsPerConnection);
     }
 
 }

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2RequesterBootstrap.java

@@ -104,6 +104,7 @@ public class H2RequesterBootstrap {
     private ConnPoolListener<HttpHost> connPoolListener;
     private IOReactorMetricsListener threadPoolListener;
     private FrameFactory frameFactory;
+    private int maxPendingCommandsPerConnection;
 
 
     private H2RequesterBootstrap() {
@@ -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.
      *
@@ -433,7 +439,8 @@ public H2AsyncRequester create() {
                 actualTlsStrategy,
                 handshakeTimeout,
                 threadPoolListener,
-                null);
+                null,
+                maxPendingCommandsPerConnection);
     }
 
 }

httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/AsyncRequesterBootstrap.java

@@ -80,6 +80,7 @@ public class AsyncRequesterBootstrap {
     private Http1StreamListener streamListener;
     private ConnPoolListener<HttpHost> connPoolListener;
     private IOReactorMetricsListener threadPoolListener;
+    private int maxPendingCommandsPerConnection;
 
     private AsyncRequesterBootstrap() {
     }
@@ -174,6 +175,26 @@ public final AsyncRequesterBootstrap setPoolConcurrencyPolicy(final PoolConcurre
         return this;
     }
 
+    /**
+     * Sets the maximum number of pending request execution commands allowed per connection.
+     * <p>
+     * This is a fail-fast back-pressure mechanism based on the per-connection I/O session command queue depth.
+     * It does <b>not</b> cap the number of concurrent in-flight HTTP exchanges on the connection.
+     * </p>
+     * <p>
+     * A value {@code <= 0} disables the limit.
+     * When the limit is enabled and exceeded, request execution fails with {@link java.util.concurrent.RejectedExecutionException}.
+     * </p>
+     *
+     * @param maxPendingCommandsPerConnection maximum number of pending commands per connection, or {@code <= 0} to disable.
+     * @return this instance.
+     */
+    public final AsyncRequesterBootstrap setMaxPendingCommandsPerConnection(final int maxPendingCommandsPerConnection) {
+        this.maxPendingCommandsPerConnection = maxPendingCommandsPerConnection;
+        return this;
+    }
+
+
     /**
      * Sets {@link TlsStrategy} instance.
      *
@@ -303,7 +324,8 @@ public HttpAsyncRequester create() {
                 tlsStrategyCopy,
                 handshakeTimeout,
                 threadPoolListener,
-                null);
+                null,
+                maxPendingCommandsPerConnection);
     }
 
 }

httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java

@@ -32,6 +32,7 @@
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.hc.core5.annotation.Internal;
@@ -98,6 +99,7 @@ public class HttpAsyncRequester extends AsyncRequester implements ConnPoolContro
     private final ManagedConnPool<HttpHost, IOSession> connPool;
     private final TlsStrategy tlsStrategy;
     private final Timeout handshakeTimeout;
+    private final int maxPendingCommandsPerConnection;
 
     /**
      * Use {@link AsyncRequesterBootstrap} to create instances of this class.
@@ -115,13 +117,15 @@ public HttpAsyncRequester(
             final TlsStrategy tlsStrategy,
             final Timeout handshakeTimeout,
             final IOReactorMetricsListener threadPoolListener,
-            final IOWorkerSelector workerSelector) {
+            final IOWorkerSelector workerSelector,
+            final int maxPendingCommandsPerConnection) {
         super(eventHandlerFactory, ioReactorConfig, ioSessionDecorator, exceptionCallback, sessionListener,
                 ShutdownCommand.GRACEFUL_IMMEDIATE_CALLBACK, DefaultAddressResolver.INSTANCE, threadPoolListener,
                 workerSelector);
         this.connPool = Args.notNull(connPool, "Connection pool");
         this.tlsStrategy = tlsStrategy;
         this.handshakeTimeout = handshakeTimeout;
+        this.maxPendingCommandsPerConnection = maxPendingCommandsPerConnection;
     }
 
     @Override
@@ -285,6 +289,21 @@ public void execute(
 
                     @Override
                     public void completed(final AsyncClientEndpoint endpoint) {
+                        final int max = maxPendingCommandsPerConnection;
+                        if (max > 0) {
+                            final IOSession ioSession = ((InternalAsyncClientEndpoint) endpoint).getIOSession();
+                            final int pending = ioSession.getPendingCommandCount();
+                            if (pending >= 0 && pending >= max) {
+                                try {
+                                    endpoint.releaseAndReuse();
+                                    exchangeHandler.failed(new RejectedExecutionException(
+                                            "Maximum number of pending requests per connection reached (max=" + max + ")"));
+                                } finally {
+                                    exchangeHandler.releaseResources();
+                                }
+                                return;
+                            }
+                        }
                         endpoint.execute(new AsyncClientExchangeHandler() {
 
                             @Override

httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSession.java

@@ -238,4 +238,13 @@ enum Status {
      */
     void updateWriteTime();
 
+    /**
+     * Returns the number of pending commands enqueued for execution, or {@code -1} if unknown.
+     *
+     * @since 5.5
+     */
+    default int getPendingCommandCount() {
+        return -1;
+    }
+
 }

httpcore5/src/main/java/org/apache/hc/core5/reactor/IOSessionImpl.java

@@ -123,6 +123,11 @@ public boolean hasCommands() {
         return !commandQueue.isEmpty();
     }
 
+    @Override
+    public int getPendingCommandCount() {
+        return commandQueue.size();
+    }
+
     @Override
     public Command poll() {
         return commandQueue.poll();

httpcore5/src/main/java/org/apache/hc/core5/reactor/InternalDataChannel.java

@@ -409,4 +409,9 @@ public String toString() {
         return Objects.toString(currentSession != null ? currentSession : ioSession, null);
     }
 
+    @Override
+    public int getPendingCommandCount() {
+        final IOSession currentSession = currentSessionRef.get();
+        return currentSession.getPendingCommandCount();
+    }
 }

httpcore5/src/main/java/org/apache/hc/core5/reactor/ssl/SSLIOSession.java

@@ -932,4 +932,8 @@ public String toString() {
         }
     }
 
+    @Override
+    public int getPendingCommandCount() {
+        return this.session.getPendingCommandCount();
+    }
 }