Improve handling of probes on timeouts

Change-Id: If94059586d4be739d6c8ae7843cfaf3bc90a5323
Signed-off-by: Ondrej Fabry <ofabry@cisco.com>

diff --git a/core/core.go b/core/core.go
index 6714f6f..4782ba1 100644 (file)
--- a/core/core.go
+++ b/core/core.go
@@ -55,7 +55,7 @@ const (
        Connected ConnectionState = iota
 
        // Disconnected connection state means that the connection to VPP has been lost.
-       Disconnected = iota
+       Disconnected
 )
 
 // ConnectionEvent is a notification about change in the VPP connection state.
@@ -85,6 +85,9 @@ type Connection struct {
        maxChannelID uint32 // maximum used client ID
        pingReqID    uint16 // ID if the ControlPing message
        pingReplyID  uint16 // ID of the ControlPingReply message
+
+       lastReplyLock sync.Mutex // lock for the last reply
+       lastReply     time.Time  // time of the last received reply from VPP
 }
 
 // channelMetadata contains core-local metadata of an API channel.
@@ -271,17 +274,19 @@ func (c *Connection) connectLoop(connChan chan ConnectionEvent) {
 // it continues with connectLoop and tries to reconnect.
 func (c *Connection) healthCheckLoop(connChan chan ConnectionEvent) {
        // create a separate API channel for health check probes
-       ch, err := conn.NewAPIChannel()
+       ch, err := conn.NewAPIChannelBuffered(1, 1)
        if err != nil {
-               log.Error("Error by creating health check API channel, health check will be disabled:", err)
+               log.Error("Failed to create health check API channel, health check will be disabled:", err)
                return
        }
 
-       failedChecks := 0
-       // send health check probes until an error occurs
+       var sinceLastReply time.Duration
+       var failedChecks int
+
+       // send health check probes until an error or timeout occurs
        for {
-               // wait for healthCheckProbeInterval
-               <-time.After(healthCheckProbeInterval)
+               // sleep until next health check probe period
+               time.Sleep(healthCheckProbeInterval)
 
                if atomic.LoadUint32(&c.connected) == 0 {
                        // Disconnect has been called in the meantime, return the healthcheck - reconnect loop
@@ -289,30 +294,56 @@ func (c *Connection) healthCheckLoop(connChan chan ConnectionEvent) {
                        return
                }
 
-               // send the control ping
-               ch.ReqChan <- &api.VppRequest{Message: msgControlPing}
-
-               // expect response within timeout period
+               // try draining probe replies from previous request before sending next one
                select {
-               case vppReply := <-ch.ReplyChan:
-                       err = vppReply.Error
-               case <-time.After(healthCheckReplyTimeout):
-                       err = errors.New("probe reply not received within the timeout period")
+               case <-ch.ReplyChan:
+                       log.Debug("drained old probe reply from reply channel")
+               default:
                }
 
-               if err != nil {
-                       failedChecks++
-                       log.Warnf("VPP health check failed (%d. time): %v", failedChecks, err)
-               } else {
-                       failedChecks = 0
+               // send the control ping request
+               ch.ReqChan <- &api.VppRequest{Message: msgControlPing}
+
+               for {
+                       // expect response within timeout period
+                       select {
+                       case vppReply := <-ch.ReplyChan:
+                               err = vppReply.Error
+
+                       case <-time.After(healthCheckReplyTimeout):
+                               err = ErrProbeTimeout
+
+                               // check if time since last reply from any other
+                               // channel is less than health check reply timeout
+                               conn.lastReplyLock.Lock()
+                               sinceLastReply = time.Since(c.lastReply)
+                               conn.lastReplyLock.Unlock()
+
+                               if sinceLastReply < healthCheckReplyTimeout {
+                                       log.Warnf("VPP health check probe timing out, but some request on other channel was received %v ago, continue waiting!", sinceLastReply)
+                                       continue
+                               }
+                       }
+                       break
                }
 
-               if failedChecks > healthCheckThreshold {
-                       // in case of error, break & disconnect
-                       log.Errorf("Number of VPP health check fails exceeded treshold (%d)", healthCheckThreshold, err)
-                       // signal disconnected event via channel
+               if err == ErrProbeTimeout {
+                       failedChecks++
+                       log.Warnf("VPP health check probe timed out after %v (%d. timeout)", healthCheckReplyTimeout, failedChecks)
+                       if failedChecks > healthCheckThreshold {
+                               // in case of exceeded treshold disconnect
+                               log.Errorf("VPP health check exceeded treshold for timeouts (>%d), assuming disconnect", healthCheckThreshold)
+                               connChan <- ConnectionEvent{Timestamp: time.Now(), State: Disconnected}
+                               break
+                       }
+               } else if err != nil {
+                       // in case of error disconnect
+                       log.Errorf("VPP health check probe failed: %v", err)
                        connChan <- ConnectionEvent{Timestamp: time.Now(), State: Disconnected}
                        break
+               } else if failedChecks > 0 {
+                       failedChecks = 0
+                       log.Infof("VPP health check probe OK")
                }
        }
 

diff --git a/core/request_handler.go b/core/request_handler.go
index 27ff4fc..8f793f5 100644 (file)
--- a/core/request_handler.go
+++ b/core/request_handler.go
@@ -18,6 +18,7 @@ import (
        "errors"
        "fmt"
        "sync/atomic"
+       "time"
 
        logger "github.com/sirupsen/logrus"
 
@@ -26,6 +27,7 @@ import (
 
 var (
        ErrNotConnected = errors.New("not connected to VPP, ignoring the request")
+       ErrProbeTimeout = errors.New("probe reply not received within timeout period")
 )
 
 // watchRequests watches for requests on the request API channel and forwards them as messages to VPP.
@@ -98,7 +100,16 @@ func (c *Connection) processRequest(ch *api.Channel, chMeta *channelMetadata, re
        }
 
        // send the request to VPP
-       c.vpp.SendMsg(chMeta.id, data)
+       err = c.vpp.SendMsg(chMeta.id, data)
+       if err != nil {
+               err = fmt.Errorf("unable to send the messge: %v", err)
+               log.WithFields(logger.Fields{
+                       "context": chMeta.id,
+                       "msg_id":  msgID,
+               }).Error(err)
+               sendReply(ch, &api.VppReply{Error: err})
+               return err
+       }
 
        if req.Multipart {
                // send a control ping to determine end of the multipart response
@@ -166,6 +177,11 @@ func msgCallback(context uint32, msgID uint16, data []byte) {
                Data:              data,
                LastReplyReceived: lastReplyReceived,
        })
+
+       // store actual time of this reply
+       conn.lastReplyLock.Lock()
+       conn.lastReply = time.Now()
+       conn.lastReplyLock.Unlock()
 }
 
 // sendReply sends the reply into the go channel, if it cannot be completed without blocking, otherwise
@@ -174,8 +190,8 @@ func sendReply(ch *api.Channel, reply *api.VppReply) {
        select {
        case ch.ReplyChan <- reply:
                // reply sent successfully
-       default:
-               // unable to write into the channel without blocking
+       case <-time.After(time.Millisecond * 100):
+               // receiver still not ready
                log.WithFields(logger.Fields{
                        "channel": ch,
                        "msg_id":  reply.MessageID,