core/request_handler.go

   1 // Copyright (c) 2017 Cisco and/or its affiliates.
   2 //
   3 // Licensed under the Apache License, Version 2.0 (the "License");
   4 // you may not use this file except in compliance with the License.
   5 // You may obtain a copy of the License at:
   6 //
   7 //     http://www.apache.org/licenses/LICENSE-2.0
   8 //
   9 // Unless required by applicable law or agreed to in writing, software
  10 // distributed under the License is distributed on an "AS IS" BASIS,
  11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 // See the License for the specific language governing permissions and
  13 // limitations under the License.
  14
  15 package core
  16
  17 import (
  18         "errors"
  19         "fmt"
  20         "sync/atomic"
  21         "time"
  22
  23         logger "github.com/sirupsen/logrus"
  24 )
  25
  26 var (
  27         ErrNotConnected = errors.New("not connected to VPP, ignoring the request")
  28         ErrProbeTimeout = errors.New("probe reply not received within timeout period")
  29 )
  30
  31 // watchRequests watches for requests on the request API channel and forwards them as messages to VPP.
  32 func (c *Connection) watchRequests(ch *Channel) {
  33         for {
  34                 select {
  35                 case req, ok := <-ch.reqChan:
  36                         // new request on the request channel
  37                         if !ok {
  38                                 // after closing the request channel, release API channel and return
  39                                 c.releaseAPIChannel(ch)
  40                                 return
  41                         }
  42                         if err := c.processRequest(ch, req); err != nil {
  43                                 sendReplyError(ch, req, err)
  44                         }
  45                 }
  46         }
  47 }
  48
  49 // processRequest processes a single request received on the request channel.
  50 func (c *Connection) processRequest(ch *Channel, req *vppRequest) error {
  51         // check whether we are connected to VPP
  52         if atomic.LoadUint32(&c.vppConnected) == 0 {
  53                 err := ErrNotConnected
  54                 log.Errorf("processing request failed: %v", err)
  55                 return err
  56         }
  57
  58         // retrieve message ID
  59         msgID, err := c.GetMessageID(req.msg)
  60         if err != nil {
  61                 log.WithFields(logger.Fields{
  62                         "msg_name": req.msg.GetMessageName(),
  63                         "msg_crc":  req.msg.GetCrcString(),
  64                         "seq_num":  req.seqNum,
  65                         "error":    err,
  66                 }).Errorf("failed to retrieve message ID")
  67                 return fmt.Errorf("unable to retrieve message ID: %v", err)
  68         }
  69
  70         // encode the message into binary
  71         data, err := c.codec.EncodeMsg(req.msg, msgID)
  72         if err != nil {
  73                 log.WithFields(logger.Fields{
  74                         "channel":  ch.id,
  75                         "msg_id":   msgID,
  76                         "msg_name": req.msg.GetMessageName(),
  77                         "seq_num":  req.seqNum,
  78                         "error":    err,
  79                 }).Errorf("failed to encode message: %#v", req.msg)
  80                 return fmt.Errorf("unable to encode the message: %v", err)
  81         }
  82
  83         context := packRequestContext(ch.id, req.multi, req.seqNum)
  84
  85         if log.Level == logger.DebugLevel { // for performance reasons - logrus does some processing even if debugs are disabled
  86                 log.WithFields(logger.Fields{
  87                         "channel":  ch.id,
  88                         "context":  context,
  89                         "is_multi": req.multi,
  90                         "msg_id":   msgID,
  91                         "msg_name": req.msg.GetMessageName(),
  92                         "msg_size": len(data),
  93                         "seq_num":  req.seqNum,
  94                         "msg_crc":  req.msg.GetCrcString(),
  95                 }).Debug(" -> Sending a message to VPP.")
  96         }
  97
  98         // send the request to VPP
  99         err = c.vppClient.SendMsg(context, data)
 100         if err != nil {
 101                 err = fmt.Errorf("unable to send the message: %v", err)
 102                 log.WithFields(logger.Fields{
 103                         "context": context,
 104                         "msg_id":  msgID,
 105                         "seq_num": req.seqNum,
 106                 }).Error(err)
 107                 return err
 108         }
 109
 110         if req.multi {
 111                 // send a control ping to determine end of the multipart response
 112                 pingData, _ := c.codec.EncodeMsg(msgControlPing, c.pingReqID)
 113
 114                 log.WithFields(logger.Fields{
 115                         "channel":  ch.id,
 116                         "context":  context,
 117                         "msg_id":   c.pingReqID,
 118                         "msg_size": len(pingData),
 119                         "seq_num":  req.seqNum,
 120                 }).Debug(" -> Sending a control ping to VPP.")
 121
 122                 if err := c.vppClient.SendMsg(context, pingData); err != nil {
 123                         log.WithFields(logger.Fields{
 124                                 "context": context,
 125                                 "msg_id":  msgID,
 126                                 "seq_num": req.seqNum,
 127                         }).Warnf("unable to send control ping: %v", err)
 128                 }
 129         }
 130
 131         return nil
 132 }
 133
 134 // msgCallback is called whenever any binary API message comes from VPP.
 135 func (c *Connection) msgCallback(msgID uint16, data []byte) {
 136         if c == nil {
 137                 log.Warn("Already disconnected, ignoring the message.")
 138                 return
 139         }
 140
 141         msg, ok := c.msgMap[msgID]
 142         if !ok {
 143                 log.Warnf("Unknown message received, ID: %d", msgID)
 144                 return
 145         }
 146
 147         // decode message context to fix for special cases of messages,
 148         // for example:
 149         // - replies that don't have context as first field (comes as zero)
 150         // - events that don't have context at all (comes as non zero)
 151         //
 152         context, err := c.codec.DecodeMsgContext(data, msg)
 153         if err != nil {
 154                 log.Errorf("decoding context failed: %v", err)
 155         }
 156
 157         chanID, isMulti, seqNum := unpackRequestContext(context)
 158         if log.Level == logger.DebugLevel { // for performance reasons - logrus does some processing even if debugs are disabled
 159                 log.WithFields(logger.Fields{
 160                         "context":  context,
 161                         "msg_id":   msgID,
 162                         "msg_name": msg.GetMessageName(),
 163                         "msg_size": len(data),
 164                         "channel":  chanID,
 165                         "is_multi": isMulti,
 166                         "seq_num":  seqNum,
 167                         "msg_crc":  msg.GetCrcString(),
 168                 }).Debug(" <- Received a message from VPP.")
 169         }
 170
 171         if context == 0 || c.isNotificationMessage(msgID) {
 172                 // process the message as a notification
 173                 c.sendNotifications(msgID, data)
 174                 return
 175         }
 176
 177         // match ch according to the context
 178         c.channelsLock.RLock()
 179         ch, ok := c.channels[chanID]
 180         c.channelsLock.RUnlock()
 181         if !ok {
 182                 log.WithFields(logger.Fields{
 183                         "channel": chanID,
 184                         "msg_id":  msgID,
 185                 }).Error("Channel ID not known, ignoring the message.")
 186                 return
 187         }
 188
 189         // if this is a control ping reply to a multipart request,
 190         // treat this as a last part of the reply
 191         lastReplyReceived := isMulti && msgID == c.pingReplyID
 192
 193         // send the data to the channel, it needs to be copied,
 194         // because it will be freed after this function returns
 195         sendReply(ch, &vppReply{
 196                 msgID:        msgID,
 197                 seqNum:       seqNum,
 198                 data:         append([]byte(nil), data...),
 199                 lastReceived: lastReplyReceived,
 200         })
 201
 202         // store actual time of this reply
 203         c.lastReplyLock.Lock()
 204         c.lastReply = time.Now()
 205         c.lastReplyLock.Unlock()
 206 }
 207
 208 // sendReply sends the reply into the go channel, if it cannot be completed without blocking, otherwise
 209 // it logs the error and do not send the message.
 210 func sendReply(ch *Channel, reply *vppReply) {
 211         select {
 212         case ch.replyChan <- reply:
 213                 // reply sent successfully
 214         case <-time.After(time.Millisecond * 100):
 215                 // receiver still not ready
 216                 log.WithFields(logger.Fields{
 217                         "channel": ch,
 218                         "msg_id":  reply.msgID,
 219                         "seq_num": reply.seqNum,
 220                 }).Warn("Unable to send the reply, reciever end not ready.")
 221         }
 222 }
 223
 224 func sendReplyError(ch *Channel, req *vppRequest, err error) {
 225         sendReply(ch, &vppReply{seqNum: req.seqNum, err: err})
 226 }
 227
 228 // isNotificationMessage returns true if someone has subscribed to provided message ID.
 229 func (c *Connection) isNotificationMessage(msgID uint16) bool {
 230         c.subscriptionsLock.RLock()
 231         defer c.subscriptionsLock.RUnlock()
 232
 233         _, exists := c.subscriptions[msgID]
 234         return exists
 235 }
 236
 237 // sendNotifications send a notification message to all subscribers subscribed for that message.
 238 func (c *Connection) sendNotifications(msgID uint16, data []byte) {
 239         c.subscriptionsLock.RLock()
 240         defer c.subscriptionsLock.RUnlock()
 241
 242         matched := false
 243
 244         // send to notification to each subscriber
 245         for _, sub := range c.subscriptions[msgID] {
 246                 log.WithFields(logger.Fields{
 247                         "msg_name": sub.event.GetMessageName(),
 248                         "msg_id":   msgID,
 249                         "msg_size": len(data),
 250                 }).Debug("Sending a notification to the subscription channel.")
 251
 252                 event := sub.msgFactory()
 253                 if err := c.codec.DecodeMsg(data, event); err != nil {
 254                         log.WithFields(logger.Fields{
 255                                 "msg_name": sub.event.GetMessageName(),
 256                                 "msg_id":   msgID,
 257                                 "msg_size": len(data),
 258                         }).Errorf("Unable to decode the notification message: %v", err)
 259                         continue
 260                 }
 261
 262                 // send the message into the go channel of the subscription
 263                 select {
 264                 case sub.notifChan <- event:
 265                         // message sent successfully
 266                 default:
 267                         // unable to write into the channel without blocking
 268                         log.WithFields(logger.Fields{
 269                                 "msg_name": sub.event.GetMessageName(),
 270                                 "msg_id":   msgID,
 271                                 "msg_size": len(data),
 272                         }).Warn("Unable to deliver the notification, reciever end not ready.")
 273                 }
 274
 275                 matched = true
 276         }
 277
 278         if !matched {
 279                 log.WithFields(logger.Fields{
 280                         "msg_id":   msgID,
 281                         "msg_size": len(data),
 282                 }).Info("No subscription found for the notification message.")
 283         }
 284 }
 285
 286 // +------------------+-------------------+-----------------------+
 287 // | 15b = channel ID | 1b = is multipart | 16b = sequence number |
 288 // +------------------+-------------------+-----------------------+
 289 func packRequestContext(chanID uint16, isMultipart bool, seqNum uint16) uint32 {
 290         context := uint32(chanID) << 17
 291         if isMultipart {
 292                 context |= 1 << 16
 293         }
 294         context |= uint32(seqNum)
 295         return context
 296 }
 297
 298 func unpackRequestContext(context uint32) (chanID uint16, isMulipart bool, seqNum uint16) {
 299         chanID = uint16(context >> 17)
 300         if ((context >> 16) & 0x1) != 0 {
 301                 isMulipart = true
 302         }
 303         seqNum = uint16(context & 0xffff)
 304         return
 305 }
 306
 307 // compareSeqNumbers returns -1, 0, 1 if sequence number <seqNum1> precedes, equals to,
 308 // or succeeds seq. number <seqNum2>.
 309 // Since sequence numbers cycle in the finite set of size 2^16, the function
 310 // must assume that the distance between compared sequence numbers is less than
 311 // (2^16)/2 to determine the order.
 312 func compareSeqNumbers(seqNum1, seqNum2 uint16) int {
 313         // calculate distance from seqNum1 to seqNum2
 314         var dist uint16
 315         if seqNum1 <= seqNum2 {
 316                 dist = seqNum2 - seqNum1
 317         } else {
 318                 dist = 0xffff - (seqNum1 - seqNum2 - 1)
 319         }
 320         if dist == 0 {
 321                 return 0
 322         } else if dist <= 0x8000 {
 323                 return -1
 324         }
 325         return 1
 326 }