HTTPCLIENT-2398 - cap async execution queue to break recursive re-entry. Add configurable maxQueuedRequests (default unlimited). Release slot on fail/cancel/close to avoid leaks

Author: arturobernalg

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,22 @@ 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
+     */
+    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 +1278,8 @@ public CloseableHttpAsyncClient build() {
                 credentialsProviderCopy,
                 contextAdaptor(),
                 defaultRequestConfig,
-                closeablesCopy);
+                closeablesCopy,
+                maxQueuedRequests);
     }
 
 }

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

@@ -75,6 +75,7 @@ public final class InternalHttpAsyncClient extends InternalAbstractHttpAsyncClie
     private final AsyncClientConnectionManager manager;
     private final HttpRoutePlanner routePlanner;
     private final TlsConfig tlsConfig;
+    private final int maxQueuedRequests;
 
     InternalHttpAsyncClient(
             final DefaultConnectingIOReactor ioReactor,
@@ -90,18 +91,20 @@ 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;
     }
 
     @Override
     AsyncExecRuntime createAsyncExecRuntime(final HandlerFactory<AsyncPushConsumer> pushHandlerFactory) {
-        return new InternalHttpAsyncExecRuntime(LOG, manager, getConnectionInitiator(), pushHandlerFactory, tlsConfig);
+        return new InternalHttpAsyncExecRuntime(LOG, manager, getConnectionInitiator(), pushHandlerFactory, tlsConfig, maxQueuedRequests);
     }
 
     @Override

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

@@ -28,6 +28,12 @@
 package org.apache.hc.client5.http.impl.async;
 
 import java.io.InterruptedIOException;
+import java.lang.reflect.Proxy;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.hc.client5.http.EndpointInfo;
@@ -66,6 +72,8 @@ static class ReUseData {
 
     }
 
+    private static final ConcurrentMap<AsyncClientConnectionManager, AtomicInteger> QUEUE_COUNTERS = new ConcurrentHashMap<>();
+
     private final Logger log;
     private final AsyncClientConnectionManager manager;
     private final ConnectionInitiator connectionInitiator;
@@ -77,13 +85,25 @@ static class ReUseData {
     private final TlsConfig tlsConfig;
     private final AtomicReference<AsyncConnectionEndpoint> endpointRef;
     private final AtomicReference<ReUseData> reuseDataRef;
+    private final int maxQueued;
+    private final AtomicInteger sharedQueued;
 
     InternalHttpAsyncExecRuntime(
             final Logger log,
             final AsyncClientConnectionManager manager,
             final ConnectionInitiator connectionInitiator,
             final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
             final TlsConfig tlsConfig) {
+        this(log, manager, connectionInitiator, pushHandlerFactory, tlsConfig, -1);
+    }
+
+    InternalHttpAsyncExecRuntime(
+            final Logger log,
+            final AsyncClientConnectionManager manager,
+            final ConnectionInitiator connectionInitiator,
+            final HandlerFactory<AsyncPushConsumer> pushHandlerFactory,
+            final TlsConfig tlsConfig,
+            final int maxQueued) {
         super();
         this.log = log;
         this.manager = manager;
@@ -92,6 +112,8 @@ static class ReUseData {
         this.tlsConfig = tlsConfig;
         this.endpointRef = new AtomicReference<>();
         this.reuseDataRef = new AtomicReference<>();
+        this.maxQueued = maxQueued;
+        this.sharedQueued = maxQueued > 0 ? QUEUE_COUNTERS.computeIfAbsent(manager, m -> new AtomicInteger(0)) : null;
     }
 
     @Override
@@ -282,10 +304,60 @@ public EndpointInfo getEndpointInfo() {
         return endpoint != null ? endpoint.getInfo() : null;
     }
 
+    private boolean tryAcquireSlot() {
+        if (sharedQueued == null) {
+            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) {
+            sharedQueued.decrementAndGet();
+        }
+    }
+
+    private AsyncClientExchangeHandler guard(final AsyncClientExchangeHandler handler) {
+        if (sharedQueued == null) {
+            return handler;
+        }
+        final AtomicBoolean released = new AtomicBoolean(false);
+        return (AsyncClientExchangeHandler) Proxy.newProxyInstance(
+                AsyncClientExchangeHandler.class.getClassLoader(),
+                new Class<?>[]{AsyncClientExchangeHandler.class},
+                (proxy, method, args) -> {
+                    if ("releaseResources".equals(method.getName())
+                            && method.getParameterCount() == 0) {
+                        try {
+                            return method.invoke(handler, args);
+                        } finally {
+                            if (released.compareAndSet(false, true)) {
+                                releaseSlot();
+                            }
+                        }
+                    }
+                    return method.invoke(handler, args);
+                });
+    }
+
     @Override
     public Cancellable execute(
             final String id, final AsyncClientExchangeHandler exchangeHandler, final HttpClientContext context) {
         final AsyncConnectionEndpoint endpoint = ensureValid();
+        if (sharedQueued != null && !tryAcquireSlot()) {
+            exchangeHandler.failed(new RejectedExecutionException(
+                    "Execution pipeline queue limit reached (max=" + maxQueued + ")"));
+            return Operations.nonCancellable();
+        }
+        final AsyncClientExchangeHandler actual = sharedQueued != null ? guard(exchangeHandler) : exchangeHandler;
         if (endpoint.isConnected()) {
             if (log.isDebugEnabled()) {
                 log.debug("{} start execution {}", ConnPoolSupport.getId(endpoint), id);
@@ -295,10 +367,10 @@ public Cancellable execute(
             if (responseTimeout != null) {
                 endpoint.setSocketTimeout(responseTimeout);
             }
-            endpoint.execute(id, exchangeHandler, pushHandlerFactory, context);
+            endpoint.execute(id, actual, pushHandlerFactory, context);
             if (context.getRequestConfigOrDefault().isHardCancellationEnabled()) {
                 return () -> {
-                    exchangeHandler.cancel();
+                    actual.cancel();
                     return true;
                 };
             }
@@ -311,20 +383,20 @@ public void completed(final AsyncExecRuntime runtime) {
                         log.debug("{} start execution {}", ConnPoolSupport.getId(endpoint), id);
                     }
                     try {
-                        endpoint.execute(id, exchangeHandler, pushHandlerFactory, context);
+                        endpoint.execute(id, actual, pushHandlerFactory, context);
                     } catch (final RuntimeException ex) {
                         failed(ex);
                     }
                 }
 
                 @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());
                 }
 
             });
@@ -344,7 +416,7 @@ public void markConnectionNonReusable() {
 
     @Override
     public AsyncExecRuntime fork() {
-        return new InternalHttpAsyncExecRuntime(log, manager, connectionInitiator, pushHandlerFactory, tlsConfig);
+        return new InternalHttpAsyncExecRuntime(log, manager, connectionInitiator, pushHandlerFactory, tlsConfig, maxQueued);
     }
 
 }