From a2625752e4c2eb49af81874dd1d78da239e453c2 Mon Sep 17 00:00:00 2001
From: Sydonian <794346190@qq.com>
Date: Tue, 5 Nov 2024 09:57:34 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E9=94=99=E8=AF=AF=E6=97=A5?=
 =?UTF-8?q?=E5=BF=97?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 pkgs/ioswitch/exec/executor.go | 46 ++++++++++++++++++++++++----------
 1 file changed, 33 insertions(+), 13 deletions(-)

diff --git a/pkgs/ioswitch/exec/executor.go b/pkgs/ioswitch/exec/executor.go
index 7f06cad..c2c17a3 100644
--- a/pkgs/ioswitch/exec/executor.go
+++ b/pkgs/ioswitch/exec/executor.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/hashicorp/go-multierror"
 	"gitlink.org.cn/cloudream/common/pkgs/future"
 	"gitlink.org.cn/cloudream/common/utils/lo2"
 	"gitlink.org.cn/cloudream/common/utils/sync2"
@@ -48,19 +49,7 @@ func (s *Executor) Run(ctx *ExecContext) (map[string]VarValue, error) {
 
 	defer cancel()
 
-	err := sync2.ParallelDo(s.plan.Ops, func(o Op, idx int) error {
-		err := o.Execute(ctx, s)
-
-		s.lock.Lock()
-		defer s.lock.Unlock()
-
-		if err != nil {
-			cancel()
-			return fmt.Errorf("%T: %w", o, err)
-		}
-
-		return nil
-	})
+	err := s.runOps(s.plan.Ops, ctx, cancel)
 	if err != nil {
 		return nil, err
 	}
@@ -68,6 +57,37 @@ func (s *Executor) Run(ctx *ExecContext) (map[string]VarValue, error) {
 	return s.store, nil
 }
 
+func (s *Executor) runOps(ops []Op, ctx *ExecContext, cancel context.CancelFunc) error {
+	lock := sync.Mutex{}
+	var err error
+
+	var wg sync.WaitGroup
+	wg.Add(len(ops))
+	for i, arg := range ops {
+		go func(arg Op, index int) {
+			defer wg.Done()
+
+			if e := arg.Execute(ctx, s); e != nil {
+				lock.Lock()
+				// 尽量不记录 ErrContextCanceled 错误，除非没有其他错误
+				if err == nil {
+					err = e
+				} else if err == sync2.ErrContextCanceled {
+					err = e
+				} else if e != sync2.ErrContextCanceled {
+					err = multierror.Append(err, e)
+				}
+				lock.Unlock()
+
+				cancel()
+			}
+		}(arg, i)
+	}
+	wg.Wait()
+
+	return err
+}
+
 func (s *Executor) BindVar(ctx context.Context, id VarID) (VarValue, error) {
 	s.lock.Lock()