mutate: every test process held to its share of the memory

cmd/mutate/main.go

@@ -27,6 +27,9 @@ func main() {
 	// Half the machine: each is a build and a test binary, and a database package's tests hold
 	// connections of their own.
 	jobs := flag.Int("j", max(runtime.NumCPU()/2, 1), "how many mutations run at once")
+	// A test binary's memory is bounded by nothing else: -timeout is time, and an edit that makes
+	// a loop append for ever has the machine long before it runs out.
+	mem := flag.Float64("mem", 0, "how many GiB one test process and what it starts may hold, 0 for its share of the memory free when the run starts, negative for no bound (Linux)")
 	staleOnly := flag.Bool("stale-exempts", false, "report every mutate-exempt reason in the tree that covers no edit this tool makes, and fail on one")
 	flag.Parse()
 
@@ -80,7 +83,7 @@ func main() {
 		os.Exit(2)
 	}
 
-	mutate.Share(*jobs)
+	mutate.Share(*jobs, int64(*mem*(1<<30)))
 	started := time.Now()
 	t, err := mutate.RunAll(chosen, importers, *timeout, *jobs, os.Stdout)
 	if err != nil {

internal/mutate/bound.go

@@ -0,0 +1,172 @@
+package mutate
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"os/exec"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+// memLimit is how many bytes a test process and every process it started may hold between them;
+// 0 is no bound. -test.timeout bounds a test's time and nothing its memory: an edit that flips the
+// only exit of a loop that appends allocates gigabytes a second, and before its minute is out it
+// has the machine, and a CI runner's job with it.
+var memLimit int64
+
+// memPoll is how often the processes are measured: at the gigabyte a second such a loop takes,
+// what one passes its bound by before it is stopped.
+const memPoll = 100 * time.Millisecond
+
+// available is the memory the machine has free; 0 where there is no /proc to read it from, which
+// bounds nothing.
+func available() int64 {
+	meminfo, _ := os.ReadFile("/proc/meminfo")
+	return memAvailable(meminfo)
+}
+
+// memAvailable is the MemAvailable line of a /proc/meminfo, in bytes; 0 where it has none.
+func memAvailable(meminfo []byte) int64 {
+	for line := range strings.Lines(string(meminfo)) {
+		// MemAvailable:   12345678 kB
+		if rest, ok := strings.CutPrefix(line, "MemAvailable:"); ok {
+			kb, _ := strconv.ParseInt(strings.TrimSuffix(strings.TrimSpace(rest), " kB"), 10, 64)
+			return kb << 10
+		}
+	}
+	return 0
+}
+
+// gib is n bytes as people read it.
+func gib(n int64) string {
+	return fmt.Sprintf("%.1f GiB", float64(n)/(1<<30))
+}
+
+// watched are the processes running under memLimit, by pid, and whether each has been stopped
+// for passing it. One scan of /proc measures all of them: a scan is milliseconds, and one per
+// process would be a fifth of a core at -j 10.
+var watched = struct {
+	mu      sync.Mutex
+	by      map[int]*watch
+	polling sync.Once
+}{by: map[int]*watch{}}
+
+type watch struct {
+	proc *os.Process
+	// limit is memLimit when it started, which the poll reads rather than memLimit: its goroutine
+	// outlives what set it.
+	limit    int64
+	exceeded bool
+}
+
+// bounded is combined, with cmd and every process it starts stopped once they hold more than
+// memLimit between them; exceeded says they were. A go test is the go command, its compiler and
+// the test binary, and the binary is the one that grows.
+func bounded(what string, cmd *exec.Cmd) (out []byte, exceeded bool, err error) {
+	if memLimit == 0 {
+		out, err := combined(what, cmd)
+		return out, false, err
+	}
+	var buf bytes.Buffer
+	cmd.Stdout, cmd.Stderr = &buf, &buf
+	started := time.Now()
+	if err := cmd.Start(); err != nil {
+		account(what, cmd, time.Since(started))
+		return nil, false, err
+	}
+	w := &watch{proc: cmd.Process, limit: memLimit}
+	watched.mu.Lock()
+	watched.by[cmd.Process.Pid] = w
+	watched.mu.Unlock()
+	watched.polling.Do(func() { go poll() })
+	err = cmd.Wait()
+	watched.mu.Lock()
+	delete(watched.by, cmd.Process.Pid)
+	exceeded = w.exceeded
+	watched.mu.Unlock()
+	account(what, cmd, time.Since(started))
+	return buf.Bytes(), exceeded, err
+}
+
+// poll measures the watched processes every memPoll, for as long as the run lasts.
+func poll() {
+	for range time.Tick(memPoll) {
+		watched.mu.Lock()
+		stopOver()
+		watched.mu.Unlock()
+	}
+}
+
+// stopOver kills every watched process whose tree holds more than its limit, and all of its
+// tree. The caller holds watched.mu.
+func stopOver() {
+	procs := processes()
+	children := map[int][]int{}
+	for pid, p := range procs {
+		children[p.ppid] = append(children[p.ppid], pid)
+	}
+	for pid, w := range watched.by {
+		tree := []int{pid}
+		var rss int64
+		for i := 0; i < len(tree); i++ {
+			rss += procs[tree[i]].rss
+			tree = append(tree, children[tree[i]]...)
+		}
+		if rss <= w.limit {
+			continue
+		}
+		w.exceeded = true
+		// The root through the handle Start gave, which never signals a pid reused after it was
+		// waited for. Its tree by pid, which a pidfd makes safe on Linux, where alone /proc names a
+		// child: go test's binary does not die with the go command, and would go on growing.
+		_ = w.proc.Kill()
+		for _, child := range tree[1:] {
+			p, _ := os.FindProcess(child)
+			_ = p.Kill()
+		}
+	}
+}
+
+type process struct {
+	ppid int
+	rss  int64
+}
+
+// processes is every process /proc lists, with its parent and resident memory; none where there
+// is no /proc, which bounds nothing.
+func processes() map[int]process {
+	out := map[int]process{}
+	entries, _ := os.ReadDir("/proc")
+	for _, e := range entries {
+		pid, err := strconv.Atoi(e.Name())
+		if err != nil {
+			continue
+		}
+		// Unread when it is gone since it was listed, which parses as nothing.
+		stat, _ := os.ReadFile("/proc/" + e.Name() + "/stat")
+		if p, ok := parseStat(stat, int64(os.Getpagesize())); ok {
+			out[pid] = p
+		}
+	}
+	return out
+}
+
+// parseStat is a /proc/<pid>/stat's parent and resident memory, which it counts in pages of page
+// bytes.
+func parseStat(stat []byte, page int64) (process, bool) {
+	// pid (comm) state ppid ... rss is the 24th field; comm may hold spaces and parentheses, so
+	// the fields are counted from its last.
+	fields := strings.Fields(string(stat[bytes.LastIndexByte(stat, ')')+1:]))
+	if len(fields) < 22 {
+		return process{}, false
+	}
+	ppid, errP := strconv.Atoi(fields[1])
+	rss, errR := strconv.ParseInt(fields[21], 10, 64)
+	if errP != nil || errR != nil {
+		return process{}, false
+	}
+	return process{ppid: ppid, rss: rss * page}, true
+}

internal/mutate/bound_internal_test.go

@@ -0,0 +1,65 @@
+package mutate
+
+import (
+	"testing"
+)
+
+// The memory free is MemAvailable's, in bytes; a meminfo without it, or with one unreadable, is
+// none, which bounds nothing.
+func TestTheMemoryFreeIsMemAvailable(t *testing.T) {
+	meminfo := "MemTotal:       32554040 kB\nMemFree:         1234567 kB\nMemAvailable:   12345678 kB\nBuffers:          102400 kB\n"
+	if got := memAvailable([]byte(meminfo)); got != 12345678<<10 {
+		t.Errorf("MemAvailable of 12345678 kB read as %d bytes", got)
+	}
+	for _, meminfo := range []string{"", "MemTotal:       32554040 kB\n", "MemAvailable:   lots kB\n"} {
+		if got := memAvailable([]byte(meminfo)); got != 0 {
+			t.Errorf("%q read as %d bytes free", meminfo, got)
+		}
+	}
+}
+
+// A process's parent and memory are counted from the last parenthesis, whatever its name holds,
+// and its memory in pages; a line cut short, or with a field that is no number, is no process.
+func TestAProcessIsReadFromItsStat(t *testing.T) {
+	stat := "4242 (a) (b c) S 17 4242 4242 0 -1 4194560 100 0 0 0 5 3 0 0 20 0 8 0 1000 2000000 300 18446744073709551615\n"
+	if p, ok := parseStat([]byte(stat), 4096); !ok || p != (process{ppid: 17, rss: 300 * 4096}) {
+		t.Errorf("read as %+v, %v; want parent 17 and 300 pages", p, ok)
+	}
+	shortest := "1 (x) S 1 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 7"
+	if p, ok := parseStat([]byte(shortest), 1); !ok || p.rss != 7 {
+		t.Errorf("a stat that ends at its memory read as %+v, %v", p, ok)
+	}
+	for _, stat := range []string{
+		"",
+		"1 (x) S 1 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0",
+		"1 (x) S parent 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 7",
+		"1 (x) S 1 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 pages",
+	} {
+		if p, ok := parseStat([]byte(stat), 1); ok {
+			t.Errorf("%q read as %+v", stat, p)
+		}
+	}
+}
+
+// Every job is held to its share of the memory free when the run starts, or to what it is given;
+// a negative is no bound.
+func TestEachJobIsHeldToItsShareOfTheMemory(t *testing.T) {
+	t.Cleanup(func() { testProcs, memLimit = 0, 0 })
+	free := available()
+	if free == 0 {
+		t.Skip("no /proc/meminfo: there is no share to take")
+	}
+	Share(1, 0)
+	// What is free moves while this runs; a tenth is far from half.
+	if diff := memLimit - free; diff < -free/10 || diff > free/10 {
+		t.Errorf("one job is held to %s, with %s free", gib(memLimit), gib(free))
+	}
+	Share(3, 5<<30)
+	if memLimit != 5<<30 {
+		t.Errorf("a job given 5 GiB is held to %s", gib(memLimit))
+	}
+	Share(3, -1)
+	if memLimit != 0 {
+		t.Errorf("a job given no bound is held to %s", gib(memLimit))
+	}
+}

internal/mutate/cover.go

@@ -114,7 +114,10 @@ func Cover(dir, timeout string, slots chan struct{}) (*Coverage, error) {
 			// A failure here is the package's own tests' to report, not this gate's; what this
 			// gate does with it is ask the test about no edit, since it fails whatever the edit -
 			// and say so, since an edit only it reaches then reads as one nothing holds.
-			said, runErr := combined("cover: run one test", run)
+			said, exceeded, runErr := bounded("cover: run one test", run)
+			if exceeded {
+				runErr = fmt.Errorf("held over the %s a test process may, and stopped", gib(memLimit))
+			}
 			failed := runErr != nil
 			if failed {
 				// mutate-exempt: how much of the failure is shown, not whether it is.

internal/mutate/mutate.go

@@ -647,6 +647,9 @@ const (
 	Unbuildable
 	// Untested: the package has no tests at all, so there was nobody to ask.
 	Untested
+	// Exceeded: the tests held more memory than a test process may, and were stopped. Refused, as
+	// a test that runs out its time is: an edit that makes a loop allocate for ever.
+	Exceeded
 )
 
 // buildFailed is go test's word that a package it was to test did not build.
@@ -720,9 +723,12 @@ func Run(m Mutation, a Asked) (Outcome, string) {
 		cmd := exec.Command("go", args...) //nolint:gosec // go test over this repository's packages
 		cmd.Dir = set.dir
 		cmd.Env = testEnv()
-		out, err := combined(set.what, cmd)
+		out, exceeded, err := bounded(set.what, cmd)
 		countTests("", out)
 		said.Write(out)
+		if exceeded {
+			return Exceeded, said.String()
+		}
 		if err == nil {
 			continue
 		}

internal/mutate/runner.go

@@ -79,6 +79,10 @@ func RunAll(chosen []Mutation, importers map[string][]string, timeout string, jo
 		case Killed:
 			t.Killed++
 			_, _ = fmt.Fprintf(w, "  refused   %s\n", m)
+		case Exceeded:
+			// Named with its bound: the edit is refused, and the loop it made is worth knowing of.
+			t.Killed++
+			_, _ = fmt.Fprintf(w, "  refused   %s (its tests held over %s, and were stopped)\n", m, gib(memLimit))
 		case Unbuildable:
 			t.Unbuildable++
 			_, _ = fmt.Fprintf(w, "  no edit   %s (does not build)\n", m)
@@ -161,14 +165,19 @@ func (p pkgRun) built(m Mutation, s *Schema, c *Coverage) Outcome {
 		if c != nil {
 			reaching = c.Reaching(m.File, m.Line)
 		}
-		if (reaching == nil || len(reaching) > 0) && ask(s.Binary, p.dir, reaching) == Killed {
-			return Killed
+		if reaching == nil || len(reaching) > 0 {
+			if outcome := ask(s.Binary, p.dir, reaching); outcome != Survived {
+				return outcome
+			}
 		}
 	}
 	for _, importer := range p.importers {
 		bin, dir := p.builds.importer(p.dir, importer, s.Overlay)
-		if bin != "" && ask(bin, dir, nil) == Killed {
-			return Killed
+		if bin == "" {
+			continue
+		}
+		if outcome := ask(bin, dir, nil); outcome != Survived {
+			return outcome
 		}
 	}
 	return Survived