Implement shared FIFO execution queue for H2 client

Introduce one shared per-client queue to cap concurrently executing requests and enqueue overflow.
Ensure queued starts release the slot on any terminal path, including synchronous start failures.

Author: arturobernalg

httpclient5-testing/src/test/java/org/apache/hc/client5/http/impl/async/InternalTestHttpAsyncExecRuntime.java

@@ -31,7 +31,6 @@
 import java.util.List;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.hc.client5.http.HttpRoute;
 import org.apache.hc.client5.http.async.AsyncExecRuntime;
@@ -64,7 +63,7 @@ public final class InternalTestHttpAsyncExecRuntime extends InternalHttpAsyncExe
     public InternalTestHttpAsyncExecRuntime(final AsyncClientConnectionManager manager,
                                             final ConnectionInitiator connectionInitiator,
                                             final TlsConfig tlsConfig) {
-        super(LOG, manager, connectionInitiator, null, tlsConfig, -1, new AtomicInteger());
+        super(LOG, manager, connectionInitiator, null, tlsConfig);
         this.cancelled = new AtomicBoolean();
     }
 

httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/H2AsyncClientBuilder.java

@@ -328,12 +328,12 @@ public final H2AsyncClientBuilder disableRequestPriority() {
     }
 
     /**
-     * Sets a hard cap on the number of requests allowed to be queued/in-flight
-     * within the internal async execution pipeline. When the limit is reached,
-     * new submissions fail fast with {@link java.util.concurrent.RejectedExecutionException}.
+     * Sets the maximum number of concurrent request executions within the
+     * internal async execution pipeline. Requests beyond this limit are queued
+     * in FIFO order and dispatched as in-flight executions complete.
      * A value {@code <= 0} means unlimited (default).
      *
-     * @param max maximum number of queued requests; {@code <= 0} to disable the cap
+     * @param max maximum number of concurrent executions; {@code <= 0} to disable the cap
      * @return this builder
      * @since 5.7
      */

httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/HttpAsyncClientBuilder.java

@@ -910,12 +910,12 @@ public final HttpAsyncClientBuilder setTlsRequired(final boolean tlsRequired) {
     }
 
     /**
-     * Sets a hard cap on the number of requests allowed to be queued/in-flight
-     * within the internal async execution pipeline. When the limit is reached,
-     * new submissions fail fast with {@link java.util.concurrent.RejectedExecutionException}.
-     * A value <= 0 means unlimited (default).
+     * Sets the maximum number of concurrent request executions within the
+     * internal async execution pipeline. Requests beyond this limit are queued
+     * in FIFO order and dispatched as in-flight executions complete.
+     * A value {@code <= 0} means unlimited (default).
      *
-     * @param max maximum number of queued requests; <= 0 to disable the cap
+     * @param max maximum number of concurrent executions; {@code <= 0} to disable the cap
      * @return this builder
      * @since 5.7
      */

httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/InternalH2AsyncClient.java

@@ -29,7 +29,6 @@
 import java.io.Closeable;
 import java.util.List;
 import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.hc.client5.http.HttpRoute;
 import org.apache.hc.client5.http.async.AsyncExecRuntime;
@@ -56,11 +55,6 @@
 
 /**
  * Internal implementation of HTTP/2 only {@link CloseableHttpAsyncClient}.
- * <p>
- * Concurrent message exchanges with the same connection route executed by
- * this client will get automatically multiplexed over a single physical HTTP/2
- * connection.
- * </p>
  *
  * @since 5.0
  */
@@ -71,8 +65,12 @@ public final class InternalH2AsyncClient extends InternalAbstractHttpAsyncClient
     private static final Logger LOG = LoggerFactory.getLogger(InternalH2AsyncClient.class);
     private final HttpRoutePlanner routePlanner;
     private final InternalH2ConnPool connPool;
-    private final int maxQueuedRequests;
-    private final AtomicInteger queuedRequests;
+
+    /**
+     * One shared FIFO queue per client instance.
+     * {@code null} means unlimited / no throttling.
+     */
+    private final SharedRequestExecutionQueue executionQueue;
 
     InternalH2AsyncClient(
             final DefaultConnectingIOReactor ioReactor,
@@ -94,20 +92,16 @@ public final class InternalH2AsyncClient extends InternalAbstractHttpAsyncClient
                 defaultConfig, closeables);
         this.connPool = connPool;
         this.routePlanner = routePlanner;
-        this.maxQueuedRequests = maxQueuedRequests;
-        this.queuedRequests = maxQueuedRequests > 0 ? new AtomicInteger(0) : null;
+        this.executionQueue = maxQueuedRequests > 0 ? new SharedRequestExecutionQueue(maxQueuedRequests) : null;
     }
 
     @Override
     AsyncExecRuntime createAsyncExecRuntime(final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) {
-        return new InternalH2AsyncExecRuntime(LOG, connPool, pushHandlerFactory, maxQueuedRequests, queuedRequests);
+        return new InternalH2AsyncExecRuntime(LOG, connPool, pushHandlerFactory, executionQueue);
     }
 
     @Override
-    HttpRoute determineRoute(
-            final HttpHost httpHost,
-            final HttpRequest request,
-            final HttpClientContext clientContext) throws HttpException {
+    HttpRoute determineRoute(final HttpHost httpHost, final HttpRequest request, final HttpClientContext clientContext) throws HttpException {
         final HttpRoute route = routePlanner.determineRoute(httpHost, request, clientContext);
         if (route.isTunnelled()) {
             throw new HttpException("HTTP/2 tunneling not supported");

httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/InternalH2AsyncExecRuntime.java

@@ -28,8 +28,6 @@
 package org.apache.hc.client5.http.impl.async;
 
 import java.io.InterruptedIOException;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.hc.client5.http.EndpointInfo;
@@ -63,30 +61,27 @@ class InternalH2AsyncExecRuntime implements AsyncExecRuntime {
     private final InternalH2ConnPool connPool;
     private final HandlerFactory<AsyncPushConsumer> pushHandlerFactory;
     private final AtomicReference<Endpoint> sessionRef;
-    private final int maxQueued;
-    private final AtomicInteger sharedQueued;
+    private final SharedRequestExecutionQueue executionQueue;
     private volatile boolean reusable;
 
     InternalH2AsyncExecRuntime(
             final Logger log,
             final InternalH2ConnPool connPool,
             final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) {
-        this(log, connPool, pushHandlerFactory, -1, null);
+        this(log, connPool, pushHandlerFactory, null);
     }
 
     InternalH2AsyncExecRuntime(
             final Logger log,
             final InternalH2ConnPool connPool,
             final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
-            final int maxQueued,
-            final AtomicInteger sharedQueued) {
+            final SharedRequestExecutionQueue executionQueue) {
         super();
         this.log = log;
         this.connPool = connPool;
         this.pushHandlerFactory = pushHandlerFactory;
         this.sessionRef = new AtomicReference<>();
-        this.maxQueued = maxQueued;
-        this.sharedQueued = sharedQueued;
+        this.executionQueue = executionQueue;
     }
 
     @Override
@@ -179,7 +174,6 @@ public boolean isEndpointConnected() {
         return endpoint != null && endpoint.session.isOpen();
     }
 
-
     Endpoint ensureValid() {
         final Endpoint endpoint = sessionRef.get();
         if (endpoint == null) {
@@ -261,41 +255,47 @@ public EndpointInfo getEndpointInfo() {
         return null;
     }
 
-    private boolean tryAcquireSlot() {
-        if (sharedQueued == null || maxQueued <= 0) {
-            return true;
+    @Override
+    public Cancellable execute(
+            final String id,
+            final AsyncClientExchangeHandler exchangeHandler,
+            final HttpClientContext context) {
+
+        final Endpoint endpoint = ensureValid();
+        final ComplexCancellable complexCancellable = new ComplexCancellable();
+
+        if (executionQueue == null) {
+            startExecution(id, endpoint, exchangeHandler, context, complexCancellable);
+            return complexCancellable;
         }
-        for (;;) {
-            final int q = sharedQueued.get();
-            if (q >= maxQueued) {
-                return false;
-            }
-            if (sharedQueued.compareAndSet(q, q + 1)) {
+
+        final Cancellable queued = executionQueue.enqueue(
+                () -> {
+                    final AsyncClientExchangeHandler wrapped =
+                            new ReleasingAsyncClientExchangeHandler(exchangeHandler, executionQueue::completed);
+                    try {
+                        startExecution(id, endpoint, wrapped, context, complexCancellable);
+                    } catch (final RuntimeException ex) {
+                        wrapped.failed(ex);
+                    }
+                },
+                exchangeHandler::cancel);
+
+        return () -> {
+            if (queued.cancel()) {
                 return true;
             }
-        }
-    }
-
-    private void releaseSlot() {
-        if (sharedQueued != null && maxQueued > 0) {
-            sharedQueued.decrementAndGet();
-        }
+            return complexCancellable.cancel();
+        };
     }
 
-    @Override
-    public Cancellable execute(
+    private void startExecution(
             final String id,
-            final AsyncClientExchangeHandler exchangeHandler, final HttpClientContext context) {
-        final Endpoint endpoint = ensureValid();
-        if (!tryAcquireSlot()) {
-            exchangeHandler.failed(new RejectedExecutionException(
-                    "Execution pipeline queue limit reached (max=" + maxQueued + ")"));
-            return Operations.nonCancellable();
-        }
-        final AsyncClientExchangeHandler actual = sharedQueued != null
-                ? new ReleasingAsyncClientExchangeHandler(exchangeHandler, this::releaseSlot)
-                : exchangeHandler;
-        final ComplexCancellable complexCancellable = new ComplexCancellable();
+            final Endpoint endpoint,
+            final AsyncClientExchangeHandler exchangeHandler,
+            final HttpClientContext context,
+            final ComplexCancellable complexCancellable) {
+
         final IOSession session = endpoint.session;
         final RequestConfig requestConfig = context.getRequestConfigOrDefault();
         final Timeout responseTimeout = requestConfig.getResponseTimeout();
@@ -306,7 +306,7 @@ public Cancellable execute(
             context.setProtocolVersion(HttpVersion.HTTP_2);
             session.enqueue(
                     new RequestExecutionCommand(
-                            actual,
+                            exchangeHandler,
                             pushHandlerFactory,
                             context,
                             streamControl -> {
@@ -330,7 +330,7 @@ public void completed(final IOSession ioSession) {
                     context.setProtocolVersion(HttpVersion.HTTP_2);
                     ioSession.enqueue(
                             new RequestExecutionCommand(
-                                    actual,
+                                    exchangeHandler,
                                     pushHandlerFactory,
                                     context,
                                     streamControl -> {
@@ -342,17 +342,16 @@ public void completed(final IOSession ioSession) {
 
                 @Override
                 public void failed(final Exception ex) {
-                    actual.failed(ex);
+                    exchangeHandler.failed(ex);
                 }
 
                 @Override
                 public void cancelled() {
-                    actual.failed(new InterruptedIOException());
+                    exchangeHandler.failed(new InterruptedIOException());
                 }
 
             });
         }
-        return complexCancellable;
     }
 
     @Override
@@ -384,7 +383,7 @@ public String getId() {
 
     @Override
     public AsyncExecRuntime fork() {
-        return new InternalH2AsyncExecRuntime(log, connPool, pushHandlerFactory, maxQueued, sharedQueued);
+        return new InternalH2AsyncExecRuntime(log, connPool, pushHandlerFactory, executionQueue);
     }
 
 }

httpclient5/src/main/java/org/apache/hc/client5/http/impl/async/InternalHttpAsyncClient.java

@@ -29,7 +29,6 @@
 import java.io.Closeable;
 import java.util.List;
 import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
 
 import org.apache.hc.client5.http.HttpRoute;
@@ -77,8 +76,12 @@ public final class InternalHttpAsyncClient extends InternalAbstractHttpAsyncClie
     private final AsyncClientConnectionManager manager;
     private final HttpRoutePlanner routePlanner;
     private final TlsConfig tlsConfig;
-    private final int maxQueuedRequests;
-    private final AtomicInteger queuedCounter;
+
+    /**
+     * One shared FIFO queue per client instance.
+     * null means "unlimited" / no throttling.
+     */
+    private final SharedRequestExecutionQueue executionQueue;
 
     InternalHttpAsyncClient(
             final DefaultConnectingIOReactor ioReactor,
@@ -103,13 +106,12 @@ public final class InternalHttpAsyncClient extends InternalAbstractHttpAsyncClie
         this.manager = manager;
         this.routePlanner = routePlanner;
         this.tlsConfig = tlsConfig;
-        this.maxQueuedRequests = maxQueuedRequests;
-        this.queuedCounter = maxQueuedRequests > 0 ? new AtomicInteger(0) : null;
+        this.executionQueue = maxQueuedRequests > 0 ? new SharedRequestExecutionQueue(maxQueuedRequests) : null;
     }
 
     @Override
     AsyncExecRuntime createAsyncExecRuntime(final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) {
-        return new InternalHttpAsyncExecRuntime(LOG, manager, getConnectionInitiator(), pushHandlerFactory, tlsConfig, maxQueuedRequests, queuedCounter);
+        return new InternalHttpAsyncExecRuntime(LOG, manager, getConnectionInitiator(), pushHandlerFactory, tlsConfig, executionQueue);
     }
 
     @Override