From 5a43b97991e8e2d942aeeb141197e96336523c39 Mon Sep 17 00:00:00 2001 From: Alexsander Hamir Date: Mon, 6 Jul 2026 17:21:32 -0700 Subject: [PATCH 1/5] refactor(collect): add bounded parallelFor helper Introduce a worker-pool helper for source_lines subprocess fan-out with a capped concurrency policy based on GOMAXPROCS. Co-authored-by: Cursor --- engine/collect/parallel.go | 64 ++++++++++++++++++++++++++++ engine/collect/parallel_test.go | 74 +++++++++++++++++++++++++++++++++ 2 files changed, 138 insertions(+) create mode 100644 engine/collect/parallel.go create mode 100644 engine/collect/parallel_test.go diff --git a/engine/collect/parallel.go b/engine/collect/parallel.go new file mode 100644 index 0000000..51b62d5 --- /dev/null +++ b/engine/collect/parallel.go @@ -0,0 +1,64 @@ +package collect + +import ( + "runtime" + "sync" +) + +// defaultSourceLinesWorkers caps concurrent go tool pprof -list subprocesses. +// Each worker loads the profile binary into memory. +const defaultSourceLinesWorkers = 8 + +// sourceLinesWorkers returns the worker count for source_lines subprocess fan-out. +func sourceLinesWorkers(jobCount int) int { + if jobCount <= 0 { + return 0 + } + max := runtime.GOMAXPROCS(0) + if max < 1 { + max = 1 + } + if jobCount < max { + max = jobCount + } + if max > defaultSourceLinesWorkers { + max = defaultSourceLinesWorkers + } + return max +} + +// parallelFor runs fn(i) for i in [0,n) with at most workers goroutines. +// It returns per-index errors; nil entries mean success. All jobs run to completion. +func parallelFor(n, workers int, fn func(i int) error) []error { + if n == 0 { + return nil + } + errs := make([]error, n) + if n == 1 || workers <= 1 { + for i := 0; i < n; i++ { + errs[i] = fn(i) + } + return errs + } + if workers > n { + workers = n + } + jobs := make(chan int, n) + for i := 0; i < n; i++ { + jobs <- i + } + close(jobs) + + var wg sync.WaitGroup + wg.Add(workers) + for w := 0; w < workers; w++ { + go func() { + defer wg.Done() + for i := range jobs { + errs[i] = fn(i) + } + }() + } + wg.Wait() + return errs +} diff --git a/engine/collect/parallel_test.go b/engine/collect/parallel_test.go new file mode 100644 index 0000000..9379b6f --- /dev/null +++ b/engine/collect/parallel_test.go @@ -0,0 +1,74 @@ +package collect + +import ( + "errors" + "runtime" + "sync/atomic" + "testing" +) + +func TestParallelFor_allIndicesRun(t *testing.T) { + t.Parallel() + const n = 20 + var seen atomic.Int64 + errs := parallelFor(n, 4, func(i int) error { + seen.Add(1 << i) + return nil + }) + if len(errs) != n { + t.Fatalf("len(errs)=%d want %d", len(errs), n) + } + var mask int64 + for i := 0; i < n; i++ { + mask |= 1 << i + } + if seen.Load() != mask { + t.Fatalf("not all indices executed: seen=%b want=%b", seen.Load(), mask) + } +} + +func TestParallelFor_preservesErrorsByIndex(t *testing.T) { + t.Parallel() + wantErr := errors.New("boom") + errs := parallelFor(8, 4, func(i int) error { + if i == 2 || i == 5 { + return wantErr + } + return nil + }) + for i, err := range errs { + if i == 2 || i == 5 { + if !errors.Is(err, wantErr) { + t.Fatalf("index %d: got %v want %v", i, err, wantErr) + } + continue + } + if err != nil { + t.Fatalf("index %d: unexpected error %v", i, err) + } + } +} + +func TestSourceLinesWorkers_capsAtDefault(t *testing.T) { + t.Parallel() + if runtime.GOMAXPROCS(0) < defaultSourceLinesWorkers { + t.Skip("GOMAXPROCS below default cap") + } + if got := sourceLinesWorkers(100); got != defaultSourceLinesWorkers { + t.Fatalf("sourceLinesWorkers(100)=%d want %d", got, defaultSourceLinesWorkers) + } +} + +func TestSourceLinesWorkers_respectsJobCount(t *testing.T) { + t.Parallel() + if got := sourceLinesWorkers(3); got != 3 { + t.Fatalf("sourceLinesWorkers(3)=%d want 3", got) + } +} + +func TestSourceLinesWorkers_zeroJobs(t *testing.T) { + t.Parallel() + if got := sourceLinesWorkers(0); got != 0 { + t.Fatalf("sourceLinesWorkers(0)=%d want 0", got) + } +} From 12c258c2fea4e90e91049fe0e95080fb000ac9df Mon Sep 17 00:00:00 2001 From: Alexsander Hamir Date: Mon, 6 Jul 2026 17:22:03 -0700 Subject: [PATCH 2/5] perf(collect): parallelize per-function pprof -list Fan out source_lines extraction across a bounded worker pool while preserving index-ordered warning semantics. Co-authored-by: Cursor --- engine/collect/artifacts_pprof.go | 21 ++++++++++-------- engine/collect/artifacts_test.go | 37 +++++++++++++++++++++++++++++++ engine/tooling/fake_runner.go | 4 ++++ 3 files changed, 53 insertions(+), 9 deletions(-) diff --git a/engine/collect/artifacts_pprof.go b/engine/collect/artifacts_pprof.go index 1cdcb31..b159d6d 100644 --- a/engine/collect/artifacts_pprof.go +++ b/engine/collect/artifacts_pprof.go @@ -87,19 +87,22 @@ func writeFunctionListPprof(runner tooling.Runner, shortStem, fullSymbol, binary func getFunctionsOutput(runner tooling.Runner, entries []parser.FunctionListEntry, binaryPath, basePath string, session *termui.Session) error { const maxPerFunctionWarnings = 3 - var skipped int - for _, e := range entries { + errs := parallelFor(len(entries), sourceLinesWorkers(len(entries)), func(i int) error { + e := entries[i] out := filepath.Join(basePath, e.OutputStem+"."+workspace.TextExtension) - if err := writeFunctionListPprof(runner, e.OutputStem, e.FullSymbol, binaryPath, out); err != nil { - skipped++ - if session != nil && session.Interactive() { - if skipped <= maxPerFunctionWarnings { - session.Warn(fmt.Sprintf("skipping per-function pprof list for %s: %v", e.OutputStem, err)) - } - } + return writeFunctionListPprof(runner, e.OutputStem, e.FullSymbol, binaryPath, out) + }) + + var skipped int + for i, err := range errs { + if err == nil { continue } + skipped++ + if session != nil && session.Interactive() && skipped <= maxPerFunctionWarnings { + session.Warn(fmt.Sprintf("skipping per-function pprof list for %s: %v", entries[i].OutputStem, err)) + } } if skipped > maxPerFunctionWarnings && session != nil && session.Interactive() { session.Warn(fmt.Sprintf("… and %d more functions skipped", skipped-maxPerFunctionWarnings)) diff --git a/engine/collect/artifacts_test.go b/engine/collect/artifacts_test.go index 87d1917..1d72377 100644 --- a/engine/collect/artifacts_test.go +++ b/engine/collect/artifacts_test.go @@ -101,3 +101,40 @@ func TestGetFunctionsOutput_fakeRunner(t *testing.T) { t.Fatalf("expected output file: %v", statErr) } } + +func TestGetFunctionsOutput_parallelFakeRunner(t *testing.T) { + t.Parallel() + cpuPath := testpaths.MustAsset(t, "fixtures", filterFixtureCPU) + entries, err := parser.GetFunctionListEntriesV2(cpuPath, config.FunctionFilter{}) + if err != nil { + t.Fatal(err) + } + const want = 5 + if len(entries) < want { + t.Fatalf("fixture should yield at least %d function entries, got %d", want, len(entries)) + } + entries = entries[:want] + + outs := make([][]byte, len(entries)) + for i, e := range entries { + outs[i] = []byte("list output for " + e.OutputStem) + } + runner := &tooling.FakeRunner{Out: outs} + dir := t.TempDir() + if outErr := getFunctionsOutput(runner, entries, cpuPath, dir, nil); outErr != nil { + t.Fatal(outErr) + } + if len(runner.Runs) != len(entries) { + t.Fatalf("expected %d pprof runs, got %d", len(entries), len(runner.Runs)) + } + for _, e := range entries { + outFile := filepath.Join(dir, e.OutputStem+"."+workspace.TextExtension) + data, readErr := os.ReadFile(outFile) + if readErr != nil { + t.Fatalf("read %s: %v", outFile, readErr) + } + if len(data) == 0 { + t.Fatalf("expected non-empty list output for %s", e.OutputStem) + } + } +} diff --git a/engine/tooling/fake_runner.go b/engine/tooling/fake_runner.go index 65b6d6f..2d30178 100644 --- a/engine/tooling/fake_runner.go +++ b/engine/tooling/fake_runner.go @@ -3,6 +3,7 @@ package tooling import ( "context" "errors" + "sync" ) // FakeRun records one invocation of [FakeRunner.Run]. @@ -13,6 +14,7 @@ type FakeRun struct { // FakeRunner implements [Runner] for tests. It records each run and returns configured results in order. type FakeRunner struct { + mu sync.Mutex Runs []FakeRun Out [][]byte Err []error @@ -25,6 +27,8 @@ func (f *FakeRunner) Run(ctx context.Context, argv []string, opts RunOpts) ([]by if f == nil { return nil, errors.New("tooling: nil FakeRunner") } + f.mu.Lock() + defer f.mu.Unlock() f.Runs = append(f.Runs, FakeRun{Argv: append([]string(nil), argv...), Opts: opts}) i := f.n f.n++ From a088edb6dd5f9edb19ccf1ac66bbd8917a86c128 Mon Sep 17 00:00:00 2001 From: Alexsander Hamir Date: Mon, 6 Jul 2026 17:22:11 -0700 Subject: [PATCH 3/5] perf(collect): parallelize source_lines across profile kinds Collect cpu, memory, and other profile kinds concurrently during step 3 of the auto pipeline. Co-authored-by: Cursor --- engine/collect/pipeline.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/engine/collect/pipeline.go b/engine/collect/pipeline.go index 3265518..b16fd18 100644 --- a/engine/collect/pipeline.go +++ b/engine/collect/pipeline.go @@ -92,7 +92,9 @@ func collectProfileFunctions(runner tooling.Runner, args *config.CollectionArgs, return err } - for _, profile := range args.Profiles { + profiles := args.Profiles + errs := parallelFor(len(profiles), sourceLinesWorkers(len(profiles)), func(i int) error { + profile := profiles[i] fnDir := layout.SourceLinesDir(profile, args.BenchmarkName) if mkdirErr := os.MkdirAll(fnDir, workspace.PermDir); mkdirErr != nil { return fmt.Errorf("failed to create output directory: %w", mkdirErr) @@ -107,7 +109,13 @@ func collectProfileFunctions(runner tooling.Runner, args *config.CollectionArgs, if fnErr := getFunctionsOutput(runner, listEntries, binPath, fnDir, session); fnErr != nil { return fmt.Errorf("getAllFunctionsPprofContents failed: %w", fnErr) } - } + return nil + }) + for i, err := range errs { + if err != nil { + return fmt.Errorf("profile %s: %w", profiles[i], err) + } + } return nil } From c7edee269fa5b939d4e9bb91db0a70627a4cc4ff Mon Sep 17 00:00:00 2001 From: Alexsander Hamir Date: Mon, 6 Jul 2026 17:22:57 -0700 Subject: [PATCH 4/5] docs: record parallel source_lines collection design Add an ADR and update contributor docs for bounded pprof -list fan-out during step 3. Co-authored-by: Cursor --- CODEBASE_DESIGN.md | 2 +- docs/collect-request-flow.md | 2 + docs/design/source-lines-parallelism.md | 64 +++++++++++++++++++++++++ engine/collect/doc.go | 1 + 4 files changed, 68 insertions(+), 1 deletion(-) create mode 100644 docs/design/source-lines-parallelism.md diff --git a/CODEBASE_DESIGN.md b/CODEBASE_DESIGN.md index 0daaa7a..0708cc6 100644 --- a/CODEBASE_DESIGN.md +++ b/CODEBASE_DESIGN.md @@ -100,7 +100,7 @@ Benchmark discovery: [`engine/collect/discovery.go`](engine/collect/discovery.go 1. [`collect.RunAuto`](engine/collect/entry.go) loads optional `prof.json` via [`config.Load`](internal/config/load.go). 2. Creates `.prof//` via [`collect/layout.go`](engine/collect/layout.go) and [`workspace.CleanOrCreateTag`](internal/workspace/tag.go). -3. Per benchmark, three TTY-gated stderr steps via [`termui.Session`](internal/termui/progress.go) in [`pipeline.go`](engine/collect/pipeline.go), preceded by a **Preparing** stage in [`entry.go`](engine/collect/entry.go): **Running benchmark** (`go test` + artifact move), **Collecting profiles** ([`processProfiles`](engine/collect/profiles.go)), **Collecting function profiles** (parser + per-function `pprof -list`). Interactive TTY keeps a persistent stage log (`✓` done lines + stage-scoped warnings); non-TTY keeps `slog` stage logs. +3. Per benchmark, three TTY-gated stderr steps via [`termui.Session`](internal/termui/progress.go) in [`pipeline.go`](engine/collect/pipeline.go), preceded by a **Preparing** stage in [`entry.go`](engine/collect/entry.go): **Running benchmark** (`go test` + artifact move), **Collecting profiles** ([`processProfiles`](engine/collect/profiles.go)), **Collecting function profiles** (parser + per-function `pprof -list`, with bounded parallel fan-out across profile kinds and functions — see [docs/design/source-lines-parallelism.md](docs/design/source-lines-parallelism.md)). Interactive TTY keeps a persistent stage log (`✓` done lines + stage-scoped warnings); non-TTY keeps `slog` stage logs. ### Manual ingest (`prof manual`) diff --git a/docs/collect-request-flow.md b/docs/collect-request-flow.md index 0f876fa..f682a4f 100644 --- a/docs/collect-request-flow.md +++ b/docs/collect-request-flow.md @@ -187,6 +187,8 @@ Resolved function filters for each benchmark come from `config.ResolveCollection - For each successfully processed profile, [`parser.GetFunctionListEntriesV2`](../parser/) reads the binary and applies config filters. - `go tool pprof -list` output is written under `source_lines/cpu/BenchmarkMatrixMultiplication/` and `source_lines/memory/BenchmarkMatrixMultiplication/`. +**Concurrency:** profile kinds and per-function `-list` subprocesses run with a bounded worker pool (`min(jobCount, GOMAXPROCS, 8)`). Artifacts, filters, and argv are unchanged; see [source-lines-parallelism.md](../design/source-lines-parallelism.md). + When all benchmarks finish, prof logs collection success and returns. ## Output layout (example) diff --git a/docs/design/source-lines-parallelism.md b/docs/design/source-lines-parallelism.md new file mode 100644 index 0000000..ec8ffab --- /dev/null +++ b/docs/design/source-lines-parallelism.md @@ -0,0 +1,64 @@ +# Parallel source_lines collection + +## Context + +Step 3 of the auto collect pipeline writes per-function `go tool pprof -list` extracts under `.prof//source_lines///`. Each extract is an independent subprocess that reads the same profile binary and writes a unique output file. + +Previously, prof ran these subprocesses sequentially: every profile kind in order, then every filtered function in order. On profiles with many in-scope symbols, this step dominated wall-clock time—especially on Windows where process spawn latency is high. + +## Decision + +Collect `source_lines` with **bounded parallelism** at two layers: + +1. **Profile kinds** (`cpu`, `memory`, …) run concurrently inside [`collectProfileFunctions`](../../engine/collect/pipeline.go). +2. **Per-function `-list` calls** run concurrently inside [`getFunctionsOutput`](../../engine/collect/artifacts_pprof.go), shared by auto and manual collection. + +Both layers use a fixed worker pool ([`parallelFor`](../../engine/collect/parallel.go)) with worker count: + +``` +min(jobCount, GOMAXPROCS, 8) +``` + +Each worker still invokes the same `go tool pprof -list=` argv as before. Filters, output paths, filenames, and warn-and-continue semantics are unchanged. + +## Alternatives considered + +| Alternative | Why not chosen | +| --- | --- | +| Unbounded goroutine per function | Risk of memory and disk contention when dozens of `pprof` processes load the same binary | +| `golang.org/x/sync/errgroup` | Adds a dependency; indexed error aggregation still requires a full result buffer | +| In-process `-list` via `github.com/google/pprof` | Output formatting may diverge from the CLI; higher implementation and maintenance cost | +| User-facing `--workers` flag | No evidence yet that the fixed cap is wrong for typical machines | + +## Consequences + +**Positive** + +- Wall-clock time for step 3 drops roughly with worker count (up to the cap) for large function lists, and across profile kinds when multiple profiles are collected. +- Auto and manual paths both benefit from function-level parallelism via shared `getFunctionsOutput`. + +**Neutral / trade-offs** + +- Up to eight concurrent `go tool pprof` processes per profile (and concurrent profiles multiply subprocess count). Disk and CPU contention are possible on very slow storage; the cap limits blast radius. +- Interactive warnings for skipped per-function lists remain capped at three plus a summary line; ordering is by function index within a profile, not subprocess completion order. + +## Invariants + +- Same `FunctionListEntry` set after `prof.json` filters. +- Same `.txt` paths and `pprof -list` subprocess argv. +- Per-function failures warn and continue; they do not fail the collect run. +- Profile-level failures (missing binary parse, mkdir error) still fail the benchmark step in profile slice order. + +## Verification + +```bash +go test ./engine/collect/ ./engine/tooling/ -count=1 +go test ./... +``` + +Optional local equivalence check: run `prof auto` before and after, then `diff -r` the `source_lines/` trees—they should match when the same filters and benchmarks are used. + +## See also + +- [Collect request flow — Step 3](../collect-request-flow.md#step-3--per-function-extracts) +- [CODEBASE_DESIGN.md — Profile pipelines](../../CODEBASE_DESIGN.md#profile-pipelines) diff --git a/engine/collect/doc.go b/engine/collect/doc.go index 5259d31..04007b5 100644 --- a/engine/collect/doc.go +++ b/engine/collect/doc.go @@ -1,4 +1,5 @@ // Package collect runs auto and manual profile collection under .prof//. // Artifacts are grouped by data domain: profiles, measurements, hotspots, // source_lines, and call_graphs (see internal/workspace.TagLayout). +// source_lines extraction fans out go tool pprof -list subprocesses with bounded parallelism. package collect From 3d90eb7fe60d04335807db661c5aa50ec1e65df0 Mon Sep 17 00:00:00 2001 From: Alexsander Hamir Date: Mon, 6 Jul 2026 17:23:24 -0700 Subject: [PATCH 5/5] fix(collect): satisfy golangci-lint in parallel helper Co-authored-by: Cursor --- engine/collect/parallel.go | 22 +++++++++++----------- engine/collect/parallel_test.go | 2 +- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/engine/collect/parallel.go b/engine/collect/parallel.go index 51b62d5..d0500f7 100644 --- a/engine/collect/parallel.go +++ b/engine/collect/parallel.go @@ -14,17 +14,17 @@ func sourceLinesWorkers(jobCount int) int { if jobCount <= 0 { return 0 } - max := runtime.GOMAXPROCS(0) - if max < 1 { - max = 1 + workers := runtime.GOMAXPROCS(0) + if workers < 1 { + workers = 1 } - if jobCount < max { - max = jobCount + if jobCount < workers { + workers = jobCount } - if max > defaultSourceLinesWorkers { - max = defaultSourceLinesWorkers + if workers > defaultSourceLinesWorkers { + workers = defaultSourceLinesWorkers } - return max + return workers } // parallelFor runs fn(i) for i in [0,n) with at most workers goroutines. @@ -35,7 +35,7 @@ func parallelFor(n, workers int, fn func(i int) error) []error { } errs := make([]error, n) if n == 1 || workers <= 1 { - for i := 0; i < n; i++ { + for i := range n { errs[i] = fn(i) } return errs @@ -44,14 +44,14 @@ func parallelFor(n, workers int, fn func(i int) error) []error { workers = n } jobs := make(chan int, n) - for i := 0; i < n; i++ { + for i := range n { jobs <- i } close(jobs) var wg sync.WaitGroup wg.Add(workers) - for w := 0; w < workers; w++ { + for range workers { go func() { defer wg.Done() for i := range jobs { diff --git a/engine/collect/parallel_test.go b/engine/collect/parallel_test.go index 9379b6f..d32266a 100644 --- a/engine/collect/parallel_test.go +++ b/engine/collect/parallel_test.go @@ -19,7 +19,7 @@ func TestParallelFor_allIndicesRun(t *testing.T) { t.Fatalf("len(errs)=%d want %d", len(errs), n) } var mask int64 - for i := 0; i < n; i++ { + for i := range n { mask |= 1 << i } if seen.Load() != mask {