Terminate subprocesses when changes are received in watch (#599).

Interestingly they now seem to be getting a SIGINT when I ctrl+c the parent, although I haven't done anything explicit to handle that?
thought-machine · Apr 19, 2019 · 594a0ec · 594a0ec · katzdm · Apr 19, 2019
1 parent b8f7ea8
commit 594a0ec
Show file tree

Hide file tree

Showing 4 changed files with 27 additions and 14 deletions.
diff --git a/src/please.go b/src/please.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	_ "net/http/pprof"
@@ -428,7 +429,7 @@ var buildFunctions = map[string]func() bool{
 	},
 	"parallel": func() bool {
 		if success, state := runBuild(opts.Run.Parallel.PositionalArgs.Targets, true, false); success {
-			os.Exit(run.Parallel(state, state.ExpandOriginalTargets(), opts.Run.Parallel.Args, opts.Run.Parallel.NumTasks, opts.Run.Parallel.Quiet, opts.Run.Env))
+			os.Exit(run.Parallel(context.Background(), state, state.ExpandOriginalTargets(), opts.Run.Parallel.Args, opts.Run.Parallel.NumTasks, opts.Run.Parallel.Quiet, opts.Run.Env))
 		}
 		return false
 	},

diff --git a/src/run/run_step.go b/src/run/run_step.go
@@ -2,6 +2,7 @@
 package run
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"os/exec"
@@ -21,13 +22,14 @@ var log = logging.MustGetLogger("run")
 
 // Run implements the running part of 'plz run'.
 func Run(state *core.BuildState, label core.BuildLabel, args []string, env bool) {
-	run(state, label, args, false, false, env)
+	run(context.Background(), state, label, args, false, false, env)
 }
 
 // Parallel runs a series of targets in parallel.
 // Returns a relevant exit code (i.e. if at least one subprocess exited unsuccessfully, it will be
 // that code, otherwise 0 if all were successful).
-func Parallel(state *core.BuildState, labels []core.BuildLabel, args []string, numTasks int, quiet, env bool) int {
+// The given context can be used to control the lifetime of the subprocesses.
+func Parallel(ctx context.Context, state *core.BuildState, labels []core.BuildLabel, args []string, numTasks int, quiet, env bool) int {
 	pool := NewGoroutinePool(numTasks)
 	var g errgroup.Group
 	for _, label := range labels {
@@ -36,7 +38,7 @@ func Parallel(state *core.BuildState, labels []core.BuildLabel, args []string, n
 			var wg sync.WaitGroup
 			wg.Add(1)
 			pool.Submit(func() {
-				if e := run(state, label, args, true, quiet, env); e != nil {
+				if e := run(ctx, state, label, args, true, quiet, env); e != nil {
 					err = e
 				}
 				wg.Done()
@@ -46,7 +48,9 @@ func Parallel(state *core.BuildState, labels []core.BuildLabel, args []string, n
 		})
 	}
 	if err := g.Wait(); err != nil {
-		log.Error("Command failed: %s", err)
+		if ctx.Err() != context.Canceled { // Don't error if the context killed the process.
+			log.Error("Command failed: %s", err)
+		}
 		return err.(*exitError).code
 	}
 	return 0
@@ -58,7 +62,7 @@ func Parallel(state *core.BuildState, labels []core.BuildLabel, args []string, n
 func Sequential(state *core.BuildState, labels []core.BuildLabel, args []string, quiet, env bool) int {
 	for _, label := range labels {
 		log.Notice("Running %s", label)
-		if err := run(state, label, args, true, quiet, env); err != nil {
+		if err := run(context.Background(), state, label, args, true, quiet, env); err != nil {
 			log.Error("%s", err)
 			return err.code
 		}
@@ -70,7 +74,7 @@ func Sequential(state *core.BuildState, labels []core.BuildLabel, args []string,
 // If fork is true then we fork to run the target and return any error from the subprocesses.
 // If it's false this function never returns (because we either win or die; it's like
 // Game of Thrones except rather less glamorous).
-func run(state *core.BuildState, label core.BuildLabel, args []string, fork, quiet, setenv bool) *exitError {
+func run(ctx context.Context, state *core.BuildState, label core.BuildLabel, args []string, fork, quiet, setenv bool) *exitError {
 	target := state.Graph.TargetOrDie(label)
 	if !target.IsBinary {
 		log.Fatalf("Target %s cannot be run; it's not marked as binary", label)
@@ -101,7 +105,7 @@ func run(state *core.BuildState, label core.BuildLabel, args []string, fork, qui
 	}
 	// Run as a normal subcommand.
 	// Note that we don't connect stdin. It doesn't make sense for multiple processes.
-	cmd := core.ExecCommand(splitCmd[0], args[1:]...) // args here don't include argv[0]
+	cmd := exec.CommandContext(ctx, splitCmd[0], args[1:]...) // args here don't include argv[0]
 	cmd.Env = env
 	if !quiet {
 		cmd.Stdout = os.Stdout

diff --git a/src/run/run_test.go b/src/run/run_test.go
@@ -1,6 +1,7 @@
 package run
 
 import (
+	"context"
 	"os"
 	"testing"
 
@@ -25,9 +26,9 @@ func TestSequential(t *testing.T) {
 
 func TestParallel(t *testing.T) {
 	state, labels1, labels2 := makeState()
-	code := Parallel(state, labels1, nil, 5, false, false)
+	code := Parallel(context.Background(), state, labels1, nil, 5, false, false)
 	assert.Equal(t, 0, code)
-	code = Parallel(state, labels2, nil, 5, true, false)
+	code = Parallel(context.Background(), state, labels2, nil, 5, true, false)
 	assert.Equal(t, 1, code)
 }
 

diff --git a/src/watch/watch.go b/src/watch/watch.go
@@ -4,6 +4,7 @@
 package watch
 
 import (
+	"context"
 	"fmt"
 	"path"
 	"time"
@@ -38,10 +39,12 @@ func Watch(state *core.BuildState, labels core.BuildLabels, callback CallbackFun
 	files := cmap.New()
 	go startWatching(watcher, state, labels, files)
 
+	ctx, cancel := context.WithCancel(context.Background())
+
 	// The initial setup only builds targets, it doesn't test or run things.
 	// Do one of those now if requested.
 	if state.NeedTests || state.NeedRun {
-		build(state, labels, callback)
+		build(ctx, state, labels, callback)
 	}
 
 	for {
@@ -52,6 +55,9 @@ func Watch(state *core.BuildState, labels core.BuildLabels, callback CallbackFun
 				log.Notice("Skipping notification for %s", event.Name)
 				continue
 			}
+			// Kill any previous process.
+			cancel()
+			ctx, cancel = context.WithCancel(context.Background())
 
 			// Quick debounce; poll and discard all events for the next brief period.
 		outer:
@@ -62,7 +68,7 @@ func Watch(state *core.BuildState, labels core.BuildLabels, callback CallbackFun
 					break outer
 				}
 			}
-			build(state, labels, callback)
+			build(ctx, state, labels, callback)
 		case err := <-watcher.Errors:
 			log.Error("Error watching files:", err)
 		}
@@ -141,7 +147,7 @@ func anyTests(state *core.BuildState, labels []core.BuildLabel) bool {
 }
 
 // build invokes a single build while watching.
-func build(state *core.BuildState, labels []core.BuildLabel, callback CallbackFunc) {
+func build(ctx context.Context, state *core.BuildState, labels []core.BuildLabel, callback CallbackFunc) {
 	// Set up a new state & copy relevant parts off the existing one.
 	ns := core.NewBuildState(state.Config.Please.NumThreads, state.Cache, state.Verbosity, state.Config)
 	ns.VerifyHashes = state.VerifyHashes
@@ -155,6 +161,7 @@ func build(state *core.BuildState, labels []core.BuildLabel, callback CallbackFu
 	ns.StartTime = time.Now()
 	callback(ns, labels)
 	if state.NeedRun {
-		run.Parallel(state, labels, nil, state.Config.Please.NumThreads, false, false)
+		// Don't wait for this, its lifetime will be controlled by the context.
+		go run.Parallel(ctx, state, labels, nil, state.Config.Please.NumThreads, false, false)
 	}
 }