diff --git a/docs/benchmarks/bounded-two-pass-results.json b/docs/benchmarks/bounded-two-pass-results.json new file mode 100644 index 000000000..34178cc8b --- /dev/null +++ b/docs/benchmarks/bounded-two-pass-results.json @@ -0,0 +1,310 @@ +{ + "platform": "macOS 26.6.2 arm64, 24 GiB RAM", + "compiler": "Apple clang 21.0.0", + "base_commit": "aa44c28ea5ea82a5f811f0bace4f7857a68cac80", + "baseline_binary_sha256": "6856b8ffb475ce0f23b9b343ca8265092b53e1b5733809e8cfb4b6b3831194a9", + "feature_binary_sha256": "d449a572586a8921cdd95b8dd2e3249ebb2a18df9204a628b8f527eb675d13a9", + "workers": 4, + "runs": [ + { + "corpus": "synthetic-typescript", + "variant": "baseline", + "run": 1, + "worker_peak_rss_bytes": 679870464, + "worker_wall_s": 2.691, + "pipeline_ms": 1448, + "phase_peak_mb": 481, + "persisted_nodes": 22533, + "persisted_edges": 57348, + "surfaces": 2049, + "parse_partial_count": 0, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "synthetic-typescript", + "variant": "baseline", + "run": 2, + "worker_peak_rss_bytes": 680443904, + "worker_wall_s": 2.638, + "pipeline_ms": 1410, + "phase_peak_mb": 481, + "persisted_nodes": 22533, + "persisted_edges": 57348, + "surfaces": 2049, + "parse_partial_count": 0, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "synthetic-typescript", + "variant": "baseline", + "run": 3, + "worker_peak_rss_bytes": 680329216, + "worker_wall_s": 2.644, + "pipeline_ms": 1423, + "phase_peak_mb": 482, + "persisted_nodes": 22533, + "persisted_edges": 57348, + "surfaces": 2049, + "parse_partial_count": 0, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "synthetic-typescript", + "variant": "feature-off", + "run": 1, + "worker_peak_rss_bytes": 681820160, + "worker_wall_s": 2.651, + "pipeline_ms": 1418, + "phase_peak_mb": 481, + "persisted_nodes": 22533, + "persisted_edges": 57348, + "surfaces": 2049, + "parse_partial_count": 0, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "synthetic-typescript", + "variant": "feature-off", + "run": 2, + "worker_peak_rss_bytes": 681574400, + "worker_wall_s": 2.669, + "pipeline_ms": 1450, + "phase_peak_mb": 482, + "persisted_nodes": 22533, + "persisted_edges": 57348, + "surfaces": 2049, + "parse_partial_count": 0, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "synthetic-typescript", + "variant": "feature-off", + "run": 3, + "worker_peak_rss_bytes": 679428096, + "worker_wall_s": 2.663, + "pipeline_ms": 1445, + "phase_peak_mb": 482, + "persisted_nodes": 22533, + "persisted_edges": 57348, + "surfaces": 2049, + "parse_partial_count": 0, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "synthetic-typescript", + "variant": "batch-128", + "run": 1, + "worker_peak_rss_bytes": 315408384, + "worker_wall_s": 3.227, + "pipeline_ms": 2002, + "phase_peak_mb": 141, + "persisted_nodes": 22533, + "persisted_edges": 57348, + "surfaces": 2049, + "parse_partial_count": 0, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "synthetic-typescript", + "variant": "batch-128", + "run": 2, + "worker_peak_rss_bytes": 315473920, + "worker_wall_s": 3.231, + "pipeline_ms": 2024, + "phase_peak_mb": 141, + "persisted_nodes": 22533, + "persisted_edges": 57348, + "surfaces": 2049, + "parse_partial_count": 0, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "synthetic-typescript", + "variant": "batch-128", + "run": 3, + "worker_peak_rss_bytes": 315654144, + "worker_wall_s": 3.242, + "pipeline_ms": 2026, + "phase_peak_mb": 141, + "persisted_nodes": 22533, + "persisted_edges": 57348, + "surfaces": 2049, + "parse_partial_count": 0, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "repomix-e3b15a4", + "variant": "baseline", + "run": 1, + "worker_peak_rss_bytes": 288817152, + "worker_wall_s": 2.055, + "pipeline_ms": 834, + "phase_peak_mb": 197, + "persisted_nodes": 12051, + "persisted_edges": 19154, + "surfaces": 1073, + "parse_partial_count": 4, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "repomix-e3b15a4", + "variant": "baseline", + "run": 2, + "worker_peak_rss_bytes": 284131328, + "worker_wall_s": 2.082, + "pipeline_ms": 835, + "phase_peak_mb": 195, + "persisted_nodes": 12051, + "persisted_edges": 19154, + "surfaces": 1073, + "parse_partial_count": 4, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "repomix-e3b15a4", + "variant": "baseline", + "run": 3, + "worker_peak_rss_bytes": 285491200, + "worker_wall_s": 2.053, + "pipeline_ms": 828, + "phase_peak_mb": 195, + "persisted_nodes": 12051, + "persisted_edges": 19154, + "surfaces": 1073, + "parse_partial_count": 4, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "repomix-e3b15a4", + "variant": "batch-128", + "run": 1, + "worker_peak_rss_bytes": 207077376, + "worker_wall_s": 2.405, + "pipeline_ms": 1193, + "phase_peak_mb": 126, + "persisted_nodes": 12051, + "persisted_edges": 19154, + "surfaces": 1073, + "parse_partial_count": 4, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "repomix-e3b15a4", + "variant": "batch-128", + "run": 2, + "worker_peak_rss_bytes": 203046912, + "worker_wall_s": 2.453, + "pipeline_ms": 1233, + "phase_peak_mb": 125, + "persisted_nodes": 12051, + "persisted_edges": 19154, + "surfaces": 1073, + "parse_partial_count": 4, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + }, + { + "corpus": "repomix-e3b15a4", + "variant": "batch-128", + "run": 3, + "worker_peak_rss_bytes": 203800576, + "worker_wall_s": 2.417, + "pipeline_ms": 1200, + "phase_peak_mb": 124, + "persisted_nodes": 12051, + "persisted_edges": 19154, + "surfaces": 1073, + "parse_partial_count": 4, + "integrity": "ok", + "equals_baseline": { + "nodes": true, + "node_properties": true, + "edges": true, + "surfaces": true + } + } + ] +} diff --git a/docs/benchmarks/bounded-two-pass.md b/docs/benchmarks/bounded-two-pass.md new file mode 100644 index 000000000..6d7a70c5b --- /dev/null +++ b/docs/benchmarks/bounded-two-pass.md @@ -0,0 +1,108 @@ +# Bounded two-pass indexing + +This is the batching-only follow-up to #1925. `CBM_STREAMING_BATCH_FILES=128` +opts into two-pass extraction when discovery finds more than 128 files. Valid +values are 1–4096; an invalid value fails the extraction phase. With the variable +unset, the existing path is used. The option works with one or multiple workers. + +The bound is on live **full extraction results**, not total process memory. +Pass A registers definitions and keeps a compact definition/import surface. +Once the whole registry, namespace map and import graph are available, it builds +cross-file LSP definitions and registries. Pass B re-extracts one batch, resolves +its relationships and releases it. The global graph and compact surface still +scale with repository size; one huge file can still be expensive. No automatic +threshold, global scheduler, admission budget or platform RSS policy is added. + +Java participates in the shared cross registry. `CBM_DISABLE_LSP_CROSS` retains +its existing presence-based semantics in both paths (including a value of `0`). +ObjectScript macros and package manifests are collected for the whole repository. +Pass B reuses definition nodes, deduplicates parse/skip diagnostics, and serial +Go implementation/override scans run once after all batches. The final manifest +and publication guards remain responsible for detecting source changes during a run. + +The TSNodeStack changes from #1925 are excluded: this branch retains main's +#2013 scratch-lifetime implementation. Result-array allocation is untouched. +The worker scoping and complexity-order changes belong to #2076 and #2079; +there is no `compare_node_qn` rewrite in this branch. + +## Measurements + +Native macOS 26.6.2 arm64, 24 GiB RAM, Apple clang 21.0.0, production `-O2`, +`CBM_WORKERS=4`, three fresh runs per row. Baseline is upstream +`aa44c28ea5ea82a5f811f0bace4f7857a68cac80`. The generated TypeScript corpus has +2,049 files / 1,893,350 source bytes. Repomix is a `git archive` of +`e3b15a406ed78d8a463620a032a059ce911bfc0e`, without dependencies or git history. + +**Whole-worker peak RSS**, measured by the parent's `wait4().ru_maxrss` at exit, +and wall seconds (including worker startup, executable verification and publication): + +- Generated TypeScript, baseline: 648.4 / 648.9 / 648.8 MiB; 2.691 / 2.638 / 2.644 s. +- Same source, feature off: 650.2 / 650.0 / 648.0 MiB; 2.651 / 2.669 / 2.663 s. +- Same source, batch 128: 300.8 / 300.9 / 301.0 MiB; 3.227 / 3.231 / 3.242 s. +- Repomix, baseline: 275.4 / 271.0 / 272.3 MiB; 2.055 / 2.082 / 2.053 s. +- Repomix, batch 128: 197.5 / 193.6 / 194.4 MiB; 2.405 / 2.453 / 2.417 s. + +The generated corpus's median process peak falls about 54%, at about 22% more +worker wall time. Repomix's median peak falls about 29%, at about 18% more wall +time. These are corpus-specific measurements, not an RSS guarantee. The +`mem.phase` figures are lower, since they only sample selected phases; they are +recorded separately and are not reported as the whole-worker peak. + +Across all 15 measured worker runs, each corpus's normalized **persisted nodes, +node properties, edges including properties, and LSP surface JSON** match its +baseline exactly. IDs are replaced with qualified names; JSON object key order +is normalized. The TypeScript database contains 22,533 nodes / 57,348 edges; +Repomix contains 12,051 nodes / 19,154 edges. Repomix's parse_partial_count stays +4; the synthetic count stays 0. SQLite integrity checks return `ok`. Persisted +counts are taken from SQLite, not inferred from the earlier CLI response counts. + +Machine-readable measurements and equality results are in +[bounded-two-pass-results.json](bounded-two-pass-results.json). Raw logs, response +JSON and database digests are produced by the reproduction script. + +## Reproduce + +Build the baseline and this branch in separate worktrees with `scripts/build.sh`. +Keep the corpus at the same absolute path for all runs: project identity and +qualified names incorporate that path. `BASE_BINARY` and `FEATURE_BINARY` below +must point to the respective production executables. `RUNS` is a new output +directory; `RUNTIME_PARENT` is a short private directory owned by your account +(short enough for Unix socket paths). Every run creates its own cache and runtime. + +```sh +python3 scripts/benchmark-streaming.py --binary "$BASE_BINARY" --repo "$CORPUS" \ + --output "$RUNS/baseline" --runtime-parent "$RUNTIME_PARENT" --runs 3 +python3 scripts/benchmark-streaming.py --binary "$FEATURE_BINARY" --repo "$CORPUS" \ + --output "$RUNS/feature-off" --runtime-parent "$RUNTIME_PARENT" --runs 3 +python3 scripts/benchmark-streaming.py --binary "$FEATURE_BINARY" --repo "$CORPUS" \ + --output "$RUNS/batch-128" --runtime-parent "$RUNTIME_PARENT" --runs 3 --batch 128 +``` + +The harness invokes the fingerprint-validated internal worker directly, so the +measured child is the indexing worker rather than a CLI waiting on a daemon. +It requires Python 3 and `os.wait4` (macOS/Linux). This measurement harness is +not evidence of a native Windows run; Windows coverage comes from the C tests +in the project's platform CI. No CI result is asserted by this document. + +Generate the synthetic corpus in an empty directory: + +```python +from pathlib import Path +r = Path("corpus") +r.mkdir() +(r / "shared.ts").write_text("export function helper(x: number): number { return x + 1; }\n") +for i in range(2048): + methods = "\n".join( + f" method{j}(x: number): number {{ let sum = 0; for (let k = 0; k < x; k++) {{ sum += helper(k); }} return sum; }}" + for j in range(8) + ) + (r / f"unit{i:04d}.ts").write_text( + f"import {{ helper }} from './shared';\nexport class Unit{i} {{\n{methods}\n}}\n" + ) +``` + +The C pipeline tests compare batch sizes 1 and 7 (one and four workers), Java +cross-file calls, Python/TypeScript inheritance across batches, disabled cross-LSP, +full graph relationship/property sets, persisted surfaces and partial-parse +counts. A separate lifetime test frees the original Go/Python/Java/Rust extraction +before rebuilding the compact surface under ASan/UBSan. diff --git a/scripts/benchmark-streaming.py b/scripts/benchmark-streaming.py new file mode 100644 index 000000000..61803f172 --- /dev/null +++ b/scripts/benchmark-streaming.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +"""Isolated batch-index A/B. Run sequentially; retain raw logs and graph digests.""" + +import argparse +import hashlib +import json +import os +import pathlib +import re +import sqlite3 +import subprocess +import sys +import tempfile +import time + +p = argparse.ArgumentParser() +p.add_argument("--binary", required=True) +p.add_argument("--repo", required=True) +p.add_argument("--output", required=True) +p.add_argument("--batch", type=int) +p.add_argument("--runs", type=int, default=3) +p.add_argument("--runtime-parent", required=True) +args = p.parse_args() +if not hasattr(os, "wait4"): + p.error("Native worker RSS measurement currently requires macOS or Linux") +if args.runs < 1 or (args.batch is not None and not 1 <= args.batch <= 4096): + p.error("runs must be positive; batch must be between 1 and 4096") +root = pathlib.Path(args.output).resolve() +root.mkdir(parents=True, exist_ok=True) +binary = pathlib.Path(args.binary).resolve() +repo = pathlib.Path(args.repo).resolve() + + +def digest_rows(db, query, normalize=None): + rows = [list(row) for row in db.execute(query)] + if normalize: + rows = [normalize(row) for row in rows] + rows.sort(key=lambda row: json.dumps(row, sort_keys=True)) + return { + "count": len(rows), + "sha256": hashlib.sha256( + json.dumps(rows, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest(), + } + + +def props(row): + row[-1] = json.loads(row[-1] or "{}") + return row + + +fingerprint = hashlib.sha256(binary.read_bytes()).hexdigest() +records = [] +for i in range(args.runs): + run = root / f"run{i+1}" + run.mkdir() + cache = run / "cache" + cache.mkdir(mode=0o700) + runtime = pathlib.Path(tempfile.mkdtemp(prefix="cbmb-", dir=args.runtime_parent)) + env = os.environ.copy() + for key in ( + "CBM_STREAMING_BATCH_FILES", + "CBM_DISABLE_LSP_CROSS", + "CBM_INDEX_SINGLE_THREAD", + ): + env.pop(key, None) + env.update( + CBM_RUNTIME_DIR=str(runtime), + CBM_CACHE_DIR=str(cache), + CBM_ALLOWED_ROOT=str(repo), + CBM_SESSION_REPO_ROOT=str(repo), + CBM_WORKERS="4", + CBM_PROFILE="1", + CBM_LOG_LEVEL="info", + ) + if args.batch: + env["CBM_STREAMING_BATCH_FILES"] = str(args.batch) + cmd = [ + str(binary), + "cli", + "--index-worker", + "--index-worker-build", + fingerprint, + "index_repository", + json.dumps({"repo_path": str(repo), "mode": "full"}), + "--response-out", + str(run / "response.json"), + ] + start = time.monotonic() + with (run / "stdout.json").open("w") as out, (run / "stderr.log").open("w") as err: + process = subprocess.Popen(cmd, cwd=repo, env=env, stdout=out, stderr=err) + while True: + pid, status, usage = os.wait4(process.pid, os.WNOHANG) + if pid: + process.returncode = os.waitstatus_to_exitcode(status) + break + if time.monotonic() - start > 600: + process.kill() + time.sleep(0.02) + result = process + elapsed = time.monotonic() - start + record = { + "binary": str(binary), + "binary_sha256": fingerprint, + "repo": str(repo), + "batch": args.batch, + "run": i + 1, + "returncode": result.returncode, + "worker_wall_s": round(elapsed, 3), + "worker_peak_rss_bytes": usage.ru_maxrss + * (1 if sys.platform == "darwin" else 1024), + "runtime": str(runtime), + "command": cmd, + } + if result.returncode: + records.append(record) + print(json.dumps(record), flush=True) + break + response = json.loads((run / "response.json").read_text()) + record["result"] = response.get("structuredContent") or ( + json.loads(response["content"][0]["text"]) + if "content" in response + else response + ) + logs = ( + (run / "stderr.log").read_text(errors="replace") + + "\n" + + "\n".join( + f.read_text(errors="replace") + for f in (cache / "logs").glob(".worker-log-*") + ) + ) + (run / "worker.log").write_text(logs) + peaks = [int(x) for x in re.findall(r"msg=mem.phase[^\n]*peak_mb=(\d+)", logs)] + timings = re.findall(r"msg=pipeline.done[^\n]*elapsed_ms=(\d+)", logs) + record["peak_mb"] = max(peaks) if peaks else None + record["pipeline_ms"] = int(timings[-1]) if timings else None + record["batch_peak_rss_mb"] = max( + [ + int(x) + for x in re.findall( + r"msg=mem.phase phase=streaming_pass_[ab]_batch rss_mb=(\d+)", logs + ) + ] + or [0] + ) + dbpath = next(f for f in cache.glob("*.db") if f.name != "_config.db") + db = sqlite3.connect(dbpath) + record["nodes"] = digest_rows( + db, "select label,name,qualified_name,file_path,start_line,end_line from nodes" + ) + record["node_properties"] = digest_rows( + db, "select qualified_name,properties from nodes", props + ) + record["edges"] = digest_rows( + db, + "select s.qualified_name,t.qualified_name,e.type,e.properties from edges e join nodes s on s.id=e.source_id join nodes t on t.id=e.target_id", + props, + ) + record["surfaces"] = digest_rows( + db, "select rel_path,surface_sha,defs_json from lsp_surface" + ) + record["integrity"] = db.execute("pragma integrity_check").fetchone()[0] + db.close() + records.append(record) + (root / "results.json").write_text(json.dumps(records, indent=2)) + print(json.dumps(record), flush=True) +(root / "results.json").write_text(json.dumps(records, indent=2)) + +if any(row["returncode"] != 0 for row in records): + sys.exit(1) diff --git a/src/pipeline/lsp_surface.c b/src/pipeline/lsp_surface.c index 369f7f7e5..6c9735d67 100644 --- a/src/pipeline/lsp_surface.c +++ b/src/pipeline/lsp_surface.c @@ -16,6 +16,7 @@ * serialization therefore IS surface equality, and the sha over the * bytes is the early-cutoff key: a body edit reserializes identically. */ +#include "foundation/arena.h" #include "pipeline/lsp_surface.h" #include @@ -36,6 +37,97 @@ enum { SURFACE_CODEC_VERSION = 1 }; * edges from dependent SQL files. KEEP IN SYNC with pxc_map_label * (pass_lsp_cross.c) and incr_label_is_registry_symbol * (pipeline_incremental.c); the codec unit test cross-checks the three. */ +/* Keep only the definition/import surface needed after pass A. In particular, + * no call sites, usage arrays, ASTs, source bytes or body fingerprints survive. + * collect_all_defs must run after ALL files have entered the name registry and + * IMPORTS graph, so serializing already-resolved defs inside a batch is lossy. */ +static const char **surface_copy_strings(CBMArena *arena, const char *const *src, int count) { + if (!src) { + return NULL; + } + if (count < 0) { + count = 0; + while (src[count]) { + count++; + } + } + const char **dst = cbm_arena_alloc(arena, ((size_t)count + 1) * sizeof(*dst)); + if (!dst) { + return NULL; + } + for (int i = 0; i < count; i++) { + dst[i] = src[i] ? cbm_arena_strdup(arena, src[i]) : NULL; + if (src[i] && !dst[i]) { + return NULL; + } + } + dst[count] = NULL; + return dst; +} + +CBMFileResult *cbm_lsp_surface_copy_result(const CBMFileResult *src) { + CBMFileResult *dst = calloc(1, sizeof(*dst)); + if (!dst) { + return NULL; + } + cbm_arena_init_sized(&dst->arena, 1024); +#define COPY_STR(to, from, field) \ + do { \ + (to)->field = (from)->field ? cbm_arena_strdup(&dst->arena, (from)->field) : NULL; \ + if ((from)->field && !(to)->field) { \ + goto fail; \ + } \ + } while (0) +#define COPY_ARRAY(field) \ + do { \ + dst->field.count = src->field.count; \ + dst->field.items = \ + cbm_arena_calloc(&dst->arena, (size_t)src->field.count * sizeof(*dst->field.items)); \ + if (src->field.count && !dst->field.items) { \ + goto fail; \ + } \ + } while (0) + COPY_STR(dst, src, namespace_name); + COPY_ARRAY(defs); + COPY_ARRAY(imports); + COPY_ARRAY(impl_traits); + for (int i = 0; i < src->defs.count; i++) { + const CBMDefinition *from = &src->defs.items[i]; + CBMDefinition *to = &dst->defs.items[i]; + COPY_STR(to, from, name); + COPY_STR(to, from, qualified_name); + COPY_STR(to, from, label); + COPY_STR(to, from, parent_class); + COPY_STR(to, from, return_type); + COPY_STR(to, from, impl_trait); + to->is_abstract = from->is_abstract; + to->signature_param_count = from->signature_param_count; + to->signature_param_types = surface_copy_strings(&dst->arena, from->signature_param_types, + from->signature_param_count); + to->base_classes = surface_copy_strings(&dst->arena, from->base_classes, -1); + to->decorators = surface_copy_strings(&dst->arena, from->decorators, -1); + if ((from->signature_param_types && !to->signature_param_types) || + (from->base_classes && !to->base_classes) || (from->decorators && !to->decorators)) { + goto fail; + } + } + for (int i = 0; i < src->imports.count; i++) { + COPY_STR(&dst->imports.items[i], &src->imports.items[i], local_name); + COPY_STR(&dst->imports.items[i], &src->imports.items[i], module_path); + } + for (int i = 0; i < src->impl_traits.count; i++) { + COPY_STR(&dst->impl_traits.items[i], &src->impl_traits.items[i], trait_name); + COPY_STR(&dst->impl_traits.items[i], &src->impl_traits.items[i], struct_name); + COPY_STR(&dst->impl_traits.items[i], &src->impl_traits.items[i], struct_qn); + } +#undef COPY_STR +#undef COPY_ARRAY + return dst; +fail: + cbm_free_result(dst); + return NULL; +} + static bool surface_reg_only_label(const char *label) { return label && (strcmp(label, "Field") == 0 || cbm_label_is_relation(label)); } diff --git a/src/pipeline/lsp_surface.h b/src/pipeline/lsp_surface.h index a1d826158..52c7da4e3 100644 --- a/src/pipeline/lsp_surface.h +++ b/src/pipeline/lsp_surface.h @@ -40,4 +40,7 @@ int cbm_lsp_surface_build_rows(const char *project, CBMFileResult **cache, * different codec version — callers route that to a full rebuild. */ int cbm_lsp_surface_defs_from_json(CBMArena *arena, const char *defs_json, CBMLSPDef **out_defs); +/* Owned minimal extraction surface for bounded two-pass indexing. */ +CBMFileResult *cbm_lsp_surface_copy_result(const CBMFileResult *src); + #endif /* CBM_PIPELINE_LSP_SURFACE_H */ diff --git a/src/pipeline/pass_parallel.c b/src/pipeline/pass_parallel.c index 9f572b2b7..16da761cb 100644 --- a/src/pipeline/pass_parallel.c +++ b/src/pipeline/pass_parallel.c @@ -210,6 +210,9 @@ static cbm_parallel_extract_opts_t cbm_parallel_extract_resolve_opts( if (opts->retain_per_file_max_bytes > 0) { resolved.retain_per_file_max_bytes = opts->retain_per_file_max_bytes; } + resolved.backpressure_futile = opts->backpressure_futile; + resolved.replay = opts->replay; + resolved.skip_pkgmap = opts->skip_pkgmap; } /* Correctness invariant: a single file can never exceed the total budget. */ @@ -666,7 +669,9 @@ typedef struct { * in-flight transients, holds the memory, so napping cannot reclaim it. * While set, pulls skip the nap (the designed soft overshoot); the cheap * over-budget probe re-arms the gate once RSS drains under budget. */ - _Atomic int bp_futile; + _Atomic int *bp_futile; + bool replay; + bool skip_pkgmap; const CBMMacroTable *macro_table; /* ObjectScript $$$macros (NULL if none) */ const CBMReturnTypeTable *return_type_table; /* ObjectScript return types (NULL if none) */ @@ -760,7 +765,7 @@ static void extract_worker(int worker_id, void *ctx_ptr) { * the gate as soon as RSS drains under budget. */ if (cbm_mem_budget() > 0) { bool over = cbm_mem_over_budget(); - bool futile = atomic_load_explicit(&ec->bp_futile, memory_order_relaxed) != 0; + bool futile = atomic_load_explicit(ec->bp_futile, memory_order_relaxed) != 0; if (over && !futile) { cbm_mem_collect(); atomic_fetch_add_explicit(&g_bp_nap_cycles, SKIP_ONE, memory_order_relaxed); @@ -775,12 +780,12 @@ static void extract_worker(int worker_id, void *ctx_ptr) { /* Log only the 0→1 transition: all workers race into the * gate before anyone latches, so a plain store would WARN * once per worker (12 lines per latch event). */ - if (atomic_exchange_explicit(&ec->bp_futile, 1, memory_order_relaxed) == 0) { + if (atomic_exchange_explicit(ec->bp_futile, 1, memory_order_relaxed) == 0) { cbm_log_warn("mem.backpressure.futile", "action", "soft_overshoot"); } } } else if (!over && futile) { - atomic_store_explicit(&ec->bp_futile, 0, memory_order_relaxed); + atomic_store_explicit(ec->bp_futile, 0, memory_order_relaxed); } } @@ -898,7 +903,7 @@ static void extract_worker(int worker_id, void *ctx_ptr) { } /* Create definition nodes in local gbuf */ - for (int d = 0; d < result->defs.count; d++) { + for (int d = 0; !ec->replay && d < result->defs.count; d++) { CBMDefinition *def = &result->defs.items[d]; if (def->qualified_name && def->name) { insert_def_into_gbuf(ws, fi, def); @@ -911,7 +916,7 @@ static void extract_worker(int worker_id, void *ctx_ptr) { cbm_free_tree(result); /* Detect and parse manifest files for package map */ - { + if (!ec->skip_pkgmap) { const char *bn = strrchr(fi->rel_path, '/'); cbm_pkgmap_try_parse(bn ? bn + SKIP_ONE : fi->rel_path, fi->rel_path, source, source_len, &ec->pkg_entries[worker_id]); @@ -1028,6 +1033,10 @@ int cbm_parallel_extract_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *file CBMFileResult **result_cache, _Atomic int64_t *shared_ids, int worker_count, const cbm_parallel_extract_opts_t *opts) { cbm_parallel_extract_opts_t resolved_opts = cbm_parallel_extract_resolve_opts(opts); + _Atomic int local_bp_futile; + atomic_init(&local_bp_futile, 0); + _Atomic int *bp_futile = + resolved_opts.backpressure_futile ? resolved_opts.backpressure_futile : &local_bp_futile; if (file_count == 0) { return 0; @@ -1102,7 +1111,8 @@ int cbm_parallel_extract_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *file /* ObjectScript macro table (NULL when no .inc include files present). */ CBMMacroTable *pp_macro_table = - cbm_build_macro_table_from_files(files, file_count, ctx->repo_path); + ctx->macro_table ? NULL + : cbm_build_macro_table_from_files(files, file_count, ctx->repo_path); extract_ctx_t ec = { .files = files, @@ -1120,15 +1130,17 @@ int cbm_parallel_extract_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *file .retain_sources = resolved_opts.retain_sources, .retain_total_budget_bytes = resolved_opts.retain_total_budget_bytes, .retain_per_file_max_bytes = resolved_opts.retain_per_file_max_bytes, - .macro_table = pp_macro_table, + .macro_table = ctx->macro_table ? ctx->macro_table : pp_macro_table, .return_type_table = ctx->return_type_table, + .bp_futile = bp_futile, + .replay = resolved_opts.replay, + .skip_pkgmap = resolved_opts.skip_pkgmap, }; atomic_init(&ec.next_worker_id, 0); atomic_init(&ec.next_file_idx, 0); atomic_init(&ec.retained_bytes, 0); atomic_init(&ec.retain_cap_warned, 0); atomic_init(&ec.oversized_warned, 0); - atomic_init(&ec.bp_futile, 0); /* Sub-phase: Dispatch workers (parse + extract per file, PARALLEL) */ CBM_PROF_START(t_dispatch); @@ -1158,9 +1170,28 @@ int cbm_parallel_extract_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *file if (err_lists) { for (int i = 0; i < worker_count; i++) { for (int j = 0; j < err_lists[i].count; j++) { - cbm_pipeline_add_file_error(ctx->pipeline, err_lists[i].items[j].path, - err_lists[i].items[j].reason, - err_lists[i].items[j].phase); + /* A second parse must not double parse_partial/skipped counts. + * Preserve a new failure if the replay sees something different. */ + cbm_file_error_t *prior = NULL; + int prior_count = 0; + bool duplicate = false; + if (resolved_opts.replay) { + cbm_pipeline_get_file_errors(ctx->pipeline, &prior, &prior_count); + for (int k = 0; k < prior_count; k++) { + if (prior[k].path && prior[k].reason && prior[k].phase && + strcmp(prior[k].path, err_lists[i].items[j].path) == 0 && + strcmp(prior[k].reason, err_lists[i].items[j].reason) == 0 && + strcmp(prior[k].phase, err_lists[i].items[j].phase) == 0) { + duplicate = true; + break; + } + } + } + if (!duplicate) { + cbm_pipeline_add_file_error(ctx->pipeline, err_lists[i].items[j].path, + err_lists[i].items[j].reason, + err_lists[i].items[j].phase); + } free(err_lists[i].items[j].path); free(err_lists[i].items[j].reason); free(err_lists[i].items[j].phase); @@ -1170,7 +1201,11 @@ int cbm_parallel_extract_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *file free(err_lists); } - merge_pkg_entries(ctx, pkg_entries, worker_count); + if (resolved_opts.skip_pkgmap) { + free(pkg_entries); /* no entries were collected */ + } else { + merge_pkg_entries(ctx, pkg_entries, worker_count); + } cbm_aligned_free(workers); free(sorted); @@ -1300,55 +1335,73 @@ static void create_channel_edges(cbm_pipeline_ctx_t *ctx, const CBMFileResult *r } } -int cbm_build_registry_from_cache(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, - int file_count, CBMFileResult **result_cache) { - cbm_log_info("parallel.registry.start", "files", itoa_log(file_count)); - +int cbm_register_definitions_from_cache(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, + int file_count, CBMFileResult **result_cache) { int reg_entries = 0; int defines_edges = 0; - int imports_edges = 0; - - /* Namespace/package → File-QN map for namespace imports (C# `using`, - * Java/Kotlin `import`, PHP `use`). Built from the full result cache so - * every declaring file is visible regardless of loop order. */ - const char **rels = (const char **)calloc((size_t)file_count, sizeof(char *)); - if (rels) { - for (int i = 0; i < file_count; i++) { - rels[i] = files[i].rel_path; - } - } - CBMHashTable *namespace_map = - cbm_pipeline_namespace_map_build(ctx->project_name, result_cache, rels, file_count); - free(rels); - for (int i = 0; i < file_count; i++) { if (cbm_pipeline_check_cancel(ctx)) { - cbm_pipeline_namespace_map_free(namespace_map); return CBM_NOT_FOUND; } - CBMFileResult *result = result_cache[i]; if (!result) { continue; } - const char *rel = files[i].rel_path; - - /* Register callable symbols + DEFINES/DEFINES_METHOD edges */ for (int d = 0; d < result->defs.count; d++) { defines_edges += register_and_link_def(ctx, &result->defs.items[d], rel, ®_entries); } + } + cbm_log_info("parallel.registry.definitions", "entries", itoa_log(reg_entries), "defines", + itoa_log(defines_edges)); + return 0; +} +int cbm_create_relationship_carriers_from_cache(cbm_pipeline_ctx_t *ctx, + const cbm_file_info_t *files, int file_count, + CBMFileResult **result_cache, + CBMHashTable *namespace_map) { + int imports_edges = 0; + for (int i = 0; i < file_count; i++) { + if (cbm_pipeline_check_cancel(ctx)) { + return CBM_NOT_FOUND; + } + CBMFileResult *result = result_cache[i]; + if (!result) { + continue; + } + const char *rel = files[i].rel_path; imports_edges += create_imports_edges(ctx, result, rel, namespace_map); create_channel_edges(ctx, result, rel); cbm_pipeline_create_env_configures_for_file(ctx, result, rel); } + cbm_log_info("parallel.registry.relationship_carriers", "imports", itoa_log(imports_edges)); + return 0; +} - cbm_pipeline_namespace_map_free(namespace_map); +int cbm_build_registry_from_cache(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, + int file_count, CBMFileResult **result_cache) { + cbm_log_info("parallel.registry.start", "files", itoa_log(file_count)); - cbm_log_info("parallel.registry.done", "entries", itoa_log(reg_entries), "defines", - itoa_log(defines_edges), "imports", itoa_log(imports_edges)); - return 0; + const char **rels = (const char **)calloc((size_t)file_count, sizeof(char *)); + if (rels) { + for (int i = 0; i < file_count; i++) { + rels[i] = files[i].rel_path; + } + } + CBMHashTable *namespace_map = + cbm_pipeline_namespace_map_build(ctx->project_name, result_cache, rels, file_count); + free(rels); + + int rc = cbm_register_definitions_from_cache(ctx, files, file_count, result_cache); + if (rc == 0) { + rc = cbm_create_relationship_carriers_from_cache(ctx, files, file_count, result_cache, + namespace_map); + } + + cbm_pipeline_namespace_map_free(namespace_map); + cbm_log_info("parallel.registry.done", "status", rc == 0 ? "ok" : "failed"); + return rc; } /* ── Phase 4: Parallel Resolution ────────────────────────────────── */ @@ -3249,11 +3302,11 @@ static void resolve_worker(int worker_id, void *ctx_ptr) { cbm_service_pattern_cache_end(); } -int cbm_parallel_resolve(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, int file_count, - CBMFileResult **result_cache, _Atomic int64_t *shared_ids, - int worker_count, CBMLSPDef *all_defs, int def_count, - char *const *def_modules, struct CBMModuleDefIndex *module_def_index, - void *cross_registries_v) { +int cbm_parallel_resolve_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, int file_count, + CBMFileResult **result_cache, _Atomic int64_t *shared_ids, + int worker_count, CBMLSPDef *all_defs, int def_count, + char *const *def_modules, struct CBMModuleDefIndex *module_def_index, + void *cross_registries_v, bool finalize_graph) { /* See header: typed as void* across the TU boundary; cast back here. */ CBMCrossLspRegistries *cross_registries = (CBMCrossLspRegistries *)cross_registries_v; if (file_count == 0) { @@ -3357,12 +3410,13 @@ int cbm_parallel_resolve(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, cbm_aligned_free(workers); - /* Go-style implicit interface satisfaction (needs full graph, serial) */ - int go_impl = cbm_pipeline_implements_go(ctx); - - /* Explicit-language override detection (same serial full-graph tail the - * sequential pipeline runs — the two venues must emit identical graphs). */ - total_lsp_overrides += cbm_pipeline_override_explicit(ctx); + int go_impl = 0; + if (finalize_graph) { + /* These scans require the complete graph and therefore run once after + * the final batch in the bounded large-repository path. */ + go_impl = cbm_pipeline_implements_go(ctx); + total_lsp_overrides += cbm_pipeline_override_explicit(ctx); + } if (atomic_load(ctx->cancelled)) { return CBM_NOT_FOUND; @@ -3530,3 +3584,21 @@ int cbm_parallel_resolve(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, "per_lookup", tp_buf, "fallback_rows", fb_buf); return 0; } + +int cbm_parallel_resolve(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, int file_count, + CBMFileResult **result_cache, _Atomic int64_t *shared_ids, + int worker_count, CBMLSPDef *all_defs, int def_count, + char *const *def_modules, struct CBMModuleDefIndex *module_def_index, + void *cross_registries_v) { + return cbm_parallel_resolve_ex(ctx, files, file_count, result_cache, shared_ids, worker_count, + all_defs, def_count, def_modules, module_def_index, + cross_registries_v, true); +} + +int cbm_parallel_resolve_finalize(cbm_pipeline_ctx_t *ctx) { + int go_impl = cbm_pipeline_implements_go(ctx); + int overrides = cbm_pipeline_override_explicit(ctx); + cbm_log_info("parallel.resolve.finalize", "go_implements", itoa_log(go_impl), "overrides", + itoa_log(overrides)); + return cbm_pipeline_check_cancel(ctx) ? CBM_NOT_FOUND : 0; +} diff --git a/src/pipeline/pipeline.c b/src/pipeline/pipeline.c index 0ffb46c0b..6fde97086 100644 --- a/src/pipeline/pipeline.c +++ b/src/pipeline/pipeline.c @@ -890,6 +890,67 @@ static bool route_sr_denied(const CBMStringRef *sr) { return is_upstream_config_key(sr->key_path); } +static CBMHashTable *cbm_pipeline_collect_infra_route_denials(const cbm_file_info_t *files, + CBMFileResult **result_cache, + int file_count, + CBMHashTable *denied) { + if (!denied) { + denied = cbm_ht_create(16); + } + if (!denied) { + return NULL; + } + for (int i = 0; i < file_count; i++) { + if (!result_cache[i] || !is_infra_file(files[i].rel_path) || + is_ci_tooling_config(files[i].rel_path)) { + continue; + } + for (int si = 0; si < result_cache[i]->string_refs.count; si++) { + const CBMStringRef *sr = &result_cache[i]->string_refs.items[si]; + if (sr->kind != CBM_STRREF_URL || !sr->value || !strstr(sr->value, "://") || + !route_sr_denied(sr) || cbm_ht_has(denied, sr->value)) { + continue; + } + char *owned = strdup(sr->value); + if (owned) { + cbm_ht_set(denied, owned, owned); + } + } + } + return denied; +} + +static void cbm_pipeline_emit_infra_routes(cbm_gbuf_t *gbuf, const cbm_file_info_t *files, + CBMFileResult **result_cache, int file_count, + CBMHashTable *denied) { + for (int i = 0; i < file_count; i++) { + if (!result_cache[i] || !is_infra_file(files[i].rel_path) || + is_ci_tooling_config(files[i].rel_path)) { + continue; + } + for (int si = 0; si < result_cache[i]->string_refs.count; si++) { + const CBMStringRef *sr = &result_cache[i]->string_refs.items[si]; + if (sr->kind == CBM_STRREF_URL && sr->value && strstr(sr->value, "://") && + (!denied || !cbm_ht_has(denied, sr->value))) { + try_upsert_infra_route(gbuf, sr, files[i].rel_path); + } + } + } +} + +static void free_owned_string_entry(const char *key, void *value, void *userdata) { + (void)value; + (void)userdata; + free((void *)key); +} + +static void cbm_pipeline_free_infra_route_denials(CBMHashTable *denied) { + if (denied) { + cbm_ht_foreach(denied, free_owned_string_entry, NULL); + cbm_ht_free(denied); + } +} + static void cbm_pipeline_extract_infra_routes(cbm_gbuf_t *gbuf, const cbm_file_info_t *files, CBMFileResult **result_cache, int file_count) { /* DENY-WINS-BY-VALUE: the same URL is often extracted as several string_refs @@ -898,29 +959,10 @@ static void cbm_pipeline_extract_infra_routes(cbm_gbuf_t *gbuf, const cbm_file_i * per-ref guard — e.g. a denied full path `registries.terraform-registry.url` * is defeated by a sibling leaf `url`. So pass 1 collects every URL value * denied under ANY of its refs; pass 2 mints only values never denied. (#521) */ - CBMHashTable *denied = cbm_ht_create(16); - for (int pass = 0; pass < 2; pass++) { - for (int i = 0; i < file_count; i++) { - if (!result_cache[i] || !is_infra_file(files[i].rel_path) || - is_ci_tooling_config(files[i].rel_path)) { - continue; - } - for (int si = 0; si < result_cache[i]->string_refs.count; si++) { - const CBMStringRef *sr = &result_cache[i]->string_refs.items[si]; - if (sr->kind != CBM_STRREF_URL || !sr->value || !strstr(sr->value, "://")) { - continue; - } - if (pass == 0) { - if (denied && route_sr_denied(sr)) { - cbm_ht_set(denied, sr->value, (void *)1); - } - } else if (!denied || !cbm_ht_has(denied, sr->value)) { - try_upsert_infra_route(gbuf, sr, files[i].rel_path); - } - } - } - } - cbm_ht_free(denied); + CBMHashTable *denied = + cbm_pipeline_collect_infra_route_denials(files, result_cache, file_count, NULL); + cbm_pipeline_emit_infra_routes(gbuf, files, result_cache, file_count, denied); + cbm_pipeline_free_infra_route_denials(denied); } /* Run decorator_tags, configlink, and route matching passes. */ @@ -1167,6 +1209,213 @@ static int run_sequential_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx, return rc; } +/* Opt-in bounded extraction. This bounds the live extraction set by file count, + * not total RSS: the graph and compact definition/import surface are global. */ +static int streaming_batch_size(void) { + char value[CBM_SZ_32]; + if (!cbm_safe_getenv("CBM_STREAMING_BATCH_FILES", value, sizeof(value), NULL)) { + return 0; + } + char *end = NULL; + errno = 0; + long parsed = strtol(value, &end, 10); + if (errno || end == value || *end || parsed <= 0 || parsed > 4096) { + cbm_log_error("pipeline.streaming.invalid_batch", "value", value); + return -1; + } + return (int)parsed; +} + +static void free_result_cache(CBMFileResult **cache, int count) { + if (cache) { + for (int i = 0; i < count; i++) { + cbm_free_result(cache[i]); + } + free(cache); + } +} + +static bool pipeline_cross_lsp_enabled(void) { + char value[CBM_SZ_16]; + bool enabled = !cbm_safe_getenv("CBM_DISABLE_LSP_CROSS", value, sizeof(value), NULL); + if (!enabled) { + cbm_log_info("lsp_cross.skipped", "reason", "CBM_DISABLE_LSP_CROSS env set"); + } + return enabled; +} + +static int run_parallel_streaming_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx, + const cbm_file_info_t *files, int file_count, + int worker_count, int batch_size, struct timespec *t) { + cbm_log_info("pipeline.mode", "mode", "parallel_streaming", "workers", itoa_buf(worker_count), + "files", itoa_buf(file_count), "batch_files", itoa_buf(batch_size)); + int rc = CBM_NOT_FOUND; + bool cross_enabled = pipeline_cross_lsp_enabled(); + CBMFileResult **surfaces = calloc((size_t)file_count, sizeof(*surfaces)); + const char **rels = calloc((size_t)file_count, sizeof(*rels)); + char **def_modules = calloc((size_t)file_count, sizeof(*def_modules)); + int *def_starts = calloc((size_t)file_count + 1, sizeof(*def_starts)); + CBMFileResult **cache = NULL; + int cache_count = 0; + CBMHashTable *infra_denied = NULL; + CBMHashTable *namespace_map = NULL; + CBMLSPDef *all_defs = NULL; + int def_count = 0; + CBMModuleDefIndex *module_index = NULL; + CBMArena cross_arena; + cbm_arena_init(&cross_arena); + CBMCrossLspRegistries cross_registries = {0}; + CBMMacroTable *macros = cbm_build_macro_table_from_files(files, file_count, ctx->repo_path); + ctx->macro_table = macros; + _Atomic int64_t shared_ids; + atomic_init(&shared_ids, cbm_gbuf_next_id(p->gbuf)); + _Atomic int bp_futile; + atomic_init(&bp_futile, 0); + cbm_parallel_extract_opts_t extract_opts = { + .retain_sources = false, + .retain_sources_set = true, + .backpressure_futile = &bp_futile, + .skip_pkgmap = true, + }; + if (!surfaces || !rels || !def_modules || !def_starts) { + goto cleanup; + } + for (int i = 0; i < file_count; i++) { + rels[i] = files[i].rel_path; + } + cbm_clock_gettime(CLOCK_MONOTONIC, t); + for (int offset = 0; offset < file_count; offset += cache_count) { + cache_count = file_count - offset < batch_size ? file_count - offset : batch_size; + cache = calloc((size_t)cache_count, sizeof(*cache)); + if (!cache || + cbm_parallel_extract_ex(ctx, files + offset, cache_count, cache, &shared_ids, + worker_count, &extract_opts) != 0 || + check_cancel(p)) { + goto cleanup; + } + cbm_gbuf_set_next_id(p->gbuf, atomic_load(&shared_ids)); + if (cbm_register_definitions_from_cache(ctx, files + offset, cache_count, cache) != 0) { + goto cleanup; + } + atomic_store(&shared_ids, cbm_gbuf_next_id(p->gbuf)); + for (int i = 0; i < cache_count; i++) { + if (cache[i]) { + surfaces[offset + i] = cbm_lsp_surface_copy_result(cache[i]); + if (!surfaces[offset + i]) { + goto cleanup; + } + } + } + infra_denied = cbm_pipeline_collect_infra_route_denials(files + offset, cache, cache_count, + infra_denied); + free_result_cache(cache, cache_count); + cache = NULL; + cbm_mem_collect(); + cbm_log_info("pipeline.streaming.pass_a_batch", "completed", itoa_buf(offset + cache_count), + "total", itoa_buf(file_count)); + log_phase_mem("streaming_pass_a_batch"); + } + cache_count = 0; + cbm_pipeline_set_pkgmap(cbm_pkgmap_build_from_repo(ctx->repo_path, files, file_count, + ctx->project_name, ctx->excluded_dirs, + ctx->excluded_count)); + namespace_map = cbm_pipeline_namespace_map_build(ctx->project_name, surfaces, rels, file_count); + /* The compact surface has imports but no channel/env carriers. Resolve + * every import against the complete graph before qualifying base types. */ + if (cbm_create_relationship_carriers_from_cache(ctx, files, file_count, surfaces, + namespace_map) != 0) { + goto cleanup; + } + if (cross_enabled) { + all_defs = cbm_pxc_collect_all_defs(ctx, surfaces, files, file_count, ctx->project_name, + def_modules, &def_count, def_starts); + cbm_lsp_surface_row_t *rows = NULL; + int row_count = 0; + if (cbm_lsp_surface_build_rows(ctx->project_name, surfaces, files, file_count, all_defs, + def_starts, &rows, &row_count) != 0) { + goto cleanup; + } + cbm_pipeline_set_lsp_surfaces(p, rows, row_count); + module_index = all_defs ? cbm_pxc_build_module_def_index(all_defs, def_count) : NULL; + if (all_defs) { + cross_registries.go = cbm_go_build_cross_registry(&cross_arena, all_defs, def_count); + cross_registries.python = + cbm_py_build_cross_registry(&cross_arena, all_defs, def_count); + cross_registries.c = cbm_c_build_cross_registry(&cross_arena, all_defs, def_count); + cross_registries.cs = cbm_cs_build_cross_registry(&cross_arena, all_defs, def_count); + cross_registries.ts = cbm_ts_build_cross_registry(&cross_arena, all_defs, def_count); + cross_registries.java = + cbm_java_build_cross_registry(&cross_arena, all_defs, def_count); + } + } + cbm_log_info("pass.timing", "pass", "streaming_pass_a", "elapsed_ms", + itoa_buf((int)elapsed_ms(*t))); + log_phase_mem("streaming_prepare"); + atomic_store(&shared_ids, cbm_gbuf_next_id(p->gbuf)); + extract_opts.replay = true; + cbm_clock_gettime(CLOCK_MONOTONIC, t); + for (int offset = 0; offset < file_count; offset += cache_count) { + cache_count = file_count - offset < batch_size ? file_count - offset : batch_size; + cache = calloc((size_t)cache_count, sizeof(*cache)); + if (!cache || + cbm_parallel_extract_ex(ctx, files + offset, cache_count, cache, &shared_ids, + worker_count, &extract_opts) != 0 || + check_cancel(p)) { + goto cleanup; + } + cbm_gbuf_set_next_id(p->gbuf, atomic_load(&shared_ids)); + if (cbm_create_relationship_carriers_from_cache(ctx, files + offset, cache_count, cache, + namespace_map) != 0) { + goto cleanup; + } + atomic_store(&shared_ids, cbm_gbuf_next_id(p->gbuf)); + if (cbm_parallel_resolve_ex(ctx, files + offset, cache_count, cache, &shared_ids, + worker_count, all_defs, def_count, def_modules + offset, + module_index, &cross_registries, false) != 0) { + goto cleanup; + } + cbm_gbuf_set_next_id(p->gbuf, atomic_load(&shared_ids)); + cbm_pipeline_emit_infra_routes(p->gbuf, files + offset, cache, cache_count, infra_denied); + cbm_pipeline_process_infra_bindings(p->gbuf, files + offset, cache, cache_count); + atomic_store(&shared_ids, cbm_gbuf_next_id(p->gbuf)); + free_result_cache(cache, cache_count); + cache = NULL; + cbm_mem_collect(); + cbm_log_info("pipeline.streaming.pass_b_batch", "completed", itoa_buf(offset + cache_count), + "total", itoa_buf(file_count)); + log_phase_mem("streaming_pass_b_batch"); + } + cache_count = 0; + if (cbm_parallel_resolve_finalize(ctx) != 0) { + goto cleanup; + } + cbm_log_info("pass.timing", "pass", "streaming_pass_b", "elapsed_ms", + itoa_buf((int)elapsed_ms(*t))); + cbm_clock_gettime(CLOCK_MONOTONIC, t); + cbm_pipeline_pass_k8s(ctx, files, file_count); + cbm_log_info("pass.timing", "pass", "k8s", "elapsed_ms", itoa_buf((int)elapsed_ms(*t))); + rc = check_cancel(p) ? CBM_NOT_FOUND : 0; +cleanup: + free_result_cache(cache, cache_count); + cbm_pipeline_namespace_map_free(namespace_map); + cbm_pipeline_free_infra_route_denials(infra_denied); + cbm_pxc_free_module_def_index(module_index); + cbm_arena_destroy(&cross_arena); + free(all_defs); + free_result_cache(surfaces, file_count); + if (def_modules) { + for (int i = 0; i < file_count; i++) { + free(def_modules[i]); + } + } + free(def_modules); + free(def_starts); + free(rels); + cbm_macro_table_free(macros); + ctx->macro_table = NULL; + return rc; +} + /* Run the parallel pipeline path: extract, registry, resolve, infra, k8s. */ static int run_parallel_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, int file_count, int worker_count, @@ -1239,12 +1488,7 @@ static int run_parallel_pipeline(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx, * on large TS projects — see #340/#344); with cross-LSP off, all_defs * stays NULL and the fused resolver simply no-ops cross-file resolution * (per-file LSP already ran during extract). */ - char cbm_lsp_cross_env[CBM_SZ_16]; - const bool run_cross_lsp = cbm_safe_getenv("CBM_DISABLE_LSP_CROSS", cbm_lsp_cross_env, - sizeof(cbm_lsp_cross_env), NULL) == NULL; - if (!run_cross_lsp) { - cbm_log_info("lsp_cross.skipped", "reason", "CBM_DISABLE_LSP_CROSS env set"); - } + const bool run_cross_lsp = pipeline_cross_lsp_enabled(); char **def_modules = NULL; int def_count = 0; CBMLSPDef *all_defs = NULL; @@ -2175,9 +2419,19 @@ static int run_extraction_phase(cbm_pipeline_t *p, cbm_pipeline_ctx_t *ctx, int worker_count = effective_worker_count(true); CBM_PROF_START(t_extract_total); - int rc = (worker_count > SKIP_ONE && file_count > MIN_FILES_FOR_PARALLEL) + int batch_size = streaming_batch_size(); + if (batch_size < 0) { + return CBM_NOT_FOUND; + } + int rc; + if (batch_size > 0 && file_count > batch_size) { + rc = run_parallel_streaming_pipeline(p, ctx, files, file_count, worker_count, batch_size, + &t); + } else { + rc = (worker_count > SKIP_ONE && file_count > MIN_FILES_FOR_PARALLEL) ? run_parallel_pipeline(p, ctx, files, file_count, worker_count, &t) : run_sequential_pipeline(p, ctx, files, file_count, &t); + } CBM_PROF_END_N("pipeline", "2_extraction_total", t_extract_total, file_count); if (check_cancel(p)) { return CBM_NOT_FOUND; diff --git a/src/pipeline/pipeline_internal.h b/src/pipeline/pipeline_internal.h index d749c695a..a08f9435a 100644 --- a/src/pipeline/pipeline_internal.h +++ b/src/pipeline/pipeline_internal.h @@ -520,6 +520,11 @@ typedef struct { bool retain_sources_set; /* false keeps the default retain_sources policy */ size_t retain_total_budget_bytes; size_t retain_per_file_max_bytes; + /* Optional run-scoped latch shared by repeated streaming batches. When + * NULL, cbm_parallel_extract_ex owns a fresh latch for this invocation. */ + _Atomic int *backpressure_futile; + bool skip_pkgmap; /* streaming driver builds the project map once */ + bool replay; /* pass B: reuse definition nodes/pkgmap; deduplicate diagnostics */ } cbm_parallel_extract_opts_t; int cbm_parallel_extract_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, int file_count, @@ -534,6 +539,12 @@ int cbm_parallel_extract(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, * Registers callable symbols (Function/Method/Class) in ctx->registry. */ int cbm_build_registry_from_cache(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, int file_count, CBMFileResult **result_cache); +int cbm_register_definitions_from_cache(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, + int file_count, CBMFileResult **result_cache); +int cbm_create_relationship_carriers_from_cache(cbm_pipeline_ctx_t *ctx, + const cbm_file_info_t *files, int file_count, + CBMFileResult **result_cache, + CBMHashTable *namespace_map); /* Phase 4: Parallel call/usage/semantic resolution. * Each worker resolves calls, usages, throws, rw, inherits, decorates, @@ -567,6 +578,12 @@ int cbm_parallel_resolve(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, * Typed as void* here to dodge the typedef/tag ordering * problem — pass_parallel.c casts back to CBMCrossLspRegistries*. */ void *cross_registries); +int cbm_parallel_resolve_ex(cbm_pipeline_ctx_t *ctx, const cbm_file_info_t *files, int file_count, + CBMFileResult **result_cache, _Atomic int64_t *shared_ids, + int worker_count, CBMLSPDef *all_defs, int def_count, + char *const *def_modules, struct CBMModuleDefIndex *module_def_index, + void *cross_registries, bool finalize_graph); +int cbm_parallel_resolve_finalize(cbm_pipeline_ctx_t *ctx); /* Post-merge: create Route nodes for HTTP_CALLS/ASYNC_CALLS edges that * have url_path in properties but point to library functions instead of routes. diff --git a/tests/test_pipeline.c b/tests/test_pipeline.c index 5608ee559..76f922f7a 100644 --- a/tests/test_pipeline.c +++ b/tests/test_pipeline.c @@ -12,6 +12,7 @@ #include "pipeline/pipeline.h" #include "pipeline/pipeline_internal.h" #include "pipeline/artifact.h" +#include "pipeline/lsp_surface.h" #include "store/store.h" #include "git/git_context.h" #include "foundation/dump_verify.h" @@ -12814,6 +12815,200 @@ TEST(pipeline_seq_ts_cross_uses_shared_registry) { PASS(); } +/* Compare natural graph keys: worker-allocated IDs are intentionally opaque. */ +static int streaming_graph_difference(const char *left_path, const char *right_path) { + sqlite3 *db = NULL; + if (sqlite3_open(left_path, &db) != SQLITE_OK) { + sqlite3_close(db); + return -1; + } + char *attach = sqlite3_mprintf("ATTACH DATABASE %Q AS other", right_path); + int rc = sqlite3_exec(db, attach, NULL, NULL, NULL); + sqlite3_free(attach); + int differences = 0; + const char *queries[] = { + "SELECT label,name,qualified_name,file_path,start_line,end_line FROM %s.nodes", + ("SELECT s.qualified_name,t.qualified_name,e.type,e.properties FROM %s.edges e " + "JOIN %s.nodes s ON s.id=e.source_id JOIN %s.nodes t ON t.id=e.target_id"), + "SELECT rel_path,surface_sha,defs_json FROM %s.lsp_surface", + }; + for (int i = 0; rc == SQLITE_OK && i < 3; i++) { + char *left = sqlite3_mprintf(queries[i], "main", "main", "main"); + char *right = sqlite3_mprintf(queries[i], "other", "other", "other"); + for (int direction = 0; direction < 2; direction++) { + char *sql = sqlite3_mprintf("SELECT count(*) FROM (%s EXCEPT %s)", + direction ? right : left, direction ? left : right); + sqlite3_stmt *stmt = NULL; + rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL); + if (rc == SQLITE_OK && sqlite3_step(stmt) == SQLITE_ROW) { + int n = sqlite3_column_int(stmt, 0); + if (n) { + fprintf(stderr, "streaming parity query=%d direction=%d differences=%d\n", i, + direction, n); + } + differences += n; + } else { + fprintf(stderr, "streaming parity SQL: %s\n", sqlite3_errmsg(db)); + rc = SQLITE_ERROR; + } + sqlite3_finalize(stmt); + sqlite3_free(sql); + } + sqlite3_free(left); + sqlite3_free(right); + } + sqlite3_close(db); + return rc == SQLITE_OK ? differences : -1; +} + +TEST(pipeline_streaming_surface_survives_extraction_release) { + const CBMLanguage languages[] = {CBM_LANG_GO, CBM_LANG_PYTHON, CBM_LANG_JAVA, CBM_LANG_RUST}; + const char *paths[] = {"types.go", "types.py", "Types.java", "types.rs"}; + const char *sources[] = { + "package types\ntype Service struct { Name string }\nfunc (s *Service) Work(x int) string " + "{ return s.Name }\n", + "from other import Parent\n@decorate\nclass Child(Parent):\n def work(self, x: int) -> " + "str:\n return str(x)\n", + "package types; public interface Types { String work(int x); }\n", + "trait Work { fn work(&self); }\nstruct Service;\nimpl Work for Service { fn work(&self) " + "{} }\n", + }; + for (int i = 0; i < 4; i++) { + CBMFileResult *original = cbm_extract_file(sources[i], strlen(sources[i]), languages[i], + "surface", paths[i], 0, NULL, NULL); + ASSERT_NOT_NULL(original); + CBMFileResult *copy = cbm_lsp_surface_copy_result(original); + ASSERT_NOT_NULL(copy); + cbm_file_info_t file = {.rel_path = (char *)paths[i], .language = languages[i]}; + char *modules[1] = {NULL}; + int starts[2] = {0}; + int count = 0; + CBMLSPDef *defs = + cbm_pxc_collect_all_defs(NULL, &original, &file, 1, "surface", modules, &count, starts); + cbm_lsp_surface_row_t *before = NULL; + int before_count = 0; + int before_rc = cbm_lsp_surface_build_rows("surface", &original, &file, 1, defs, starts, + &before, &before_count); + free(defs); + free(modules[0]); + modules[0] = NULL; + cbm_free_result(original); + /* All original strings and arrays are now dead. ASan checks the copy's + * transitive ownership, and codec equality checks the complete surface. */ + defs = cbm_pxc_collect_all_defs(NULL, ©, &file, 1, "surface", modules, &count, starts); + cbm_lsp_surface_row_t *after = NULL; + int after_count = 0; + int after_rc = cbm_lsp_surface_build_rows("surface", ©, &file, 1, defs, starts, &after, + &after_count); + bool same = before_rc == 0 && after_rc == 0 && before_count == 1 && after_count == 1 && + strcmp(before[0].defs_json, after[0].defs_json) == 0; + bool no_body = copy->calls.count == 0 && copy->usages.count == 0 && + copy->cached_tree == NULL && copy->defs.items[0].body_tokens == NULL; + cbm_store_free_lsp_surfaces(before, before_count); + cbm_store_free_lsp_surfaces(after, after_count); + free(defs); + free(modules[0]); + cbm_free_result(copy); + ASSERT_TRUE(same); + ASSERT_TRUE(no_body); + } + PASS(); +} + +TEST(pipeline_streaming_cross_batch_graph_and_diagnostics) { + /* Windows expands /tmp/ to the per-user TEMP path in place. */ + char tmp[CBM_SZ_512] = "/tmp/cbm_streaming_XXXXXX"; + ASSERT_NOT_NULL(cbm_mkdtemp(tmp)); + write_temp_file(tmp, "Base.java", + "package batch; public class Base { " + "public void work() {} }\n"); + write_temp_file(tmp, "Factory.java", + "package batch; public class Factory { " + "public static Base create() { return new Base(); } }\n"); + write_temp_file(tmp, "Caller.java", + "package batch; public class Caller { " + "public void run(batch.Base value) { value.work(); } }\n"); + write_temp_file(tmp, "child.py", + "from parent import Parent\nclass Child(Parent):\n" + " def invoke(self):\n self.inherited()\n"); + write_temp_file(tmp, "parent.py", "class Parent:\n def inherited(self):\n pass\n"); + write_temp_file(tmp, "derived.ts", + "import { ParentTS } from './parent';\n" + "export class ChildTS extends ParentTS { invokeTS() { this.baseTS(); } }\n"); + write_temp_file(tmp, "parent.ts", "export class ParentTS { baseTS() {} }\n"); + write_temp_file(tmp, "broken.py", "def good():\n pass\n\ndef broken(\n"); + for (int i = 0; i < 48; i++) { + char name[64]; + char body[128]; + snprintf(name, sizeof(name), "pad%02d.ts", i); + snprintf(body, sizeof(body), "export const pad%d = %d;\n", i, i); + write_temp_file(tmp, name, body); + } + const char *env_names[] = {"CBM_WORKERS", "CBM_INDEX_SINGLE_THREAD", + "CBM_STREAMING_BATCH_FILES", "CBM_DISABLE_LSP_CROSS"}; + char *saved[4]; + for (int i = 0; i < 4; i++) { + const char *old = getenv(env_names[i]); + saved[i] = old ? strdup(old) : NULL; + cbm_unsetenv(env_names[i]); + } + cbm_setenv("CBM_WORKERS", "4", 1); + char paths[5][512]; + int run_rc[5]; + int errors[5] = {0}; + int java_calls[5] = {0}; + for (int run = 0; run < 5; run++) { + if (run == 1) { + cbm_setenv("CBM_STREAMING_BATCH_FILES", "1", 1); + cbm_setenv("CBM_WORKERS", "1", 1); + } else if (run == 2 || run == 4) { + cbm_setenv("CBM_STREAMING_BATCH_FILES", "7", 1); + cbm_setenv("CBM_WORKERS", "4", 1); + } else { + cbm_unsetenv("CBM_STREAMING_BATCH_FILES"); + } + if (run >= 3) { + cbm_setenv("CBM_DISABLE_LSP_CROSS", "1", 1); + } + snprintf(paths[run], sizeof(paths[run]), "%s/run%d.db", tmp, run); + cbm_pipeline_t *p = cbm_pipeline_new(tmp, paths[run], CBM_MODE_FULL); + run_rc[run] = p ? cbm_pipeline_run(p) : -1; + cbm_file_error_t *file_errors = NULL; + cbm_pipeline_get_file_errors(p, &file_errors, &errors[run]); + cbm_store_t *store = cbm_store_open_path(paths[run]); + if (store && p) { + java_calls[run] = + named_edge_count(store, cbm_pipeline_project_name(p), "CALLS", "run", "work"); + } + cbm_store_close(store); + cbm_pipeline_free(p); + } + for (int i = 0; i < 4; i++) { + if (saved[i]) { + cbm_setenv(env_names[i], saved[i], 1); + } else { + cbm_unsetenv(env_names[i]); + } + free(saved[i]); + } + int single_diff = streaming_graph_difference(paths[0], paths[1]); + int batch_diff = streaming_graph_difference(paths[0], paths[2]); + int disabled_diff = streaming_graph_difference(paths[3], paths[4]); + th_rmtree(tmp); + for (int i = 0; i < 5; i++) { + ASSERT_EQ(run_rc[i], 0); + ASSERT_EQ(errors[i], errors[0]); + } + ASSERT_GT(errors[0], 0); + ASSERT_GT(java_calls[0], 0); + ASSERT_EQ(java_calls[1], java_calls[0]); + ASSERT_EQ(java_calls[2], java_calls[0]); + ASSERT_EQ(single_diff, 0); + ASSERT_EQ(batch_diff, 0); + ASSERT_EQ(disabled_diff, 0); + PASS(); +} + /* The closure-repair route lives and dies by two properties of the persisted * per-file LSP surface: a BODY edit must leave the surface_sha unchanged (the * early cutoff -- no dependent recomputation owed), while a SIGNATURE edit @@ -13344,7 +13539,6 @@ TEST(pipeline_delta_patch_indexes_docstring_into_fts_body) { PASS(); } - /* End-to-end for #518/#519: source → docstring → properties JSON → nodes_fts * `body` → findable. Each layer has its own test; this one proves they connect. * It is also the guard on the size budget: build_def_props drops an oversized @@ -13491,6 +13685,8 @@ TEST(pipeline_objectscript_export_range_join_keeps_one_trailing_marker) { #endif SUITE(pipeline) { + RUN_TEST(pipeline_streaming_surface_survives_extraction_release); + RUN_TEST(pipeline_streaming_cross_batch_graph_and_diagnostics); RUN_TEST(pipeline_lsp_surface_persisted_and_body_edit_invariant); /* Index lock */ RUN_TEST(pipeline_lock_try_acquire);