import (
"errors"
"fmt"
+ "reflect"
"sync/atomic"
"time"
logger "github.com/sirupsen/logrus"
+
+ "git.fd.io/govpp.git/api"
)
+var ReplyChannelTimeout = time.Millisecond * 100
+
var (
ErrNotConnected = errors.New("not connected to VPP, ignoring the request")
ErrProbeTimeout = errors.New("probe reply not received within timeout period")
c.releaseAPIChannel(ch)
return
}
- c.processRequest(ch, req)
+ if err := c.processRequest(ch, req); err != nil {
+ sendReplyError(ch, req, err)
+ }
}
}
}
// processRequest processes a single request received on the request channel.
func (c *Connection) processRequest(ch *Channel, req *vppRequest) error {
// check whether we are connected to VPP
- if atomic.LoadUint32(&c.connected) == 0 {
+ if atomic.LoadUint32(&c.vppConnected) == 0 {
err := ErrNotConnected
log.Errorf("processing request failed: %v", err)
- sendReplyError(ch, req, err)
return err
}
// retrieve message ID
msgID, err := c.GetMessageID(req.msg)
if err != nil {
- err = fmt.Errorf("unable to retrieve message ID: %v", err)
log.WithFields(logger.Fields{
"msg_name": req.msg.GetMessageName(),
"msg_crc": req.msg.GetCrcString(),
"seq_num": req.seqNum,
- }).Error(err)
- sendReplyError(ch, req, err)
- return err
+ "error": err,
+ }).Errorf("failed to retrieve message ID")
+ return fmt.Errorf("unable to retrieve message ID: %v", err)
}
// encode the message into binary
data, err := c.codec.EncodeMsg(req.msg, msgID)
if err != nil {
- err = fmt.Errorf("unable to encode the messge: %v", err)
log.WithFields(logger.Fields{
"channel": ch.id,
"msg_id": msgID,
"msg_name": req.msg.GetMessageName(),
"seq_num": req.seqNum,
- }).Error(err)
- sendReplyError(ch, req, err)
- return err
+ "error": err,
+ }).Errorf("failed to encode message: %#v", req.msg)
+ return fmt.Errorf("unable to encode the message: %v", err)
}
- // get context
context := packRequestContext(ch.id, req.multi, req.seqNum)
+
if log.Level == logger.DebugLevel { // for performance reasons - logrus does some processing even if debugs are disabled
log.WithFields(logger.Fields{
"channel": ch.id,
"context": context,
"is_multi": req.multi,
"msg_id": msgID,
- "msg_name": req.msg.GetMessageName(),
"msg_size": len(data),
"seq_num": req.seqNum,
- }).Debug(" -> Sending a message to VPP.")
+ "msg_crc": req.msg.GetCrcString(),
+ }).Debugf("--> govpp SEND: %s %+v", req.msg.GetMessageName(), req.msg)
}
// send the request to VPP
- err = c.vpp.SendMsg(context, data)
+ err = c.vppClient.SendMsg(context, data)
if err != nil {
err = fmt.Errorf("unable to send the message: %v", err)
log.WithFields(logger.Fields{
"msg_id": msgID,
"seq_num": req.seqNum,
}).Error(err)
- sendReplyError(ch, req, err)
return err
}
if req.multi {
// send a control ping to determine end of the multipart response
- pingData, _ := c.codec.EncodeMsg(msgControlPing, c.pingReqID)
+ pingData, _ := c.codec.EncodeMsg(c.msgControlPing, c.pingReqID)
log.WithFields(logger.Fields{
"channel": ch.id,
"msg_id": c.pingReqID,
"msg_size": len(pingData),
"seq_num": req.seqNum,
- }).Debug(" -> Sending a control ping to VPP.")
+ }).Debug(" -> sending control ping")
- if err := c.vpp.SendMsg(context, pingData); err != nil {
+ if err := c.vppClient.SendMsg(context, pingData); err != nil {
log.WithFields(logger.Fields{
"context": context,
"msg_id": msgID,
// msgCallback is called whenever any binary API message comes from VPP.
func (c *Connection) msgCallback(msgID uint16, data []byte) {
- connLock.RLock()
- defer connLock.RUnlock()
-
if c == nil {
log.Warn("Already disconnected, ignoring the message.")
return
}
chanID, isMulti, seqNum := unpackRequestContext(context)
+
if log.Level == logger.DebugLevel { // for performance reasons - logrus does some processing even if debugs are disabled
+ msg = reflect.New(reflect.TypeOf(msg).Elem()).Interface().(api.Message)
+
+ // decode the message
+ if err = c.codec.DecodeMsg(data, msg); err != nil {
+ err = fmt.Errorf("decoding message failed: %w", err)
+ return
+ }
+
log.WithFields(logger.Fields{
"context": context,
"msg_id": msgID,
- "msg_name": msg.GetMessageName(),
"msg_size": len(data),
"channel": chanID,
"is_multi": isMulti,
"seq_num": seqNum,
- }).Debug(" <- Received a message from VPP.")
+ "msg_crc": msg.GetCrcString(),
+ }).Debugf("<-- govpp RECEIVE: %s %+v", msg.GetMessageName(), msg)
}
if context == 0 || c.isNotificationMessage(msgID) {
// treat this as a last part of the reply
lastReplyReceived := isMulti && msgID == c.pingReplyID
- // send the data to the channel
+ // send the data to the channel, it needs to be copied,
+ // because it will be freed after this function returns
sendReply(ch, &vppReply{
msgID: msgID,
seqNum: seqNum,
- data: data,
+ data: append([]byte(nil), data...),
lastReceived: lastReplyReceived,
})
func sendReply(ch *Channel, reply *vppReply) {
select {
case ch.replyChan <- reply:
- // reply sent successfully
- case <-time.After(time.Millisecond * 100):
+ // reply sent successfully
+ case <-time.After(ReplyChannelTimeout):
// receiver still not ready
log.WithFields(logger.Fields{
"channel": ch,
Code Review / govpp.git / blobdiff