From ead654e06cdd2189446f7fffb417bab4e4cb540f Mon Sep 17 00:00:00 2001
From: Robert Varga <rovarga@cisco.com>
Date: Mon, 27 Mar 2017 16:19:25 +0200
Subject: [PATCH] BUG-5280: make sure we arm the request timer

The timer which is supposed to timeout requests and detect
overall badness of the backeend was not being armed. Fix that
by scheduling it whenever we make the queue non-empty.

Change-Id: I9d8be694e3ed5154b66baca76c0788840a38c2f7
Signed-off-by: Robert Varga <rovarga@cisco.com>
---
 .../client/AbstractClientConnection.java      | 35 +++++++++++++++----
 .../cluster/access/client/TransmitQueue.java  |  4 +++
 2 files changed, 32 insertions(+), 7 deletions(-)

diff --git a/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/AbstractClientConnection.java b/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/AbstractClientConnection.java
index ac4ac785d6..69deb01910 100644
--- a/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/AbstractClientConnection.java
+++ b/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/AbstractClientConnection.java
@@ -44,12 +44,18 @@ public abstract class AbstractClientConnection<T extends BackendInfo> {
     @VisibleForTesting
     static final long REQUEST_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(30);
 
+    private static final FiniteDuration REQUEST_TIMEOUT_DURATION = FiniteDuration.apply(REQUEST_TIMEOUT_NANOS,
+        TimeUnit.NANOSECONDS);
+
     private final Lock lock = new ReentrantLock();
     private final ClientActorContext context;
     @GuardedBy("lock")
     private final TransmitQueue queue;
     private final Long cookie;
 
+    @GuardedBy("lock")
+    private boolean haveTimer;
+
     private volatile RequestException poisoned;
 
     // Do not allow subclassing outside of this package
@@ -108,7 +114,7 @@ public abstract class AbstractClientConnection<T extends BackendInfo> {
 
     @GuardedBy("lock")
     final void finishReplay(final ReconnectForwarder forwarder) {
-        queue.setForwarder(forwarder, readTime());
+        setForwarder(forwarder);
         lock.unlock();
     }
 
@@ -127,6 +133,10 @@ public abstract class AbstractClientConnection<T extends BackendInfo> {
     final long enqueueEntry(final ConnectionEntry entry, final long now) {
         lock.lock();
         try {
+            if (queue.isEmpty()) {
+                // The queue is becoming non-empty, schedule a timer
+                scheduleTimer(REQUEST_TIMEOUT_DURATION);
+            }
             return queue.enqueue(entry, now);
         } finally {
             lock.unlock();
@@ -138,7 +148,7 @@ public abstract class AbstractClientConnection<T extends BackendInfo> {
         try {
             TimeUnit.NANOSECONDS.sleep(delay);
         } catch (InterruptedException e) {
-            LOG.debug("Interrupted while sleeping");
+            LOG.debug("Interrupted while sleeping", e);
         }
     }
 
@@ -147,9 +157,19 @@ public abstract class AbstractClientConnection<T extends BackendInfo> {
      *
      * @param delay Delay, in nanoseconds
      */
+    @GuardedBy("lock")
     private void scheduleTimer(final FiniteDuration delay) {
+        if (haveTimer) {
+            LOG.debug("{}: timer already scheduled", context.persistenceId());
+            return;
+        }
+        if (queue.hasSuccessor()) {
+            LOG.debug("{}: connection has successor, not scheduling timer", context.persistenceId());
+            return;
+        }
         LOG.debug("{}: scheduling timeout in {}", context.persistenceId(), delay);
         context.executeInActor(this::runTimer, delay);
+        haveTimer = true;
     }
 
     /**
@@ -165,6 +185,7 @@ public abstract class AbstractClientConnection<T extends BackendInfo> {
 
         lock.lock();
         try {
+            haveTimer = false;
             final long now = readTime();
             // The following line is only reliable when queue is not forwarding, but such state should not last long.
             final long ticksSinceProgress = queue.ticksStalling(now);
@@ -185,15 +206,15 @@ public abstract class AbstractClientConnection<T extends BackendInfo> {
                 // We have timed out. There is no point in scheduling a timer
                 return reconnectConnection(current);
             }
+
+            if (delay.isPresent()) {
+                // If there is new delay, schedule a timer
+                scheduleTimer(delay.get());
+            }
         } finally {
             lock.unlock();
         }
 
-        if (delay.isPresent()) {
-            // If there is new delay, schedule a timer
-            scheduleTimer(delay.get());
-        }
-
         return current;
     }
 
diff --git a/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/TransmitQueue.java b/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/TransmitQueue.java
index 4a1b3a2f29..d384ba4736 100644
--- a/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/TransmitQueue.java
+++ b/opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/TransmitQueue.java
@@ -112,6 +112,10 @@ abstract class TransmitQueue {
         return tracker.ticksStalling(now);
     }
 
+    final boolean hasSuccessor() {
+        return successor != null;
+    }
+
     // If a matching request was found, this will track a task was closed.
     final Optional<TransmittedConnectionEntry> complete(final ResponseEnvelope<?> envelope, final long now) {
         Optional<TransmittedConnectionEntry> maybeEntry = findMatchingEntry(inflight, envelope);
-- 
2.36.6