perf(collect): parallelize source_lines pprof -list extraction

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/<tag>/` 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`)
 

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)

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/<tag>/source_lines/<profile>/<benchmark>/`. 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=<pattern>` 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)

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))

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)
+		}
+	}
+}

engine/collect/doc.go

@@ -1,4 +1,5 @@
 // Package collect runs auto and manual profile collection under .prof/<tag>/.
 // 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

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
+	}
+	workers := runtime.GOMAXPROCS(0)
+	if workers < 1 {
+		workers = 1
+	}
+	if jobCount < workers {
+		workers = jobCount
+	}
+	if workers > defaultSourceLinesWorkers {
+		workers = defaultSourceLinesWorkers
+	}
+	return workers
+}
+
+// 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 := range n {
+			errs[i] = fn(i)
+		}
+		return errs
+	}
+	if workers > n {
+		workers = n
+	}
+	jobs := make(chan int, n)
+	for i := range n {
+		jobs <- i
+	}
+	close(jobs)
+
+	var wg sync.WaitGroup
+	wg.Add(workers)
+	for range workers {
+		go func() {
+			defer wg.Done()
+			for i := range jobs {
+				errs[i] = fn(i)
+			}
+		}()
+	}
+	wg.Wait()
+	return errs
+}

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 := range n {
+		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)
+	}
+}

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
 }

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++