core/core.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 //go:generate binapi-generator --input-dir=bin_api --output-dir=bin_api
  16
  17 package core
  18
  19 import (
  20         "errors"
  21         "os"
  22         "sync"
  23         "sync/atomic"
  24         "time"
  25
  26         logger "github.com/sirupsen/logrus"
  27
  28         "git.fd.io/govpp.git/adapter"
  29         "git.fd.io/govpp.git/api"
  30         "git.fd.io/govpp.git/core/bin_api/vpe"
  31 )
  32
  33 var (
  34         msgControlPing      api.Message = &vpe.ControlPing{}
  35         msgControlPingReply api.Message = &vpe.ControlPingReply{}
  36 )
  37
  38 const (
  39         requestChannelBufSize      = 100 // default size of the request channel buffers
  40         replyChannelBufSize        = 100 // default size of the reply channel buffers
  41         notificationChannelBufSize = 100 // default size of the notification channel buffers
  42 )
  43
  44 var (
  45         healthCheckProbeInterval = time.Second * 1        // default health check probe interval
  46         healthCheckReplyTimeout  = time.Millisecond * 100 // timeout for reply to a health check probe
  47         healthCheckThreshold     = 1                      // number of failed healthProbe until the error is reported
  48 )
  49
  50 // ConnectionState holds the current state of the connection to VPP.
  51 type ConnectionState int
  52
  53 const (
  54         // Connected connection state means that the connection to VPP has been successfully established.
  55         Connected ConnectionState = iota
  56
  57         // Disconnected connection state means that the connection to VPP has been lost.
  58         Disconnected
  59 )
  60
  61 // ConnectionEvent is a notification about change in the VPP connection state.
  62 type ConnectionEvent struct {
  63         // Timestamp holds the time when the event has been generated.
  64         Timestamp time.Time
  65
  66         // State holds the new state of the connection to VPP at the time when the event has been generated.
  67         State ConnectionState
  68 }
  69
  70 // Connection represents a shared memory connection to VPP via vppAdapter.
  71 type Connection struct {
  72         vpp       adapter.VppAdapter // VPP adapter
  73         connected uint32             // non-zero if the adapter is connected to VPP
  74         codec     *MsgCodec          // message codec
  75
  76         msgIDsLock sync.RWMutex      // lock for the message IDs map
  77         msgIDs     map[string]uint16 // map of message IDs indexed by message name + CRC
  78
  79         channelsLock sync.RWMutex            // lock for the channels map
  80         channels     map[uint16]*api.Channel // map of all API channels indexed by the channel ID
  81
  82         notifSubscriptionsLock sync.RWMutex                        // lock for the subscriptions map
  83         notifSubscriptions     map[uint16][]*api.NotifSubscription // map od all notification subscriptions indexed by message ID
  84
  85         maxChannelID uint32 // maximum used channel ID (the real limit is 2^15, 32-bit is used for atomic operations)
  86         pingReqID    uint16 // ID if the ControlPing message
  87         pingReplyID  uint16 // ID of the ControlPingReply message
  88
  89         lastReplyLock sync.Mutex // lock for the last reply
  90         lastReply     time.Time  // time of the last received reply from VPP
  91 }
  92
  93 var (
  94         log      *logger.Logger // global logger
  95         conn     *Connection    // global handle to the Connection (used in the message receive callback)
  96         connLock sync.RWMutex   // lock for the global connection
  97 )
  98
  99 // init initializes global logger, which logs debug level messages to stdout.
 100 func init() {
 101         log = logger.New()
 102         log.Out = os.Stdout
 103         log.Level = logger.DebugLevel
 104 }
 105
 106 // SetLogger sets global logger to provided one.
 107 func SetLogger(l *logger.Logger) {
 108         log = l
 109 }
 110
 111 // SetHealthCheckProbeInterval sets health check probe interval.
 112 // Beware: Function is not thread-safe. It is recommended to setup this parameter
 113 // before connecting to vpp.
 114 func SetHealthCheckProbeInterval(interval time.Duration) {
 115         healthCheckProbeInterval = interval
 116 }
 117
 118 // SetHealthCheckReplyTimeout sets timeout for reply to a health check probe.
 119 // If reply arrives after the timeout, check is considered as failed.
 120 // Beware: Function is not thread-safe. It is recommended to setup this parameter
 121 // before connecting to vpp.
 122 func SetHealthCheckReplyTimeout(timeout time.Duration) {
 123         healthCheckReplyTimeout = timeout
 124 }
 125
 126 // SetHealthCheckThreshold sets the number of failed healthProbe checks until the error is reported.
 127 // Beware: Function is not thread-safe. It is recommended to setup this parameter
 128 // before connecting to vpp.
 129 func SetHealthCheckThreshold(threshold int) {
 130         healthCheckThreshold = threshold
 131 }
 132
 133 // SetControlPingMessages sets the messages for ControlPing and ControlPingReply
 134 func SetControlPingMessages(controPing, controlPingReply api.Message) {
 135         msgControlPing = controPing
 136         msgControlPingReply = controlPingReply
 137 }
 138
 139 // Connect connects to VPP using specified VPP adapter and returns the connection handle.
 140 // This call blocks until VPP is connected, or an error occurs. Only one connection attempt will be performed.
 141 func Connect(vppAdapter adapter.VppAdapter) (*Connection, error) {
 142         // create new connection handle
 143         c, err := newConnection(vppAdapter)
 144         if err != nil {
 145                 return nil, err
 146         }
 147
 148         // blocking attempt to connect to VPP
 149         err = c.connectVPP()
 150         if err != nil {
 151                 return nil, err
 152         }
 153
 154         return conn, nil
 155 }
 156
 157 // AsyncConnect asynchronously connects to VPP using specified VPP adapter and returns the connection handle
 158 // and ConnectionState channel. This call does not block until connection is established, it
 159 // returns immediately. The caller is supposed to watch the returned ConnectionState channel for
 160 // Connected/Disconnected events. In case of disconnect, the library will asynchronously try to reconnect.
 161 func AsyncConnect(vppAdapter adapter.VppAdapter) (*Connection, chan ConnectionEvent, error) {
 162         // create new connection handle
 163         c, err := newConnection(vppAdapter)
 164         if err != nil {
 165                 return nil, nil, err
 166         }
 167
 168         // asynchronously attempt to connect to VPP
 169         connChan := make(chan ConnectionEvent, notificationChannelBufSize)
 170         go c.connectLoop(connChan)
 171
 172         return conn, connChan, nil
 173 }
 174
 175 // Disconnect disconnects from VPP and releases all connection-related resources.
 176 func (c *Connection) Disconnect() {
 177         if c == nil {
 178                 return
 179         }
 180         connLock.Lock()
 181         defer connLock.Unlock()
 182
 183         if c != nil && c.vpp != nil {
 184                 c.disconnectVPP()
 185         }
 186         conn = nil
 187 }
 188
 189 // newConnection returns new connection handle.
 190 func newConnection(vppAdapter adapter.VppAdapter) (*Connection, error) {
 191         connLock.Lock()
 192         defer connLock.Unlock()
 193
 194         if conn != nil {
 195                 return nil, errors.New("only one connection per process is supported")
 196         }
 197
 198         conn = &Connection{
 199                 vpp:                vppAdapter,
 200                 codec:              &MsgCodec{},
 201                 channels:           make(map[uint16]*api.Channel),
 202                 msgIDs:             make(map[string]uint16),
 203                 notifSubscriptions: make(map[uint16][]*api.NotifSubscription),
 204         }
 205
 206         conn.vpp.SetMsgCallback(msgCallback)
 207         return conn, nil
 208 }
 209
 210 // connectVPP performs one blocking attempt to connect to VPP.
 211 func (c *Connection) connectVPP() error {
 212         log.Debug("Connecting to VPP...")
 213
 214         // blocking connect
 215         err := c.vpp.Connect()
 216         if err != nil {
 217                 log.Warn(err)
 218                 return err
 219         }
 220
 221         // store control ping IDs
 222         if c.pingReqID, err = c.GetMessageID(msgControlPing); err != nil {
 223                 c.vpp.Disconnect()
 224                 return err
 225         }
 226         if c.pingReplyID, err = c.GetMessageID(msgControlPingReply); err != nil {
 227                 c.vpp.Disconnect()
 228                 return err
 229         }
 230
 231         // store connected state
 232         atomic.StoreUint32(&c.connected, 1)
 233
 234         log.Info("Connected to VPP.")
 235         return nil
 236 }
 237
 238 // disconnectVPP disconnects from VPP in case it is connected.
 239 func (c *Connection) disconnectVPP() {
 240         if atomic.CompareAndSwapUint32(&c.connected, 1, 0) {
 241                 c.vpp.Disconnect()
 242         }
 243 }
 244
 245 // connectLoop attempts to connect to VPP until it succeeds.
 246 // Then it continues with healthCheckLoop.
 247 func (c *Connection) connectLoop(connChan chan ConnectionEvent) {
 248         // loop until connected
 249         for {
 250                 if err := c.vpp.WaitReady(); err != nil {
 251                         log.Warnf("wait ready failed: %v", err)
 252                 }
 253                 if err := c.connectVPP(); err == nil {
 254                         // signal connected event
 255                         connChan <- ConnectionEvent{Timestamp: time.Now(), State: Connected}
 256                         break
 257                 } else {
 258                         log.Errorf("connecting to VPP failed: %v", err)
 259                         time.Sleep(time.Second)
 260                 }
 261         }
 262
 263         // we are now connected, continue with health check loop
 264         c.healthCheckLoop(connChan)
 265 }
 266
 267 // healthCheckLoop checks whether connection to VPP is alive. In case of disconnect,
 268 // it continues with connectLoop and tries to reconnect.
 269 func (c *Connection) healthCheckLoop(connChan chan ConnectionEvent) {
 270         // create a separate API channel for health check probes
 271         ch, err := conn.NewAPIChannelBuffered(1, 1)
 272         if err != nil {
 273                 log.Error("Failed to create health check API channel, health check will be disabled:", err)
 274                 return
 275         }
 276
 277         var sinceLastReply time.Duration
 278         var failedChecks int
 279
 280         // send health check probes until an error or timeout occurs
 281         for {
 282                 // sleep until next health check probe period
 283                 time.Sleep(healthCheckProbeInterval)
 284
 285                 if atomic.LoadUint32(&c.connected) == 0 {
 286                         // Disconnect has been called in the meantime, return the healthcheck - reconnect loop
 287                         log.Debug("Disconnected on request, exiting health check loop.")
 288                         return
 289                 }
 290
 291                 // try draining probe replies from previous request before sending next one
 292                 select {
 293                 case <-ch.ReplyChan:
 294                         log.Debug("drained old probe reply from reply channel")
 295                 default:
 296                 }
 297
 298                 // send the control ping request
 299                 ch.ReqChan <- &api.VppRequest{Message: msgControlPing}
 300
 301                 for {
 302                         // expect response within timeout period
 303                         select {
 304                         case vppReply := <-ch.ReplyChan:
 305                                 err = vppReply.Error
 306
 307                         case <-time.After(healthCheckReplyTimeout):
 308                                 err = ErrProbeTimeout
 309
 310                                 // check if time since last reply from any other
 311                                 // channel is less than health check reply timeout
 312                                 conn.lastReplyLock.Lock()
 313                                 sinceLastReply = time.Since(c.lastReply)
 314                                 conn.lastReplyLock.Unlock()
 315
 316                                 if sinceLastReply < healthCheckReplyTimeout {
 317                                         log.Warnf("VPP health check probe timing out, but some request on other channel was received %v ago, continue waiting!", sinceLastReply)
 318                                         continue
 319                                 }
 320                         }
 321                         break
 322                 }
 323
 324                 if err == ErrProbeTimeout {
 325                         failedChecks++
 326                         log.Warnf("VPP health check probe timed out after %v (%d. timeout)", healthCheckReplyTimeout, failedChecks)
 327                         if failedChecks > healthCheckThreshold {
 328                                 // in case of exceeded treshold disconnect
 329                                 log.Errorf("VPP health check exceeded treshold for timeouts (>%d), assuming disconnect", healthCheckThreshold)
 330                                 connChan <- ConnectionEvent{Timestamp: time.Now(), State: Disconnected}
 331                                 break
 332                         }
 333                 } else if err != nil {
 334                         // in case of error disconnect
 335                         log.Errorf("VPP health check probe failed: %v", err)
 336                         connChan <- ConnectionEvent{Timestamp: time.Now(), State: Disconnected}
 337                         break
 338                 } else if failedChecks > 0 {
 339                         failedChecks = 0
 340                         log.Infof("VPP health check probe OK")
 341                 }
 342         }
 343
 344         // cleanup
 345         ch.Close()
 346         c.disconnectVPP()
 347
 348         // we are now disconnected, start connect loop
 349         c.connectLoop(connChan)
 350 }
 351
 352 // NewAPIChannel returns a new API channel for communication with VPP via govpp core.
 353 // It uses default buffer sizes for the request and reply Go channels.
 354 func (c *Connection) NewAPIChannel() (*api.Channel, error) {
 355         if c == nil {
 356                 return nil, errors.New("nil connection passed in")
 357         }
 358         return c.NewAPIChannelBuffered(requestChannelBufSize, replyChannelBufSize)
 359 }
 360
 361 // NewAPIChannelBuffered returns a new API channel for communication with VPP via govpp core.
 362 // It allows to specify custom buffer sizes for the request and reply Go channels.
 363 func (c *Connection) NewAPIChannelBuffered(reqChanBufSize, replyChanBufSize int) (*api.Channel, error) {
 364         if c == nil {
 365                 return nil, errors.New("nil connection passed in")
 366         }
 367
 368         chID := uint16(atomic.AddUint32(&c.maxChannelID, 1) & 0x7fff)
 369         ch := api.NewChannelInternal(chID)
 370         ch.MsgDecoder = c.codec
 371         ch.MsgIdentifier = c
 372
 373         // create the communication channels
 374         ch.ReqChan = make(chan *api.VppRequest, reqChanBufSize)
 375         ch.ReplyChan = make(chan *api.VppReply, replyChanBufSize)
 376         ch.NotifSubsChan = make(chan *api.NotifSubscribeRequest, reqChanBufSize)
 377         ch.NotifSubsReplyChan = make(chan error, replyChanBufSize)
 378
 379         // store API channel within the client
 380         c.channelsLock.Lock()
 381         c.channels[chID] = ch
 382         c.channelsLock.Unlock()
 383
 384         // start watching on the request channel
 385         go c.watchRequests(ch)
 386
 387         return ch, nil
 388 }
 389
 390 // releaseAPIChannel releases API channel that needs to be closed.
 391 func (c *Connection) releaseAPIChannel(ch *api.Channel) {
 392         log.WithFields(logger.Fields{
 393                 "ID": ch.ID,
 394         }).Debug("API channel closed.")
 395
 396         // delete the channel from channels map
 397         c.channelsLock.Lock()
 398         delete(c.channels, ch.ID)
 399         c.channelsLock.Unlock()
 400 }