[dev.fuzz] cmd/go: implement -fuzztime flag and support cancellation

fuzz.CoordinateFuzzing and RunFuzzWorker now accept a context.Context
parameter. They should terminate gracefully when the context is
cancelled. The worker should exit quickly without processing more
inputs. The coordinator should save interesting inputs to the cache.

The testing package can't import context directly, so it provides a
timeout argument to testdeps.CoordinateFuzzing instead. The testdeps
wrapper sets the timeout and installs an interrupt handler (for SIGINT
on POSIX and the equivalent on Windows) that cancels the context when
^C is pressed.

Note that on POSIX platforms, pressing ^C causes the shell to deliver
SIGINT to all processes in the active group: so 'go test', the
coordinator, and the workers should all react to that. On Windows,
pressing ^C only interrupts 'go test'. We may want to look at that
separately.

Change-Id: I924d3be2905f9685dae82ff3c047ca3d6b5e2357
Reviewed-on: https://go-review.googlesource.com/c/go/+/279487
Run-TryBot: Jay Conrod <jayconrod@google.com>
TryBot-Result: Go Bot <gobot@golang.org>
Reviewed-by: Katie Hockman <katie@golang.org>
Trust: Katie Hockman <katie@golang.org>
Trust: Jay Conrod <jayconrod@google.com>

2 files changed, 70 insertions, 35 deletions

diff --git a/src/internal/fuzz/fuzz.go b/src/internal/fuzz/fuzz.go
index 2ab16b1189..aacc053682 100644
--- a/src/internal/fuzz/fuzz.go
+++ b/src/internal/fuzz/fuzz.go
@@ -8,6 +8,7 @@
 package fuzz
 
 import (
+	"context"
 	"crypto/sha256"
 	"fmt"
 	"io/ioutil"
@@ -15,7 +16,6 @@ import (
 	"path/filepath"
 	"runtime"
 	"sync"
-	"time"
 )
 
 // CoordinateFuzzing creates several worker processes and communicates with
@@ -39,14 +39,13 @@ import (
 //
 // If a crash occurs, the function will return an error containing information
 // about the crash, which can be reported to the user.
-func CoordinateFuzzing(parallel int, seed [][]byte, corpusDir, cacheDir string) (err error) {
+func CoordinateFuzzing(ctx context.Context, parallel int, seed [][]byte, corpusDir, cacheDir string) (err error) {
+	if err := ctx.Err(); err != nil {
+		return err
+	}
 	if parallel == 0 {
 		parallel = runtime.GOMAXPROCS(0)
 	}
-	// TODO(jayconrod): support fuzzing indefinitely or with a given duration.
-	// The value below is just a placeholder until we figure out how to handle
-	// interrupts.
-	duration := 5 * time.Second
 
 	corpus, err := readCorpusAndCache(seed, corpusDir, cacheDir)
 	if err != nil {
@@ -121,26 +120,28 @@ func CoordinateFuzzing(parallel int, seed [][]byte, corpusDir, cacheDir string)
 	defer func() {
 		close(c.doneC)
 		wg.Wait()
-		if err == nil {
-			for _, err = range workerErrs {
-				if err != nil {
-					// Return the first error found.
-					return
+		if err == nil || err == ctx.Err() {
+			for _, werr := range workerErrs {
+				if werr != nil {
+					// Return the first error found, replacing ctx.Err() if a more
+					// interesting error is found.
+					err = werr
 				}
 			}
 		}
 	}()
 
 	// Main event loop.
-	stopC := time.After(duration)
 	i := 0
 	for {
 		select {
-		// TODO(jayconrod): handle interruptions like SIGINT.
-
-		case <-stopC:
-			// Time's up.
-			return nil
+		case <-ctx.Done():
+			// Interrupted, cancelled, or timed out.
+			// TODO(jayconrod,katiehockman): On Windows, ^C only interrupts 'go test',
+			// not the coordinator or worker processes. 'go test' will stop running
+			// actions, but it won't interrupt its child processes. This makes it
+			// difficult to stop fuzzing on Windows without a timeout.
+			return ctx.Err()
 
 		case crasher := <-c.crasherC:
 			// A worker found a crasher. Write it to testdata and return it.
diff --git a/src/internal/fuzz/worker.go b/src/internal/fuzz/worker.go
index 4658687106..ef2a9303ef 100644
--- a/src/internal/fuzz/worker.go
+++ b/src/internal/fuzz/worker.go
@@ -5,6 +5,7 @@
 package fuzz
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -105,15 +106,26 @@ func (w *worker) runFuzzing() error {
 				args := fuzzArgs{Duration: workerFuzzDuration}
 				value, resp, err := w.client.fuzz(input.b, args)
 				if err != nil {
-					// TODO(jayconrod): if we get an error here, something failed between
-					// main and the call to testing.F.Fuzz. The error here won't
-					// be useful. Collect stderr, clean it up, and return that.
-					// TODO(jayconrod): we can get EPIPE if w.stop is called concurrently
-					// and it kills the worker process. Suppress this message in
-					// that case.
+					// Error communicating with worker.
+					select {
+					case <-w.termC:
+						// Worker terminated, perhaps unexpectedly.
+						// We expect I/O errors due to partially sent or received RPCs,
+						// so ignore this error.
+					case <-w.coordinator.doneC:
+						// Timeout or interruption. Worker may also be interrupted.
+						// Again, ignore I/O errors.
+					default:
+						// TODO(jayconrod): if we get an error here, something failed between
+						// main and the call to testing.F.Fuzz. The error here won't
+						// be useful. Collect stderr, clean it up, and return that.
+						// TODO(jayconrod): we can get EPIPE if w.stop is called concurrently
+						// and it kills the worker process. Suppress this message in
+						// that case.
+						fmt.Fprintf(os.Stderr, "communicating with worker: %v\n", err)
+					}
 					// TODO(jayconrod): what happens if testing.F.Fuzz is never called?
 					// TODO(jayconrod): time out if the test process hangs.
-					fmt.Fprintf(os.Stderr, "communicating with worker: %v\n", err)
 				} else if resp.Err != "" {
 					// The worker found a crasher. Inform the coordinator.
 					crasher := crasherEntry{
@@ -301,13 +313,13 @@ func (w *worker) stop() error {
 //
 // RunFuzzWorker returns an error if it could not communicate with the
 // coordinator process.
-func RunFuzzWorker(fn func([]byte) error) error {
+func RunFuzzWorker(ctx context.Context, fn func([]byte) error) error {
 	comm, err := getWorkerComm()
 	if err != nil {
 		return err
 	}
 	srv := &workerServer{workerComm: comm, fuzzFn: fn}
-	return srv.serve()
+	return srv.serve(ctx)
 }
 
 // call is serialized and sent from the coordinator on fuzz_in. It acts as
@@ -370,21 +382,41 @@ type workerServer struct {
 // serve returns errors that occurred when communicating over pipes. serve
 // does not return errors from method calls; those are passed through serialized
 // responses.
-func (ws *workerServer) serve() error {
+func (ws *workerServer) serve(ctx context.Context) error {
+	// Stop handling messages when ctx.Done() is closed. This normally happens
+	// when the worker process receives a SIGINT signal, which on POSIX platforms
+	// is sent to the process group when ^C is pressed.
+	//
+	// Ordinarily, the coordinator process may stop a worker by closing fuzz_in.
+	// We simulate that and interrupt a blocked read here.
+	doneC := make(chan struct{})
+	defer func() { close(doneC) }()
+	go func() {
+		select {
+		case <-ctx.Done():
+			ws.fuzzIn.Close()
+		case <-doneC:
+		}
+	}()
+
 	enc := json.NewEncoder(ws.fuzzOut)
 	dec := json.NewDecoder(ws.fuzzIn)
 	for {
 		var c call
-		if err := dec.Decode(&c); err == io.EOF {
-			return nil
-		} else if err != nil {
-			return err
+		if err := dec.Decode(&c); err != nil {
+			if ctx.Err() != nil {
+				return ctx.Err()
+			} else if err == io.EOF {
+				return nil
+			} else {
+				return err
+			}
 		}
 
 		var resp interface{}
 		switch {
 		case c.Fuzz != nil:
-			resp = ws.fuzz(*c.Fuzz)
+			resp = ws.fuzz(ctx, *c.Fuzz)
 		default:
 			return errors.New("no arguments provided for any call")
 		}
@@ -398,11 +430,13 @@ func (ws *workerServer) serve() error {
 // fuzz runs the test function on random variations of a given input value for
 // a given amount of time. fuzz returns early if it finds an input that crashes
 // the fuzz function or an input that expands coverage.
-func (ws *workerServer) fuzz(args fuzzArgs) fuzzResponse {
-	t := time.NewTimer(args.Duration)
+func (ws *workerServer) fuzz(ctx context.Context, args fuzzArgs) fuzzResponse {
+	ctx, cancel := context.WithTimeout(ctx, args.Duration)
+	defer cancel()
+
 	for {
 		select {
-		case <-t.C:
+		case <-ctx.Done():
 			// TODO(jayconrod,katiehockman): this value is not interesting. Use a
 			// real heuristic once we have one.
 			return fuzzResponse{Interesting: true}