diff --git a/docs/changes/unreleased/2039-darwin-process-inspection.md b/docs/changes/unreleased/2039-darwin-process-inspection.md new file mode 100644 index 0000000000..5a91eb265e --- /dev/null +++ b/docs/changes/unreleased/2039-darwin-process-inspection.md @@ -0,0 +1,19 @@ +--- +kind: fixed +title: macOS reads process groups for left-running services and the job subtree bound +pr: 2039 +surface: [engine, docs] +invalidates: + - "On macOS, a server left running by a shell command was never reported, and service records named the shell without its server's argv or listening ports. These records now inspect the surviving processes." + - "The job subtree CPU and process bound never cut on macOS because subtree usage could not be read. macOS now measures the whole tree and applies the existing bound." +--- + +On macOS, a registered bash job that sustains more than three fifths of the +machine's logical cores for thirty seconds, or grows past 128 processes for +that long, is now ended like on Linux. The CPU bound applies when no CPU quota +or affinity mask holds codeaf below the machine's cores. Both limits require +15 consecutive readings two seconds apart, and the run is told why it ended. + +Process inspection works without cgo, including release and sidecar builds. +Service descriptions read each surviving member's sockets once, select the +first listening member's argv and collect the sorted, unique listening ports. diff --git a/internal/darwinproc/doc.go b/internal/darwinproc/doc.go new file mode 100644 index 0000000000..88c2428641 --- /dev/null +++ b/internal/darwinproc/doc.go @@ -0,0 +1,3 @@ +// Package darwinproc reads macOS process identity, descendants, CPU usage, +// arguments and listening sockets without cgo. +package darwinproc diff --git a/internal/darwinproc/procinfo_darwin.go b/internal/darwinproc/procinfo_darwin.go new file mode 100644 index 0000000000..b190e614d3 --- /dev/null +++ b/internal/darwinproc/procinfo_darwin.go @@ -0,0 +1,267 @@ +//go:build darwin + +package darwinproc + +import ( + "bytes" + "encoding/binary" + "slices" + "sync" + "unsafe" + + "golang.org/x/sys/unix" +) + +// The raw proc_info trap uses Go's existing per-architecture syscall stubs, +// avoiding assembly, dynamic imports and linkname dependencies that garbling +// could break. Unlike libproc's wrappers, the trap preserves errno directly. +// Apple promises stability for neither the trap numbers nor libproc, but both +// have remained unchanged since macOS 10.5. The private XNU call selectors and +// SDK layouts below are identical on arm64 and amd64; both are little-endian. +const ( + callListPIDs = 1 + callPIDInfo = 2 + callPIDFDInfo = 3 + ppidOnly = 6 + pidListFDs = 1 + pidTaskAllInfo = 2 + pidFDSocketInfo = 3 + fdTypeSocket = 2 + socketInfoTCP = 2 + tcpStateListen = 1 + familyIPv4 = 2 + familyIPv6 = 30 + + // StatusZombie marks exited processes, which cannot contribute live work. + StatusZombie = 5 + + pidSize = 4 + taskAllInfoSize = 232 + bsdInfoOffset = 0 + bsdInfoSize = 136 + bsdStatusOffset = bsdInfoOffset + 4 + bsdPIDOffset = bsdInfoOffset + 12 + bsdPPIDOffset = bsdInfoOffset + 16 + bsdPGIDOffset = bsdInfoOffset + 100 + bsdStartSecondsOffset = bsdInfoOffset + 120 + bsdStartMicrosOffset = bsdInfoOffset + 128 + taskInfoOffset = bsdInfoSize + taskInfoSize = 96 + taskResidentOffset = taskInfoOffset + 8 + taskUserOffset = taskInfoOffset + 16 + taskSystemOffset = taskInfoOffset + 24 + fdInfoSize = 8 + fdNumberOffset = 0 + fdTypeOffset = 4 + socketFDInfoSize = 792 + socketInfoOffset = 24 + socketInfoSize = 768 + socketFamilyOffset = socketInfoOffset + 160 + socketKindOffset = socketInfoOffset + 232 + socketProtocolOffset = socketInfoOffset + 240 + socketLocalPortOffset = socketProtocolOffset + 4 + socketTCPStateOffset = socketProtocolOffset + 80 + + childrenInitialPIDs = 64 + childrenGrowth = 4 + childrenAttempts = 4 + + // procArgsAlign is the boundary the kernel pads the executable path to + // before the arguments begin: the pointer size of every supported Mac. + procArgsAlign = 8 +) + +// procInfo keeps pointer conversion in the syscall expression so the buffer +// remains alive and cannot move while the kernel writes it. A nil buffer is +// used only by interfaces that return a sizing estimate. +func procInfo(call, a1, a2, a3 int, buf []byte) (int, error) { + if len(buf) > 1<<31-1 { + return 0, unix.EINVAL + } + var n uintptr + var errno unix.Errno + if len(buf) == 0 { + n, _, errno = unix.Syscall6(unix.SYS_PROC_INFO, uintptr(call), uintptr(a1), uintptr(a2), uintptr(a3), 0, 0) + } else { + n, _, errno = unix.Syscall6(unix.SYS_PROC_INFO, uintptr(call), uintptr(a1), uintptr(a2), uintptr(a3), uintptr(unsafe.Pointer(&buf[0])), uintptr(len(buf))) + } + if errno != 0 { + return 0, errno + } + if n > 1<<31-1 || (len(buf) > 0 && n > uintptr(len(buf))) { + return 0, unix.EIO + } + return int(n), nil +} + +// Children lists immediate children, bounding retries when the tree grows +// between reads. When even the last buffer fills, the children it holds are +// returned with ENOMEM: they are a lower bound on the children, never the +// whole list, and a caller counting them must say so. +func Children(ppid int) ([]int, error) { + if ppid <= 0 { + return nil, unix.EINVAL + } + var initial [childrenInitialPIDs * pidSize]byte + buf := initial[:] + for attempt := 0; attempt < childrenAttempts; attempt++ { + n, err := procInfo(callListPIDs, ppidOnly, ppid, 0, buf) + if err != nil { + return nil, err + } + if n%pidSize != 0 { + return nil, unix.EIO + } + if n < len(buf) { + return childPIDs(buf[:n]), nil + } + if attempt+1 < childrenAttempts { + buf = make([]byte, len(buf)*childrenGrowth) + } + } + return childPIDs(buf), unix.ENOMEM +} + +func childPIDs(buf []byte) []int { + pids := make([]int, 0, len(buf)/pidSize) + for offset := 0; offset+pidSize <= len(buf); offset += pidSize { + if pid := int(int32(binary.LittleEndian.Uint32(buf[offset:]))); pid > 0 { + pids = append(pids, pid) + } + } + return pids +} + +// TaskInfo combines identity and CPU from one kernel snapshot so callers can +// validate ancestry before attributing a process's accumulated CPU time. +type TaskInfo struct { + PID, PPID, PGID, Status int + // Start is the process's start time in microseconds. A pid handed out + // again names a process with another start, so a reader that must not + // confuse the two compares it. + Start uint64 + CPUSeconds float64 + Resident uint64 +} + +var ( + frequencyOnce sync.Once + tickFrequency uint64 +) + +// Task reads one process. Kernel task times are absolute ticks even under +// Rosetta; hw.tbfrequency converts them correctly on both supported CPUs. +func Task(pid int) (TaskInfo, error) { + if pid <= 0 { + return TaskInfo{}, unix.EINVAL + } + frequencyOnce.Do(func() { tickFrequency, _ = unix.SysctlUint64("hw.tbfrequency") }) + if tickFrequency == 0 { + return TaskInfo{}, unix.EIO + } + var buf [taskAllInfoSize]byte + n, err := procInfo(callPIDInfo, pid, pidTaskAllInfo, 0, buf[:]) + if err != nil { + return TaskInfo{}, err + } + if n != taskAllInfoSize { + return TaskInfo{}, unix.EIO + } + return TaskInfo{ + PID: int(int32(binary.LittleEndian.Uint32(buf[bsdPIDOffset:]))), + PPID: int(int32(binary.LittleEndian.Uint32(buf[bsdPPIDOffset:]))), + PGID: int(int32(binary.LittleEndian.Uint32(buf[bsdPGIDOffset:]))), + Status: int(binary.LittleEndian.Uint32(buf[bsdStatusOffset:])), + Start: binary.LittleEndian.Uint64(buf[bsdStartSecondsOffset:])*1_000_000 + binary.LittleEndian.Uint64(buf[bsdStartMicrosOffset:]), + CPUSeconds: float64(binary.LittleEndian.Uint64(buf[taskUserOffset:]))/float64(tickFrequency) + float64(binary.LittleEndian.Uint64(buf[taskSystemOffset:]))/float64(tickFrequency), + Resident: binary.LittleEndian.Uint64(buf[taskResidentOffset:]), + }, nil +} + +// ListenPorts inspects only this process's socket descriptors. Descriptor races +// are skipped individually, and a shared socket buffer avoids one allocation +// per descriptor. +func ListenPorts(pid int) []int { + if pid <= 0 { + return nil + } + size, err := procInfo(callPIDInfo, pid, pidListFDs, 0, nil) + if err != nil || size == 0 || size%fdInfoSize != 0 { + return nil + } + fds := make([]byte, size) + n, err := procInfo(callPIDInfo, pid, pidListFDs, 0, fds) + if err != nil || n%fdInfoSize != 0 { + return nil + } + var socket [socketFDInfoSize]byte + var ports []int + for offset := 0; offset < n; offset += fdInfoSize { + fd := fds[offset : offset+fdInfoSize] + if binary.LittleEndian.Uint32(fd[fdTypeOffset:]) != fdTypeSocket { + continue + } + number := int(int32(binary.LittleEndian.Uint32(fd[fdNumberOffset:]))) + count, err := procInfo(callPIDFDInfo, pid, pidFDSocketInfo, number, socket[:]) + if err != nil || count != socketFDInfoSize { + continue + } + family := binary.LittleEndian.Uint32(socket[socketFamilyOffset:]) + if (family != familyIPv4 && family != familyIPv6) || binary.LittleEndian.Uint32(socket[socketKindOffset:]) != socketInfoTCP || binary.LittleEndian.Uint32(socket[socketTCPStateOffset:]) != tcpStateListen { + continue + } + // The low sixteen bits contain network-order bytes on both macOS CPUs. + port := int(binary.BigEndian.Uint16(socket[socketLocalPortOffset:])) + ports = append(ports, port) + } + slices.Sort(ports) + return slices.Compact(ports) +} + +// Argv discards the raw sysctl response after parsing because it also contains +// environment data that must never enter a service record or diagnostic. +func Argv(pid int) []string { + if pid <= 0 { + return nil + } + raw, err := unix.SysctlRaw("kern.procargs2", pid) + if err != nil { + return nil + } + return parseProcArgs(raw) +} + +func parseProcArgs(raw []byte) []string { + if len(raw) < pidSize { + return nil + } + argc := int(int32(binary.LittleEndian.Uint32(raw))) + if argc <= 0 || argc > len(raw)-pidSize { + return nil + } + rest := raw[pidSize:] + end := bytes.IndexByte(rest, 0) + if end < 0 { + return nil + } + // The kernel pads the executable path with NULs so the arguments start on + // a pointer-size boundary of the string area. Skipping every NUL instead + // would swallow an empty argv[0] and shift the first environment string, + // which can hold a secret, into the arguments. Padding that is not all NUL + // is a layout this parser does not know, and it reads as nothing. + start := (end + 1 + procArgsAlign - 1) &^ (procArgsAlign - 1) + if start > len(rest) || len(bytes.Trim(rest[end+1:start], "\x00")) != 0 { + return nil + } + rest = rest[start:] + argv := make([]string, 0, argc) + for range argc { + end = bytes.IndexByte(rest, 0) + if end < 0 { + return nil + } + argv = append(argv, string(rest[:end])) + rest = rest[end+1:] + } + return argv +} diff --git a/internal/darwinproc/procinfo_darwin_test.go b/internal/darwinproc/procinfo_darwin_test.go new file mode 100644 index 0000000000..f4ce0b1c5c --- /dev/null +++ b/internal/darwinproc/procinfo_darwin_test.go @@ -0,0 +1,257 @@ +//go:build darwin + +package darwinproc + +import ( + "encoding/binary" + "math" + "net" + "os" + "os/exec" + "path/filepath" + "reflect" + "slices" + "syscall" + "testing" + "time" +) + +// procArgsFixture lays out a kern.procargs2 answer the way the kernel does: +// argc, the executable path padded with NULs to the next eight-byte boundary, +// then the arguments and the environment. +func procArgsFixture(argc int32, exec string, strings ...string) []byte { + raw := make([]byte, 4) + binary.LittleEndian.PutUint32(raw, uint32(argc)) + raw = append(raw, exec...) + raw = append(raw, 0) + for (len(raw)-4)%procArgsAlign != 0 { + raw = append(raw, 0) + } + for _, s := range strings { + raw = append(raw, s...) + raw = append(raw, 0) + } + return raw +} + +func TestDarwinProcArgsStopsBeforeEnvironment(t *testing.T) { + header := func(argc int32, body string) []byte { + raw := make([]byte, 4, 4+len(body)) + binary.LittleEndian.PutUint32(raw, uint32(argc)) + return append(raw, body...) + } + for _, tc := range []struct { + name string + raw []byte + want []string + }{ + {"padding and environment", procArgsFixture(4, "/bin/tool", "tool", "", "two words", "last", "PRIVATE=value"), []string{"tool", "", "two words", "last"}}, + {"empty argv[0] keeps the environment out", procArgsFixture(2, "/bin/tool", "", "argument", "PRIVATE=value"), []string{"", "argument"}}, + {"path that fills its boundary", procArgsFixture(1, "/bin/to", "tool", "PRIVATE=value"), []string{"tool"}}, + {"padding that is not NUL", header(1, "/bin/tool\x00\x00tool\x00PRIVATE=value\x00"), nil}, + {"zero", procArgsFixture(0, "/bin/tool", "tool"), nil}, + {"negative", procArgsFixture(-1, "/bin/tool", "tool"), nil}, + {"absurd", procArgsFixture(1<<30, "/bin/tool", "tool"), nil}, + {"missing argument", procArgsFixture(2, "/bin/tool", "tool"), nil}, + {"truncated argument", header(1, "/bin/tool\x00\x00\x00\x00\x00\x00\x00tool"), nil}, + {"truncated padding", header(1, "/bin/tool\x00\x00"), nil}, + {"truncated executable", header(1, "/bin/tool"), nil}, + {"short header", []byte{1, 0, 0}, nil}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := parseProcArgs(tc.raw); !reflect.DeepEqual(got, tc.want) { + t.Fatalf("argv = %q, want %q", got, tc.want) + } + }) + } +} + +// An empty argv[0] is read in place on a live process, with the environment +// left out, whatever the length of the executable path. +func TestDarwinArgvKeepsAnEmptyFirstArgument(t *testing.T) { + dir := t.TempDir() + for _, name := range []string{"s", "sl", "sleeper", "sleeperx"} { + link := filepath.Join(dir, name) + if err := os.Symlink("/bin/sleep", link); err != nil { + t.Fatal(err) + } + cmd := exec.Command(link, "30") + cmd.Args[0] = "" + cmd.Env = []string{"PRIVATE=value"} + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }) + var got []string + for deadline := time.Now().Add(2 * time.Second); time.Now().Before(deadline); time.Sleep(5 * time.Millisecond) { + if got = Argv(cmd.Process.Pid); reflect.DeepEqual(got, []string{"", "30"}) { + break + } + } + if !reflect.DeepEqual(got, []string{"", "30"}) { + t.Fatalf("%s: argv = %q, want [\"\" \"30\"]", name, got) + } + } +} + +func sleepChild(t testing.TB) *exec.Cmd { + t.Helper() + cmd := exec.Command("sleep", "30") + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = cmd.Process.Kill(); _ = cmd.Wait() }) + return cmd +} + +func TestDarwinArgvReadsALiveChild(t *testing.T) { + cmd := sleepChild(t) + if got := Argv(cmd.Process.Pid); !reflect.DeepEqual(got, []string{"sleep", "30"}) { + t.Fatalf("argv = %q", got) + } +} + +func TestDarwinTaskMatchesSelf(t *testing.T) { + deadline := time.Now().Add(150 * time.Millisecond) + var work uint64 + for time.Now().Before(deadline) { + for range 10000 { + work = work*1664525 + 1013904223 + } + } + if work == 0 { + t.Log("bounded loop finished at zero") + } + task, err := Task(os.Getpid()) + if err != nil { + t.Fatal(err) + } + if task.PID != os.Getpid() || task.PPID != os.Getppid() { + t.Fatalf("identity = %+v", task) + } + var usage syscall.Rusage + if err := syscall.Getrusage(syscall.RUSAGE_SELF, &usage); err != nil { + t.Fatal(err) + } + seconds := float64(usage.Utime.Sec+usage.Stime.Sec) + float64(usage.Utime.Usec+usage.Stime.Usec)/1e6 + if math.Abs(task.CPUSeconds-seconds) > math.Max(seconds*.2, .03) { + t.Fatalf("CPU seconds = %f, getrusage = %f", task.CPUSeconds, seconds) + } + gone := exec.Command("true") + gone.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + if err := gone.Run(); err != nil { + t.Fatal(err) + } + if _, err := Task(gone.Process.Pid); err == nil { + t.Fatal("gone process returned a task") + } +} + +func TestDarwinChildrenListsASpawnedChild(t *testing.T) { + cmd := sleepChild(t) + pids, err := Children(os.Getpid()) + if err != nil { + t.Fatal(err) + } + if !slices.Contains(pids, cmd.Process.Pid) { + t.Fatalf("children %v omit %d", pids, cmd.Process.Pid) + } +} + +func TestDarwinListeningPorts(t *testing.T) { + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = listener.Close() }) + want := []int{listener.Addr().(*net.TCPAddr).Port} + listener6, err := net.Listen("tcp6", "[::1]:0") + if err == nil { + t.Cleanup(func() { _ = listener6.Close() }) + want = append(want, listener6.Addr().(*net.TCPAddr).Port) + } else { + t.Logf("IPv6 listener unavailable: %v", err) + } + conn, err := net.DialTimeout("tcp4", listener.Addr().String(), time.Second) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + _ = listener.(*net.TCPListener).SetDeadline(time.Now().Add(time.Second)) + accepted, err := listener.Accept() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = accepted.Close() }) + udp, err := net.ListenPacket("udp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = udp.Close() }) + // A short directory keeps the socket path within macOS's sockaddr_un limit. + socketDir, err := os.MkdirTemp("", "dp-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.RemoveAll(socketDir) }) + local, err := net.Listen("unix", filepath.Join(socketDir, "s")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = local.Close() }) + got := ListenPorts(os.Getpid()) + for _, p := range want { + if !slices.Contains(got, p) { + t.Fatalf("ports %v omit listener %d", got, p) + } + } + for _, p := range []int{conn.LocalAddr().(*net.TCPAddr).Port, udp.LocalAddr().(*net.UDPAddr).Port} { + // A UDP or ephemeral TCP port can coincide with an independently allocated listener. + if !slices.Contains(want, p) && slices.Contains(got, p) { + t.Fatalf("ports %v include non-listener %d", got, p) + } + } + if !slices.IsSorted(got) || len(slices.Compact(slices.Clone(got))) != len(got) { + t.Fatalf("ports not sorted and unique: %v", got) + } +} + +func BenchmarkTask(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + if _, err := Task(os.Getpid()); err != nil { + b.Fatal(err) + } + } +} +func BenchmarkChildren(b *testing.B) { + _ = sleepChild(b) + b.ReportAllocs() + for b.Loop() { + if _, err := Children(os.Getpid()); err != nil { + b.Fatal(err) + } + } +} +func BenchmarkListenPorts(b *testing.B) { + listener, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + b.Fatal(err) + } + b.Cleanup(func() { _ = listener.Close() }) + b.ReportAllocs() + for b.Loop() { + _ = ListenPorts(os.Getpid()) + } +} +func BenchmarkArgv(b *testing.B) { + cmd := sleepChild(b) + b.ReportAllocs() + for b.Loop() { + if len(Argv(cmd.Process.Pid)) == 0 { + b.Fatal("argv unavailable") + } + } +} diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 8fd16ec67c..34dfb48de2 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -115,10 +115,10 @@ type Chunk struct { // Service is a process group still alive after its call returned. type Service struct { - Name string // basename of the first member's argv[0] - Argv []string // the first member's command line (the request's off Linux) + Name string // Basename of the naming member's argv[0]. + Argv []string // The naming member's command line on Linux and macOS; otherwise the request's. PGID int - Ports []int // listening TCP ports, when known (Linux only) + Ports []int // Listening TCP ports, when known on Linux and macOS. DataDir string // cell-relative data directory, when the service keeps one } diff --git a/internal/executor/service.go b/internal/executor/service.go index 42ac58bf38..d0e13c9871 100644 --- a/internal/executor/service.go +++ b/internal/executor/service.go @@ -4,6 +4,7 @@ import ( "context" "os/exec" "path/filepath" + "slices" "strings" "github.com/Agent-Field/codeaf/internal/processgroup" @@ -37,16 +38,42 @@ func Left(call Call) []Service { // listens names it, because it is the server; with none listening the first // member does. The ports of all members are its ports. func describe(req ExecRequest, root string, pgid int, members []int) Service { + service := describeWith(req, root, pgid, members, pidPorts, cmdline) + if len(service.Ports) == 0 { + // An empty query preserves each platform's unknown-versus-empty result + // without inspecting any member's sockets again. + service.Ports = listenPorts(nil) + } + return service +} + +// describeWith uses one socket snapshot for naming and aggregation so a member +// is not inspected twice while a service is recorded. +func describeWith(req ExecRequest, root string, pgid int, members []int, ports func(int) []int, cmdline func(int) []string) Service { + allPorts := make([]int, 0) + naming := 0 + for _, member := range members { + listening := ports(member) + if naming == 0 && len(listening) > 0 { + naming = member + } + allPorts = append(allPorts, listening...) + } + if naming == 0 && len(members) > 0 { + naming = members[0] + } + slices.Sort(allPorts) + allPorts = slices.Compact(allPorts) argv := req.Argv - if name := namingMember(members); name > 0 { - if a := cmdline(name); len(a) > 0 { + if naming > 0 { + if a := cmdline(naming); len(a) > 0 { argv = a } } name := filepath.Base(argv[0]) return Service{ Name: name, Argv: argv, PGID: pgid, - Ports: listenPorts(members), + Ports: allPorts, DataDir: dataDir(name, argv, root, req.Dir), } } @@ -100,18 +127,3 @@ func inside(root, dir, path string) string { } return filepath.ToSlash(rel) } - -// namingMember is the member that names a service: the first that listens on a -// port, else the first member, else 0. A shell that started a server lingers -// beside it, and the shell's command line says nothing of what is being served. -func namingMember(members []int) int { - for _, m := range members { - if len(listenPorts([]int{m})) > 0 { - return m - } - } - if len(members) > 0 { - return members[0] - } - return 0 -} diff --git a/internal/executor/service_darwin.go b/internal/executor/service_darwin.go new file mode 100644 index 0000000000..aa748b30b0 --- /dev/null +++ b/internal/executor/service_darwin.go @@ -0,0 +1,22 @@ +//go:build darwin + +package executor + +import ( + "slices" + + "github.com/Agent-Field/codeaf/internal/darwinproc" +) + +func cmdline(pid int) []string { return darwinproc.Argv(pid) } + +func pidPorts(pid int) []int { return darwinproc.ListenPorts(pid) } + +func listenPorts(pids []int) []int { + ports := make([]int, 0) + for _, pid := range pids { + ports = append(ports, pidPorts(pid)...) + } + slices.Sort(ports) + return slices.Compact(ports) +} diff --git a/internal/executor/service_describe_test.go b/internal/executor/service_describe_test.go new file mode 100644 index 0000000000..4ac262a0cc --- /dev/null +++ b/internal/executor/service_describe_test.go @@ -0,0 +1,64 @@ +package executor + +import ( + "reflect" + "testing" +) + +func TestDescribeInspectsEachMemberOnce(t *testing.T) { + members := []int{10, 11, 12} + wantArgv := []string{"postgres", "-D", "data"} + reads := make(map[int]int) + ports := func(pid int) []int { + reads[pid]++ + return map[int][]int{11: {8080, 5432}, 12: {8080}}[pid] + } + argv := func(pid int) []string { + if pid != 11 { + t.Fatalf("argv read for member %d, want first listener 11", pid) + } + return wantArgv + } + got := describeWith(ExecRequest{Argv: []string{"sh"}}, "/w", 10, members, ports, argv) + for _, pid := range members { + if reads[pid] != 1 { + t.Errorf("member %d inspected %d times, want once", pid, reads[pid]) + } + } + if got.Name != "postgres" || !reflect.DeepEqual(got.Argv, wantArgv) || + !reflect.DeepEqual(got.Ports, []int{5432, 8080}) || got.DataDir != "data" || got.PGID != 10 { + t.Fatalf("service = %+v", got) + } + + t.Run("no listeners", func(t *testing.T) { + for _, fallback := range []bool{false, true} { + reads := make(map[int]int) + req := ExecRequest{Argv: []string{"/bin/request", "argument"}} + wantArgv := []string{"/bin/sleep", "30"} + wantName := "sleep" + if fallback { + wantArgv, wantName = req.Argv, "request" + } + got := describeWith(req, "/w", 10, members, func(pid int) []int { + reads[pid]++ + return nil + }, func(pid int) []string { + if pid != members[0] { + t.Fatalf("argv read for member %d, want first member %d", pid, members[0]) + } + if fallback { + return nil + } + return wantArgv + }) + for _, pid := range members { + if reads[pid] != 1 { + t.Errorf("member %d inspected %d times, want once", pid, reads[pid]) + } + } + if got.Name != wantName || !reflect.DeepEqual(got.Argv, wantArgv) || len(got.Ports) != 0 { + t.Fatalf("fallback=%t: service = %+v", fallback, got) + } + } + }) +} diff --git a/internal/executor/service_linux.go b/internal/executor/service_linux.go index 559a307076..f51c85e82f 100644 --- a/internal/executor/service_linux.go +++ b/internal/executor/service_linux.go @@ -26,11 +26,8 @@ func cmdline(pid int) []string { func listenPorts(pids []int) []int { seen := map[int]bool{} for _, pid := range pids { - listening := listeningInodes(pid) - for inode := range socketInodes(pid) { - if port, ok := listening[inode]; ok { - seen[port] = true - } + for _, port := range pidPorts(pid) { + seen[port] = true } } ports := make([]int, 0, len(seen)) @@ -41,6 +38,17 @@ func listenPorts(pids []int) []int { return ports } +func pidPorts(pid int) []int { + listening := listeningInodes(pid) + ports := make([]int, 0) + for inode := range socketInodes(pid) { + if port, ok := listening[inode]; ok { + ports = append(ports, port) + } + } + return ports +} + func listeningInodes(pid int) map[uint64]int { out := map[uint64]int{} for _, table := range []string{"tcp", "tcp6"} { diff --git a/internal/executor/service_linux_test.go b/internal/executor/service_linux_test.go index 0eaf13ae71..b913825cb6 100644 --- a/internal/executor/service_linux_test.go +++ b/internal/executor/service_linux_test.go @@ -5,12 +5,8 @@ package executor import ( "reflect" "testing" - - "github.com/Agent-Field/codeaf/internal/processgroup" ) -func membersOf(pgid int) []int { return processgroup.Members(pgid) } - const tcpFixture = ` sl local_address rem_address st tx_queue rx_queue tr tm->when retrnsmt uid timeout inode 0: 0100007F:1F90 00000000:0000 0A 00000000:00000000 00:00000000 00000000 1000 0 12345 1 0000 100 0 0 10 0 1: 0100007F:0050 0100007F:9C40 01 00000000:00000000 00:00000000 00000000 1000 0 22222 1 0000 100 0 0 10 0 diff --git a/internal/executor/service_other.go b/internal/executor/service_other.go index 0a4ffe3f19..f16013f462 100644 --- a/internal/executor/service_other.go +++ b/internal/executor/service_other.go @@ -1,8 +1,10 @@ -//go:build !linux +//go:build !linux && !darwin package executor -// Off Linux the record keeps the request's argv and no ports. +// Where process inspection is unavailable, the record keeps the request's argv and no ports. func cmdline(int) []string { return nil } +func pidPorts(int) []int { return nil } + func listenPorts([]int) []int { return nil } diff --git a/internal/executor/service_other_test.go b/internal/executor/service_other_test.go deleted file mode 100644 index 328bd36598..0000000000 --- a/internal/executor/service_other_test.go +++ /dev/null @@ -1,5 +0,0 @@ -//go:build !linux - -package executor - -func membersOf(int) []int { return nil } diff --git a/internal/executor/service_unix_test.go b/internal/executor/service_unix_test.go index 80e1a8d473..679d03064a 100644 --- a/internal/executor/service_unix_test.go +++ b/internal/executor/service_unix_test.go @@ -1,17 +1,22 @@ -//go:build !windows +//go:build linux || darwin -// K6 keeps Unix process-group service inspection on Unix. +// K6 keeps process-group service inspection on Linux and macOS. package executor import ( "context" "syscall" "testing" - "time" ) func TestLocalRecordsAListeningServiceFully(t *testing.T) { - script := "exec python3 -c 'import socket,time;s=socket.socket();s.bind((\"127.0.0.1\",0));s.listen();print(s.getsockname()[1],flush=True);time.sleep(60)' >port.txt 2>&1 port.txt 2>&1 0 && int(record.Eproc.Pgid) == pgid && record.Proc.P_stat != darwinproc.StatusZombie { + pids = append(pids, int(record.Proc.P_pid)) + } + } + sort.Ints(pids) + return pids +} + +// The shell is reaped before a sweep on macOS (internal/exec/shellwait_other.go), +// so the existing signal probe has no zombie leader to mistake for live work. +func groupHasLiveMember(pgid int) bool { + return Alive(pgid) +} diff --git a/internal/processgroup/members_darwin_test.go b/internal/processgroup/members_darwin_test.go new file mode 100644 index 0000000000..f9b1c5b1e0 --- /dev/null +++ b/internal/processgroup/members_darwin_test.go @@ -0,0 +1,71 @@ +//go:build darwin + +package processgroup + +import ( + "os/exec" + "reflect" + "strings" + "testing" + + "github.com/Agent-Field/codeaf/internal/darwinproc" + "golang.org/x/sys/unix" +) + +func TestDarwinMembersExcludeZombies(t *testing.T) { + records := make([]unix.KinfoProc, 6) + for i, pid := range []int32{30, 10, 20, 40, 0, -1} { + records[i].Proc.P_pid = pid + records[i].Proc.P_stat = 3 + records[i].Eproc.Pgid = 100 + } + records[0].Proc.P_stat = 4 + records[2].Proc.P_stat = darwinproc.StatusZombie + records[3].Eproc.Pgid = 101 + if got := darwinMemberPIDs(100, records); !reflect.DeepEqual(got, []int{10, 30}) { + t.Fatalf("Members = %v, want sleeping and stopped members [10 30]", got) + } +} + +func TestDarwinMembersReadsALiveGroup(t *testing.T) { + cmd := startDarwinGroup(t, "sleep", "30") + pid := cmd.Process.Pid + if got := Members(pid); !reflect.DeepEqual(got, []int{pid}) { + t.Fatalf("Members(%d) = %v, want [%d]", pid, got, pid) + } + if err := cmd.Process.Kill(); err != nil { + t.Fatal(err) + } + _ = cmd.Wait() + if got := Members(pid); len(got) != 0 { + t.Fatalf("Members(%d) after wait = %v, want empty", pid, got) + } +} + +func startDarwinGroup(tb testing.TB, name string, args ...string) *exec.Cmd { + tb.Helper() + cmd := exec.Command(name, args...) + Configure(cmd) + if err := cmd.Start(); err != nil { + tb.Fatal(err) + } + tb.Cleanup(func() { + if cmd.ProcessState == nil { + _ = Kill(cmd.Process.Pid) + _ = cmd.Wait() + } + }) + return cmd +} + +func BenchmarkMembers(b *testing.B) { + cmd := startDarwinGroup(b, "sh", "-c", strings.Repeat("sleep 30 & ", 9)+"wait") + waitForDarwinUsage(b, CaptureGroup(cmd.Process.Pid), 10) + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + if got := Members(cmd.Process.Pid); len(got) != 10 || got[0] != cmd.Process.Pid { + b.Fatalf("Members = %v, want the leader and nine sleeping children", got) + } + } +} diff --git a/internal/processgroup/members_other.go b/internal/processgroup/members_other.go index a0777f0b37..137975cd77 100644 --- a/internal/processgroup/members_other.go +++ b/internal/processgroup/members_other.go @@ -1,14 +1,14 @@ -//go:build !windows && !linux +//go:build !windows && !linux && !darwin package processgroup -// groupHasLiveMember is the signal probe off Linux: there the shell is reaped -// before any sweep looks at its group (internal/exec, shellwait_other.go), so -// no zombie leader is ever in the group when this is asked, and kill(-pgid, 0) -// answers correctly. +// groupHasLiveMember retains the signal probe on other Unix platforms: the +// shell is reaped before any sweep looks at its group (internal/exec, +// shellwait_other.go), so no zombie leader is in the group when this is asked, +// and kill(-pgid, 0) answers correctly. func groupHasLiveMember(pgid int) bool { return Alive(pgid) } -// Members is unknown off Linux; callers fall back to what they started. +// Members is unknown off Linux and macOS; callers fall back to what they started. func Members(pgid int) []int { return nil } diff --git a/internal/processgroup/usage.go b/internal/processgroup/usage.go index b3d418024b..7eb8c794f0 100644 --- a/internal/processgroup/usage.go +++ b/internal/processgroup/usage.go @@ -16,11 +16,9 @@ package processgroup // cores the subtree held, which is the reading this product bounds. One // reading on its own is not a rate and must not be read as one. // -// THE READING IS /proc, WHICH IS LINUX'S. A platform without it answers "cannot -// say" rather than a guess, and a bound that cannot say never cuts — the same -// silence rule the admission governor keeps (session's task_pressure.go): a -// person on a machine this cannot measure gets exactly the scheduler they had -// before the bound existed. +// Linux reads /proc and macOS reads the kernel's process-info interface. Other +// platforms answer "cannot say" rather than guess, and a bound that cannot say +// never cuts, preserving the scheduler the person had before the bound existed. // Usage is what one job subtree is using at one instant. type Usage struct { @@ -30,4 +28,11 @@ type Usage struct { // every process in it. Subtract two readings and divide by the wall between // them to get the cores the subtree held. CPUSeconds float64 + // CPUPartial says CPUSeconds left out part of the tree: the walk stopped at + // its limit, a listing was truncated, or a process could not be read. Such + // a sum is not comparable with another reading, because a process missing + // from one and present in the next would read as a burst of its whole + // lifetime's CPU, so no rate may be taken from it. Processes is still a + // lower bound. + CPUPartial bool } diff --git a/internal/processgroup/usage_darwin.go b/internal/processgroup/usage_darwin.go new file mode 100644 index 0000000000..cd87c8d9df --- /dev/null +++ b/internal/processgroup/usage_darwin.go @@ -0,0 +1,125 @@ +//go:build darwin + +package processgroup + +import ( + "errors" + + "github.com/Agent-Field/codeaf/internal/darwinproc" + "golang.org/x/sys/unix" +) + +// usageWalkLimit is how many processes one reading looks at most. Past it the +// process count is a lower bound and the CPU sum is marked partial, so a tree +// that keeps growing cannot make every two-second sample a longer walk. A storm +// that size normally still counts far over the job bound's 128-process ceiling +// and is cut on the count; a lower count can only hold a cut back. +const usageWalkLimit = 1024 + +// usageReader is the kernel's answer for one process and for its children. It +// is a variable so the walk's limits and refusals can be proven on a scripted +// tree instead of real load. +var usageReader = struct { + task func(pid int) (darwinproc.TaskInfo, error) + children func(ppid int) ([]int, error) +}{darwinproc.Task, darwinproc.Children} + +// Usage reports the recorded leader's whole subtree, or says it cannot say. +// Identity is checked around the walk because a reading of a reused pid could +// cause the bound to cut somebody else's work. +func (g Group) Usage() (Usage, bool) { + if ok, _ := g.owned(); !ok { + return Usage{}, false + } + usage, ok := subtreeUsage(g.pid) + if !ok { + return Usage{}, false + } + if ok, _ := g.owned(); !ok { + return Usage{}, false + } + return usage, true +} + +// subtreeUsage walks the tree below root through each process's children, so +// a child that changed process groups is still the job's. +// +// A PARENT IS RE-READ AFTER ITS CHILDREN ARE READ. A queued process can exit +// and its pid be handed to another process while its children are listed or +// read; the children of that stranger would pass the parent-pid check. A pid is +// never handed on while its process lives, so a parent that reads with the same +// start time after its children were read was their parent while they were +// read, and only then are they admitted. +// +// A listing that filled even the largest buffer, a walk that reached +// [usageWalkLimit], and a process that could not be read for any reason other +// than having exited are counted as far as they got, with the CPU sum marked +// partial ([Usage.CPUPartial]): the count is a lower bound and no rate is taken +// from the sum. A process that exited between two reads is skipped, as the +// Linux reading skips it. A listing that fails outright makes the whole +// reading unknown. +func subtreeUsage(root int) (Usage, bool) { + top, err := usageReader.task(root) + if err != nil || top.PID != root || top.Status == darwinproc.StatusZombie { + return Usage{}, false + } + usage := Usage{} + seen := map[int]bool{root: true} + queue := []darwinproc.TaskInfo{top} + for next := 0; next < len(queue); next++ { + parent := queue[next] + usage.Processes++ + usage.CPUSeconds += parent.CPUSeconds + if len(seen) >= usageWalkLimit { + usage.CPUPartial = true + continue + } + children, err := usageReader.children(parent.PID) + switch { + case err == nil: + case errors.Is(err, unix.ENOMEM): + usage.CPUPartial = true + case errors.Is(err, unix.ESRCH): + continue + default: + return Usage{}, false + } + var read []darwinproc.TaskInfo + for _, pid := range children { + if len(seen) >= usageWalkLimit { + usage.CPUPartial = true + break + } + if pid <= 0 || seen[pid] { + continue + } + seen[pid] = true + child, err := usageReader.task(pid) + if err != nil && !errors.Is(err, unix.ESRCH) { + usage.CPUPartial = true + continue + } + // A child can exit, be reparented or have its pid reused between reads. + // Skip that snapshot rather than charging an unrelated process to us. + if err != nil || child.PID != pid || child.PPID != parent.PID || child.Status == darwinproc.StatusZombie { + continue + } + read = append(read, child) + } + again, err := usageReader.task(parent.PID) + if err != nil || again.Start != parent.Start || again.PID != parent.PID { + if next == 0 { + return Usage{}, false + } + // A parent that exited or whose pid changed hands left the tree with + // its children; one that merely could not be read leaves them out of + // this sum only, so the sum is partial. + if err != nil && !errors.Is(err, unix.ESRCH) { + usage.CPUPartial = true + } + continue + } + queue = append(queue, read...) + } + return usage, true +} diff --git a/internal/processgroup/usage_darwin_test.go b/internal/processgroup/usage_darwin_test.go new file mode 100644 index 0000000000..ec769fff9d --- /dev/null +++ b/internal/processgroup/usage_darwin_test.go @@ -0,0 +1,80 @@ +//go:build darwin + +package processgroup + +import ( + "os" + "strings" + "testing" + "time" +) + +func TestDarwinUsageReadsDescendants(t *testing.T) { + cmd := startDarwinGroup(t, "sh", "-c", `sleep 30 & sleep 30 & sh -c 'sleep 30 & wait' & wait`) + group := CaptureGroup(cmd.Process.Pid) + usage := waitForDarwinUsage(t, group, 5) + if usage.CPUSeconds < 0 { + t.Fatalf("Usage CPUSeconds = %v, want nonnegative", usage.CPUSeconds) + } +} + +func TestDarwinUsageRefusesUnownedGroup(t *testing.T) { + if usage, ok := (Group{}).Usage(); ok || usage != (Usage{}) { + t.Fatalf("unowned Usage = %+v, %v, want zero, false", usage, ok) + } + cmd := startDarwinGroup(t, "sleep", "30") + id, ok := CaptureGroup(cmd.Process.Pid).Identity() + if !ok { + t.Fatal("could not capture sleeping leader identity") + } + id.Start++ + if usage, ok := Restore(id).Usage(); ok || usage != (Usage{}) { + t.Fatalf("wrong identity Usage = %+v, %v, want zero, false", usage, ok) + } + + t.Run("identity changes during sample", func(t *testing.T) { + group := CaptureGroup(os.Getpid()) + saved := processStart + t.Cleanup(func() { processStart = saved }) + reads := 0 + processStart = func(pid int) (uint64, bool) { + reads++ + return group.start, reads == 1 + } + if usage, ok := group.Usage(); ok || usage != (Usage{}) { + t.Fatalf("changed identity Usage = %+v, %v, want zero, false", usage, ok) + } + if reads != 2 { + t.Fatalf("identity read %d times, want before and after the sample", reads) + } + }) +} + +func waitForDarwinUsage(tb testing.TB, group Group, processes int) Usage { + tb.Helper() + deadline := time.Now().Add(5 * time.Second) + var usage Usage + var ok bool + for time.Now().Before(deadline) { + usage, ok = group.Usage() + if ok && usage.Processes >= processes { + return usage + } + time.Sleep(10 * time.Millisecond) + } + tb.Fatalf("Usage = %+v, %v, want at least %d processes", usage, ok, processes) + return Usage{} +} + +func BenchmarkUsage(b *testing.B) { + cmd := startDarwinGroup(b, "sh", "-c", strings.Repeat("sleep 30 & ", 9)+"wait") + group := CaptureGroup(cmd.Process.Pid) + waitForDarwinUsage(b, group, 10) + b.ReportAllocs() + b.ResetTimer() + for b.Loop() { + if usage, ok := group.Usage(); !ok || usage.Processes < 10 || usage.CPUSeconds < 0 { + b.Fatalf("Usage = %+v, %v, want the leader and sleeping children", usage, ok) + } + } +} diff --git a/internal/processgroup/usage_darwin_walk_test.go b/internal/processgroup/usage_darwin_walk_test.go new file mode 100644 index 0000000000..cf521e767a --- /dev/null +++ b/internal/processgroup/usage_darwin_walk_test.go @@ -0,0 +1,230 @@ +//go:build darwin + +package processgroup + +import ( + "testing" + + "github.com/Agent-Field/codeaf/internal/darwinproc" + "golang.org/x/sys/unix" +) + +// scriptedTree answers the walk from a fixed table, so its limits and refusals +// are proven on a tree that exists only as numbers and never as real load. +type scriptedTree struct { + tasks map[int]darwinproc.TaskInfo + children map[int][]int + listErr map[int]error + // restart gives a pid a new start time after its children are listed: the + // pid was handed to another process in between. + restart map[int]bool + listed map[int]bool + taskRead int +} + +func (s *scriptedTree) use(t *testing.T) { + t.Helper() + saved := usageReader + t.Cleanup(func() { usageReader = saved }) + s.listed = map[int]bool{} + usageReader.task = func(pid int) (darwinproc.TaskInfo, error) { + s.taskRead++ + task, ok := s.tasks[pid] + if !ok { + return darwinproc.TaskInfo{}, unix.ESRCH + } + if s.restart[pid] && s.listed[pid] { + task.Start++ + } + return task, nil + } + usageReader.children = func(ppid int) ([]int, error) { + s.listed[ppid] = true + return s.children[ppid], s.listErr[ppid] + } +} + +func task(pid, ppid, pgid int, cpu float64) darwinproc.TaskInfo { + return darwinproc.TaskInfo{PID: pid, PPID: ppid, PGID: pgid, Start: uint64(pid) * 1000, CPUSeconds: cpu, Status: 2} +} + +// The job's tree is its children's children too, including one that moved to +// a process group of its own; each live process is counted once. +func TestDarwinUsageCountsTheWholeTreeOnce(t *testing.T) { + tree := &scriptedTree{ + tasks: map[int]darwinproc.TaskInfo{ + 10: task(10, 1, 10, 1), 11: task(11, 10, 10, 2), 12: task(12, 10, 99, 3), 13: task(13, 11, 10, 4), + }, + children: map[int][]int{10: {11, 12, 11}, 11: {13}, 12: {13}}, + } + tree.use(t) + usage, ok := subtreeUsage(10) + if !ok || usage.Processes != 4 || usage.CPUSeconds != 10 || usage.CPUPartial { + t.Fatalf("usage = %+v, %v, want 4 processes and a complete 10 CPU seconds", usage, ok) + } +} + +// A queued process whose pid went to another process before its children were +// listed contributes itself but none of the stranger's children. +func TestDarwinUsageKeepsAReusedParentsChildrenOut(t *testing.T) { + tree := &scriptedTree{ + tasks: map[int]darwinproc.TaskInfo{ + 10: task(10, 1, 10, 1), 11: task(11, 10, 10, 1), 30: task(30, 11, 30, 50), + }, + children: map[int][]int{10: {11}, 11: {30}}, + restart: map[int]bool{11: true}, + } + tree.use(t) + usage, ok := subtreeUsage(10) + if !ok || usage.Processes != 2 || usage.CPUSeconds != 2 { + t.Fatalf("usage = %+v, %v, want the leader and its own child only", usage, ok) + } + tree.restart, tree.listed = map[int]bool{10: true}, map[int]bool{} + if usage, ok := subtreeUsage(10); ok { + t.Fatalf("a root whose pid changed hands mid-walk was read: %+v", usage) + } +} + +// A listing that filled the largest buffer is a lower bound and still counts; +// an unexpected failure makes the reading unknown rather than partial. +func TestDarwinUsageCountsATruncatedListingAndRefusesAFailedOne(t *testing.T) { + tasks := map[int]darwinproc.TaskInfo{10: task(10, 1, 10, 0)} + var kids []int + for pid := 100; pid < 400; pid++ { + tasks[pid] = task(pid, 10, 10, 0) + kids = append(kids, pid) + } + tree := &scriptedTree{tasks: tasks, children: map[int][]int{10: kids}, listErr: map[int]error{10: unix.ENOMEM}} + tree.use(t) + if usage, ok := subtreeUsage(10); !ok || usage.Processes != 301 || !usage.CPUPartial { + t.Fatalf("truncated listing: usage = %+v, %v, want the 301 it named and a partial CPU sum", usage, ok) + } + tree.listErr = map[int]error{10: unix.EPERM} + if usage, ok := subtreeUsage(10); ok { + t.Fatalf("a listing that failed was read as a partial tree: %+v", usage) + } +} + +// A tree past the walk limit is read as far as the limit, with a bounded +// number of kernel reads, and is still far over the job bound's ceiling. +func TestDarwinUsageStopsAtTheWalkLimit(t *testing.T) { + tasks := map[int]darwinproc.TaskInfo{10: task(10, 1, 10, 0)} + children := map[int][]int{} + parent, next := 10, 11 + for i := 0; i < 5000; i++ { + tasks[next] = task(next, parent, 10, 0) + children[parent] = append(children[parent], next) + if i%3 == 2 { + parent = next + } + next++ + } + tree := &scriptedTree{tasks: tasks, children: children} + tree.use(t) + usage, ok := subtreeUsage(10) + if !ok || usage.Processes != usageWalkLimit || !usage.CPUPartial { + t.Fatalf("usage = %+v, %v, want exactly the walk limit %d and a partial CPU sum", usage, ok, usageWalkLimit) + } + if tree.taskRead > 2*usageWalkLimit+1 { + t.Fatalf("%d task reads for a walk limited to %d processes", tree.taskRead, usageWalkLimit) + } + if len(tree.listed) > usageWalkLimit { + t.Fatalf("%d child listings for a walk limited to %d processes", len(tree.listed), usageWalkLimit) + } +} + +// A child that exited between two reads leaves the tree, as on Linux; one that +// could not be read for another reason makes the CPU sum partial. +func TestDarwinUsageMarksAnUnreadableChildPartial(t *testing.T) { + tree := &scriptedTree{ + tasks: map[int]darwinproc.TaskInfo{10: task(10, 1, 10, 1)}, + children: map[int][]int{10: {11}}, + } + tree.use(t) + if usage, ok := subtreeUsage(10); !ok || usage.Processes != 1 || usage.CPUPartial { + t.Fatalf("an exited child: usage = %+v, %v, want the leader alone and a complete sum", usage, ok) + } + saved := usageReader.task + usageReader.task = func(pid int) (darwinproc.TaskInfo, error) { + if pid == 11 { + return darwinproc.TaskInfo{}, unix.EPERM + } + return saved(pid) + } + if usage, ok := subtreeUsage(10); !ok || usage.Processes != 1 || !usage.CPUPartial { + t.Fatalf("an unreadable child: usage = %+v, %v, want the leader and a partial sum", usage, ok) + } +} + +// A parent that read once and then could not be re-read leaves its children out +// of this reading only, so the sum is partial; one that exited does not. +func TestDarwinUsageMarksAnUnreadableParentPartial(t *testing.T) { + tree := &scriptedTree{ + tasks: map[int]darwinproc.TaskInfo{10: task(10, 1, 10, 1), 11: task(11, 10, 10, 1), 12: task(12, 11, 10, 9)}, + children: map[int][]int{10: {11}, 11: {12}}, + } + tree.use(t) + saved := usageReader.task + reads := map[int]int{} + failWith := unix.EPERM + usageReader.task = func(pid int) (darwinproc.TaskInfo, error) { + reads[pid]++ + if pid == 11 && reads[pid] == 2 { + return darwinproc.TaskInfo{}, failWith + } + return saved(pid) + } + if usage, ok := subtreeUsage(10); !ok || usage.Processes != 2 || !usage.CPUPartial { + t.Fatalf("a parent that could not be re-read: usage = %+v, %v, want 2 processes and a partial sum", usage, ok) + } + reads, failWith = map[int]int{}, unix.ESRCH + if usage, ok := subtreeUsage(10); !ok || usage.Processes != 2 || usage.CPUPartial { + t.Fatalf("a parent that exited: usage = %+v, %v, want 2 processes and a complete sum", usage, ok) + } +} + +// A parent's pid and its listed child's pid can both change hands right after +// the parent read the same again. The child read after that swap is a +// stranger's busy child and must not be charged to the job: a child is +// admitted only when its parent still reads as the same process after it. +func TestDarwinUsageAdmitsAChildOnlyWhileItsParentIsTheSame(t *testing.T) { + saved := usageReader + t.Cleanup(func() { usageReader = saved }) + leader, own := task(10, 1, 10, 1), task(11, 10, 10, 1) + strangerLeader, strangerChild := task(10, 1, 50, 0), task(11, 10, 50, 500) + strangerLeader.Start, strangerChild.Start = 90000, 91000 + strangers := map[int]darwinproc.TaskInfo{12: task(12, 11, 50, 300), 13: task(13, 11, 50, 300)} + leaderReads, swapped := 0, false + usageReader.task = func(pid int) (darwinproc.TaskInfo, error) { + switch { + case pid == 10 && swapped: + return strangerLeader, nil + case pid == 10: + // The swap lands just after the leader's second read. + if leaderReads++; leaderReads == 2 { + swapped = true + } + return leader, nil + case pid == 11 && swapped: + return strangerChild, nil + case pid == 11: + return own, nil + case swapped && strangers[pid].PID == pid: + return strangers[pid], nil + } + return darwinproc.TaskInfo{}, unix.ESRCH + } + usageReader.children = func(ppid int) ([]int, error) { + switch { + case ppid == 10: + return []int{11}, nil + case ppid == 11 && swapped: + return []int{12, 13}, nil + } + return nil, nil + } + usage, ok := subtreeUsage(10) + if !ok || usage.Processes != 2 || usage.CPUSeconds != 2 { + t.Fatalf("usage = %+v, %v, want the leader and its own child only (2 processes, 2 CPU seconds)", usage, ok) + } +} diff --git a/internal/processgroup/usage_other.go b/internal/processgroup/usage_other.go index 4ecf7f23a7..84077706f8 100644 --- a/internal/processgroup/usage_other.go +++ b/internal/processgroup/usage_other.go @@ -1,10 +1,8 @@ -//go:build !linux +//go:build !linux && !darwin package processgroup -// Usage has no answer where there is no /proc to read: the subtree's CPU time -// and process count are Linux's to report here. A platform that cannot say -// answers so, and a bound that cannot say never cuts — it gets the scheduler it -// had before the bound existed rather than a guess at somebody else's process -// group. +// Usage has no answer where the subtree's CPU time and process count cannot be +// read. A bound that cannot say never cuts, preserving the existing scheduler +// rather than guessing at somebody else's process group. func (g Group) Usage() (Usage, bool) { return Usage{}, false } diff --git a/internal/session/jobbound.go b/internal/session/jobbound.go index 8248be1fc7..d0ef847d13 100644 --- a/internal/session/jobbound.go +++ b/internal/session/jobbound.go @@ -58,8 +58,8 @@ package session // // The watch is this program's own and runs everywhere. It reads the subtree // through the process-group handle #1155 gave, which works wherever a process -// group is a process group; the CPU-and-process reading itself is /proc, so off -// Linux it says it cannot say and the bound never cuts ([processgroup.Usage]). +// group is a process group. Linux and macOS read the subtree's CPU and process +// count; other platforms cannot say and the bound never cuts ([processgroup.Usage]). // The subtree's USAGE needs no cgroup — a job is a process group and not a // cgroup. Whether a quota BINDS is a different question — how many cores the // subtree could ever reach — answered from the process's own cgroup CPU quota and @@ -179,10 +179,18 @@ func (b *subtreeBound) strike(now time.Time, usage processgroup.Usage, ok bool) return false } cores := 0.0 - if b.have && now.After(b.last) { - cores = (usage.CPUSeconds - b.lastCPU) / now.Sub(b.last).Seconds() + if usage.CPUPartial { + // A sum that left part of the tree out yields no rate, and it is no + // baseline for the next reading either: only the process count is + // judged, and the next complete reading arms the CPU half again. + b.have = false + } else { + if b.have && now.After(b.last) { + cores = (usage.CPUSeconds - b.lastCPU) / now.Sub(b.last).Seconds() + } + b.last, b.lastCPU, b.have = now, usage.CPUSeconds, true } - b.last, b.lastCPU, b.have, b.lastCores = now, usage.CPUSeconds, true, cores + b.lastCores = cores if usage.Processes > jobBoundProcesses || cores > b.ceilingCores() { b.strikes++ } else { diff --git a/internal/session/jobbound_test.go b/internal/session/jobbound_test.go index cfa0898075..2228ce7108 100644 --- a/internal/session/jobbound_test.go +++ b/internal/session/jobbound_test.go @@ -230,3 +230,39 @@ func TestASubtreeThatCannotBeReadIsNeverCut(t *testing.T) { t.Fatal("a subtree whose usage could not be read was cut: silence is not pressure") } } + +// TestAPartialCPUSumNeverMakesARate: a reading that left part of the tree out +// (macOS's walk limit, a truncated listing) still counts processes, but its CPU +// sum is not comparable with the next one. A storm of sleepers read partially +// for fourteen readings that then shrinks to a leader and one old child — whose +// lifetime CPU the partial sums never held — must not be cut on that jump. The +// storm starts after one complete, quiet reading, so the partial readings must +// also drop that earlier baseline rather than leave it waiting for the jump. +func TestAPartialCPUSumNeverMakesARate(t *testing.T) { + registry := &jobRegistry{} + one := &job{id: 1, kind: jobKindBash, state: jobRunning, done: make(chan struct{})} + readings := []processgroup.Usage{{Processes: 2, CPUSeconds: 1}} + for i := 0; i < jobBoundStrikes-1; i++ { + readings = append(readings, processgroup.Usage{Processes: 1100, CPUSeconds: 1, CPUPartial: true}) + } + readings = append(readings, processgroup.Usage{Processes: 2, CPUSeconds: 1001}, processgroup.Usage{Processes: 2, CPUSeconds: 1001}) + subtreeReadings(t, readings, 2*time.Second) + if driveBound(registry, one, newTestBound(), len(readings)) { + t.Fatal("a jump between a partial and a complete CPU sum was read as a storm and cut") + } +} + +// TestAPartialReadingStillCountsTowardsTheProcessBackstop: the count from a +// partial reading is a lower bound, and a fork storm read that way is cut. +func TestAPartialReadingStillCountsTowardsTheProcessBackstop(t *testing.T) { + registry := &jobRegistry{} + one := &job{id: 1, kind: jobKindBash, state: jobRunning, done: make(chan struct{})} + readings := storm(0.5, 1024, long, time.Second) + for i := range readings { + readings[i].CPUPartial = true + } + subtreeReadings(t, readings, time.Second) + if !driveBound(registry, one, newTestBound(), long) { + t.Fatal("a fork storm read partially was never cut by the process backstop") + } +}