Off-lock disposal in PoolingHttpClientConnectionManager: enqueue GRACEFUL, run IMMEDIATE inline

Drain queue after lease/release/closeExpired/closeIdle to avoid holding the pool lock

Author: arturobernalg

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

@@ -111,6 +111,8 @@ public class PoolingHttpClientConnectionManager
 
     private static final Logger LOG = LoggerFactory.getLogger(PoolingHttpClientConnectionManager.class);
 
+    private final OffLockDisposalCallback<ManagedHttpClientConnection> disposer; // off-lock draining
+
     public static final int DEFAULT_MAX_TOTAL_CONNECTIONS = 25;
     public static final int DEFAULT_MAX_CONNECTIONS_PER_ROUTE = 5;
 
@@ -218,14 +220,18 @@ public PoolingHttpClientConnectionManager(
             final HttpConnectionFactory<ManagedHttpClientConnection> connFactory) {
         super();
         this.connectionOperator = Args.notNull(httpClientConnectionOperator, "Connection operator");
+
+        // off-lock disposer wraps the default closer; we explicitly drain after pool ops
+        this.disposer = new OffLockDisposalCallback<>(new DefaultDisposalCallback<>());
+
         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<>(),
+                        disposer,
                         null) {
 
                     @Override
@@ -240,6 +246,7 @@ public void closeExpired() {
                         DEFAULT_MAX_CONNECTIONS_PER_ROUTE,
                         timeToLive,
                         poolReusePolicy,
+                        disposer,
                         null) {
 
                     @Override
@@ -266,6 +273,7 @@ protected PoolingHttpClientConnectionManager(
         this.pool = Args.notNull(pool, "Connection pool");
         this.connFactory = connFactory != null ? connFactory : ManagedHttpClientConnectionFactory.INSTANCE;
         this.closed = new AtomicBoolean(false);
+        this.disposer = null;
     }
 
     @Override
@@ -280,6 +288,7 @@ public void close(final CloseMode closeMode) {
                 LOG.debug("Shutdown connection pool {}", closeMode);
             }
             this.pool.close(closeMode);
+            drainDisposals();
             LOG.debug("Connection pool shut down");
         }
     }
@@ -409,6 +418,7 @@ public ConnectionEndpoint get(
                     }
                 } finally {
                     lock.unlock();
+                    drainDisposals();
                 }
             }
 
@@ -472,6 +482,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 +552,7 @@ public void closeIdle(final TimeValue idleTime) {
             return;
         }
         this.pool.closeIdle(idleTime);
+        drainDisposals();
     }
 
     @Override
@@ -550,6 +562,7 @@ public void closeExpired() {
         }
         LOG.debug("Closing expired connections");
         this.pool.closeExpired();
+        drainDisposals();
     }
 
     @Override
@@ -837,4 +850,11 @@ public boolean isClosed() {
     }
 
 
+    private void drainDisposals() {
+        if (disposer != null) {
+            disposer.drain();
+        }
+    }
+
+
 }

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
+    }
+}