Gomemif improvemets (#228)

* gomemif: fix the packet reader function

Packet reader function does not consider the length of
the packet array to accommodate the number of receiving
packets. It results in crash, if receiving packets are
more than packet array length. This patch fixes this
issue.

Signed-off-by: Mohsin Kazmi <sykazmi@cisco.com>

* gomemif: improve the memif interface interrupt handling

Previously, nested iterations over interfaces and queues
were used to handle receive interrupts. This patch replaces
that approach with a lookup-based solution.

Signed-off-by: Mohsin Kazmi <sykazmi@cisco.com>

* gomemif: fix packet starvation in receive queue

Previously, packets could starve in the receive queue
if `rx_burst` read fewer packets than were enqueued.
This patch ensures the queue remains interrupted,
allowing the event handler to continue processing
until all packets in the receive queue have been read.

Signed-off-by: Mohsin Kazmi <sykazmi@cisco.com>

---------

Signed-off-by: Mohsin Kazmi <sykazmi@cisco.com>

extras/gomemif/memif/control_channel.go

@@ -61,16 +61,17 @@ type controlChannel struct {
 // Socket represents a UNIX domain socket used for communication
 // between memif peers
 type Socket struct {
-	appName       string
-	filename      string
-	listener      *listener
-	interfaceList *list.List
-	ccList        *list.List
-	epfd          int
-	interruptfd   int
-	wakeEvent     syscall.EpollEvent
-	stopPollChan  chan struct{}
-	wg            sync.WaitGroup
+	appName           string
+	filename          string
+	intfQueueEventMap map[int]memifIntfQueue
+	listener          *listener
+	interfaceList     *list.List
+	ccList            *list.List
+	epfd              int
+	interruptfd       int
+	wakeEvent         syscall.EpollEvent
+	stopPollChan      chan struct{}
+	wg                sync.WaitGroup
 }
 
 type interrupt struct {
@@ -83,6 +84,11 @@ type memifInterrupt struct {
 	qid        uint16
 }
 
+type memifIntfQueue struct {
+	intf *Interface
+	qid  uint16
+}
+
 // StopPolling stops polling events on the socket
 func (socket *Socket) StopPolling() error {
 	if socket.stopPollChan != nil {
@@ -216,6 +222,7 @@ func NewSocket(appName string, filename string) (socket *Socket, err error) {
 		socket.filename = DefaultSocketFilename
 	}
 
+	socket.intfQueueEventMap = make(map[int]memifIntfQueue)
 	socket.epfd, _ = syscall.EpollCreate1(0)
 
 	efd, err := eventFd()
@@ -236,18 +243,10 @@ func (socket *Socket) handleEvent(event *syscall.EpollEvent) error {
 	if socket.listener != nil && socket.listener.event.Fd == event.Fd {
 		return socket.listener.handleEvent(event)
 	}
-	for elt := socket.interfaceList.Front(); elt != nil; elt = elt.Next() {
-		intf := elt.Value.(*Interface)
-		if intf.args.InterruptFunc != nil {
-			for rx_qid := 0; rx_qid < int(intf.GetMemoryConfig().NumQueuePairs); rx_qid++ {
-				queue, _ := intf.GetRxQueue(rx_qid)
-				interruptFd, _ := queue.GetEventFd()
-				if int(event.Fd) == interruptFd {
-					intf.onInterrupt(intf, rx_qid)
-					return nil
-				}
-			}
-		}
+
+	intfQueue, found := socket.intfQueueEventMap[int(event.Fd)]
+	if found {
+		return intfQueue.intf.onInterrupt(intfQueue.intf, int(intfQueue.qid))
 	}
 
 	for elt := socket.ccList.Front(); elt != nil; elt = elt.Next() {

extras/gomemif/memif/interface.go

@@ -525,7 +525,7 @@ func (i *Interface) connect() (err error) {
 		q.lastTail = 0
 	}
 
-	for _, q := range i.rxQueues {
+	for qid, q := range i.rxQueues {
 		q.updateRing()
 
 		if q.ring.getCookie() != cookie {
@@ -534,6 +534,11 @@ func (i *Interface) connect() (err error) {
 
 		q.lastHead = 0
 		q.lastTail = 0
+
+		if i.args.InterruptFunc != nil {
+			interruptFd, _ := q.GetEventFd()
+			i.socket.intfQueueEventMap[interruptFd] = memifIntfQueue{i, uint16(qid)}
+		}
 	}
 
 	return i.args.ConnectedFunc(i)

extras/gomemif/memif/packet_reader.go

@@ -107,6 +107,7 @@ func (q *Queue) Rx_burst(pkt []MemifPacketBuffer) (uint16, error) {
 	var length uint32
 	var offset uint32
 	var nSlots uint16
+	var readQueueInterrupt bool = true
 	var desc descBuf = newDescBuf()
 
 	if q.i.args.IsMaster {
@@ -124,6 +125,11 @@ func (q *Queue) Rx_burst(pkt []MemifPacketBuffer) (uint16, error) {
 		return 0, nil
 	}
 
+	if nSlots > uint16(len(pkt)) {
+		nSlots = uint16(len(pkt))
+		readQueueInterrupt = false
+	}
+
 	rx := 0
 	for nSlots > 0 {
 		// copy descriptor from shm
@@ -144,8 +150,10 @@ func (q *Queue) Rx_burst(pkt []MemifPacketBuffer) (uint16, error) {
 
 	}
 
-	b := make([]byte, 8)
-	syscall.Read(int(q.interruptFd), b)
+	if readQueueInterrupt {
+		b := make([]byte, 8)
+		syscall.Read(int(q.interruptFd), b)
+	}
 
 	return uint16(rx), nil
 }