Improve default client parameters

[netconf.git] / netconf / netconf-netty-util / src / main / java / org / opendaylight / netconf / nettyutil / handler / ssh / client / AsyncSshHandler.java

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 b47522b61e58ffbdc50b3f18b5f681e2dccf1678..b659187b8732dc37dd582318c6854005f491f850 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,11 @@ 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.time.Duration;
 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;
@@ -35,8 +37,17 @@ import org.slf4j.LoggerFactory;
 /**
  * Netty SSH handler class. Acts as interface between Netty and SSH library.
  */
-public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
+public final 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";
 
@@ -45,12 +56,13 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
     public static final NetconfSshClient DEFAULT_CLIENT;
 
     static {
-        final NetconfSshClient c = new NetconfClientBuilder().build();
+        final var c = new NetconfClientBuilder().build();
         // Disable default timeouts from mina sshd
-        c.getProperties().put(CoreModuleProperties.AUTH_TIMEOUT.getName(), "0");
-        c.getProperties().put(CoreModuleProperties.IDLE_TIMEOUT.getName(), "0");
-        c.getProperties().put(CoreModuleProperties.NIO2_READ_TIMEOUT.getName(), "0");
-        c.getProperties().put(CoreModuleProperties.TCP_NODELAY.getName(), true);
+        final var zero = Duration.ofMillis(0);
+        CoreModuleProperties.AUTH_TIMEOUT.set(c, zero);
+        CoreModuleProperties.IDLE_TIMEOUT.set(c, zero);
+        CoreModuleProperties.NIO2_READ_TIMEOUT.set(c, zero);
+        CoreModuleProperties.TCP_NODELAY.set(c, true);
 
         // TODO make configurable, or somehow reuse netty threadpool
         c.setNioWorkers(SSH_DEFAULT_NIO_WORKERS);
@@ -58,7 +70,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 +83,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 +181,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());
@@ -182,15 +199,22 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
             return;
         }
 
-        openFuture.addListener(future -> onOpenComplete(future, ctx));
+        openFuture.addListener(future -> ctx.executor().execute(() -> onOpenComplete(future, ctx)));
     }
 
+    // This callback has to run on the channel's executor because it runs fireChannelActive(), which needs to be
+    // delivered synchronously. If we were to execute on some other thread we would end up delaying the event,
+    // potentially creating havoc in the pipeline.
     private synchronized void onOpenComplete(final OpenFuture openFuture, final ChannelHandlerContext ctx) {
         final var cause = openFuture.getException();
         if (cause != null) {
             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) {
@@ -221,7 +245,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));
         }
     }