From a101d966133a70b8a76526be25070436d14fcf9f Mon Sep 17 00:00:00 2001 From: Rastislav Szabo Date: Thu, 4 May 2017 11:09:03 +0200 Subject: initial commit Signed-off-by: Rastislav Szabo --- core/core.go | 340 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 340 insertions(+) create mode 100644 core/core.go (limited to 'core/core.go') diff --git a/core/core.go b/core/core.go new file mode 100644 index 0000000..e11a30f --- /dev/null +++ b/core/core.go @@ -0,0 +1,340 @@ +// Copyright (c) 2017 Cisco and/or its affiliates. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at: +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package core + +//go:generate binapi_generator --input-file=/usr/share/vpp/api/vpe.api.json --output-dir=./bin_api + +import ( + "errors" + "fmt" + "os" + "sync" + "sync/atomic" + + logger "github.com/Sirupsen/logrus" + + "gerrit.fd.io/r/govpp/adapter" + "gerrit.fd.io/r/govpp/api" + "gerrit.fd.io/r/govpp/core/bin_api/vpe" +) + +const ( + requestChannelBufSize = 100 // default size of the request channel buffers + replyChannelBufSize = 100 // default size of the reply channel buffers +) + +// Connection represents a shared memory connection to VPP via vppAdapter. +type Connection struct { + vpp adapter.VppAdapter // VPP adapter + codec *MsgCodec // message codec + + msgIDs map[string]uint16 // map os message IDs indexed by message name + CRC + msgIDsLock sync.RWMutex // lock for the message IDs map + + channels map[uint32]*api.Channel // map of all API channels indexed by the channel ID + channelsLock sync.RWMutex // lock for the channels map + + notifSubscriptions map[uint16][]*api.NotifSubscription // map od all notification subscriptions indexed by message ID + notifSubscriptionsLock sync.RWMutex // lock for the subscriptions map + + maxChannelID uint32 // maximum used client ID + pingReqID uint16 // ID if the ControlPing message + pingReplyID uint16 // ID of the ControlPingReply message +} + +// channelMetadata contains core-local metadata of an API channel. +type channelMetadata struct { + id uint32 // channel ID + multipart uint32 // 1 if multipart request is being processed, 0 otherwise +} + +var ( + log *logger.Logger // global logger + conn *Connection // global handle to the Connection (used in the message receive callback) + connLock sync.RWMutex // lock for the global connection +) + +// init initializes global logger, which logs debug level messages to stdout. +func init() { + log = logger.New() + log.Out = os.Stdout + log.Level = logger.DebugLevel +} + +// SetLogger sets global logger to provided one. +func SetLogger(l *logger.Logger) { + log = l +} + +// Connect connects to VPP using specified VPP adapter and returns the connection handle. +func Connect(vppAdapter adapter.VppAdapter) (*Connection, error) { + connLock.Lock() + defer connLock.Unlock() + + if conn != nil { + return nil, errors.New("only one connection per process is supported") + } + + conn = &Connection{vpp: vppAdapter, codec: &MsgCodec{}} + conn.channels = make(map[uint32]*api.Channel) + conn.msgIDs = make(map[string]uint16) + conn.notifSubscriptions = make(map[uint16][]*api.NotifSubscription) + + conn.vpp.SetMsgCallback(msgCallback) + + logger.Debug("Connecting to VPP...") + + err := conn.vpp.Connect() + if err != nil { + return nil, err + } + + // store control ping IDs + conn.pingReqID, _ = conn.GetMessageID(&vpe.ControlPing{}) + conn.pingReplyID, _ = conn.GetMessageID(&vpe.ControlPingReply{}) + + logger.Debug("VPP connected.") + + return conn, nil +} + +// Disconnect disconnects from VPP. +func (c *Connection) Disconnect() { + connLock.Lock() + defer connLock.Unlock() + + if c != nil && c.vpp != nil { + c.vpp.Disconnect() + } + conn = nil +} + +// NewAPIChannel returns a new API channel for communication with VPP via govpp core. +// It uses default buffer sizes for the request and reply Go channels. +func (c *Connection) NewAPIChannel() (*api.Channel, error) { + return c.NewAPIChannelBuffered(requestChannelBufSize, replyChannelBufSize) +} + +// NewAPIChannelBuffered returns a new API channel for communication with VPP via govpp core. +// It allows to specify custom buffer sizes for the request and reply Go channels. +func (c *Connection) NewAPIChannelBuffered(reqChanBufSize, replyChanBufSize int) (*api.Channel, error) { + chID := atomic.AddUint32(&c.maxChannelID, 1) + chMeta := &channelMetadata{id: chID} + + ch := api.NewChannelInternal(chMeta) + ch.MsgDecoder = c.codec + ch.MsgIdentifier = c + + // create the communication channels + ch.ReqChan = make(chan *api.VppRequest, reqChanBufSize) + ch.ReplyChan = make(chan *api.VppReply, replyChanBufSize) + ch.NotifSubsChan = make(chan *api.NotifSubscribeRequest, reqChanBufSize) + ch.NotifSubsReplyChan = make(chan error, replyChanBufSize) + + // store API channel within the client + c.channelsLock.Lock() + c.channels[chID] = ch + c.channelsLock.Unlock() + + // start watching on the request channel + go c.watchRequests(ch, chMeta) + + return ch, nil +} + +// watchRequests watches for requests on the request API channel and forwards them as messages to VPP. +func (c *Connection) watchRequests(ch *api.Channel, chMeta *channelMetadata) { + for { + select { + case req, ok := <-ch.ReqChan: + // new request on the request channel + if !ok { + // after closing the request channel, release API channel and return + c.releaseAPIChannel(ch, chMeta) + return + } + c.processRequest(ch, chMeta, req) + + case req := <-ch.NotifSubsChan: + // new request on the notification subscribe channel + c.processNotifSubscribeRequest(ch, req) + } + } +} + +// processRequest processes a single request received on the request channel. +func (c *Connection) processRequest(ch *api.Channel, chMeta *channelMetadata, req *api.VppRequest) error { + // retrieve message ID + msgID, err := c.GetMessageID(req.Message) + if err != nil { + error := fmt.Errorf("unable to retrieve message ID: %v", err) + log.WithFields(logger.Fields{ + "msg_name": req.Message.GetMessageName(), + "msg_crc": req.Message.GetCrcString(), + }).Errorf("unable to retrieve message ID: %v", err) + sendReply(ch, &api.VppReply{Error: error}) + return error + } + + // encode the message into binary + data, err := c.codec.EncodeMsg(req.Message, msgID) + if err != nil { + error := fmt.Errorf("unable to encode the messge: %v", err) + log.WithFields(logger.Fields{ + "context": chMeta.id, + "msg_id": msgID, + }).Errorf("%v", error) + sendReply(ch, &api.VppReply{Error: error}) + return error + } + + // send the message + log.WithFields(logger.Fields{ + "context": chMeta.id, + "msg_id": msgID, + "msg_size": len(data), + }).Debug("Sending a message to VPP.") + + c.vpp.SendMsg(chMeta.id, data) + + if req.Multipart { + // multipart request + atomic.StoreUint32(&chMeta.multipart, 1) + + // send a control ping + ping := &vpe.ControlPing{} + pingData, _ := c.codec.EncodeMsg(ping, c.pingReqID) + + log.WithFields(logger.Fields{ + "context": chMeta.id, + "msg_id": c.pingReqID, + "msg_size": len(pingData), + }).Debug("Sending a control ping to VPP.") + + c.vpp.SendMsg(chMeta.id, pingData) + } + + return nil +} + +// releaseAPIChannel releases API channel that needs to be closed. +func (c *Connection) releaseAPIChannel(ch *api.Channel, chMeta *channelMetadata) { + log.WithFields(logger.Fields{ + "context": chMeta.id, + }).Debug("API channel closed.") + + // delete the channel from channels map + c.channelsLock.Lock() + delete(c.channels, chMeta.id) + c.channelsLock.Unlock() +} + +// msgCallback is called whenever any binary API message comes from VPP. +func msgCallback(context uint32, msgID uint16, data []byte) { + connLock.RLock() + defer connLock.RUnlock() + + if conn == nil { + log.Warn("Already disconnected, ignoring the message.") + return + } + + log.WithFields(logger.Fields{ + "context": context, + "msg_id": msgID, + "msg_size": len(data), + }).Debug("Received a message from VPP.") + + if context == 0 || conn.isNotificationMessage(msgID) { + // process the message as a notification + conn.sendNotifications(msgID, data) + return + } + + // match ch according to the context + conn.channelsLock.RLock() + ch, ok := conn.channels[context] + conn.channelsLock.RUnlock() + + if !ok { + log.WithFields(logger.Fields{ + "context": context, + "msg_id": msgID, + }).Error("Context ID not known, ignoring the message.") + return + } + + chMeta := ch.Metadata().(*channelMetadata) + lastReplyReceived := false + // if this is a control ping reply and multipart request is being processed, treat this as a last part of the reply + if msgID == conn.pingReplyID && atomic.CompareAndSwapUint32(&chMeta.multipart, 1, 0) { + lastReplyReceived = true + } + + // send the data to the channel + sendReply(ch, &api.VppReply{ + MessageID: msgID, + Data: data, + LastReplyReceived: lastReplyReceived, + }) +} + +// sendReply sends the reply into the go channel, if it cannot be completed without blocking, otherwise +// it logs the error and do not send the message. +func sendReply(ch *api.Channel, reply *api.VppReply) { + select { + case ch.ReplyChan <- reply: + // reply sent successfully + default: + // unable to write into the channel without blocking + log.WithFields(logger.Fields{ + "channel": ch, + "msg_id": reply.MessageID, + }).Warn("Unable to send the reply, reciever end not ready.") + } +} + +// GetMessageID returns message identifier of given API message. +func (c *Connection) GetMessageID(msg api.Message) (uint16, error) { + return c.messageNameToID(msg.GetMessageName(), msg.GetCrcString()) +} + +// messageNameToID returns message ID of a message identified by its name and CRC. +func (c *Connection) messageNameToID(msgName string, msgCrc string) (uint16, error) { + // try to get the ID from the map + c.msgIDsLock.RLock() + id, ok := c.msgIDs[msgName+msgCrc] + c.msgIDsLock.RUnlock() + if ok { + return id, nil + } + + // get the ID using VPP API + id, err := c.vpp.GetMsgID(msgName, msgCrc) + if err != nil { + error := fmt.Errorf("unable to retrieve message ID: %v", err) + log.WithFields(logger.Fields{ + "msg_name": msgName, + "msg_crc": msgCrc, + }).Errorf("unable to retrieve message ID: %v", err) + return id, error + } + + c.msgIDsLock.Lock() + c.msgIDs[msgName+msgCrc] = id + c.msgIDsLock.Unlock() + + return id, nil +} -- cgit 1.2.3-korg