Pair requests with replies using sequence numbers

Requests are given sequence numbers (cycling over a finite set of 2^16
integers) that are stored into the lower 16bits of the context.
1bit is also allocated for isMultipart boolean flag and the remaining
15bits are used to store the channel ID. The sequence numbers allow
to reliably pair replies with requests, even in scenarious with timeouted
requests or ignored (unread) replies.

Sequencing is not used with asynchronous messaging as it is implemented
by methods of the Channel structure, i.e. above ReqChan and ReplyChan
channels.

Change-Id: I7ca0e8489c7ffcc388c3cfef6d05c02f9500931c
Signed-off-by: Milan Lenco <milan.lenco@pantheon.tech>

1 files changed, 9 insertions, 16 deletions

diff --git a/core/core.go b/core/core.go
index 4782ba1..052eb0b 100644
--- a/core/core.go
+++ b/core/core.go
@@ -77,12 +77,12 @@ type Connection struct {
 	msgIDs     map[string]uint16 // map of message IDs indexed by message name + CRC
 
 	channelsLock sync.RWMutex            // lock for the channels map
-	channels     map[uint32]*api.Channel // map of all API channels indexed by the channel ID
+	channels     map[uint16]*api.Channel // map of all API channels indexed by the channel ID
 
 	notifSubscriptionsLock sync.RWMutex                        // lock for the subscriptions map
 	notifSubscriptions     map[uint16][]*api.NotifSubscription // map od all notification subscriptions indexed by message ID
 
-	maxChannelID uint32 // maximum used client ID
+	maxChannelID uint32 // maximum used channel ID (the real limit is 2^15, 32-bit is used for atomic operations)
 	pingReqID    uint16 // ID if the ControlPing message
 	pingReplyID  uint16 // ID of the ControlPingReply message
 
@@ -90,12 +90,6 @@ type Connection struct {
 	lastReply     time.Time  // time of the last received reply from VPP
 }
 
-// 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)
@@ -204,7 +198,7 @@ func newConnection(vppAdapter adapter.VppAdapter) (*Connection, error) {
 	conn = &Connection{
 		vpp:                vppAdapter,
 		codec:              &MsgCodec{},
-		channels:           make(map[uint32]*api.Channel),
+		channels:           make(map[uint16]*api.Channel),
 		msgIDs:             make(map[string]uint16),
 		notifSubscriptions: make(map[uint16][]*api.NotifSubscription),
 	}
@@ -370,10 +364,9 @@ func (c *Connection) NewAPIChannelBuffered(reqChanBufSize, replyChanBufSize int)
 	if c == nil {
 		return nil, errors.New("nil connection passed in")
 	}
-	chID := atomic.AddUint32(&c.maxChannelID, 1)
-	chMeta := &channelMetadata{id: chID}
 
-	ch := api.NewChannelInternal(chMeta)
+	chID := uint16(atomic.AddUint32(&c.maxChannelID, 1) & 0x7fff)
+	ch := api.NewChannelInternal(chID)
 	ch.MsgDecoder = c.codec
 	ch.MsgIdentifier = c
 
@@ -389,19 +382,19 @@ func (c *Connection) NewAPIChannelBuffered(reqChanBufSize, replyChanBufSize int)
 	c.channelsLock.Unlock()
 
 	// start watching on the request channel
-	go c.watchRequests(ch, chMeta)
+	go c.watchRequests(ch)
 
 	return ch, nil
 }
 
 // releaseAPIChannel releases API channel that needs to be closed.
-func (c *Connection) releaseAPIChannel(ch *api.Channel, chMeta *channelMetadata) {
+func (c *Connection) releaseAPIChannel(ch *api.Channel) {
 	log.WithFields(logger.Fields{
-		"context": chMeta.id,
+		"ID": ch.ID,
 	}).Debug("API channel closed.")
 
 	// delete the channel from channels map
 	c.channelsLock.Lock()
-	delete(c.channels, chMeta.id)
+	delete(c.channels, ch.ID)
 	c.channelsLock.Unlock()
 }