msg concurrent refacted

dtmsvr/trans_type_msg.go

@@ -74,68 +74,34 @@ func (t *transMsgProcessor) ProcessOnce(branches []TransBranch) error {
 		t.touchCronTime(cronKeep, cmc.Delay)
 		return nil
 	}
-	execBranch := func(current int) (bool, error) {
+	var started int
-		branch := &branches[current]
+	resultsChan := make(chan error, len(branches))
-		if branch.Op != dtmcli.BranchAction || branch.Status != dtmcli.StatusPrepared {
+	var err error
-			return true, nil
+	for i, _ := range branches {
-		}
+		b := &branches[i]
-		err := t.execBranch(branch, current)
+		if b.Op != dtmcli.BranchAction || b.Status != dtmcli.StatusPrepared {
-		if err != nil {
+			continue
-			if !errors.Is(err, dtmcli.ErrOngoing) {
-				logger.Errorf("exec branch error: %v", err)
-			}
-			return false, err
-		}
-		if branch.Status != dtmcli.StatusSucceed {
-			return false, nil
-		}
-		return true, nil
-	}
-	type branchResult struct {
-		success bool
-		err     error
-	}
-	waitChan := make(chan branchResult, len(branches))
-	consumeWork := func(i int) error {
-		success, err := execBranch(i)
-		waitChan <- branchResult{
-			success: success,
-			err:     err,
 		}
-		return err
+		if t.Concurrent {
-	}
+			started++
-	produceWork := func() {
+			go func(pos int) {
-		for i := 0; i < len(branches); i++ {
+				resultsChan <- t.execBranch(b, pos)
-			if t.Concurrent {
+			}(i)
-				go func(i int) {
+		} else {
-					_ = consumeWork(i)
+			err = t.execBranch(b, i)
-				}(i)
-				continue
-			}
-			err := consumeWork(i)
 			if err != nil {
-				return
+				break
 			}
 		}
 	}
-	go produceWork()
+	for i := 0; i < started && err == nil; i++ {
-	successCnt := 0
+		err = <-resultsChan
-	var err error
-	for i := 0; i < len(branches); i++ {
-		result := <-waitChan
-		if result.err != nil {
-			err = result.err
-			if !t.Concurrent {
-				return err
-			}
-		}
-		if result.success {
-			successCnt++
-		}
 	}
-	if successCnt == len(branches) { // msg 事务完成
+	if err == dtmcli.ErrOngoing {
-		t.changeStatus(dtmcli.StatusSucceed)
 		return nil
+	} else if err != nil {
+		return err
 	}
-	panic("msg go pass all branch")
+	t.changeStatus(dtmcli.StatusSucceed)
+	return nil
 }