#1412 Wait for async close tasks before close completes

client/src/main/java/org/asynchttpclient/AsyncHttpClientState.java

@@ -17,13 +17,13 @@
 
 public class AsyncHttpClientState {
 
-    private final AtomicBoolean closed;
+    private final AtomicBoolean closeTriggered;
 
-    public AsyncHttpClientState(AtomicBoolean closed) {
-        this.closed = closed;
+    public AsyncHttpClientState(AtomicBoolean closeTriggered) {
+        this.closeTriggered = closeTriggered;
     }
 
-    public boolean isClosed() {
-        return closed.get();
+    public boolean isCloseTriggered() {
+        return closeTriggered.get();
     }
 }

client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java

@@ -19,8 +19,10 @@
 import static org.asynchttpclient.util.Assertions.assertNotNull;
 import io.netty.channel.EventLoopGroup;
 import io.netty.util.HashedWheelTimer;
+import io.netty.util.ThreadDeathWatcher;
 import io.netty.util.Timer;
 
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Predicate;
 
@@ -42,6 +44,7 @@ public class DefaultAsyncHttpClient implements AsyncHttpClient {
 
     private final static Logger LOGGER = LoggerFactory.getLogger(DefaultAsyncHttpClient.class);
     private final AsyncHttpClientConfig config;
+    private final AtomicBoolean closeTriggered = new AtomicBoolean(false);
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final ChannelManager channelManager;
     private final ConnectionSemaphore connectionSemaphore;
@@ -87,7 +90,7 @@ public DefaultAsyncHttpClient(AsyncHttpClientConfig config) {
 
         channelManager = new ChannelManager(config, nettyTimer);
         connectionSemaphore = new ConnectionSemaphore(config);
-        requestSender = new NettyRequestSender(config, channelManager, connectionSemaphore, nettyTimer, new AsyncHttpClientState(closed));
+        requestSender = new NettyRequestSender(config, channelManager, connectionSemaphore, nettyTimer, new AsyncHttpClientState(closeTriggered));
         channelManager.configureBootstraps(requestSender);
     }
 
@@ -99,7 +102,7 @@ private Timer newNettyTimer() {
 
     @Override
     public void close() {
-        if (closed.compareAndSet(false, true)) {
+        if (closeTriggered.compareAndSet(false, true)) {
             try {
                 channelManager.close();
             } catch (Throwable t) {
@@ -112,6 +115,15 @@ public void close() {
                     LOGGER.warn("Unexpected error on HashedWheelTimer close", t);
                 }
             }
+
+            //see https://github.com/netty/netty/issues/2084#issuecomment-44822314
+            try {
+                ThreadDeathWatcher.awaitInactivity(config.getShutdownTimeout(), TimeUnit.MILLISECONDS);
+            } catch(InterruptedException t) {
+                // Ignore
+            }
+
+            closed.compareAndSet(false, true);
         }
     }
 

client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

@@ -43,6 +43,7 @@
 import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -99,11 +100,15 @@ public class ChannelManager {
     private final ChannelPool channelPool;
     private final ChannelGroup openChannels;
 
+    private final CountDownLatch closeLatch;
+
     private AsyncHttpClientHandler wsHandler;
 
     public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) {
 
         this.config = config;
+
+        closeLatch = new CountDownLatch(2);
 
         this.sslEngineFactory = config.getSslEngineFactory() != null ? config.getSslEngineFactory() : new DefaultSslEngineFactory();
         try {
@@ -300,16 +305,30 @@ public boolean removeAll(Channel connection) {
     }
 
     private void doClose() {
-        openChannels.close();
+        openChannels.close().addListener(future -> closeLatch.countDown());
         channelPool.destroy();
+
+        //see https://github.com/netty/netty/issues/2084#issuecomment-44822314
+        try {
+            GlobalEventExecutor.INSTANCE.awaitInactivity(config.getShutdownTimeout(), TimeUnit.MILLISECONDS);
+        } catch(InterruptedException t) {
+            // Ignore
+        } finally {
+            closeLatch.countDown();
+        }
     }
 
     public void close() {
         if (allowReleaseEventLoopGroup) {
             eventLoopGroup.shutdownGracefully(config.getShutdownQuietPeriod(), config.getShutdownTimeout(), TimeUnit.MILLISECONDS)//
                     .addListener(future -> doClose());
-        } else {
+        } else
             doClose();
+
+        try {
+            closeLatch.await(config.getShutdownTimeout(), TimeUnit.MILLISECONDS);
+        } catch (InterruptedException e) {
+           // Ignore
         }
     }
 

client/src/main/java/org/asynchttpclient/netty/request/NettyChannelConnector.java

@@ -72,7 +72,7 @@ public void connect(final Bootstrap bootstrap, final NettyConnectListener<?> con
         try {
             connect0(bootstrap, connectListener, remoteAddress);
         } catch (RejectedExecutionException e) {
-            if (clientState.isClosed()) {
+            if (clientState.isCloseTriggered()) {
                 LOGGER.info("Connect crash but engine is shutting down");
             } else {
                 connectListener.onFailure(null, e);

client/src/main/java/org/asynchttpclient/netty/request/NettyRequestSender.java

@@ -597,7 +597,7 @@ public void replayRequest(final NettyResponseFuture<?> future, FilterContext fc,
     }
 
     public boolean isClosed() {
-        return clientState.isClosed();
+        return clientState.isCloseTriggered();
     }
 
     public void drainChannelAndExecuteNextRequest(final Channel channel, final NettyResponseFuture<?> future, Request nextRequest) {