Bugfix: wait for processing all message before complete

Signed-off-by: Weiqiang Tang <[email protected]>

ofctrl/transaction.go

@@ -6,6 +6,7 @@ package ofctrl
 
 import (
 	"fmt"
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -118,28 +119,22 @@ func (tx *Transaction) createBundleAddFlowMessage(flowMod *openflow13.FlowMod) (
 }
 
 func (tx *Transaction) listenReply() {
-	for {
-		select {
-		case reply, ok := <-tx.controlReplyCh:
-			if !ok { // controlReplyCh closed.
-				return
+	for reply := range tx.controlReplyCh {
+		switch reply.msgType {
+		case BundleControlMessage:
+			select {
+			case tx.controlIntCh <- reply:
+			case <-time.After(messageTimeout):
+				log.Warningln("BundleControlMessage reply message accept timeout")
 			}
-			switch reply.msgType {
-			case BundleControlMessage:
-				select {
-				case tx.controlIntCh <- reply:
-				case <-time.After(messageTimeout):
-					log.Warningln("BundleControlMessage reply message accept timeout")
-				}
-			case BundleAddMessage:
-				if !reply.succeed {
-					func() {
-						tx.lock.Lock()
-						defer tx.lock.Unlock()
-						// Remove failed add message from successAdd.
-						delete(tx.successAdd, reply.xID)
-					}()
-				}
+		case BundleAddMessage:
+			if !reply.succeed {
+				func() {
+					tx.lock.Lock()
+					defer tx.lock.Unlock()
+					// Remove failed add message from successAdd.
+					delete(tx.successAdd, reply.xID)
+				}()
 			}
 		}
 	}
@@ -194,6 +189,9 @@ func (tx *Transaction) Complete() (int, error) {
 		}
 		tx.closed = true
 	}
+	for len(tx.controlReplyCh) > 0 {
+		runtime.Gosched()
+	}
 	tx.lock.Lock()
 	defer tx.lock.Unlock()
 	return len(tx.successAdd), nil