Ignore invalid message ID if last request timed out

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

diff --git a/adapter/vppapiclient/vppapiclient_adapter.go b/adapter/vppapiclient/vppapiclient_adapter.go
index 1cdaf72..10ec53b 100644 (file)
--- a/adapter/vppapiclient/vppapiclient_adapter.go
+++ b/adapter/vppapiclient/vppapiclient_adapter.go
@@ -77,7 +77,6 @@ govpp_get_msg_index(char *name_and_crc)
 import "C"
 
 import (
-       "errors"
        "fmt"
        "os"
        "reflect"
@@ -130,7 +129,7 @@ func (a *vppAPIClientAdapter) GetMsgID(msgName string, msgCrc string) (uint16, e
 
        msgID := uint16(C.govpp_get_msg_index(nameAndCrc))
        if msgID == ^uint16(0) {
-               return msgID, errors.New("unkonwn message")
+               return msgID, fmt.Errorf("unknown message: %v (crc: %v)", msgName, msgCrc)
        }
 
        return msgID, nil

diff --git a/api/api.go b/api/api.go
index 98932fa..60508cd 100644 (file)
--- a/api/api.go
+++ b/api/api.go
@@ -18,6 +18,8 @@ import (
        "errors"
        "fmt"
        "time"
+
+       "github.com/sirupsen/logrus"
 )
 
 // MessageType represents the type of a VPP message.
@@ -61,7 +63,7 @@ type ChannelProvider interface {
        // It uses default buffer sizes for the request and reply Go channels.
        NewAPIChannel() (*Channel, error)
 
-       // NewAPIChannel returns a new channel for communication with VPP via govpp core.
+       // NewAPIChannelBuffered returns a new channel for communication with VPP via govpp core.
        // It allows to specify custom buffer sizes for the request and reply Go channels.
        NewAPIChannelBuffered() (*Channel, error)
 }
@@ -94,6 +96,7 @@ type Channel struct {
 
        replyTimeout time.Duration // maximum time that the API waits for a reply from VPP before returning an error, can be set with SetReplyTimeout
        metadata     interface{}   // opaque metadata of the API channel
+       lastTimedOut bool          // wether last request timed out
 }
 
 // VppRequest is a request that will be sent to VPP.
@@ -214,34 +217,46 @@ func (ch *Channel) receiveReplyInternal(msg Message) (LastReplyReceived bool, er
        if msg == nil {
                return false, errors.New("nil message passed in")
        }
-       select {
-       // blocks until a reply comes to ReplyChan or until timeout expires
-       case vppReply := <-ch.ReplyChan:
-               if vppReply.Error != nil {
-                       err = vppReply.Error
-                       return
-               }
-               if vppReply.LastReplyReceived {
-                       LastReplyReceived = true
-                       return
-               }
-               // message checks
-               expMsgID, err := ch.MsgIdentifier.GetMessageID(msg)
-               if err != nil {
-                       err = fmt.Errorf("message %s with CRC %s is not compatible with the VPP we are connected to",
-                               msg.GetMessageName(), msg.GetCrcString())
+
+       timer := time.NewTimer(ch.replyTimeout)
+       for {
+               select {
+               // blocks until a reply comes to ReplyChan or until timeout expires
+               case vppReply := <-ch.ReplyChan:
+                       if vppReply.Error != nil {
+                               err = vppReply.Error
+                               return
+                       }
+                       if vppReply.LastReplyReceived {
+                               LastReplyReceived = true
+                               return
+                       }
+                       // message checks
+                       expMsgID, err := ch.MsgIdentifier.GetMessageID(msg)
+                       if err != nil {
+                               err = fmt.Errorf("message %s with CRC %s is not compatible with the VPP we are connected to",
+                                       msg.GetMessageName(), msg.GetCrcString())
+                               return false, err
+                       }
+                       if vppReply.MessageID != expMsgID {
+                               if ch.lastTimedOut {
+                                       logrus.Warnf("received invalid message ID, expected %d (%s), but got %d (probably timed out reply from previous request)",
+                                               expMsgID, msg.GetMessageName(), vppReply.MessageID)
+                                       continue
+                               }
+                               err = fmt.Errorf("received invalid message ID, expected %d (%s), but got %d (check if multiple goroutines are not sharing single GoVPP channel)",
+                                       expMsgID, msg.GetMessageName(), vppReply.MessageID)
+                               return false, err
+                       }
+                       ch.lastTimedOut = false
+                       // decode the message
+                       err = ch.MsgDecoder.DecodeMsg(vppReply.Data, msg)
                        return false, err
-               }
-               if vppReply.MessageID != expMsgID {
-                       err = fmt.Errorf("received invalid message ID, expected %d (%s), but got %d (check if multiple goroutines are not sharing single GoVPP channel)",
-                               expMsgID, msg.GetMessageName(), vppReply.MessageID)
+               case <-timer.C:
+                       ch.lastTimedOut = true
+                       err = fmt.Errorf("no reply received within the timeout period %s", ch.replyTimeout)
                        return false, err
                }
-               // decode the message
-               err = ch.MsgDecoder.DecodeMsg(vppReply.Data, msg)
-
-       case <-time.After(ch.replyTimeout):
-               err = fmt.Errorf("no reply received within the timeout period %s", ch.replyTimeout)
        }
        return
 }

diff --git a/api/api_test.go b/api/api_test.go
index 9af6e71..62541ab 100644 (file)
--- a/api/api_test.go
+++ b/api/api_test.go
@@ -49,8 +49,6 @@ func setupTest(t *testing.T) *testCtx {
        ctx.ch, err = ctx.conn.NewAPIChannel()
        Expect(err).ShouldNot(HaveOccurred())
 
-       ctx.ch.SetReplyTimeout(time.Millisecond)
-
        return ctx
 }
 
@@ -328,6 +326,8 @@ func TestSetReplyTimeout(t *testing.T) {
        ctx := setupTest(t)
        defer ctx.teardownTest()
 
+       ctx.ch.SetReplyTimeout(time.Millisecond)
+
        // first one request should work
        ctx.mockVpp.MockReply(&vpe.ControlPingReply{})
        err := ctx.ch.SendRequest(&vpe.ControlPing{}).ReceiveReply(&vpe.ControlPingReply{})
@@ -343,6 +343,8 @@ func TestSetReplyTimeoutMultiRequest(t *testing.T) {
        ctx := setupTest(t)
        defer ctx.teardownTest()
 
+       ctx.ch.SetReplyTimeout(time.Millisecond)
+
        for i := 1; i <= 3; i++ {
                ctx.mockVpp.MockReply(&interfaces.SwInterfaceDetails{
                        SwIfIndex:     uint32(i),
@@ -424,7 +426,7 @@ func TestMultiRequestDouble(t *testing.T) {
        ctx.mockVpp.MockReply(&vpe.ControlPingReply{})
 
        cnt := 0
-       sendMultiRequest := func() error {
+       var sendMultiRequest = func() error {
                reqCtx := ctx.ch.SendMultiRequest(&interfaces.SwInterfaceDump{})
                for {
                        msg := &interfaces.SwInterfaceDetails{}
@@ -448,3 +450,98 @@ func TestMultiRequestDouble(t *testing.T) {
 
        Expect(cnt).To(BeEquivalentTo(6))
 }
+
+func TestReceiveReplyAfterTimeout(t *testing.T) {
+       ctx := setupTest(t)
+       defer ctx.teardownTest()
+
+       ctx.ch.SetReplyTimeout(time.Millisecond)
+
+       // first one request should work
+       ctx.mockVpp.MockReply(&vpe.ControlPingReply{})
+       err := ctx.ch.SendRequest(&vpe.ControlPing{}).ReceiveReply(&vpe.ControlPingReply{})
+       Expect(err).ShouldNot(HaveOccurred())
+
+       err = ctx.ch.SendRequest(&vpe.ControlPing{}).ReceiveReply(&vpe.ControlPingReply{})
+       Expect(err).Should(HaveOccurred())
+       Expect(err.Error()).To(ContainSubstring("timeout"))
+
+       // simulating late reply
+       ctx.mockVpp.MockReply(&vpe.ControlPingReply{})
+
+       // normal reply for next request
+       ctx.mockVpp.MockReply(&tap.TapConnectReply{})
+
+       req := &tap.TapConnect{
+               TapName:      []byte("test-tap-name"),
+               UseRandomMac: 1,
+       }
+       reply := &tap.TapConnectReply{}
+
+       // should succeed
+       err = ctx.ch.SendRequest(req).ReceiveReply(reply)
+       Expect(err).ShouldNot(HaveOccurred())
+}
+
+/*
+       TODO: fix mock adapter
+       This test will fail because mock adapter will stop sending replies
+       when it encounters control_ping_reply from multi request,
+       thus never sending reply for next request
+
+func TestReceiveReplyAfterTimeoutMultiRequest(t *testing.T) {
+       ctx := setupTest(t)
+       defer ctx.teardownTest()
+
+       ctx.ch.SetReplyTimeout(time.Millisecond * 100)
+
+       // first one request should work
+       ctx.mockVpp.MockReply(&vpe.ControlPingReply{})
+       err := ctx.ch.SendRequest(&vpe.ControlPing{}).ReceiveReply(&vpe.ControlPingReply{})
+       Expect(err).ShouldNot(HaveOccurred())
+
+       cnt := 0
+       var sendMultiRequest = func() error {
+               reqCtx := ctx.ch.SendMultiRequest(&interfaces.SwInterfaceDump{})
+               for {
+                       msg := &interfaces.SwInterfaceDetails{}
+                       stop, err := reqCtx.ReceiveReply(msg)
+                       if stop {
+                               break // break out of the loop
+                       }
+                       if err != nil {
+                               return err
+                       }
+                       cnt++
+               }
+               return nil
+       }
+
+       err = sendMultiRequest()
+       Expect(err).Should(HaveOccurred())
+       Expect(err.Error()).To(ContainSubstring("timeout"))
+       Expect(cnt).To(BeEquivalentTo(0))
+
+       // simulating late replies
+       for i := 1; i <= 3; i++ {
+               ctx.mockVpp.MockReply(&interfaces.SwInterfaceDetails{
+                       SwIfIndex:     uint32(i),
+                       InterfaceName: []byte("if-name-test"),
+               })
+       }
+       ctx.mockVpp.MockReply(&vpe.ControlPingReply{})
+
+       // normal reply for next request
+       ctx.mockVpp.MockReply(&tap.TapConnectReply{})
+
+       req := &tap.TapConnect{
+               TapName:      []byte("test-tap-name"),
+               UseRandomMac: 1,
+       }
+       reply := &tap.TapConnectReply{}
+
+       // should succeed
+       err = ctx.ch.SendRequest(req).ReceiveReply(reply)
+       Expect(err).ShouldNot(HaveOccurred())
+}
+*/