Check disconnected flag on each operation

We are performing a staggered setup during which we may encounter an
asynchronous disconnect. We have an atomic flag guarding this, which is
always flipped before disconnect enters critical region.

Check the flag at each synchronized connect stage and abort connection
if it is observed as flipped -- ensuring we cannot accidentally signal
channel activation after the channel has been close()d.

Also ditch the AtomicBoolean and instead use a simple VarHandle, saving
a few bytes of overhead.

JIRA: NETCONF-905
Change-Id: I857b1768b083c13ac5060a803a9ba6d4008c5e4b
Signed-off-by: Robert Varga <robert.varga@pantheon.tech>
(cherry picked from commit bdc0153931463070df6d03c75357e08f65bb9783)

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 e006a812f5e7d623e7711fa06851d9781058990f..141bd43551cd6c13aa429726cd0093966fb335ae 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,9 +16,10 @@ import io.netty.channel.ChannelPromise;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.FutureListener;
 import java.io.IOException;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
 import java.net.SocketAddress;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
 import org.checkerframework.checker.lock.qual.GuardedBy;
 import org.checkerframework.checker.lock.qual.Holding;
 import org.eclipse.jdt.annotation.Nullable;
@@ -37,6 +38,15 @@ import org.slf4j.LoggerFactory;
  */
 public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
     private static final Logger LOG = LoggerFactory.getLogger(AsyncSshHandler.class);
+    private static final VarHandle DISCONNECTED;
+
+    static {
+        try {
+            DISCONNECTED = MethodHandles.lookup().findVarHandle(AsyncSshHandler.class, "disconnected", boolean.class);
+        } catch (NoSuchFieldException | IllegalAccessException e) {
+            throw new ExceptionInInitializerError(e);
+        }
+    }
 
     public static final String SUBSYSTEM = "netconf";
 
@@ -58,7 +68,6 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
         DEFAULT_CLIENT = c;
     }
 
-    private final AtomicBoolean isDisconnected = new AtomicBoolean();
     private final AuthenticationHandler authenticationHandler;
     private final Future<?> negotiationFuture;
     private final NetconfSshClient sshClient;
@@ -72,6 +81,8 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
     private ClientSession session;
     private FutureListener<Object> negotiationFutureListener;
 
+    private volatile boolean disconnected;
+
     public AsyncSshHandler(final AuthenticationHandler authenticationHandler, final NetconfSshClient sshClient,
             final Future<?> negotiationFuture) {
         this.authenticationHandler = requireNonNull(authenticationHandler);
@@ -168,6 +179,10 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
             onOpenFailure(ctx, new AuthenticationFailedException("Authentication failed", cause));
             return;
         }
+        if (disconnected) {
+            LOG.debug("Skipping SSH subsystem allocation, channel: {}", ctx.channel());
+            return;
+        }
 
         LOG.debug("SSH session authenticated on channel: {}, server version: {}", ctx.channel(),
             clientSession.getServerVersion());
@@ -194,6 +209,10 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
             onOpenFailure(ctx, cause);
             return;
         }
+        if (disconnected) {
+            LOG.trace("Skipping activation, channel: {}", ctx.channel());
+            return;
+        }
 
         LOG.trace("SSH subsystem channel opened successfully on channel: {}", ctx.channel());
         if (negotiationFuture == null) {
@@ -224,7 +243,7 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
 
     @Override
     public void disconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) {
-        if (isDisconnected.compareAndSet(false, true)) {
+        if (DISCONNECTED.compareAndSet(this, false, true)) {
             ctx.executor().execute(() -> safelyDisconnect(ctx, promise));
         }
     }