ahc-1412 added a new close method to wait on shared netty resources

MiErnst · MiErnst · commit 7026fc237167 · 2017-09-26T11:02:41.000+02:00
diff --git a/client/src/main/java/org/asynchttpclient/AsyncHttpClient.java b/client/src/main/java/org/asynchttpclient/AsyncHttpClient.java
@@ -287,4 +287,12 @@ public interface AsyncHttpClient extends Closeable {
      * @return the config associated to this client.
      */
     AsyncHttpClientConfig getConfig();
+
+    /**
+     * Similar to calling the method {@link #close()} but additionally waits for inactivity on shared resources between
+     * multiple instances of netty. Calling this method instead of the method {@link #close()} might be helpful 
+     * on application shutdown to prevent errors like a {@link ClassNotFoundException} because the class loader was
+     * already removed but there are still some active tasks on this shared resources which want to access these classes.
+     */
+    void closeAndAwaitInactivity();
 }
diff --git a/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java b/client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClient.java
@@ -21,8 +21,12 @@
 import io.netty.util.HashedWheelTimer;
 import io.netty.util.ThreadDeathWatcher;
 import io.netty.util.Timer;
+import io.netty.util.concurrent.GlobalEventExecutor;
 
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Predicate;
 
@@ -102,31 +106,60 @@ private Timer newNettyTimer() {
 
     @Override
     public void close() {
+        closeInternal(false);
+    }
+    
+    public void closeAndAwaitInactivity() {
+        closeInternal(true);
+    }
+    
+    private void closeInternal(boolean awaitInactivity) {
         if (closeTriggered.compareAndSet(false, true)) {
-            try {
-                channelManager.close();
-            } catch (Throwable t) {
-                LOGGER.warn("Unexpected error on ChannelManager close", t);
-            }
-            if (allowStopNettyTimer) {
-                try {
-                    nettyTimer.stop();
-                } catch (Throwable t) {
-                    LOGGER.warn("Unexpected error on HashedWheelTimer close", t);
+            CompletableFuture<Void> handledCloseFuture = channelManager.close().whenComplete((v, t) -> {
+                if(t != null) {
+                    LOGGER.warn("Unexpected error on ChannelManager close", t);
+                }
+                if (allowStopNettyTimer) {
+                    try {
+                        nettyTimer.stop();
+                    } catch (Throwable th) {
+                        LOGGER.warn("Unexpected error on HashedWheelTimer close", th);
+                    }
                 }
+            });
+
+            if(awaitInactivity) {
+                handledCloseFuture = handledCloseFuture.thenCombine(awaitInactivity(), (v1,v2) -> null) ;
             }
-            
-            //see https://github.com/netty/netty/issues/2084#issuecomment-44822314
+
             try {
-                ThreadDeathWatcher.awaitInactivity(config.getShutdownTimeout(), TimeUnit.MILLISECONDS);
-            } catch(InterruptedException t) {
-                // Ignore
+                handledCloseFuture.get(config.getShutdownTimeout(), TimeUnit.MILLISECONDS);
+            } catch (InterruptedException | TimeoutException t) {
+                LOGGER.warn("Unexpected error on AsyncHttpClient close", t);
+            } catch (ExecutionException e) {
+                // already handled and could be ignored
             }
-            
             closed.compareAndSet(false, true);
         }
     }
 
+    private CompletableFuture<Void> awaitInactivity() {
+        //see https://github.com/netty/netty/issues/2084#issuecomment-44822314
+        CompletableFuture<Void> wait1 = CompletableFuture.runAsync(() -> {
+        try {
+            GlobalEventExecutor.INSTANCE.awaitInactivity(config.getShutdownTimeout(), TimeUnit.MILLISECONDS);
+        } catch(InterruptedException t) {
+            // Ignore
+        }});
+        CompletableFuture<Void> wait2 = CompletableFuture.runAsync(() -> {
+        try {
+            ThreadDeathWatcher.awaitInactivity(config.getShutdownTimeout(), TimeUnit.MILLISECONDS);
+        } catch(InterruptedException t) {
+            // Ignore
+        }});  
+        return wait1.thenCombine(wait2, (v1,v2) -> null);
+    }
+
     @Override
     public boolean isClosed() {
         return closed.get();
diff --git a/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java b/client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java
@@ -38,12 +38,13 @@
 import io.netty.handler.stream.ChunkedWriteHandler;
 import io.netty.util.Timer;
 import io.netty.util.concurrent.DefaultThreadFactory;
+import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GlobalEventExecutor;
 
 import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -100,15 +101,11 @@ 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 {
@@ -304,32 +301,25 @@ public boolean removeAll(Channel connection) {
         return channelPool.removeAll(connection);
     }
 
-    private void doClose() {
-        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() {
+    public CompletableFuture<Void> close() {
+        CompletableFuture<Void> closeFuture = CompletableFuture.completedFuture(null);
         if (allowReleaseEventLoopGroup) {
-            eventLoopGroup.shutdownGracefully(config.getShutdownQuietPeriod(), config.getShutdownTimeout(), TimeUnit.MILLISECONDS)//
-                    .addListener(future -> doClose());
-        } else
-            doClose();
-
-        try {
-            closeLatch.await(config.getShutdownTimeout(), TimeUnit.MILLISECONDS);
-        } catch (InterruptedException e) {
-           // Ignore
+            closeFuture = toCompletableFuture(
+                eventLoopGroup.shutdownGracefully(config.getShutdownQuietPeriod(), config.getShutdownTimeout(), TimeUnit.MILLISECONDS));
         }
+        return closeFuture.thenCompose(v -> toCompletableFuture(openChannels.close())).whenComplete((v,t) -> channelPool.destroy());
+    }
+    
+    private static CompletableFuture<Void> toCompletableFuture(final Future<?> future) {
+        final CompletableFuture<Void> completableFuture = new CompletableFuture<>();
+        future.addListener(r  -> {
+            if(r.isSuccess()) {
+                completableFuture.complete(null);
+            } else {
+                completableFuture.completeExceptionally(r.cause());
+            }
+        });
+        return completableFuture;
     }
 
     public void closeChannel(Channel channel) {
diff --git a/extras/registry/src/test/java/org/asynchttpclient/extras/registry/BadAsyncHttpClient.java b/extras/registry/src/test/java/org/asynchttpclient/extras/registry/BadAsyncHttpClient.java
@@ -143,4 +143,9 @@ public void flushChannelPoolPartitions(Predicate<Object> predicate) {
     public AsyncHttpClientConfig getConfig() {
         return null;
     }
+
+	@Override
+	public void closeAndAwaitInactivity() {
+		
+	}
 }
diff --git a/extras/registry/src/test/java/org/asynchttpclient/extras/registry/TestAsyncHttpClient.java b/extras/registry/src/test/java/org/asynchttpclient/extras/registry/TestAsyncHttpClient.java
@@ -139,4 +139,8 @@ public void flushChannelPoolPartitions(Predicate<Object> predicate) {
     public AsyncHttpClientConfig getConfig() {
         return null;
     }
+
+	@Override
+	public void closeAndAwaitInactivity() {
+	}
 }

Original file line number	Diff line number	Diff line change
`@@ -143,4 +143,9 @@ public void flushChannelPoolPartitions(Predicate<Object> predicate) {`
`143`	`143`	`public AsyncHttpClientConfig getConfig() {`
`144`	`144`	`return null;`
`145`	`145`	`}`
	`146`	`+`
	`147`	`+ @Override`
	`148`	`+ public void closeAndAwaitInactivity() {`
	`149`	`+`
	`150`	`+ }`
`146`	`151`	`}`
Original file line number	Diff line number	Diff line change
`@@ -139,4 +139,8 @@ public void flushChannelPoolPartitions(Predicate<Object> predicate) {`
`139`	`139`	`public AsyncHttpClientConfig getConfig() {`
`140`	`140`	`return null;`
`141`	`141`	`}`
	`142`	`+`
	`143`	`+ @Override`
	`144`	`+ public void closeAndAwaitInactivity() {`
	`145`	`+ }`
`142`	`146`	`}`