diff --git a/packages/python/e2e-attestations/claude-1325-parallel-walk.json b/packages/python/e2e-attestations/claude-1325-parallel-walk.json new file mode 100644 index 000000000..23c3ee014 --- /dev/null +++ b/packages/python/e2e-attestations/claude-1325-parallel-walk.json @@ -0,0 +1,7 @@ +{ + "command": "just test-e2e", + "ran_at": 1791242399, + "exit_code": 0, + "commit": "05c2e1ad49fa161f2538efdad2c7e42468f001a7", + "branch": "claude/1325-parallel-walk" +} diff --git a/packages/rust/changelog.d/2026-10-05-path-table-parallel-walk.md b/packages/rust/changelog.d/2026-10-05-path-table-parallel-walk.md new file mode 100644 index 000000000..800c73eff --- /dev/null +++ b/packages/rust/changelog.d/2026-10-05-path-table-parallel-walk.md @@ -0,0 +1,4 @@ +**Changed** Path-table scans and the startup scan read sibling directories on +spare cores. `SELECT count(*) FROM './**/*.md'` over a 1.6M-file tree spread +across 2.3k directories drops from 0.72s to 0.23s, and `SELECT path` from +0.80s to 0.33s. The files returned and their order are unchanged. diff --git a/packages/rust/migrations.d/2026-10-05-path-table-parallel-walk.md b/packages/rust/migrations.d/2026-10-05-path-table-parallel-walk.md new file mode 100644 index 000000000..dfde35a87 --- /dev/null +++ b/packages/rust/migrations.d/2026-10-05-path-table-parallel-walk.md @@ -0,0 +1,28 @@ +### Core: directory walks read sibling directories in parallel + +#### Summary + +Path-table scans and the startup scan read sibling directories on spare +cores. The files returned and their order are unchanged; nothing breaks. + +#### Required changes + +_None._ + +#### Deprecations removed + +_None._ + +#### Behavior changes without code changes + +_None._ + +#### Verification + +```bash +cd "$(mktemp -d)" +mkdir -p a b +touch a/2.md a/1.md b/1.md +dirsql query "SELECT path FROM './**/*.md'" +# expected: [{"path":"a/1.md"},{"path":"a/2.md"},{"path":"b/1.md"}] +``` diff --git a/packages/rust/src/lib.rs b/packages/rust/src/lib.rs index 9965779cb..be7d0be59 100644 --- a/packages/rust/src/lib.rs +++ b/packages/rust/src/lib.rs @@ -43,6 +43,7 @@ pub mod row_event_flat; pub mod scanner; #[doc(hidden)] pub mod sql_literal; +mod tree_walk; #[doc(hidden)] pub mod vtab; mod vtab_scaffold; diff --git a/packages/rust/src/scanner.rs b/packages/rust/src/scanner.rs index e52724209..d19dda7a1 100644 --- a/packages/rust/src/scanner.rs +++ b/packages/rust/src/scanner.rs @@ -1,10 +1,12 @@ use crate::listing::{Listing, Seen}; use crate::matcher::{GlobError, Pattern, TableMatcher}; +use crate::tree_walk::{Step, walk_in_order}; use ignore::Match; use ignore::gitignore::{Gitignore, GitignoreBuilder}; use std::ffi::OsStr; use std::fs; use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::thread; /// Top-level directory name reserved for `dirsql`'s own metadata (e.g. the @@ -75,25 +77,27 @@ fn scan_below( // Match against relative path so globs like "comments/**/*.jsonl" work // regardless of the absolute root directory. + let walker = Walk { + ignore: matcher, + glob: None, + gitignore: false, + in_repo: false, + frames: Vec::new(), + }; walk( root, start, - matcher, - None, - false, - None, - &mut |rel_path, dir, name| { - if matcher.is_ignored(&rel_path) { - return; - } - + walker, + &|rel_path| !matcher.is_ignored(rel_path), + &mut |rel_path, dir| { seen += 1; on_file(seen); // Fan-out: a file matching N tables' globs yields N (path, table) // pairs, one per matching table, in declaration order. + let path = dir.join(rel_path.file_name().unwrap_or_default()); for m in matcher.match_all(&rel_path) { - results.push((dir.join(name), m.table_name)); + results.push((path.clone(), m.table_name)); } }, ); @@ -134,20 +138,20 @@ pub fn scan_glob( } else { None }; - let repo_frames = repo.map(|top| gitignores_above(root, top)); + let walker = Walk { + ignore, + glob: Some(glob), + gitignore, + in_repo: repo.is_some(), + frames: repo.map_or_else(Vec::new, |top| gitignores_above(root, top)), + }; let mut results = Vec::new(); walk( root, root, - ignore, - Some(glob), - gitignore, - repo_frames, - &mut |rel_path, _, _| { - if is_wanted(glob, ignore, &rel_path) { - results.push(rel_path); - } - }, + walker, + &|rel_path| is_wanted(glob, ignore, rel_path), + &mut |rel_path, _| results.push(rel_path), ); results } @@ -308,109 +312,133 @@ pub(crate) fn parse_gitignore_arg(arg: &str) -> Result { } } -/// The shared traversal: every file under `start`, visited with its -/// `root`-relative path, siblings in name order, so the whole walk comes out -/// in path order without a sort at the end. Prunes the reserved top-level -/// `.dirsql/` subtree and any directory the skip rules ignore wholesale, so an -/// ignored tree is never read at all. With `glob` given, a dot-named entry it -/// does not spell is skipped; `None` admits them all. With `gitignore` set, -/// entries a `.gitignore` in force ignores are pruned/skipped too, starting -/// from `repo_frames`: the ones in force above `start` when a repo encloses -/// it, `None` when none does. Inside a directory holding `.git`, its -/// `.gitignore` files apply. Symlinks are followed only as `glob` allows, and -/// not at all without one; a broken link or an unreadable directory -/// contributes nothing. +/// The shared traversal: every file under `start` that `keep` takes, visited +/// with its `root`-relative path and its directory, siblings in name order, +/// so the whole walk comes out in path order without a sort at the end. +/// Directories are read on spare cores ahead of the visiting. Prunes the +/// reserved top-level `.dirsql/` subtree and any directory the skip rules +/// ignore wholesale, so an ignored tree is never read at all. With a glob, a +/// dot-named entry it does not spell is skipped; without one all are +/// admitted. With `gitignore` set, entries a `.gitignore` in force ignores +/// are pruned/skipped too, starting from the walker's frames: the ones in +/// force above `start` when a repo encloses it. Inside a directory holding +/// `.git`, its `.gitignore` files apply. Symlinks are followed only as the +/// glob allows, and not at all without one; a broken link or an unreadable +/// directory contributes nothing. fn walk( root: &Path, start: &Path, - ignore: &TableMatcher, - glob: Option<&PathGlob>, - gitignore: bool, - repo_frames: Option>, - visit: &mut dyn FnMut(PathBuf, &Path, &OsStr), + walker: Walk<'_>, + keep: &(dyn Fn(&Path) -> bool + Sync), + visit: &mut dyn FnMut(PathBuf, &Path), ) { let rel = start.strip_prefix(root).unwrap_or(start); - // Depth below `root`, not below `start`: the reserved-directory rule is - // about the tree's top level wherever the walk begins. - let depth = rel.components().count(); - let mut walk = Walk { - ignore, - glob, - gitignore, - in_repo: repo_frames.is_some(), - frames: repo_frames.unwrap_or_default(), + let states = walker.glob.map_or_else(Vec::new, PathGlob::start); + let place = Place { + walk: walker, + dir: start.to_path_buf(), + rel: rel.to_path_buf(), + // Depth below `root`, not below `start`: the reserved-directory rule + // is about the tree's top level wherever the walk begins. + depth: rel.components().count(), + states, + linked: false, }; - let states = glob.map_or_else(Vec::new, PathGlob::start); - walk.descend(start, rel, depth, &states, false, visit); + walk_in_order( + Box::new(place), + &|place: Box>| (*place).explore(keep), + &mut |(rel, dir)| { + visit(rel, &dir); + }, + ); } -struct Walk<'a> { - ignore: &'a TableMatcher, - glob: Option<&'a PathGlob>, - gitignore: bool, - /// Whether a repo encloses the walk's current position, which is what - /// puts a `.gitignore` in force. - in_repo: bool, - /// The `.gitignore` files in force at the walk's current position, root - /// first; a directory's own file is pushed on entry and popped on exit. - frames: Vec, +/// A directory the walk enters, and how it got there: the glob `states` +/// it is entered with, and `linked` as whether its path crosses a followed +/// symlink. +struct Place<'a> { + walk: Walk<'a>, + dir: PathBuf, + rel: PathBuf, + depth: usize, + states: Vec, + linked: bool, } -impl Walk<'_> { - fn descend( - &mut self, - dir: &Path, - rel: &Path, - depth: usize, - states: &[usize], - linked: bool, - visit: &mut dyn FnMut(PathBuf, &Path, &OsStr), - ) { - let entered_repo = self.gitignore && !self.in_repo && holds_git(dir); - if entered_repo { - self.in_repo = true; - } - let mut pushed = false; - if self.in_repo - && let Some(matcher) = load_gitignore(dir) +type Found = (PathBuf, Arc); + +impl<'a> Place<'a> { + /// The directory's entries in walk order: each file `keep` takes, and + /// each directory to enter. + fn explore(self, keep: &(dyn Fn(&Path) -> bool + Sync)) -> Vec>, Found>> { + let Place { + mut walk, + dir, + rel, + depth, + states, + linked, + } = self; + walk.in_repo = walk.in_repo || (walk.gitignore && holds_git(&dir)); + if walk.in_repo + && let Some(matcher) = load_gitignore(&dir) { - self.frames.push(matcher); - pushed = true; + walk.frames.push(Arc::new(matcher)); } - let listing = Listing::read(dir); + let listing = Listing::read(&dir); let taken = judge_all(listing.entries(), &|listed| { - self.take( - dir, + walk.take( + &dir, listing.name(listed), listed.seen, - rel, + &rel, depth + 1, - states, + &states, linked, ) + .filter(|taken| match taken { + Taken::File(child) => keep(child), + Taken::Dir { .. } => true, + }) }); - for (listed, taken) in listing.entries().iter().zip(taken) { - let name = listing.name(listed); - match taken { - Some(Taken::Dir { + let dir: Arc = Arc::from(dir); + listing + .entries() + .iter() + .zip(taken) + .filter_map(|(listed, taken)| match taken? { + Taken::Dir { child, next, linked, - }) => { - self.descend(&dir.join(name), &child, depth + 1, &next, linked, visit); - } - Some(Taken::File(child)) => visit(child, dir, name), - None => {} - } - } - if pushed { - self.frames.pop(); - } - if entered_repo { - self.in_repo = false; - } + } => Some(Step::Dir(Box::new(Place { + walk: walk.clone(), + dir: dir.join(listing.name(listed)), + rel: child, + depth: depth + 1, + states: next, + linked, + }))), + Taken::File(child) => Some(Step::Leaf((child, Arc::clone(&dir)))), + }) + .collect() } +} +#[derive(Clone)] +struct Walk<'a> { + ignore: &'a TableMatcher, + glob: Option<&'a PathGlob>, + gitignore: bool, + /// Whether a repo encloses the walk's current position, which is what + /// puts a `.gitignore` in force. + in_repo: bool, + /// The `.gitignore` files in force at the walk's current position, root + /// first. + frames: Vec>, +} + +impl Walk<'_> { /// What the walk makes of entry `name` of `dir`, seen as `seen`, the /// directory at `rel` whose entries sit at `depth`, reached with glob /// `states` and `linked` as whether its path already crosses a followed @@ -570,10 +598,11 @@ fn classify(is_dir: bool, is_file: bool, linked: bool) -> Option { /// The `.gitignore` files above `start`, up to and including the repo root /// `top`, outermost first. -fn gitignores_above(start: &Path, top: &Path) -> Vec { +fn gitignores_above(start: &Path, top: &Path) -> Vec> { dirs_above(start, top) .into_iter() .filter_map(load_gitignore) + .map(Arc::new) .collect() } @@ -610,7 +639,7 @@ fn load_gitignore(dir: &Path) -> Option { /// Whether the `.gitignore` files in force mark `path` ignored. Deeper files /// take precedence (git's rule), and a whitelisting `!pattern` un-ignores. -fn is_gitignored(frames: &[Gitignore], path: &Path, is_dir: bool) -> bool { +fn is_gitignored(frames: &[Arc], path: &Path, is_dir: bool) -> bool { for frame in frames.iter().rev() { match frame.matched(path, is_dir) { Match::Ignore(_) => return true, @@ -840,12 +869,12 @@ mod tests { } /// A gitignore frame compiled from in-memory lines; no filesystem. - fn frame(dir: &str, lines: &[&str]) -> Gitignore { + fn frame(dir: &str, lines: &[&str]) -> Arc { let mut builder = GitignoreBuilder::new(Path::new(dir)); for line in lines { builder.add_line(None, line).unwrap(); } - builder.build().unwrap() + Arc::new(builder.build().unwrap()) } #[test] @@ -909,7 +938,7 @@ mod tests { fn walk_with<'a>( ignore: &'a TableMatcher, glob: Option<&'a PathGlob>, - frames: Vec, + frames: Vec>, ) -> Walk<'a> { Walk { ignore, diff --git a/packages/rust/src/tree_walk.rs b/packages/rust/src/tree_walk.rs new file mode 100644 index 000000000..86b84597f --- /dev/null +++ b/packages/rust/src/tree_walk.rs @@ -0,0 +1,239 @@ +//! A depth-first walk whose directories are read on spare cores. +//! +//! Exploring a directory (reading and judging its entries) is independent +//! of every other directory, so workers explore ahead of the walk while the +//! calling thread hands out what they found in depth-first order. + +use std::sync::{Condvar, Mutex}; +use std::thread; + +/// What exploring a directory yields: a leaf to hand out, or a directory to +/// explore in its place. +pub(crate) enum Step { + Dir(D), + Leaf(L), +} + +/// Once a walk has found this many directories it shares them across the +/// cores; below it, spawning workers costs more than the exploring. +const PARALLEL_DIRS: usize = 16; + +enum Item { + Dir(usize), + Leaf(L), +} + +struct Queue { + /// Each directory found, until a thread claims it. + unclaimed: Vec>, + /// Each explored directory's items, until the walk reaches it. + explored: Vec>>>, + /// Unclaimed directories, the next one the walk will want on top. + open: Vec, + done: bool, +} + +struct Shared<'e, D, L> { + queue: Mutex>, + changed: Condvar, + explore: &'e (dyn Fn(D) -> Vec> + Sync), +} + +/// Hand `visit` every leaf under `root`, depth first in the order `explore` +/// yields each directory's steps, exploring directories on every core. +pub(crate) fn walk_in_order( + root: D, + explore: &(dyn Fn(D) -> Vec> + Sync), + visit: &mut dyn FnMut(L), +) { + let shared = Shared { + queue: Mutex::new(Queue { + unclaimed: vec![Some(root)], + explored: vec![None], + open: Vec::new(), + done: false, + }), + changed: Condvar::new(), + explore, + }; + thread::scope(|scope| { + let mut spawned = false; + let mut stack = vec![shared.obtain(0).into_iter()]; + while let Some(items) = stack.last_mut() { + match items.next() { + Some(Item::Leaf(leaf)) => visit(leaf), + Some(Item::Dir(id)) => { + if !spawned && shared.found() >= PARALLEL_DIRS { + spawned = true; + let workers = thread::available_parallelism().map_or(1, usize::from); + for _ in 1..workers { + scope.spawn(|| shared.work()); + } + } + stack.push(shared.obtain(id).into_iter()); + } + None => { + stack.pop(); + } + } + } + shared.lock().done = true; + shared.changed.notify_all(); + }); +} + +impl Shared<'_, D, L> { + fn lock(&self) -> std::sync::MutexGuard<'_, Queue> { + self.queue + .lock() + .expect("exploring a directory does not panic") + } + + fn found(&self) -> usize { + self.lock().unclaimed.len() + } + + /// Directory `id`'s items: explored here if no worker has claimed it, + /// else once the worker that has is through. + fn obtain(&self, id: usize) -> Vec> { + let mut queue = self.lock(); + loop { + if let Some(items) = queue.explored[id].take() { + return items; + } + if let Some(dir) = queue.unclaimed[id].take() { + drop(queue); + return self.explore(dir); + } + queue = self + .changed + .wait(queue) + .expect("exploring a directory does not panic"); + } + } + + fn work(&self) { + let mut queue = self.lock(); + loop { + if queue.done { + return; + } + let claimed = queue.open.pop().map(|id| (id, queue.unclaimed[id].take())); + match claimed { + Some((id, Some(dir))) => { + drop(queue); + let items = self.explore(dir); + queue = self.lock(); + queue.explored[id] = Some(items); + self.changed.notify_all(); + } + Some((_, None)) => {} + None => { + queue = self + .changed + .wait(queue) + .expect("exploring a directory does not panic"); + } + } + } + } + + /// Explore `dir`, registering the directories it holds for any thread + /// to claim. + fn explore(&self, dir: D) -> Vec> { + let steps = (self.explore)(dir); + let mut queue = self.lock(); + let mut dirs = Vec::new(); + let items = steps + .into_iter() + .map(|step| match step { + Step::Leaf(leaf) => Item::Leaf(leaf), + Step::Dir(dir) => { + let id = queue.unclaimed.len(); + queue.unclaimed.push(Some(dir)); + queue.explored.push(None); + dirs.push(id); + Item::Dir(id) + } + }) + .collect(); + queue.open.extend(dirs.into_iter().rev()); + self.changed.notify_all(); + items + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashSet; + use std::time::Duration; + + /// A tree of `fanout` directories per level, `depth` levels deep, each + /// holding one leaf named for its path, with the leaf before its + /// subdirectories. + fn explore_tree(fanout: usize, depth: usize) -> impl Fn(String) -> Vec> { + move |path: String| { + let mut steps = vec![Step::Leaf(format!("{path}/f"))]; + if path.matches('/').count() < depth { + steps.extend((0..fanout).map(|i| Step::Dir(format!("{path}/{i}")))); + } + steps + } + } + + fn depth_first(path: &str, fanout: usize, depth: usize, out: &mut Vec) { + out.push(format!("{path}/f")); + if path.matches('/').count() < depth { + for i in 0..fanout { + depth_first(&format!("{path}/{i}"), fanout, depth, out); + } + } + } + + #[test] + fn leaves_come_out_depth_first_in_the_order_explored() { + let explore = explore_tree(3, 4); + let mut leaves = Vec::new(); + walk_in_order(String::from("r"), &explore, &mut |leaf| leaves.push(leaf)); + let mut expected = Vec::new(); + depth_first("r", 3, 4, &mut expected); + assert_eq!(leaves, expected); + } + + #[test] + fn a_lone_directory_yields_its_leaves() { + let explore = |(): ()| vec![Step::Leaf(1), Step::Leaf(2)]; + let mut leaves = Vec::new(); + walk_in_order((), &explore, &mut |leaf| leaves.push(leaf)); + assert_eq!(leaves, [1, 2]); + } + + #[test] + fn sibling_directories_are_explored_on_more_than_one_thread() { + assert!(threads_exploring(4, 3) > 1); + } + + fn threads_exploring(fanout: usize, depth: usize) -> usize { + let tree = explore_tree(fanout, depth); + let threads = Mutex::new(HashSet::new()); + let explore = |path: String| { + threads.lock().unwrap().insert(thread::current().id()); + thread::sleep(Duration::from_millis(2)); + tree(path) + }; + walk_in_order(String::from("r"), &explore, &mut |_| {}); + threads.into_inner().unwrap().len() + } + + #[test] + fn a_walk_of_few_directories_stays_on_the_calling_thread() { + assert_eq!(threads_exploring(3, 1), 1); + } + + #[test] + fn a_walk_spawns_its_workers_once() { + let cores = thread::available_parallelism().map_or(1, usize::from); + assert!(threads_exploring(4, 3) <= cores); + } +} diff --git a/packages/rust/tests/path_table_parallel_scan.rs b/packages/rust/tests/path_table_parallel_scan.rs index 7f69898d4..7919f2e54 100644 --- a/packages/rust/tests/path_table_parallel_scan.rs +++ b/packages/rust/tests/path_table_parallel_scan.rs @@ -17,14 +17,17 @@ const ROUNDS: usize = 4; /// A tree whose walk costs CPU rather than disk: every directory carries a /// `.gitignore` the walker compiles on entry (`.git` puts them in force), so a /// scan takes the same time whether or not the page cache already holds the -/// tree. +/// tree. Each directory nests in the one before, so no one walk has two +/// directories to read at once and only walking both trees together can +/// overlap. fn tree(root: &Path, name: &str) { fs::create_dir_all(root.join(".git")).unwrap(); let ignore: String = (0..PATTERNS) .map(|i| format!("build-{i}/**/*.tmp\n")) .collect(); - for i in 0..DIRS { - let dir = root.join(name).join(format!("d{i:05}")); + let mut dir = root.join(name); + for _ in 0..DIRS { + dir.push("d"); fs::create_dir_all(&dir).unwrap(); fs::write(dir.join(".gitignore"), &ignore).unwrap(); fs::write(dir.join("f.txt"), "x").unwrap(); diff --git a/packages/ts/e2e-attestations/claude-1325-parallel-walk.json b/packages/ts/e2e-attestations/claude-1325-parallel-walk.json new file mode 100644 index 000000000..cdaa650eb --- /dev/null +++ b/packages/ts/e2e-attestations/claude-1325-parallel-walk.json @@ -0,0 +1,7 @@ +{ + "command": "pnpm test:e2e", + "ran_at": 1791242461, + "exit_code": 0, + "commit": "2e00e1c1c2a267d0c89a766b8de9fa0b03fd123f", + "branch": "claude/1325-parallel-walk" +} diff --git a/testing-conventions.toml b/testing-conventions.toml index 27c3f2707..e363c8558 100644 --- a/testing-conventions.toml +++ b/testing-conventions.toml @@ -126,7 +126,7 @@ reason = "equivalent: RDONLY is zero and DIRECTORY and CLOEXEC are disjoint bits [[rust.exempt]] path = "src/scanner.rs" rules = ["mutation"] -lines = [73, 133, 332, 547] +lines = [75, 133, 332, 547] reason = "scan_below, scan_glob, walk, descend and kind_of read the filesystem; tests/scanner.rs (scan_finds_matching_files, scan_recurses_into_subdirectories), tests/vtab.rs (glob_scopes_the_walk, rowids_count_up_from_zero_in_scan_order) and tests/path_table_symlinks.rs assert what they return" [[rust.exempt]] @@ -138,7 +138,7 @@ reason = "whether a walked entry is a symlink, and whether the walk is below one [[rust.exempt]] path = "src/scanner.rs" rules = ["mutation"] -lines = [370, 374, 409, 446, 574, 587] +lines = [370, 382, 409, 446, 574, 587, 602] reason = "whether a repo encloses the walk, and which .gitignore files are in force, is known only from a real tree holding .git; the choice of directories is unit-tested in enclosing_repo and dirs_above, and tests/path_table_gitignore.rs (with_no_repo_a_gitignore_does_not_apply_at_the_index_root, inside_a_repo_the_same_gitignore_hides_everything, a_repo_root_gitignore_above_the_index_root_applies) asserts every outcome" [[rust.exempt]] @@ -150,7 +150,7 @@ reason = "the walk's running file count is observable only through the CLI's pro [[rust.exempt]] path = "src/scanner.rs" rules = ["mutation"] -lines = [387, 400] +lines = [387, 395, 400, 418] reason = "a child's depth decides the reserved top-level .dirsql rule for a real tree; tests/scanner.rs scan_excludes_top_level_dirsql_directory and tests/vtab.rs reserved_dirsql_directory_is_skipped assert it" [[rust.exempt]]