Merge pull request #9 from ash2k/ash2k/fix-regression

Signal processes to stop on error and don't start new ones

Author: ash2k

multirun/multirun.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"sync"
 )
 
 type multirun struct {
@@ -29,6 +30,8 @@ func (m multirun) run(ctx context.Context) error {
 
 // unboundedExecution execute multiple commands without concurrency limit
 func (m multirun) unboundedExecution(ctx context.Context) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
 	errs := make(chan error)
 
 	for _, cmd := range m.commands {
@@ -48,7 +51,9 @@ func (m multirun) unboundedExecution(ctx context.Context) error {
 
 	var firstError error
 	for range m.commands {
-		if err := <-errs; firstError == nil {
+		err := <-errs
+		if err != nil && firstError == nil {
+			cancel()
 			firstError = err
 		}
 	}
@@ -58,34 +63,50 @@ func (m multirun) unboundedExecution(ctx context.Context) error {
 
 // boundedExecution execute multiple commands using a sized worker pool
 func (m multirun) boundedExecution(ctx context.Context) error {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
 	// errs should be buffered to avoid blocking
 	// when len(m.commands) > m.jobs
 	errs := make(chan error, len(m.commands))
-	commands := make(chan command)
-
-	// start worker pool
-	for w := 0; w < m.jobs; w++ {
-		go m.spawnWorker(ctx, commands, errs)
-	}
+	commands := make(chan command, len(m.commands))
 
 	// send command to worker pool
 	for _, cmd := range m.commands {
 		commands <- cmd
 	}
 	close(commands)
 
-	var firstError error
-	for range m.commands {
-		if err := <-errs; firstError == nil {
-			firstError = err
-		}
+	// start worker pool
+	var wg sync.WaitGroup
+	wg.Add(m.jobs)
+	for w := 0; w < m.jobs; w++ {
+		go func() {
+			defer wg.Done()
+			err := m.spawnWorker(ctx, commands)
+			if err != nil {
+				errs <- err // first error must go first, cancel after it has been sent.
+				cancel()
+			}
+		}()
 	}
 
-	return firstError
+	wg.Wait()
+	close(errs)
+
+	for err := range errs {
+		return err
+	}
+
+	return nil
 }
 
-func (m multirun) spawnWorker(ctx context.Context, commands <-chan command, errs chan<- error) {
+func (m multirun) spawnWorker(ctx context.Context, commands <-chan command) error {
 	for cmd := range commands {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
 		p := process{
 			tag:        cmd.Tag,
 			path:       cmd.Path,
@@ -103,6 +124,10 @@ func (m multirun) spawnWorker(ctx context.Context, commands <-chan command, errs
 			fmt.Fprintf(m.stderrSink, "Running %s\n", cmd.Tag)
 		}
 
-		errs <- p.run(ctx)
+		err := p.run(ctx)
+		if err != nil {
+			return err
+		}
 	}
+	return nil
 }

multirun/test/BUILD.bazel

@@ -130,12 +130,12 @@ multirun(
         ":command_echo_2",
         ":command_echo_1",
     ],
-    parallel = True,
+    jobs = 0,
 )
 
 multirun(
     name = "multirun_parallel_tagged_success",
-    parallel = True,
+    jobs = 0,
     tagged_commands = {
         ":command_echo_2": "echo-2",
         ":command_echo_1": "echo-1",
@@ -182,10 +182,10 @@ multirun(
     name = "multirun_parallel_failure",
     commands = [
         ":command_echo_2",
-        ":command_exit",
-        ":command_echo_3",
+        ":command_sleep_and_exit_0",
+        ":command_sleep_and_exit_5",
     ],
-    parallel = True,
+    jobs = 0,
 )
 
 command(
@@ -269,22 +269,30 @@ sh_binary(
 multirun(
     name = "multirun_signal_handling",
     commands = [
-        ":command_sleep_and_exit",
+        ":command_sleep_and_exit_0",
     ],
 )
 
 multirun(
     name = "multirun_parallel_signal_handling",
     commands = [
-        ":command_sleep_and_exit",
+        ":command_sleep_and_exit_0",
+    ],
+    jobs = 0,
+)
+
+command(
+    name = "command_sleep_and_exit_0",
+    arguments = [
+        "4", "0",
     ],
-    parallel = True,
+    command = ":sleep_and_exit",
 )
 
 command(
-    name = "command_sleep_and_exit",
+    name = "command_sleep_and_exit_5",
     arguments = [
-        "4",
+        "1", "5",
     ],
     command = ":sleep_and_exit",
 )

multirun/test/expected_parallel_failure.txt

@@ -1,5 +1,6 @@
 [//multirun/test:command_echo_2] command_2 a
 [//multirun/test:command_echo_2] command_2 b
 [//multirun/test:command_echo_2] command_2 c
-[//multirun/test:command_echo_3] command_3 a
-[//multirun/test:command_exit] exiting 5
+[//multirun/test:command_sleep_and_exit_0] sleeping before exiting 4
+[//multirun/test:command_sleep_and_exit_5] exiting: 5
+[//multirun/test:command_sleep_and_exit_5] sleeping before exiting 1

multirun/test/expected_parallel_sigint.txt

@@ -1,5 +1,5 @@
 RAK>>running multirun/test/multirun_parallel_signal_handling.bash
 RAK>>sleeping 2
-[//multirun/test:command_sleep_and_exit] sleeping before exiting 4
+[//multirun/test:command_sleep_and_exit_0] sleeping before exiting 4
 RAK>>killing with signal SIGINT
 RAK>>done

multirun/test/expected_parallel_sigterm.txt

@@ -1,5 +1,5 @@
 RAK>>running multirun/test/multirun_parallel_signal_handling.bash
 RAK>>sleeping 2
-[//multirun/test:command_sleep_and_exit] sleeping before exiting 4
+[//multirun/test:command_sleep_and_exit_0] sleeping before exiting 4
 RAK>>killing with signal SIGTERM
 RAK>>done

multirun/test/expected_sigint.txt

@@ -1,6 +1,6 @@
 RAK>>running multirun/test/multirun_signal_handling.bash
 RAK>>sleeping 2
-Running //multirun/test:command_sleep_and_exit
+Running //multirun/test:command_sleep_and_exit_0
 sleeping before exiting 4
 RAK>>killing with signal SIGINT
 RAK>>done

multirun/test/expected_sigterm.txt

@@ -1,6 +1,6 @@
 RAK>>running multirun/test/multirun_signal_handling.bash
 RAK>>sleeping 2
-Running //multirun/test:command_sleep_and_exit
+Running //multirun/test:command_sleep_and_exit_0
 sleeping before exiting 4
 RAK>>killing with signal SIGTERM
 RAK>>done

multirun/test/sleep_and_exit.sh

@@ -2,4 +2,5 @@
 
 echo "sleeping before exiting $1"
 sleep "$1"
-echo "exiting"
+echo "exiting: $2"
+exit "$2"