From 1fe167dab65a941eef91642c53f837ee43cdd9c7 Mon Sep 17 00:00:00 2001
From: Robert Varga <robert.varga@pantheon.tech>
Date: Wed, 12 Oct 2022 13:54:06 +0200
Subject: [PATCH] Expire negotiation on event loop

Rather than having to completely synchronize state transitions, make
sure we run expiry on the event loop.

JIRA: NETCONF-827
Change-Id: I8d02c025ceaf78d547848e7e92861cb125405141
Signed-off-by: Robert Varga <robert.varga@pantheon.tech>
(cherry picked from commit 0df65029004fe5b4766caf2b47db23e47f955332)
(cherry picked from commit 0f23eb998f0bea7e12be4172593c24cbb3afebed)
---
 .../AbstractNetconfSessionNegotiator.java     | 36 ++++++++++++++-----
 .../AbstractNetconfSessionNegotiatorTest.java |  1 +
 2 files changed, 28 insertions(+), 9 deletions(-)

diff --git a/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSessionNegotiator.java b/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSessionNegotiator.java
index 9c713f1b5c..21fb3b8e86 100644
--- a/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSessionNegotiator.java
+++ b/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSessionNegotiator.java
@@ -23,6 +23,7 @@ import io.netty.util.concurrent.GenericFutureListener;
 import io.netty.util.concurrent.Promise;
 import java.util.concurrent.TimeUnit;
 import org.checkerframework.checker.lock.qual.GuardedBy;
+import org.checkerframework.checker.lock.qual.Holding;
 import org.eclipse.jdt.annotation.Nullable;
 import org.opendaylight.netconf.api.NetconfDocumentedException;
 import org.opendaylight.netconf.api.NetconfMessage;
@@ -63,8 +64,8 @@ public abstract class AbstractNetconfSessionNegotiator<P extends NetconfSessionP
     private final L sessionListener;
     private final Timer timer;
 
+    @GuardedBy("this")
     private Timeout timeoutTask;
-
     @GuardedBy("this")
     private State state = State.IDLE;
 
@@ -118,15 +119,26 @@ public abstract class AbstractNetconfSessionNegotiator<P extends NetconfSessionP
         sendMessage(helloMessage);
 
         replaceHelloMessageOutboundHandler();
-        changeState(State.OPEN_WAIT);
 
-        timeoutTask = this.timer.newTimeout(this::timeoutExpired, connectionTimeoutMillis, TimeUnit.MILLISECONDS);
+        synchronized (this) {
+            lockedChangeState(State.OPEN_WAIT);
+
+            // Service the timeout on channel's eventloop, so that we do not get state transition problems
+            timeoutTask = timer.newTimeout(unused -> channel.eventLoop().execute(this::timeoutExpired),
+                connectionTimeoutMillis, TimeUnit.MILLISECONDS);
+        }
     }
 
-    private synchronized void timeoutExpired(final Timeout timeout) {
+    private synchronized void timeoutExpired() {
+        if (timeoutTask == null) {
+            // cancelTimeout() between expiry and execution on the loop
+            return;
+        }
+        timeoutTask = null;
+
         if (state != State.ESTABLISHED) {
-            LOG.debug("Connection timeout after {}, session backed by channel {} is in state {}", timeout, channel,
-                state);
+            LOG.debug("Connection timeout after {}ms, session backed by channel {} is in state {}",
+                connectionTimeoutMillis, channel, state);
 
             // Do not fail negotiation if promise is done or canceled
             // It would result in setting result of the promise second time and that throws exception
@@ -150,9 +162,10 @@ public abstract class AbstractNetconfSessionNegotiator<P extends NetconfSessionP
 
     @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
             justification = "https://github.com/spotbugs/spotbugs/issues/811")
-    private void cancelTimeout() {
-        if (timeoutTask != null) {
-            timeoutTask.cancel();
+    private synchronized void cancelTimeout() {
+        if (timeoutTask != null && !timeoutTask.cancel()) {
+            // Late-coming cancel: make sure
+            timeoutTask = null;
         }
     }
 
@@ -226,6 +239,11 @@ public abstract class AbstractNetconfSessionNegotiator<P extends NetconfSessionP
             throws NetconfDocumentedException;
 
     private synchronized void changeState(final State newState) {
+        lockedChangeState(newState);
+    }
+
+    @Holding("this")
+    private void lockedChangeState(final State newState) {
         LOG.debug("Changing state from : {} to : {} for channel: {}", state, newState, channel);
         checkState(isStateChangePermitted(state, newState),
                 "Cannot change state from %s to %s for channel %s", state, newState, channel);
diff --git a/netconf/netconf-netty-util/src/test/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSessionNegotiatorTest.java b/netconf/netconf-netty-util/src/test/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSessionNegotiatorTest.java
index a5a4a2395c..50db35b136 100644
--- a/netconf/netconf-netty-util/src/test/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSessionNegotiatorTest.java
+++ b/netconf/netconf-netty-util/src/test/java/org/opendaylight/netconf/nettyutil/AbstractNetconfSessionNegotiatorTest.java
@@ -131,6 +131,7 @@ public class AbstractNetconfSessionNegotiatorTest {
         negotiator.startNegotiation();
 
         captor.getValue().run(timeout);
+        channel.runPendingTasks();
         verify(closedDetector).close(any(), any());
     }
 
-- 
2.36.6