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_size": len(data),
  92                         "seq_num":  req.seqNum,
  93                         "msg_crc":  req.msg.GetCrcString(),
  94                 }).Debugf(" --> sending msg: %s", req.msg.GetMessageName())
  95         }
  96
  97         // send the request to VPP
  98         err = c.vppClient.SendMsg(context, data)
  99         if err != nil {
 100                 err = fmt.Errorf("unable to send the message: %v", err)
 101                 log.WithFields(logger.Fields{
 102                         "context": context,
 103                         "msg_id":  msgID,
 104                         "seq_num": req.seqNum,
 105                 }).Error(err)
 106                 return err
 107         }
 108
 109         if req.multi {
 110                 // send a control ping to determine end of the multipart response
 111                 pingData, _ := c.codec.EncodeMsg(msgControlPing, c.pingReqID)
 112
 113                 log.WithFields(logger.Fields{
 114                         "channel":  ch.id,
 115                         "context":  context,
 116                         "msg_id":   c.pingReqID,
 117                         "msg_size": len(pingData),
 118                         "seq_num":  req.seqNum,
 119                 }).Debug("  -> sending control ping")
 120
 121                 if err := c.vppClient.SendMsg(context, pingData); err != nil {
 122                         log.WithFields(logger.Fields{
 123                                 "context": context,
 124                                 "msg_id":  msgID,
 125                                 "seq_num": req.seqNum,
 126                         }).Warnf("unable to send control ping: %v", err)
 127                 }
 128         }
 129
 130         return nil
 131 }
 132
 133 // msgCallback is called whenever any binary API message comes from VPP.
 134 func (c *Connection) msgCallback(msgID uint16, data []byte) {
 135         if c == nil {
 136                 log.Warn("Already disconnected, ignoring the message.")
 137                 return
 138         }
 139
 140         msg, ok := c.msgMap[msgID]
 141         if !ok {
 142                 log.Warnf("Unknown message received, ID: %d", msgID)
 143                 return
 144         }
 145
 146         // decode message context to fix for special cases of messages,
 147         // for example:
 148         // - replies that don't have context as first field (comes as zero)
 149         // - events that don't have context at all (comes as non zero)
 150         //
 151         context, err := c.codec.DecodeMsgContext(data, msg)
 152         if err != nil {
 153                 log.Errorf("decoding context failed: %v", err)
 154         }
 155
 156         chanID, isMulti, seqNum := unpackRequestContext(context)
 157         if log.Level == logger.DebugLevel { // for performance reasons - logrus does some processing even if debugs are disabled
 158                 log.WithFields(logger.Fields{
 159                         "context":  context,
 160                         "msg_id":   msgID,
 161                         "msg_size": len(data),
 162                         "channel":  chanID,
 163                         "is_multi": isMulti,
 164                         "seq_num":  seqNum,
 165                         "msg_crc":  msg.GetCrcString(),
 166                 }).Debugf(" <- received msg: %s", msg.GetMessageName())
 167         }
 168
 169         if context == 0 || c.isNotificationMessage(msgID) {
 170                 // process the message as a notification
 171                 c.sendNotifications(msgID, data)
 172                 return
 173         }
 174
 175         // match ch according to the context
 176         c.channelsLock.RLock()
 177         ch, ok := c.channels[chanID]
 178         c.channelsLock.RUnlock()
 179         if !ok {
 180                 log.WithFields(logger.Fields{
 181                         "channel": chanID,
 182                         "msg_id":  msgID,
 183                 }).Error("Channel ID not known, ignoring the message.")
 184                 return
 185         }
 186
 187         // if this is a control ping reply to a multipart request,
 188         // treat this as a last part of the reply
 189         lastReplyReceived := isMulti && msgID == c.pingReplyID
 190
 191         // send the data to the channel, it needs to be copied,
 192         // because it will be freed after this function returns
 193         sendReply(ch, &vppReply{
 194                 msgID:        msgID,
 195                 seqNum:       seqNum,
 196                 data:         append([]byte(nil), data...),
 197                 lastReceived: lastReplyReceived,
 198         })
 199
 200         // store actual time of this reply
 201         c.lastReplyLock.Lock()
 202         c.lastReply = time.Now()
 203         c.lastReplyLock.Unlock()
 204 }
 205
 206 // sendReply sends the reply into the go channel, if it cannot be completed without blocking, otherwise
 207 // it logs the error and do not send the message.
 208 func sendReply(ch *Channel, reply *vppReply) {
 209         select {
 210         case ch.replyChan <- reply:
 211                 // reply sent successfully
 212         case <-time.After(time.Millisecond * 100):
 213                 // receiver still not ready
 214                 log.WithFields(logger.Fields{
 215                         "channel": ch,
 216                         "msg_id":  reply.msgID,
 217                         "seq_num": reply.seqNum,
 218                 }).Warn("Unable to send the reply, reciever end not ready.")
 219         }
 220 }
 221
 222 func sendReplyError(ch *Channel, req *vppRequest, err error) {
 223         sendReply(ch, &vppReply{seqNum: req.seqNum, err: err})
 224 }
 225
 226 // isNotificationMessage returns true if someone has subscribed to provided message ID.
 227 func (c *Connection) isNotificationMessage(msgID uint16) bool {
 228         c.subscriptionsLock.RLock()
 229         defer c.subscriptionsLock.RUnlock()
 230
 231         _, exists := c.subscriptions[msgID]
 232         return exists
 233 }
 234
 235 // sendNotifications send a notification message to all subscribers subscribed for that message.
 236 func (c *Connection) sendNotifications(msgID uint16, data []byte) {
 237         c.subscriptionsLock.RLock()
 238         defer c.subscriptionsLock.RUnlock()
 239
 240         matched := false
 241
 242         // send to notification to each subscriber
 243         for _, sub := range c.subscriptions[msgID] {
 244                 log.WithFields(logger.Fields{
 245                         "msg_name": sub.event.GetMessageName(),
 246                         "msg_id":   msgID,
 247                         "msg_size": len(data),
 248                 }).Debug("Sending a notification to the subscription channel.")
 249
 250                 event := sub.msgFactory()
 251                 if err := c.codec.DecodeMsg(data, event); err != nil {
 252                         log.WithFields(logger.Fields{
 253                                 "msg_name": sub.event.GetMessageName(),
 254                                 "msg_id":   msgID,
 255                                 "msg_size": len(data),
 256                         }).Errorf("Unable to decode the notification message: %v", err)
 257                         continue
 258                 }
 259
 260                 // send the message into the go channel of the subscription
 261                 select {
 262                 case sub.notifChan <- event:
 263                         // message sent successfully
 264                 default:
 265                         // unable to write into the channel without blocking
 266                         log.WithFields(logger.Fields{
 267                                 "msg_name": sub.event.GetMessageName(),
 268                                 "msg_id":   msgID,
 269                                 "msg_size": len(data),
 270                         }).Warn("Unable to deliver the notification, reciever end not ready.")
 271                 }
 272
 273                 matched = true
 274         }
 275
 276         if !matched {
 277                 log.WithFields(logger.Fields{
 278                         "msg_id":   msgID,
 279                         "msg_size": len(data),
 280                 }).Info("No subscription found for the notification message.")
 281         }
 282 }
 283
 284 // +------------------+-------------------+-----------------------+
 285 // | 15b = channel ID | 1b = is multipart | 16b = sequence number |
 286 // +------------------+-------------------+-----------------------+
 287 func packRequestContext(chanID uint16, isMultipart bool, seqNum uint16) uint32 {
 288         context := uint32(chanID) << 17
 289         if isMultipart {
 290                 context |= 1 << 16
 291         }
 292         context |= uint32(seqNum)
 293         return context
 294 }
 295
 296 func unpackRequestContext(context uint32) (chanID uint16, isMulipart bool, seqNum uint16) {
 297         chanID = uint16(context >> 17)
 298         if ((context >> 16) & 0x1) != 0 {
 299                 isMulipart = true
 300         }
 301         seqNum = uint16(context & 0xffff)
 302         return
 303 }
 304
 305 // compareSeqNumbers returns -1, 0, 1 if sequence number <seqNum1> precedes, equals to,
 306 // or succeeds seq. number <seqNum2>.
 307 // Since sequence numbers cycle in the finite set of size 2^16, the function
 308 // must assume that the distance between compared sequence numbers is less than
 309 // (2^16)/2 to determine the order.
 310 func compareSeqNumbers(seqNum1, seqNum2 uint16) int {
 311         // calculate distance from seqNum1 to seqNum2
 312         var dist uint16
 313         if seqNum1 <= seqNum2 {
 314                 dist = seqNum2 - seqNum1
 315         } else {
 316                 dist = 0xffff - (seqNum1 - seqNum2 - 1)
 317         }
 318         if dist == 0 {
 319                 return 0
 320         } else if dist <= 0x8000 {
 321                 return -1
 322         }
 323         return 1
 324 }