optimize(component):remove useless chan

actuator_test.go

@@ -38,3 +38,10 @@ func TestPanic(t *testing.T) {
 	c := NewActuator()
 	testPanic(t, c)
 }
+
+func TestManyError(t *testing.T) {
+	timeout := time.Second
+	opt := &Options{TimeOut:&timeout}
+	c := NewActuator(opt)
+	testManyError(t, c)
+}

common_test.go

@@ -15,7 +15,7 @@ func testTimeout(t *testing.T, c TimedActuator) {
 
 	Equal(t, err, ErrorTimeOut)
 	et := time.Now().UnixNano()
-	t.Logf("used time:%ds", (et-st)/1000000)
+	t.Logf("used time:%dms", (et-st)/1000000)
 	time.Sleep(time.Millisecond * 500)
 }
 
@@ -26,7 +26,26 @@ func testError(t *testing.T, c TimedActuator) {
 
 	Equal(t, err, te)
 	et := time.Now().UnixNano()
-	t.Logf("used time:%ds", (et-st)/1000000)
+	t.Logf("used time:%dms", (et-st)/1000000)
+	time.Sleep(time.Millisecond * 500)
+}
+
+func testManyError(t *testing.T, c TimedActuator) {
+	tasks := make([]Task, 0)
+	tmp, te := getErrorTask()
+	tasks = append(tasks, tmp...)
+
+	for i := 0; i < 100; i++ {
+		tmp, _ = getErrorTask()
+		tasks = append(tasks, tmp...)
+	}
+
+	st := time.Now().UnixNano()
+	err := c.Exec(tasks...)
+
+	Equal(t, err, te)
+	et := time.Now().UnixNano()
+	t.Logf("used time:%dms", (et-st)/1000000)
 	time.Sleep(time.Millisecond * 500)
 }
 

exec.go

@@ -11,29 +11,41 @@ import (
 )
 
 // wait waits for the notification of execution result
-func wait(c TimedActuator, ctx context.Context, resChan chan error) error {
+func wait(c TimedActuator, ctx context.Context,
+	resChan chan error, cancel context.CancelFunc) error {
 	if timeout := c.GetTimeout(); timeout != nil {
-		return waitWithTimeout(ctx, resChan, *timeout)
+		return waitWithTimeout(ctx, resChan, *timeout, cancel)
 	}
 
-	select {
-	case <-ctx.Done():
-		return nil
-	case err := <-resChan:
-		return err
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case err := <-resChan:
+			if err != nil {
+				cancel()
+				return err
+			}
+		}
 	}
 }
 
 // waitWithTimeout waits for the notification of execution result
 // when the timeout is set
-func waitWithTimeout(ctx context.Context, resChan chan error, timeout time.Duration) error {
-	select {
-	case <-time.After(timeout):
-		return ErrorTimeOut
-	case <-ctx.Done():
-		return nil
-	case err := <-resChan:
-		return err
+func waitWithTimeout(ctx context.Context, resChan chan error,
+	timeout time.Duration, cancel context.CancelFunc) error {
+	for {
+		select {
+		case <-time.After(timeout):
+			return ErrorTimeOut
+		case <-ctx.Done():
+			return nil
+		case err := <-resChan:
+			if err != nil {
+				cancel()
+				return err
+			}
+		}
 	}
 }
 
@@ -68,7 +80,7 @@ func execTasks(c TimedActuator, parent context.Context,
 		execFunc(f)
 	}
 
-	return wait(c, ctx, resChan)
+	return wait(c, ctx, resChan, cancel)
 }
 
 // simplyRun uses a new goroutine to run the function

util.go

@@ -18,29 +18,20 @@ func DurationPtr(t time.Duration) *time.Duration {
 func wrapperTask(ctx context.Context, task Task,
 	wg *sync.WaitGroup, resChan chan error) func() {
 	return func() {
-		tmpChan := make(chan error, 1)
 		defer func() {
 			if r := recover(); r != nil {
 				err := fmt.Errorf("conexec panic:%v\n%s", r, string(debug.Stack()))
 				resChan <- err
 			}
 
 			wg.Done()
-			close(tmpChan)
 		}()
 
 		select {
 		case <-ctx.Done():
 			return // fast return
-		case tmpChan <- task():
-			if err := <-tmpChan; err != nil {
-				resChan <- err
-			}
+		case resChan <- task():
 		}
-
-		// if err := task(); err != nil {
-		// 	resChan <- err
-		// }
 	}
 }