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

@@ -26,10 +26,8 @@
  */
 package org.apache.hc.client5.http.impl.io;
 
-import java.util.ArrayDeque;
-import java.util.Deque;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.ReentrantLock;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 
 import org.apache.hc.core5.annotation.Internal;
 import org.apache.hc.core5.io.CloseMode;
@@ -40,65 +38,24 @@
 final class OffLockDisposalCallback<T extends ModalCloseable> implements DisposalCallback<T> {
 
     private final DisposalCallback<T> delegate;
-    private final Deque<Item<T>> queue = new ArrayDeque<>();
-    private final ReentrantLock qlock = new ReentrantLock();
-    private final AtomicBoolean draining = new AtomicBoolean();
+    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) {
-        qlock.lock();
-        try {
-            queue.addLast(new Item<>(closeable, mode));
-        } finally {
-            qlock.unlock();
+        if (mode == CloseMode.IMMEDIATE) {
+            delegate.execute(closeable, CloseMode.IMMEDIATE);
+        } else {
+            gracefulQueue.offer(closeable);
         }
     }
 
     void drain() {
-        for (;;) {
-            if (!draining.compareAndSet(false, true)) {
-                return;
-            }
-            try {
-                for (;;) {
-                    final Item<T> it;
-                    qlock.lock();
-                    try {
-                        it = queue.pollFirst();
-                    } finally {
-                        qlock.unlock();
-                    }
-                    if (it == null) {
-                        break;
-                    }
-                    delegate.execute(it.closeable, it.mode);
-                }
-            } finally {
-                draining.set(false);
-            }
-            qlock.lock();
-            try {
-                if (queue.isEmpty()) {
-                    return;
-                }
-            } finally {
-                qlock.unlock();
-            }
+        for (T c; (c = gracefulQueue.poll()) != null; ) {
+            delegate.execute(c, CloseMode.GRACEFUL);
         }
     }
-
-    static final class Item<T> {
-        final T closeable;
-        final CloseMode mode;
-
-        Item(final T c, final CloseMode m) {
-            this.closeable = c;
-            this.mode = m;
-        }
-    }
-}
-
+}

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

@@ -415,7 +415,6 @@ public ConnectionEndpoint get(
                     }
                 } finally {
                     lock.unlock();
-                    drainDisposals();
                 }
             }
 

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

@@ -65,14 +65,13 @@ public void bind(final Socket socket) throws IOException {
     @Override
     public void close(final CloseMode closeMode) {
         try {
-            Thread.sleep(sleepMillis);
+            if (closeMode == CloseMode.GRACEFUL) {
+                Thread.sleep(sleepMillis);   // slow only for graceful
+            }
         } catch (final InterruptedException ignore) {
             Thread.currentThread().interrupt();
         } finally {
             open = false;
-            if (onClose != null) {
-                onClose.run(); // capture thread in test
-            }
         }
     }