Use shutdown executor to avoid deadlock

acogoluegnes · acogoluegnes · commit 98cfd3be41cd · 2016-09-05T14:33:58.000+02:00
Use the shutdown executor only if it's been set. Fixes #194
diff --git a/src/main/java/com/rabbitmq/client/ConnectionFactory.java b/src/main/java/com/rabbitmq/client/ConnectionFactory.java
@@ -642,7 +642,7 @@ public MetricsCollector getMetricsCollector() {
     }
 
     protected FrameHandlerFactory createFrameHandlerFactory() throws IOException {
-        return new FrameHandlerFactory(connectionTimeout, factory, socketConf, isSSL());
+        return new FrameHandlerFactory(connectionTimeout, factory, socketConf, isSSL(), this.shutdownExecutor);
     }
 
     /**
diff --git a/src/main/java/com/rabbitmq/client/impl/AMQConnection.java b/src/main/java/com/rabbitmq/client/impl/AMQConnection.java
@@ -393,6 +393,7 @@ public void start()
     protected ChannelManager instantiateChannelManager(int channelMax, ThreadFactory threadFactory) {
         ChannelManager result = new ChannelManager(this._workService, channelMax, threadFactory, this.metricsCollector);
         result.setShutdownExecutor(this.shutdownExecutor);
+        result.setChannelShutdownTimeout((int) (requestedHeartbeat * 1000.0 + requestedHeartbeat * 1000.0 * (5.0 / 100.0)));
         return result;
     }
 
diff --git a/src/main/java/com/rabbitmq/client/impl/ChannelManager.java b/src/main/java/com/rabbitmq/client/impl/ChannelManager.java
@@ -15,6 +15,7 @@
 
 package com.rabbitmq.client.impl;
 
+import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.MetricsCollector;
 import com.rabbitmq.client.NoOpMetricsCollector;
 import com.rabbitmq.client.ShutdownSignalException;
@@ -51,6 +52,8 @@ public class ChannelManager {
     private ExecutorService shutdownExecutor;
     private final ThreadFactory threadFactory;
 
+    private int channelShutdownTimeout = (int) (ConnectionFactory.DEFAULT_HEARTBEAT + ConnectionFactory.DEFAULT_HEARTBEAT * (5.0 / 100.0)) * 1000;
+
     protected final MetricsCollector metricsCollector;
 
     public int getChannelMax(){
@@ -103,28 +106,30 @@ public void handleSignal(final ShutdownSignalException signal) {
         synchronized(this.monitor) {
             channels = new HashSet<ChannelN>(_channelMap.values());
         }
-        ExecutorService executorService = Executors.newSingleThreadExecutor();
-        try {
-            for (final ChannelN channel : channels) {
-                releaseChannelNumber(channel);
-                Future<?> channelShutdownTask = executorService.submit(new Runnable() {
-                    @Override
-                    public void run() {
-                        channel.processShutdownSignal(signal, true, true);
-                        shutdownSet.add(channel.getShutdownLatch());
-                    }
-                });
+
+        for (final ChannelN channel : channels) {
+            releaseChannelNumber(channel);
+            // async shutdown if possible
+            // see https://github.com/rabbitmq/rabbitmq-java-client/issues/194
+            Runnable channelShutdownRunnable = new Runnable() {
+                @Override
+                public void run() {
+                    channel.processShutdownSignal(signal, true, true);
+                }
+            };
+            if(this.shutdownExecutor == null) {
+                channelShutdownRunnable.run();
+            } else {
+                Future<?> channelShutdownTask = this.shutdownExecutor.submit(channelShutdownRunnable);
                 try {
-                    // FIXME make the timeout configurable
-                    channelShutdownTask.get(1, TimeUnit.SECONDS);
+                    channelShutdownTask.get(channelShutdownTimeout, TimeUnit.MILLISECONDS);
                 } catch (Exception e) {
-                    LOGGER.warn("Couldn't properly close channel {}", channel.getChannelNumber());
-                } finally {
-                    channel.notifyListeners();
+                    LOGGER.warn("Couldn't properly close channel {} on shutdown after waiting for {} ms", channel.getChannelNumber(), channelShutdownTimeout);
+                    channelShutdownTask.cancel(true);
                 }
             }
-        } finally {
-            executorService.shutdownNow();
+            shutdownSet.add(channel.getShutdownLatch());
+            channel.notifyListeners();
         }
         scheduleShutdownProcessing();
     }
@@ -243,4 +248,16 @@ public ExecutorService getShutdownExecutor() {
     public void setShutdownExecutor(ExecutorService shutdownExecutor) {
         this.shutdownExecutor = shutdownExecutor;
     }
+
+    /**
+     * Set the shutdown timeout for channels.
+     * This is the amount of time the manager waits for a channel to
+     * shutdown before giving up.
+     * Works only when the {@code shutdownExecutor} property is set.
+     * Default to {@link com.rabbitmq.client.ConnectionFactory#DEFAULT_HEARTBEAT} + 5 % seconds
+     * @param channelShutdownTimeout shutdown timeout in milliseconds
+     */
+    public void setChannelShutdownTimeout(int channelShutdownTimeout) {
+        this.channelShutdownTimeout = channelShutdownTimeout;
+    }
 }
diff --git a/src/main/java/com/rabbitmq/client/impl/FrameHandlerFactory.java b/src/main/java/com/rabbitmq/client/impl/FrameHandlerFactory.java
@@ -23,18 +23,25 @@
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.net.Socket;
+import java.util.concurrent.ExecutorService;
 
 public class FrameHandlerFactory {
     private final int connectionTimeout;
     private final SocketFactory factory;
     private final SocketConfigurator configurator;
+    private final ExecutorService shutdownExecutor;
     private final boolean ssl;
 
     public FrameHandlerFactory(int connectionTimeout, SocketFactory factory, SocketConfigurator configurator, boolean ssl) {
+        this(connectionTimeout, factory, configurator, ssl, null);
+    }
+
+    public FrameHandlerFactory(int connectionTimeout, SocketFactory factory, SocketConfigurator configurator, boolean ssl, ExecutorService shutdownExecutor) {
         this.connectionTimeout = connectionTimeout;
         this.factory = factory;
         this.configurator = configurator;
         this.ssl = ssl;
+        this.shutdownExecutor = shutdownExecutor;
     }
 
     public FrameHandler create(Address addr) throws IOException {
@@ -55,7 +62,7 @@ public FrameHandler create(Address addr) throws IOException {
 
     public FrameHandler create(Socket sock) throws IOException
     {
-        return new SocketFrameHandler(sock);
+        return new SocketFrameHandler(sock, this.shutdownExecutor);
     }
 
     private static void quietTrySocketClose(Socket socket) {
diff --git a/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java b/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java
@@ -35,6 +35,11 @@ public class SocketFrameHandler implements FrameHandler {
     /** The underlying socket */
     private final Socket _socket;
 
+    /**
+     * Optional {@link ExecutorService} for final flush.
+     */
+    private final ExecutorService _shutdownExecutor;
+
     /** Socket's inputstream - data from the broker - synchronized on */
     private final DataInputStream _inputStream;
 
@@ -48,7 +53,15 @@ public class SocketFrameHandler implements FrameHandler {
      * @param socket the socket to use
      */
     public SocketFrameHandler(Socket socket) throws IOException {
+        this(socket, null);
+    }
+
+    /**
+     * @param socket the socket to use
+     */
+    public SocketFrameHandler(Socket socket, ExecutorService shutdownExecutor) throws IOException {
         _socket = socket;
+        _shutdownExecutor = shutdownExecutor;
 
         _inputStream = new DataInputStream(new BufferedInputStream(socket.getInputStream()));
         _outputStream = new DataOutputStream(new BufferedOutputStream(socket.getOutputStream()));
@@ -153,20 +166,24 @@ public void flush() throws IOException {
     @SuppressWarnings("unused")
     public void close() {
         try { _socket.setSoLinger(true, SOCKET_CLOSING_TIMEOUT); } catch (Exception _e) {}
-        ExecutorService executorService = Executors.newSingleThreadExecutor();
+        // async flush if possible
+        // see https://github.com/rabbitmq/rabbitmq-java-client/issues/194
+        Callable<Void> flushCallable = new Callable<Void>() {
+            @Override
+            public Void call() throws Exception {
+                flush();
+                return null;
+            }
+        };
         try {
-            Future<Void> flushTask = executorService.submit(new Callable<Void>() {
-                @Override
-                public Void call() throws Exception {
-                    flush();
-                    return null;
-                }
-            });
-            flushTask.get(SOCKET_CLOSING_TIMEOUT, TimeUnit.SECONDS);
-        } catch (Exception _e) {
-
-        } finally {
-            executorService.shutdownNow();
+            if(this._shutdownExecutor == null) {
+                flushCallable.call();
+            } else {
+                Future<Void> flushTask = this._shutdownExecutor.submit(flushCallable);
+                flushTask.get(SOCKET_CLOSING_TIMEOUT, TimeUnit.SECONDS);
+            }
+        } catch(Exception e) {
+
         }
         try { _socket.close();                                   } catch (Exception _e) {}
     }

Original file line number	Diff line number	Diff line change
`@@ -642,7 +642,7 @@ public MetricsCollector getMetricsCollector() {`
`642`	`642`	`}`
`643`	`643`
`644`	`644`	`protected FrameHandlerFactory createFrameHandlerFactory() throws IOException {`
`645`		`- return new FrameHandlerFactory(connectionTimeout, factory, socketConf, isSSL());`
	`645`	`+ return new FrameHandlerFactory(connectionTimeout, factory, socketConf, isSSL(), this.shutdownExecutor);`
`646`	`646`	`}`
`647`	`647`
`648`	`648`	`/**`
Original file line number	Diff line number	Diff line change
`@@ -393,6 +393,7 @@ public void start()`
`393`	`393`	`protected ChannelManager instantiateChannelManager(int channelMax, ThreadFactory threadFactory) {`
`394`	`394`	`ChannelManager result = new ChannelManager(this._workService, channelMax, threadFactory, this.metricsCollector);`
`395`	`395`	`result.setShutdownExecutor(this.shutdownExecutor);`
	`396`	`+ result.setChannelShutdownTimeout((int) (requestedHeartbeat * 1000.0 + requestedHeartbeat * 1000.0 * (5.0 / 100.0)));`
`396`	`397`	`return result;`
`397`	`398`	`}`
`398`	`399`