HTTPCLIENT-2398 - Cap async execution queue to break recursive re-entry

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

@@ -0,0 +1,86 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.client5.http.impl.async;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.hc.core5.annotation.Internal;
+import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
+
+@Internal
+final class AsyncClientExchangeHandlerProxy implements InvocationHandler {
+
+    private final AsyncClientExchangeHandler handler;
+    private final Runnable onRelease;
+    private final AtomicBoolean released;
+
+    private AsyncClientExchangeHandlerProxy(
+            final AsyncClientExchangeHandler handler,
+            final Runnable onRelease) {
+        this.handler = handler;
+        this.onRelease = onRelease;
+        this.released = new AtomicBoolean(false);
+    }
+
+    static AsyncClientExchangeHandler newProxy(
+            final AsyncClientExchangeHandler handler,
+            final Runnable onRelease) {
+        return (AsyncClientExchangeHandler) Proxy.newProxyInstance(
+                AsyncClientExchangeHandler.class.getClassLoader(),
+                new Class<?>[]{AsyncClientExchangeHandler.class},
+                new AsyncClientExchangeHandlerProxy(handler, onRelease));
+    }
+
+    @Override
+    public Object invoke(
+            final Object proxy,
+            final Method method,
+            final Object[] args) throws Throwable {
+        if ("releaseResources".equals(method.getName())
+                && method.getParameterCount() == 0) {
+            try {
+                return method.invoke(handler, args);
+            } catch (final InvocationTargetException ex) {
+                throw ex.getCause();
+            } finally {
+                if (released.compareAndSet(false, true)) {
+                    onRelease.run();
+                }
+            }
+        }
+        try {
+            return method.invoke(handler, args);
+        } catch (final InvocationTargetException ex) {
+            throw ex.getCause();
+        }
+    }
+
+}

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

@@ -218,6 +218,8 @@ private ExecInterceptorEntry(
 
     private boolean priorityHeaderDisabled;
 
+    private int maxQueuedRequests = -1;
+
     public static H2AsyncClientBuilder create() {
         return new H2AsyncClientBuilder();
     }
@@ -324,6 +326,22 @@ public final H2AsyncClientBuilder disableRequestPriority() {
         return this;
     }
 
+    /**
+     * 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 {@code <= 0} means unlimited (default).
+     *
+     * @param max maximum number of queued requests; {@code <= 0} to disable the cap
+     * @return this builder
+     * @since 5.6
+     */
+    @Experimental
+    public final H2AsyncClientBuilder setMaxQueuedRequests(final int max) {
+        this.maxQueuedRequests = max;
+        return this;
+    }
+
     /**
      * Adds this protocol interceptor to the head of the protocol processing list.
      *
@@ -976,7 +994,9 @@ public CloseableHttpAsyncClient build() {
                 cookieStoreCopy,
                 credentialsProviderCopy,
                 defaultRequestConfig,
-                closeablesCopy);
+                closeablesCopy,
+                maxQueuedRequests);
+
     }
 
     static class IdleConnectionEvictor implements Closeable {

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

@@ -267,6 +267,8 @@ private ExecInterceptorEntry(
 
     private ProxySelector proxySelector;
 
+    private int maxQueuedRequests = -1;
+
     private EarlyHintsListener earlyHintsListener;
 
     private boolean priorityHeaderDisabled;
@@ -899,6 +901,23 @@ public HttpAsyncClientBuilder disableContentCompression() {
         return this;
     }
 
+    /**
+     * 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).
+     *
+     * @param max maximum number of queued requests; <= 0 to disable the cap
+     * @return this builder
+     * @since 5.6
+     */
+    @Experimental
+    public HttpAsyncClientBuilder setMaxQueuedRequests(final int max) {
+        this.maxQueuedRequests = max;
+        return this;
+    }
+
+
     /**
      * Disable installing the HTTP/2 Priority header interceptor by default.
      * @since 5.6
@@ -1260,7 +1279,8 @@ public CloseableHttpAsyncClient build() {
                 credentialsProviderCopy,
                 contextAdaptor(),
                 defaultRequestConfig,
-                closeablesCopy);
+                closeablesCopy,
+                maxQueuedRequests);
     }
 
 }

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

@@ -29,6 +29,7 @@
 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;
@@ -69,6 +70,8 @@ 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;
 
     InternalH2AsyncClient(
             final DefaultConnectingIOReactor ioReactor,
@@ -82,21 +85,27 @@ public final class InternalH2AsyncClient extends InternalAbstractHttpAsyncClient
             final CookieStore cookieStore,
             final CredentialsProvider credentialsProvider,
             final RequestConfig defaultConfig,
-            final List<Closeable> closeables) {
+            final List<Closeable> closeables,
+            final int maxQueuedRequests) {
         super(ioReactor, pushConsumerRegistry, threadFactory, execChain,
                 cookieSpecRegistry, authSchemeRegistry, cookieStore, credentialsProvider, HttpClientContext::castOrCreate,
                 defaultConfig, closeables);
         this.connPool = connPool;
         this.routePlanner = routePlanner;
+        this.maxQueuedRequests = maxQueuedRequests;
+        this.queuedRequests = maxQueuedRequests > 0 ? new AtomicInteger(0) : null;
     }
 
     @Override
     AsyncExecRuntime createAsyncExecRuntime(final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) {
-        return new InternalH2AsyncExecRuntime(LOG, connPool, pushHandlerFactory);
+        return new InternalH2AsyncExecRuntime(LOG, connPool, pushHandlerFactory, maxQueuedRequests, queuedRequests);
     }
 
     @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,6 +28,8 @@
 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;
@@ -61,17 +63,30 @@ 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 volatile boolean reusable;
 
     InternalH2AsyncExecRuntime(
             final Logger log,
             final InternalH2ConnPool connPool,
             final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) {
+        this(log, connPool, pushHandlerFactory, -1, null);
+    }
+
+    InternalH2AsyncExecRuntime(
+            final Logger log,
+            final InternalH2ConnPool connPool,
+            final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
+            final int maxQueued,
+            final AtomicInteger sharedQueued) {
         super();
         this.log = log;
         this.connPool = connPool;
         this.pushHandlerFactory = pushHandlerFactory;
         this.sessionRef = new AtomicReference<>();
+        this.maxQueued = maxQueued;
+        this.sharedQueued = sharedQueued;
     }
 
     @Override
@@ -246,20 +261,49 @@ public EndpointInfo getEndpointInfo() {
         return null;
     }
 
+    private boolean tryAcquireSlot() {
+        if (sharedQueued == null || maxQueued <= 0) {
+            return true;
+        }
+        for (;;) {
+            final int q = sharedQueued.get();
+            if (q >= maxQueued) {
+                return false;
+            }
+            if (sharedQueued.compareAndSet(q, q + 1)) {
+                return true;
+            }
+        }
+    }
+
+    private void releaseSlot() {
+        if (sharedQueued != null && maxQueued > 0) {
+            sharedQueued.decrementAndGet();
+        }
+    }
+
     @Override
     public Cancellable execute(
             final String id,
             final AsyncClientExchangeHandler exchangeHandler, final HttpClientContext context) {
-        final ComplexCancellable complexCancellable = new ComplexCancellable();
         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
+                ? AsyncClientExchangeHandlerProxy.newProxy(exchangeHandler, this::releaseSlot)
+                : exchangeHandler;
+        final ComplexCancellable complexCancellable = new ComplexCancellable();
         final IOSession session = endpoint.session;
         if (session.isOpen()) {
             if (log.isDebugEnabled()) {
                 log.debug("{} start execution {}", ConnPoolSupport.getId(endpoint), id);
             }
             context.setProtocolVersion(HttpVersion.HTTP_2);
             session.enqueue(
-                    new RequestExecutionCommand(exchangeHandler, pushHandlerFactory, complexCancellable, context),
+                    new RequestExecutionCommand(actual, pushHandlerFactory, complexCancellable, context),
                     Command.Priority.NORMAL);
         } else {
             final HttpRoute route = endpoint.route;
@@ -276,19 +320,19 @@ public void completed(final IOSession ioSession) {
                         log.debug("{} start execution {}", ConnPoolSupport.getId(endpoint), id);
                     }
                     context.setProtocolVersion(HttpVersion.HTTP_2);
-                    session.enqueue(
-                            new RequestExecutionCommand(exchangeHandler, pushHandlerFactory, complexCancellable, context),
+                    ioSession.enqueue(
+                            new RequestExecutionCommand(actual, pushHandlerFactory, complexCancellable, context),
                             Command.Priority.NORMAL);
                 }
 
                 @Override
                 public void failed(final Exception ex) {
-                    exchangeHandler.failed(ex);
+                    actual.failed(ex);
                 }
 
                 @Override
                 public void cancelled() {
-                    exchangeHandler.failed(new InterruptedIOException());
+                    actual.failed(new InterruptedIOException());
                 }
 
             });
@@ -325,7 +369,7 @@ public String getId() {
 
     @Override
     public AsyncExecRuntime fork() {
-        return new InternalH2AsyncExecRuntime(log, connPool, pushHandlerFactory);
+        return new InternalH2AsyncExecRuntime(log, connPool, pushHandlerFactory, maxQueued, sharedQueued);
     }
 
 }

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

@@ -29,6 +29,7 @@
 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;
@@ -75,6 +76,8 @@ 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;
 
     InternalHttpAsyncClient(
             final DefaultConnectingIOReactor ioReactor,
@@ -90,18 +93,21 @@ public final class InternalHttpAsyncClient extends InternalAbstractHttpAsyncClie
             final CredentialsProvider credentialsProvider,
             final Function<HttpContext, HttpClientContext> contextAdaptor,
             final RequestConfig defaultConfig,
-            final List<Closeable> closeables) {
+            final List<Closeable> closeables,
+            final int maxQueuedRequests) {
         super(ioReactor, pushConsumerRegistry, threadFactory, execChain,
                 cookieSpecRegistry, authSchemeRegistry, cookieStore, credentialsProvider, contextAdaptor,
                 defaultConfig, closeables);
         this.manager = manager;
         this.routePlanner = routePlanner;
         this.tlsConfig = tlsConfig;
+        this.maxQueuedRequests = maxQueuedRequests;
+        this.queuedCounter = maxQueuedRequests > 0 ? new AtomicInteger(0) : null;
     }
 
     @Override
     AsyncExecRuntime createAsyncExecRuntime(final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) {
-        return new InternalHttpAsyncExecRuntime(LOG, manager, getConnectionInitiator(), pushHandlerFactory, tlsConfig);
+        return new InternalHttpAsyncExecRuntime(LOG, manager, getConnectionInitiator(), pushHandlerFactory, tlsConfig, maxQueuedRequests, queuedCounter);
     }
 
     @Override