diff --git a/CHANGELOG.md b/CHANGELOG.md index dc14f8f9..d09d761e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,36 @@ All notable changes to the Toolpath workspace are documented here. +## Parallel `path query` execution — 2026-07-21 + +`path query` now evaluates cache documents on a thread pool. For the +plans that treat every file independently (`PerFileStream`, +`Decompose`), the whole per-file pipeline — read, parse, wrap, filter, +render — runs on rayon workers, with only ordered output assembly on +the main thread; `Slurp` plans parallelize the parse/wrap phase only +(the merged jaq array is `Rc`-based and must be built on one thread). +Output is byte-identical to the sequential scan — same ordering, same +warnings, same error precedence — enforced by tests pinning the +parallel building blocks against the sequential engine. + +Measured on a real 97-doc / 114 MB cache (M-series, medians of 5): +streamed and decomposed queries drop from ~0.95 s to ~0.31 s (~3.1×), +slurped queries improve ~1.2×. On an even 60-doc / 56 MB synthetic +cache the same queries run ~4.7× faster (e.g. `length` 966 ms → +206 ms). + +- **`path-cli`** (0.17.0): + - `query/mod.rs`: plan dispatch (`execute_plan`) and a chunked + parallel driver (`for_each_file`) that preserves selection order, + per-file warning order, and explicit-file error positions. + - `query/filter.rs`: per-file worker evaluation (`render_file`, + `partials_file` — partials cross threads as compact JSON bytes and + are reparsed on the consuming thread), a per-thread compiled-filter + cache, and `finish_decompose` shared by the sequential and parallel + drivers so the zero-file rule lives once. + - The emscripten (playground) build keeps the sequential engine — + no threads there. + ## `path p cache sync` — incremental session ingestion — 2026-07-16 Adds `path p cache sync [types…]`, the first step toward a cache that diff --git a/CLAUDE.md b/CLAUDE.md index 8cfcfdb5..57c7eaa0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -212,7 +212,7 @@ Tests live alongside the code (`#[cfg(test)] mod tests`), plus `path-cli` has in - `toolpath-cursor`: 78 unit + 8 integration round-trip + 1 real-DB sanity + 1 doc test (state.vscdb SQLite reader, bubble store + composer header parsing, content-addressed blob lookup, projector with full TOOL_TABLE coverage, JSONL transcript ingest in `examples/dump_fixture.rs`) - `toolpath-pi`: 133 unit + 26 integration + 5 doc tests (types, paths, error, reader, io, provider) - `toolpath-dot`: 30 unit + 2 doc tests (render, visual conventions, escaping) -- `path-cli`: 358 unit + 121 integration tests (import/export/cache, track sessions, merge, validate, roundtrip, render-md snapshots, deprecation aliases, pathbase HTTP mock-server tests, fzf-friendly TSV output, `path resume` orchestration with injectable `ExecStrategy`, `path query`/`path kind` jaq filters + kind-selector matching + step wrapping over a `$TOOLPATH_CONFIG_DIR` cache sandbox, streaming-planner recognition + streamed-output-equals-slurp equality checks, `p cache sync` incremental ingestion — stat fingerprints, refresh-overwrite, per-type filtering, failure tallying — over resolver-injected provider fixtures, and import/share manifest recording end-to-end). For an end-to-end check against a real Pathbase deployment, run `scripts/test-pathbase-live.sh ` — it does an anon round-trip in a sandboxed config dir and, if you're logged into that URL, an authed pathstash round-trip too. +- `path-cli`: 361 unit + 121 integration tests (import/export/cache, track sessions, merge, validate, roundtrip, render-md snapshots, deprecation aliases, pathbase HTTP mock-server tests, fzf-friendly TSV output, `path resume` orchestration with injectable `ExecStrategy`, `path query`/`path kind` jaq filters + kind-selector matching + step wrapping over a `$TOOLPATH_CONFIG_DIR` cache sandbox, streaming-planner recognition + streamed-output-equals-slurp equality checks, `p cache sync` incremental ingestion — stat fingerprints, refresh-overwrite, per-type filtering, failure tallying — over resolver-injected provider fixtures, and import/share manifest recording end-to-end). For an end-to-end check against a real Pathbase deployment, run `scripts/test-pathbase-live.sh ` — it does an anon round-trip in a sandboxed config dir and, if you're logged into that URL, an authed pathstash round-trip too. - `toolpath-cli`: 0 tests (it's a one-line `path_cli::run()` shim crate that exists only so `cargo install toolpath-cli` keeps installing the `path` binary) Validate example documents: `for f in examples/*.json; do cargo run -p path-cli -- p validate --input "$f"; done` @@ -281,6 +281,6 @@ Build the site after changes: `cd site && pnpm run build` (should produce 11 pag - Conversation metadata title field: `toolpath-claude::ConversationMetadata`, `toolpath-gemini::ConversationMetadata`, and `toolpath-pi::SessionMeta` all expose `first_user_message: Option` — the first non-empty user-prompt text. Populated cheaply during the metadata pass (single-pass for Claude/Gemini; one extra short read for Pi). Used by the picker UI but useful for any "list sessions by topic" surface. - `path share` is the one-shot equivalent of `path p import | path p export pathbase`. It probes installed agent harnesses (claude/gemini/codex/opencode/pi), aggregates their sessions into a single fzf picker, and ranks rows whose project (claude/gemini/pi) or recorded cwd (codex/opencode) canonicalizes to the current directory at the top. `--harness` narrows the picker to one provider; `--harness X --session Y` (and `--project P` for keyed providers) skips the picker entirely. Pathbase flags (`--url`, `--anon`, `--repo`, `--slug`, `--public`) match `path export pathbase`. By default the derived doc is written to the cache like `import` does; pass `--no-cache` to skip. When the manifest shows the picked session unchanged since its last sync (`sync::fresh_cache_id`: stamps match, doc present; the freshness stat targets that one artifact directly, no sibling enumeration), share uploads the cached doc directly instead of re-deriving. The cache ingests maximally (thinking always included), and uploads carry the same full derivation as local projection (`resume`, `p export `) — there is no egress stripping. - `path resume ` is the inverse of `path share`. It accepts a Pathbase URL, an `owner/repo/slug` shorthand, a local toolpath JSON file, or a cache id; resolves it (caching URL fetches under `~/.toolpath/documents/` unless `--no-cache`); validates that the document is a single agent-bearing `Path`; then opens an `fzf` harness picker (skipped with `--harness X`). The picker pre-selects the source harness inferred from `path.meta.source` (`claude-code`/`gemini-cli`/`codex`/`opencode`/`pi`) when it's installed. After picking, `path resume` projects the session into the harness's on-disk layout under the chosen working directory (default: shell cwd; override with `-C, --cwd P`) and `execvp`'s the harness's resume command (`claude -r ` / `gemini --resume ` / `codex resume ` / `opencode --session ` / `pi --session `). On Windows it spawns and waits, propagating the exit code. The exec is mockable via `cmd_resume::ExecStrategy` — production uses `RealExec`; integration tests use `RecordingExec` to capture the recipe without launching a real harness. -- `path query` does not load the whole cache into memory when it can avoid it. `crates/path-cli/src/query/plan.rs` parses the jaq filter into jaq's own AST (`jaq_core::load::parse::Term`) and classifies it into a `Plan`: `PerFileStream` (`.[] | g` element-wise work — run per document, print as you go), `Decompose { reduce }` (algebraic aggregations — run the whole filter per file, concatenate the per-file outputs, then run a derived combine: `map`→`add` (array concat), top-N `sort_by(k)|.[:N]`→`add | sort_by(k)|.[:N]`, `length`→`add` over exact integer counts), or `Slurp` (the always-correct whole-array fallback). Recognition is conservative — a non-distributive prefix like `unique`/`group_by` slurps, and so do scalar `add` (float sums re-associate across per-file partials), `min`/`max` (`[] | min == null` poisons the merge), and any unrecognized tail — so **the planner never changes an answer** — `crates/path-cli/src/query/filter.rs` tests assert streamed output equals slurp byte-for-byte. `filter::execute` compiles the filter once (jaq's compiled `Filter` is fully owned, so it's reused across files) and drives the plan; `mod.rs::stream_files` yields one document's wrapped steps at a time. `TOOLPATH_QUERY_EXPLAIN=1` prints the chosen plan to stderr. No user-facing flag — it's automatic. Tie-break caveat: a streamed top-N matches slurp's *ranking*, but boundary ties may resolve to different specific rows. +- `path query` does not load the whole cache into memory when it can avoid it. `crates/path-cli/src/query/plan.rs` parses the jaq filter into jaq's own AST (`jaq_core::load::parse::Term`) and classifies it into a `Plan`: `PerFileStream` (`.[] | g` element-wise work — run per document, print as you go), `Decompose { reduce }` (algebraic aggregations — run the whole filter per file, concatenate the per-file outputs, then run a derived combine: `map`→`add` (array concat), top-N `sort_by(k)|.[:N]`→`add | sort_by(k)|.[:N]`, `length`→`add` over exact integer counts), or `Slurp` (the always-correct whole-array fallback). Recognition is conservative — a non-distributive prefix like `unique`/`group_by` slurps, and so do scalar `add` (float sums re-associate across per-file partials), `min`/`max` (`[] | min == null` poisons the merge), and any unrecognized tail — so **the planner never changes an answer** — `crates/path-cli/src/query/filter.rs` tests assert streamed output equals slurp byte-for-byte. `filter::execute` compiles the filter once (jaq's compiled `Filter` is fully owned, so it's reused across files) and drives the plan; `mod.rs::stream_files` yields one document's wrapped steps at a time. Execution is parallel (`mod.rs::execute_plan`/`for_each_file`): `PerFileStream`/`Decompose` run the whole per-file pipeline (parse → wrap → filter → render/pack) on rayon workers in chunks — partials cross threads as compact JSON bytes since jaq `Val`s are `Rc`-based — while `Slurp` parallelizes only parse/wrap; output stays byte-identical to a sequential scan (ordering, warnings, error precedence), and the emscripten build stays fully sequential. `TOOLPATH_QUERY_EXPLAIN=1` prints the chosen plan to stderr. No user-facing flag — it's automatic. Tie-break caveat: a streamed top-N matches slurp's *ranking*, but boundary ties may resolve to different specific rows. - Cache sync: `path p cache sync [types…]` (`crates/path-cli/src/artifact.rs`: `ArtifactType` + `ArtifactRef` + the stamp helpers; `sync/engine.rs`: manifest + ingestion loop; `sync/sources.rs`: an `ArtifactSource` trait — enumerate / stamp / derive / peek-dir / scope-match — with one impl per provider, so the engine never matches on artifact type) incrementally ingests artifacts into the cache — no args syncs every artifact type. Change detection is **stat-level**: each artifact is enumerated as an `ArtifactRef` whose fingerprint is the source file's mtime + size (claude: the *whole session chain* — max segment mtime + summed segment sizes via `claude_chain_stamp`, because Claude Code rotates to a new file on continuation while the chain keeps its oldest segment's id, so appends land in the newest file, not the head; the chain comes from the same cached index `list_conversations` builds; codex: rollout file, id from the stem's trailing UUID; pi: session file, id from a one-line header peek; copilot: `session-state//events.jsonl`, pure read-dir + stat) or the DB row's updated-at (opencode: header-only `SELECT time_updated`; cursor: composer headers' `lastUpdatedAt`, bubble-less drafts skipped, workspace-less composers *included* unlike `share`). Gemini enumerates via `PathResolver::list_session_entries` (`toolpath-gemini` 0.6.1), whose identity peek is bounded to the first 4 KiB of a main file. Deciding "nothing changed" reads no session bodies — a no-op sync is milliseconds. Changed/new artifacts derive through the same provider managers (each source calls the `derive_*_session_with` helpers in `derive.rs`). Manifest at `~/.toolpath/manifest.json`: artifact type → artifact id → `{path?, cache_id, modified?, size?, synced_at}`; atomic temp+rename writes, `0600`, checkpointed every 10 writes (interruption-safe: a killed run keeps nearly everything it derived, and derives run newest-first so partial progress covers the sessions that matter most); writers serialize on an advisory lock (`manifest.json.lock`) and every write is a locked read-merge-save — checkpoints merge only the records the run wrote — so concurrent invocations (query auto-syncs, imports) union their records instead of clobbering each other. Pending work reports progress on stderr (`\r`-updating ` done/total` on a TTY, a plain line every 25 items otherwise; no-op syncs stay silent). Sync always writes the cache with force — refresh semantics — and never deletes: artifacts removed upstream keep their cache docs and manifest records (archive, not mirror). Derivation failures warn and tally, they don't abort. A record's `cache_id` is *optional*: a record without one is "known, not materialized" — created when a `--parent-dir` constraint excluded a peeked artifact, or when `p cache rm` evicts a doc (rm downgrades the record; the next in-scope sync re-materializes it, and sync also verifies the doc file actually exists before skipping, so even out-of-band deletions self-heal). `--parent-dir ` / `-d` on both `p cache sync` and `path query` restricts ingestion to artifacts under that directory (subtree): path-keyed providers prune whole projects before enumerating (claude compares in *slug space* — its dir slugs are lossy, `/`/`_`/`.` all became `-`), cwd-keyed ones check the directory their cheap headers carry, and codex/copilot — whose cwd lives inside the session file — get a one-line peek only when new/changed, memoized into the record's `path`. The stat gate always runs first: unchanged+cached artifacts skip before any scope check. Out-of-scope work is tallied separately (`N out of scope`) and never touches a materialized record's stamp. Claude derives leave `DeriveConfig.project_path` unset so `path.base` comes from the session's own recorded cwd rather than the lossy slug. `path query` runs this sync implicitly before reading, scoped to its flags (`--source X` → that type; `--id`s → their prefixes; bare query → all types; `--input`-only → none), quiet unless something was ingested, degrading to the cache as-is if sync fails; `--no-sync` opts out. `p import` and `share` record what they write: every session derive carries a provenance `ArtifactRef` (stamped *before* the source is read, in `DerivedDoc.provenance`), and the cache-write sites call `sync::record_artifact` so the next sync sees those artifacts as unchanged instead of re-deriving them. Every import flow — explicit `--session`, picker multi-select, `--all`, and the most-recent fallbacks — loops the per-session helpers, so every session write is recorded; there is no bulk `derive_project` path in the CLI anymore, and `p import pi --all` now emits one Path per session like every other provider (it used to emit a single combined Graph). `--no-cache` paths record nothing: the manifest describes the cache. - `ArtifactType` (`crates/path-cli/src/artifact.rs`) is the general enum naming artifact sources — the seven agent harnesses (incl. copilot) plus `Git` (8 variants). Git artifacts are *recorded* in the manifest by `p import git` (id `-`, `path` = the repo directory) but never *discovered* — there is no machine-wide registry of repos — so sync reports them and leaves them alone. Github and pathbase are deliberately not artifact types: they are remote services, not local artifact sources, and their imports stay out of the manifest. It derives `clap::ValueEnum` and is used by `p cache sync` types, the sync manifest keys, `ArtifactRow.artifact_type`, and `cmd_import`'s cache-id prefixes (`name()` is both the manifest key and the `make_id` source string). The deliberately parallel `Harness` enum (`crates/path-cli/src/harness.rs`, alongside `HarnessBundle`) names the seven agent *runtimes* — things sessions can be shared from and resumed into — and is what `share`/`resume` `--harness` take, so future non-harness artifact types stay unrepresentable there (you can't resume into a git repo). `Harness::artifact_type()` maps into the general enum; `ArtifactType::harness()` is the partial inverse. Keep new code on `ArtifactType` unless it's genuinely harness-only. diff --git a/Cargo.lock b/Cargo.lock index 6cdb1e62..3305f9a8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -543,6 +543,31 @@ dependencies = [ "libc", ] +[[package]] +name = "crossbeam-deque" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" + [[package]] name = "crossterm" version = "0.29.0" @@ -2441,7 +2466,7 @@ dependencies = [ [[package]] name = "path-cli" -version = "0.16.0" +version = "0.17.0" dependencies = [ "anyhow", "assert_cmd", @@ -2457,6 +2482,7 @@ dependencies = [ "pathbase-client", "predicates", "rand 0.9.4", + "rayon", "regex", "reqwest", "rusqlite", @@ -3032,6 +3058,26 @@ dependencies = [ "bitflags 2.11.1", ] +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "recvmsg" version = "1.0.0" diff --git a/Cargo.toml b/Cargo.toml index 23223581..d7103cdd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,7 +37,7 @@ toolpath-github = { version = "0.6.0", path = "crates/toolpath-github" } toolpath-dot = { version = "0.5.0", path = "crates/toolpath-dot" } toolpath-md = { version = "0.7.0", path = "crates/toolpath-md" } toolpath-pi = { version = "0.6.1", path = "crates/toolpath-pi" } -path-cli = { version = "0.16.0", path = "crates/path-cli" } +path-cli = { version = "0.17.0", path = "crates/path-cli" } pathbase-client = { version = "0.2.0", path = "crates/pathbase-client" } reqwest = { version = "0.13", default-features = false, features = ["blocking", "json", "rustls"] } @@ -54,6 +54,7 @@ similar = "2" tempfile = "3.15" insta = "1" rusqlite = { version = "0.32", features = ["bundled"] } +rayon = "1.10" uuid = { version = "1", features = ["serde"] } [profile.wasm] diff --git a/crates/path-cli/Cargo.toml b/crates/path-cli/Cargo.toml index 575c972f..b42638a5 100644 --- a/crates/path-cli/Cargo.toml +++ b/crates/path-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "path-cli" -version = "0.16.0" +version = "0.17.0" edition.workspace = true license.workspace = true repository = "https://github.com/empathic/toolpath" @@ -56,6 +56,7 @@ git2 = { workspace = true } reqwest = { workspace = true } tokio = { workspace = true } rusqlite = { workspace = true } +rayon = { workspace = true } uuid = { workspace = true, features = ["v4"] } # Embedded fuzzy picker — used when external `fzf` isn't on PATH. # Disable default features to avoid pulling in skim's CLI deps diff --git a/crates/path-cli/src/query/filter.rs b/crates/path-cli/src/query/filter.rs index d7a68a81..159fe520 100644 --- a/crates/path-cli/src/query/filter.rs +++ b/crates/path-cli/src/query/filter.rs @@ -59,7 +59,6 @@ pub fn execute( eval_print(&main, merged, out, &pp, raw)?; } Plan::Decompose { reduce } => { - let reducer = compile(reduce)?; let mut partials: Vec = Vec::new(); let mut saw_file = false; let mut emit = |val: Val| { @@ -68,23 +67,109 @@ pub fn execute( Ok(()) }; run_files(&mut emit)?; - if saw_file { - let merged: Val = partials.into_iter().collect(); - eval_print(&reducer, merged, out, &pp, raw)?; - } else { - // No document contributed a partial: the decomposition - // identity `reduce(⋃ main(fᵢ)) == main(⋃ fᵢ)` degenerates to - // `main([])`. Run the *main* filter over an empty array so the - // answer matches slurp (`length` → 0, `sort_by|.[:N]` → []), - // not the reducer over `[]` (which would give null / error). - let empty: Val = std::iter::empty().collect(); - eval_print(&main, empty, out, &pp, raw)?; - } + finish_decompose(main_src, reduce, partials, saw_file, compact, raw, out)?; } } Ok(()) } +/// Finish a `Decompose` plan over the gathered per-file partials — shared by +/// the sequential and parallel drivers so the zero-file rule lives once. +/// +/// With no document contributing a partial, the decomposition identity +/// `reduce(⋃ main(fᵢ)) == main(⋃ fᵢ)` degenerates to `main([])`. Run the +/// *main* filter over an empty array so the answer matches slurp (`length` +/// → 0, `sort_by|.[:N]` → []), not the reducer over `[]` (which would give +/// null / error). +pub fn finish_decompose( + main_src: &str, + reduce_src: &str, + partials: Vec, + saw_file: bool, + compact: bool, + raw: bool, + out: &mut dyn Write, +) -> Result<()> { + let pp = pretty(compact); + if saw_file { + let reducer = compile(reduce_src)?; + let merged: Val = partials.into_iter().collect(); + eval_print(&reducer, merged, out, &pp, raw) + } else { + let main = compile(main_src)?; + let empty: Val = std::iter::empty().collect(); + eval_print(&main, empty, out, &pp, raw) + } +} + +/// Surface a filter's load/compile errors without running it. The parallel +/// drivers call this before touching any file, preserving the sequential +/// path's error precedence (a bad filter beats a missing `--id`). +#[cfg(not(target_os = "emscripten"))] +pub fn compile_check(code: &str) -> Result<()> { + compile(code).map(|_| ()) +} + +/// Compile `code` at most once per thread. Rayon workers are long-lived, so +/// per-file evaluation pays one compile per pool thread, not per file. +#[cfg(not(target_os = "emscripten"))] +fn with_compiled(code: &str, f: impl FnOnce(&Program) -> Result) -> Result { + use std::cell::RefCell; + thread_local! { + static CACHE: RefCell> = const { RefCell::new(None) }; + } + CACHE.with(|cache| { + let mut cache = cache.borrow_mut(); + if !matches!(&*cache, Some((cached, _)) if cached == code) { + *cache = Some((code.to_string(), compile(code)?)); + } + let (_, prog) = cache.as_ref().expect("just populated"); + f(prog) + }) +} + +/// Evaluate `code` over one file's steps and render its outputs exactly as +/// [`execute`]'s streaming printer would. Runs on worker threads. +#[cfg(not(target_os = "emscripten"))] +pub fn render_file( + code: &str, + steps: Vec, + compact: bool, + raw: bool, +) -> Result> { + with_compiled(code, |prog| { + let mut buf = Vec::new(); + eval_print(prog, steps_to_val(steps)?, &mut buf, &pretty(compact), raw)?; + Ok(buf) + }) +} + +/// Evaluate `code` over one file's steps and pack the partial outputs into +/// one compact JSON array. Jaq values are `Rc`-based and thread-local, so +/// partials cross threads as bytes; [`unpack_partials`] reparses them on the +/// consuming thread. +#[cfg(not(target_os = "emscripten"))] +pub fn partials_file(code: &str, steps: Vec) -> Result> { + with_compiled(code, |prog| { + let vals = eval_collect(prog, steps_to_val(steps)?)?; + let arr: Val = vals.into_iter().collect(); + let mut buf = Vec::new(); + jaq_json::write::write(&mut buf, &pretty(true), 0, &arr) + .map_err(|e| anyhow!("internal: could not pack partials: {e}"))?; + Ok(buf) + }) +} + +/// Reparse one file's packed partials (the inverse of [`partials_file`]). +#[cfg(not(target_os = "emscripten"))] +pub fn unpack_partials(bytes: &[u8]) -> Result> { + match jaq_json::read::parse_single(bytes) { + Ok(Val::Arr(items)) => Ok(items.iter().cloned().collect()), + Ok(_) => Err(anyhow!("internal: packed partials were not an array")), + Err(e) => Err(anyhow!("internal: could not reparse partials: {e}")), + } +} + /// Convert one file's wrapped steps into a jaq array value. The byte buffer is /// per-file (bounded), so no whole-cache serialization is ever held. pub fn steps_to_val(steps: Vec) -> Result { @@ -402,4 +487,93 @@ mod tests { // filter. assert_slurps("map({id: .step.id}) | (sort_by(.id)) | .[:1]"); } + + // ── The parallel drivers' building blocks ───────────────────────── + // + // `render_file`/`partials_file` + `finish_decompose` are what the + // parallel per-file drivers run on worker threads. Their output must be + // byte-identical to the sequential engine, which is itself pinned to + // slurp above. + + fn file_steps(f: &serde_json::Value) -> Vec { + f.as_array().cloned().unwrap_or_default() + } + + #[test] + fn render_file_concat_matches_sequential_stream() { + for (compact, raw) in [(true, false), (false, false), (true, true)] { + for code in [ + ".[] | select(.dead_end)", + r#".[] | select(.step.actor | startswith("agent:")) | .step.id"#, + "map(select(.tokens > 40))", + ] { + let mut seq: Vec = Vec::new(); + execute(&Plan::PerFileStream, code, compact, raw, &mut seq, |emit| { + for f in &fixture() { + emit(steps_to_val(file_steps(f))?)?; + } + Ok(()) + }) + .unwrap(); + + let mut par: Vec = Vec::new(); + for f in &fixture() { + par.extend(render_file(code, file_steps(f), compact, raw).unwrap()); + } + assert_eq!( + String::from_utf8(par).unwrap(), + String::from_utf8(seq).unwrap(), + "parallel render must equal sequential for `{code}` (compact={compact}, raw={raw})" + ); + } + } + } + + #[test] + fn partials_roundtrip_matches_sequential_decompose() { + for code in [ + "length", + "map(select(.dead_end)) | length", + "sort_by(-.tokens) | .[:2]", + "map({id: .step.id, t: .tokens})", + ] { + let plan = crate::query::plan::analyze(code); + let Plan::Decompose { reduce } = &plan else { + panic!("`{code}` should decompose"); + }; + let seq = run_with(&plan, code, &fixture()); + + let mut partials: Vec = Vec::new(); + let mut saw_file = false; + for f in &fixture() { + saw_file = true; + let bytes = partials_file(code, file_steps(f)).unwrap(); + partials.extend(unpack_partials(&bytes).unwrap()); + } + let mut par: Vec = Vec::new(); + finish_decompose(code, reduce, partials, saw_file, true, false, &mut par).unwrap(); + assert_eq!( + String::from_utf8(par).unwrap(), + seq, + "parallel decompose must equal sequential for `{code}`" + ); + } + } + + #[test] + fn finish_decompose_zero_files_matches_slurp() { + for code in ["length", "sort_by(.tokens) | .[:2]"] { + let plan = crate::query::plan::analyze(code); + let Plan::Decompose { reduce } = &plan else { + panic!("`{code}` should decompose"); + }; + let mut par: Vec = Vec::new(); + finish_decompose(code, reduce, Vec::new(), false, true, false, &mut par).unwrap(); + let none: &[serde_json::Value] = &[]; + assert_eq!( + String::from_utf8(par).unwrap(), + run_with(&Plan::Slurp, code, none) + ); + } + } } diff --git a/crates/path-cli/src/query/mod.rs b/crates/path-cli/src/query/mod.rs index 635c81bb..f833de1e 100644 --- a/crates/path-cli/src/query/mod.rs +++ b/crates/path-cli/src/query/mod.rs @@ -61,12 +61,120 @@ pub fn run(scope: &Scope, code: &str, compact: bool, raw: bool) -> Result<()> { // raw `StdoutLock` is line-buffered (a syscall per line). let stdout = std::io::stdout(); let mut out = std::io::BufWriter::new(stdout.lock()); - filter::execute(&plan, code, compact, raw, &mut out, |emit| { - stream_files(scope, emit) - })?; + execute_plan(scope, &plan, code, compact, raw, &mut out)?; out.flush().context("flush stdout") } +/// Dispatch a plan to its driver. `PerFileStream` and `Decompose` treat every +/// file independently, so their whole per-file pipeline — parse, wrap, filter, +/// render/pack — runs on worker threads and only ordered consumption stays +/// here. `Slurp` needs all steps as one jaq array on one thread, so it +/// parallelizes parsing only. +#[cfg(not(target_os = "emscripten"))] +fn execute_plan( + scope: &Scope, + plan: &plan::Plan, + code: &str, + compact: bool, + raw: bool, + out: &mut dyn Write, +) -> Result<()> { + match plan { + plan::Plan::Slurp => filter::execute(plan, code, compact, raw, out, |emit| { + stream_files(scope, emit) + }), + plan::Plan::PerFileStream => { + filter::compile_check(code)?; + for_each_file( + scope, + |steps| filter::render_file(code, steps, compact, raw), + |bytes| { + out.write_all(&bytes)?; + Ok(()) + }, + ) + } + plan::Plan::Decompose { reduce } => { + filter::compile_check(code)?; + let mut partials: Vec = Vec::new(); + let mut saw_file = false; + for_each_file( + scope, + |steps| filter::partials_file(code, steps), + |bytes| { + saw_file = true; + partials.extend(filter::unpack_partials(&bytes)?); + Ok(()) + }, + )?; + filter::finish_decompose(code, reduce, partials, saw_file, compact, raw, out) + } + } +} + +/// The playground build has no threads (emscripten without pthreads): every +/// plan runs on the sequential engine. +#[cfg(target_os = "emscripten")] +fn execute_plan( + scope: &Scope, + plan: &plan::Plan, + code: &str, + compact: bool, + raw: bool, + out: &mut dyn Write, +) -> Result<()> { + filter::execute(plan, code, compact, raw, out, |emit| { + stream_files(scope, emit) + }) +} + +/// Drive the per-file parallel plans: evaluate `per_file` for each selected, +/// scoped document on a thread pool in chunks, consuming products in +/// selection order on this thread. Chunking keeps output (and per-file +/// warnings) byte-identical to a sequential scan while holding at most one +/// chunk of products in memory. Explicit-file failures abort at the file's +/// position; a corrupt cache file skips with a warning, exactly like the +/// sequential scan. +#[cfg(not(target_os = "emscripten"))] +fn for_each_file( + scope: &Scope, + per_file: impl Fn(Vec) -> Result + Sync, + mut consume: impl FnMut(T) -> Result<()>, +) -> Result<()> { + use rayon::prelude::*; + + let kind_sel = scope.kind.as_deref().map(kinds::parse_kind_selector); + let project = scope.project.as_deref().map(canonicalize_or_self); + let parent_dir = scope.parent_dir.as_deref().map(canonicalize_or_self); + let sources = select_files(scope)?; + + let chunk = rayon::current_num_threads().max(1) * 2; + for batch in sources.chunks(chunk) { + let products: Vec> = batch + .par_iter() + .map(|src| { + load_and_wrap( + src, + kind_sel.as_ref(), + project.as_deref(), + parent_dir.as_deref(), + ) + .and_then(&per_file) + }) + .collect(); + for (src, res) in batch.iter().zip(products) { + match res { + Ok(product) => consume(product)?, + Err(e) if src.explicit => { + return Err(e.context(format!("read {}", src.label()))); + } + Err(e) => eprintln!("warning: skipping {}: {e:#}", src.label()), + } + } + } + Ok(()) +} + /// Where a document came from, and the `cache_id` to stamp on its steps. struct DocSource { cache_id: String, @@ -91,43 +199,87 @@ impl DocSource { } } -/// Stream each selected, scoped document to `emit` as a jaq array value, -/// one file at a time. Only one document's `Graph` and step values are alive -/// per iteration; the executor decides how to combine them. +/// Stream each selected, scoped document to `emit` as a jaq array value, in +/// selection order. Reading and parsing — the dominant cost — runs on a +/// thread pool in chunks; conversion to `Val` and emission stay on this +/// thread because jaq values are `Rc`-based, not `Send`. Chunking keeps +/// output (and per-file warnings) byte-identical to a sequential scan while +/// holding at most one chunk of parsed documents in memory. fn stream_files(scope: &Scope, emit: &mut dyn FnMut(Val) -> Result<()>) -> Result<()> { let kind_sel = scope.kind.as_deref().map(kinds::parse_kind_selector); let project = scope.project.as_deref().map(canonicalize_or_self); let parent_dir = scope.parent_dir.as_deref().map(canonicalize_or_self); + let sources = select_files(scope)?; - for src in select_files(scope)? { - let graph = match read_source(&src) { - Ok(g) => g, - // An explicitly named document that won't read/parse is a hard - // error (a typo'd `--input`, bad stdin). A corrupt file merely - // encountered while scanning the cache is skipped with a warning. - Err(e) if src.explicit => { - return Err(e.context(format!("read {}", src.label()))); - } - Err(e) => { - eprintln!("warning: skipping {}: {e:#}", src.label()); - continue; + #[cfg(not(target_os = "emscripten"))] + { + use rayon::prelude::*; + let chunk = rayon::current_num_threads().max(1) * 2; + for batch in sources.chunks(chunk) { + let parsed: Vec>> = batch + .par_iter() + .map(|src| { + load_and_wrap( + src, + kind_sel.as_ref(), + project.as_deref(), + parent_dir.as_deref(), + ) + }) + .collect(); + for (src, res) in batch.iter().zip(parsed) { + emit_wrapped(src, res, emit)?; } - }; - let mut steps = Vec::new(); - wrap_graph( - &src, - &graph, + } + } + + // The playground build has no threads (emscripten without pthreads). + #[cfg(target_os = "emscripten")] + for src in &sources { + let res = load_and_wrap( + src, kind_sel.as_ref(), project.as_deref(), parent_dir.as_deref(), - &mut steps, ); - drop(graph); - emit(filter::steps_to_val(steps)?)?; + emit_wrapped(src, res, emit)?; } + Ok(()) } +/// Read, parse, and wrap one document's steps. +fn load_and_wrap( + src: &DocSource, + kind_sel: Option<&KindSelector>, + project: Option<&FsPath>, + parent_dir: Option<&FsPath>, +) -> Result> { + let graph = read_source(src)?; + let mut steps = Vec::new(); + wrap_graph(src, &graph, kind_sel, project, parent_dir, &mut steps); + Ok(steps) +} + +/// Emit one file's outcome. An explicitly named document that won't +/// read/parse is a hard error (a typo'd `--input`, bad stdin); a corrupt +/// file merely encountered while scanning the cache is skipped with a +/// warning. +fn emit_wrapped( + src: &DocSource, + res: Result>, + emit: &mut dyn FnMut(Val) -> Result<()>, +) -> Result<()> { + match res { + Ok(steps) => emit(filter::steps_to_val(steps)?), + Err(e) if src.explicit => Err(e.context(format!("read {}", src.label()))), + Err(e) => { + eprintln!("warning: skipping {}: {e:#}", src.label()); + Ok(()) + } + } +} + /// Resolve the scope's file-selection flags to a deterministic list of /// documents to load. /// diff --git a/site/_data/crates.json b/site/_data/crates.json index 99bcad0b..5ff8c063 100644 --- a/site/_data/crates.json +++ b/site/_data/crates.json @@ -113,7 +113,7 @@ }, { "name": "path-cli", - "version": "0.16.0", + "version": "0.17.0", "description": "Unified CLI (binary: path)", "docs": "https://docs.rs/path-cli", "crate": "https://crates.io/crates/path-cli",