Pooling managers: offload connection disposal from the pool lock.

httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/OffLockDisposalCallback.java

@@ -0,0 +1,61 @@
+/*
+ * ====================================================================
+ * 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.io;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.apache.hc.core5.annotation.Internal;
+import org.apache.hc.core5.io.CloseMode;
+import org.apache.hc.core5.io.ModalCloseable;
+import org.apache.hc.core5.pool.DisposalCallback;
+
+@Internal
+final class OffLockDisposalCallback<T extends ModalCloseable> implements DisposalCallback<T> {
+
+    private final DisposalCallback<T> delegate;
+    private final Queue<T> gracefulQueue = new ConcurrentLinkedQueue<>();
+
+    OffLockDisposalCallback(final DisposalCallback<T> delegate) {
+        this.delegate = delegate;
+    }
+
+    @Override
+    public void execute(final T closeable, final CloseMode mode) {
+        if (mode == CloseMode.IMMEDIATE) {
+            delegate.execute(closeable, CloseMode.IMMEDIATE);
+        } else {
+            gracefulQueue.offer(closeable);
+        }
+    }
+
+    void drain() {
+        for (T c; (c = gracefulQueue.poll()) != null; ) {
+            delegate.execute(c, CloseMode.GRACEFUL);
+        }
+    }
+}

httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/PoolingHttpClientConnectionManager.java

@@ -72,6 +72,7 @@
 import org.apache.hc.core5.io.CloseMode;
 import org.apache.hc.core5.pool.ConnPoolControl;
 import org.apache.hc.core5.pool.DefaultDisposalCallback;
+import org.apache.hc.core5.pool.DisposalCallback;
 import org.apache.hc.core5.pool.LaxConnPool;
 import org.apache.hc.core5.pool.ManagedConnPool;
 import org.apache.hc.core5.pool.PoolConcurrencyPolicy;
@@ -107,10 +108,13 @@
  */
 @Contract(threading = ThreadingBehavior.SAFE_CONDITIONAL)
 public class PoolingHttpClientConnectionManager
-    implements HttpClientConnectionManager, ConnPoolControl<HttpRoute> {
+        implements HttpClientConnectionManager, ConnPoolControl<HttpRoute> {
 
     private static final Logger LOG = LoggerFactory.getLogger(PoolingHttpClientConnectionManager.class);
 
+    private final DisposalCallback<ManagedHttpClientConnection> defaultDisposal;
+    private final OffLockDisposalCallback<ManagedHttpClientConnection> offLockDisposer;
+
     public static final int DEFAULT_MAX_TOTAL_CONNECTIONS = 25;
     public static final int DEFAULT_MAX_CONNECTIONS_PER_ROUTE = 5;
 
@@ -216,16 +220,34 @@ public PoolingHttpClientConnectionManager(
             final PoolReusePolicy poolReusePolicy,
             final TimeValue timeToLive,
             final HttpConnectionFactory<ManagedHttpClientConnection> connFactory) {
+        this(httpClientConnectionOperator,poolConcurrencyPolicy,poolReusePolicy,timeToLive,connFactory,false);
+
+    }
+
+    @Internal
+    public PoolingHttpClientConnectionManager(
+            final HttpClientConnectionOperator httpClientConnectionOperator,
+            final PoolConcurrencyPolicy poolConcurrencyPolicy,
+            final PoolReusePolicy poolReusePolicy,
+            final TimeValue timeToLive,
+            final HttpConnectionFactory<ManagedHttpClientConnection> connFactory,
+            final boolean offLockDisposalEnabled) {
         super();
         this.connectionOperator = Args.notNull(httpClientConnectionOperator, "Connection operator");
+
+        this.defaultDisposal = new DefaultDisposalCallback<>();
+        this.offLockDisposer = offLockDisposalEnabled ? new OffLockDisposalCallback<>(this.defaultDisposal) : null;
+        final DisposalCallback<ManagedHttpClientConnection> callbackForPool = offLockDisposalEnabled ? this.offLockDisposer : this.defaultDisposal;
+
+
         switch (poolConcurrencyPolicy != null ? poolConcurrencyPolicy : PoolConcurrencyPolicy.STRICT) {
             case STRICT:
                 this.pool = new StrictConnPool<HttpRoute, ManagedHttpClientConnection>(
                         DEFAULT_MAX_CONNECTIONS_PER_ROUTE,
                         DEFAULT_MAX_TOTAL_CONNECTIONS,
                         timeToLive,
                         poolReusePolicy,
-                        new DefaultDisposalCallback<>(),
+                        callbackForPool,
                         null) {
 
                     @Override
@@ -240,6 +262,7 @@ public void closeExpired() {
                         DEFAULT_MAX_CONNECTIONS_PER_ROUTE,
                         timeToLive,
                         poolReusePolicy,
+                        callbackForPool,
                         null) {
 
                     @Override
@@ -266,6 +289,8 @@ protected PoolingHttpClientConnectionManager(
         this.pool = Args.notNull(pool, "Connection pool");
         this.connFactory = connFactory != null ? connFactory : ManagedHttpClientConnectionFactory.INSTANCE;
         this.closed = new AtomicBoolean(false);
+        this.defaultDisposal = null;
+        this.offLockDisposer = null;
     }
 
     @Override
@@ -280,6 +305,7 @@ public void close(final CloseMode closeMode) {
                 LOG.debug("Shutdown connection pool {}", closeMode);
             }
             this.pool.close(closeMode);
+            drainDisposals();
             LOG.debug("Connection pool shut down");
         }
     }
@@ -386,6 +412,10 @@ public ConnectionEndpoint get(
                                 }
                             }
                         }
+
+                        // Single drain point under the lease lock.
+                        drainDisposals();
+
                         final ManagedHttpClientConnection conn = poolEntry.getConnection();
                         if (conn != null) {
                             conn.activate();
@@ -472,6 +502,7 @@ public void release(final ConnectionEndpoint endpoint, final Object state, final
             if (LOG.isDebugEnabled()) {
                 LOG.debug("{} connection released {}", ConnPoolSupport.getId(endpoint), ConnPoolSupport.formatStats(entry.getRoute(), entry.getState(), pool));
             }
+            drainDisposals();
         }
     }
 
@@ -541,6 +572,7 @@ public void closeIdle(final TimeValue idleTime) {
             return;
         }
         this.pool.closeIdle(idleTime);
+        drainDisposals();
     }
 
     @Override
@@ -550,6 +582,7 @@ public void closeExpired() {
         }
         LOG.debug("Closing expired connections");
         this.pool.closeExpired();
+        drainDisposals();
     }
 
     @Override
@@ -825,16 +858,17 @@ public HttpConnection get() {
     }
 
     /**
-     * Method that can be called to determine whether the connection manager has been shut down and
-     * is closed or not.
+     * Returns whether this connection manager has been shut down.
      *
-     * @return {@code true} if the connection manager has been shut down and is closed, otherwise
-     * return {@code false}.
      * @since 5.4
      */
     public boolean isClosed() {
         return this.closed.get();
     }
 
-
+    private void drainDisposals() {
+        if (offLockDisposer != null) {
+            offLockDisposer.drain();
+        }
+    }
 }

httpclient5/src/main/java/org/apache/hc/client5/http/impl/io/PoolingHttpClientConnectionManagerBuilder.java

@@ -38,6 +38,7 @@
 import org.apache.hc.client5.http.io.ManagedHttpClientConnection;
 import org.apache.hc.client5.http.ssl.DefaultClientTlsStrategy;
 import org.apache.hc.client5.http.ssl.TlsSocketStrategy;
+import org.apache.hc.core5.annotation.Experimental;
 import org.apache.hc.core5.annotation.Internal;
 import org.apache.hc.core5.function.Resolver;
 import org.apache.hc.core5.http.HttpHost;
@@ -92,6 +93,8 @@ public class PoolingHttpClientConnectionManagerBuilder {
     private int maxConnTotal;
     private int maxConnPerRoute;
 
+    private boolean offLockDisposalEnabled;
+
     public static PoolingHttpClientConnectionManagerBuilder create() {
         return new PoolingHttpClientConnectionManagerBuilder();
     }
@@ -304,6 +307,16 @@ public final PoolingHttpClientConnectionManagerBuilder useSystemProperties() {
         return this;
     }
 
+    /**
+     * Enable/disable off-lock disposal.
+     * @since 5.6
+     */
+    @Experimental
+    public final PoolingHttpClientConnectionManagerBuilder setOffLockDisposalEnabled(final boolean enabled) {
+        this.offLockDisposalEnabled = enabled;
+        return this;
+    }
+
     @Internal
     protected HttpClientConnectionOperator createConnectionOperator(
             final SchemePortResolver schemePortResolver,
@@ -332,7 +345,8 @@ public PoolingHttpClientConnectionManager build() {
                 poolConcurrencyPolicy,
                 poolReusePolicy,
                 null,
-                connectionFactory);
+                connectionFactory,
+                offLockDisposalEnabled);
         poolingmgr.setSocketConfigResolver(socketConfigResolver);
         poolingmgr.setConnectionConfigResolver(connectionConfigResolver);
         poolingmgr.setTlsConfigResolver(tlsConfigResolver);

httpclient5/src/test/java/org/apache/hc/client5/http/impl/io/NoopOperator.java

@@ -0,0 +1,59 @@
+/*
+ * ====================================================================
+ * 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.io;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+
+import org.apache.hc.client5.http.io.HttpClientConnectionOperator;
+import org.apache.hc.client5.http.io.ManagedHttpClientConnection;
+import org.apache.hc.core5.http.HttpHost;
+import org.apache.hc.core5.http.io.SocketConfig;
+import org.apache.hc.core5.http.protocol.HttpContext;
+import org.apache.hc.core5.util.TimeValue;
+
+public final class NoopOperator implements HttpClientConnectionOperator {
+
+    @Override
+    public void connect(
+            final ManagedHttpClientConnection conn,
+            final HttpHost host,
+            final InetSocketAddress localAddress,
+            final TimeValue connectTimeout,
+            final SocketConfig socketConfig,
+            final HttpContext context) throws IOException {
+        // no-op for tests
+    }
+
+    @Override
+    public void upgrade(
+            final ManagedHttpClientConnection conn,
+            final HttpHost host,
+            final HttpContext context) throws IOException {
+        // no-op for tests
+    }
+}