Merge "Fixed deadlock between AsyncSshHandlerReader and AsyncSshHandler" into stable...

diff --git a/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java b/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java
index 149ccf16f9910dfbf38a866f5c0d89fbda50c7b6..15882ded63ec8d1acfdf09f0f9284e5d59a65f8e 100644 (file)
--- a/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java
+++ b/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java
@@ -16,6 +16,7 @@ import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.GenericFutureListener;
 import java.io.IOException;
 import java.net.SocketAddress;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.apache.sshd.client.SshClient;
 import org.apache.sshd.client.channel.ClientChannel;
 import org.apache.sshd.client.future.AuthFuture;
@@ -52,6 +53,7 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
 
     private final AuthenticationHandler authenticationHandler;
     private final SshClient sshClient;
+    private final AtomicBoolean isDisconnected = new AtomicBoolean();
     private Future<?> negotiationFuture;
 
     private AsyncSshHandlerReader sshReadAsyncListener;
@@ -213,11 +215,17 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
         disconnect(ctx, promise);
     }
 
-    @SuppressWarnings("checkstyle:IllegalCatch")
     @Override
-    public synchronized void disconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) {
+    public void disconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) {
+        if (isDisconnected.compareAndSet(false, true)) {
+            safelyDisconnect(ctx, promise);
+        }
+    }
+
+    @SuppressWarnings("checkstyle:IllegalCatch")
+    private synchronized void safelyDisconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) {
         LOG.trace("Closing SSH session on channel: {} with connect promise in state: {}",
-                ctx.channel(),connectPromise);
+                ctx.channel(), connectPromise);
 
         // If we have already succeeded and the session was dropped after,
         // we need to fire inactive to notify reconnect logic
@@ -272,5 +280,4 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
         promise.setSuccess();
         LOG.debug("SSH session closed on channel: {}", ctx.channel());
     }
-
 }

diff --git a/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandlerReader.java b/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandlerReader.java
index b905f7ad2145a126a07396af2f2065ecb840014a..da5bb7482815c77c5d4452cd716ad383de434437 100644 (file)
--- a/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandlerReader.java
+++ b/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandlerReader.java
@@ -47,12 +47,17 @@ public final class AsyncSshHandlerReader implements SshFutureListener<IoReadFutu
     }
 
     @Override
-    public synchronized void operationComplete(final IoReadFuture future) {
-        if (future.getException() != null) {
+    public void operationComplete(final IoReadFuture future) {
+        if (checkDisconnect(future)) {
+            invokeDisconnect();
+        }
+    }
 
+    private synchronized boolean checkDisconnect(final IoReadFuture future) {
+        if (future.getException() != null) {
             //if asyncout is already set to null by close method, do nothing
             if (asyncOut == null) {
-                return;
+                return false;
             }
 
             if (asyncOut.isClosed() || asyncOut.isClosing()) {
@@ -61,11 +66,8 @@ public final class AsyncSshHandlerReader implements SshFutureListener<IoReadFutu
             } else {
                 LOG.warn("Exception while reading from SSH remote on channel {}", channelId, future.getException());
             }
-            invokeDisconnect();
-            return;
-        }
-
-        if (future.getRead() > 0) {
+            return true;
+        } else if (future.getRead() > 0) {
             final ByteBuf msg = Unpooled.wrappedBuffer(buf.array(), 0, future.getRead());
             if (LOG.isTraceEnabled()) {
                 LOG.trace("Reading message on channel: {}, message: {}",
@@ -78,8 +80,13 @@ public final class AsyncSshHandlerReader implements SshFutureListener<IoReadFutu
             currentReadFuture = asyncOut.read(buf);
             currentReadFuture.addListener(this);
         }
+        return false;
     }
 
+    /**
+     * Closing of the {@link AsyncSshHandlerReader}. This method should never be called with any locks held since
+     * call to {@link AutoCloseable#close()} can be a source of ABBA deadlock.
+     */
     @SuppressWarnings("checkstyle:IllegalCatch")
     private void invokeDisconnect() {
         try {