From 358b0ae59d4198eca0a74868e7d6a8366d69a160 Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 10:28:24 -0800 Subject: [PATCH 1/3] git: resolve full commit ids without a prefix lookup `resolve_to_commit()` passes every revspec to `rev_parse_single()`, and gix treats a 40-hex id there as a prefix to disambiguate. A prefix lookup always ends in an object-store refresh: it takes the store's write lock and rescans the pack directory and alternates, whether the object was found or not. Commit-metadata extraction resolves full ids three times per commit and the range generators once more, so metadata extraction alone rescans about 57,000 times on the v7.3..v7.4-rc1 merge window. The next commit shares one store between the workers, which turns that write lock into a queue; it depends on this change. A revspec that parses as a full object id and names an object in the repository now resolves directly to that object, peeled to a commit; a tag id peels to its commit as before. Anything else, a missing id included, goes through `rev_parse_single()` unchanged. A full id of a tree or blob now fails with "'' does not resolve to a commit" instead of gix's peel error. Indexing only the commits of that merge window (`semcode-index --commits`, 19,780 commits, 316 workers) took 15.1 s instead of 17.6 s and stored the same `git_commits` rows. `resolve_to_commit_takes_full_ids_directly` covers commit, tag, tree, missing, short and ancestry specs. `resolve_to_commit_full_id_does_not_refresh_the_store` reads the store's refresh counter: unchanged after resolving full ids, and moving for the same commits by short id. The git test helper moves to `git::test_git()`, which also clears `GIT_DIR`, `GIT_WORK_TREE` and `GIT_INDEX_FILE`, so a test run from a hook or `git rebase --exec` cannot touch the enclosing repository. --- src/git.rs | 114 +++++++++++++++++++++++++++++++++++++++++++++++++ src/indexer.rs | 20 +-------- 2 files changed, 115 insertions(+), 19 deletions(-) diff --git a/src/git.rs b/src/git.rs index 97e69db..a66b42c 100644 --- a/src/git.rs +++ b/src/git.rs @@ -7,6 +7,18 @@ use gix::bstr::ByteSlice; /// Helper function to resolve a revspec (SHA, tag, etc.) to a commit object /// Ensures that tags are properly dereferenced to their target commits pub fn resolve_to_commit<'a>(repo: &'a gix::Repository, revspec: &str) -> Result> { + // A full object id names its object directly. `rev_parse_single` would + // treat it as a prefix to disambiguate, and a prefix lookup ends in an + // object-store refresh that takes the store-wide write lock and rescans + // the pack directory, once per call, whether or not the id was found. + if let Ok(id) = gix::ObjectId::from_hex(revspec.as_bytes()) { + if let Ok(object) = repo.find_object(id) { + return object + .peel_to_commit() + .map_err(|_| anyhow::anyhow!("'{}' does not resolve to a commit", revspec)); + } + } + // Add ^{commit} to dereference tags to commits, but insert it before ancestry operators // For example: v6.16~8 becomes v6.16^{commit}~8 let commit_spec = if revspec.contains("^{") { @@ -219,6 +231,57 @@ pub fn get_git_file_hash_with_fallback>(file_path: P) -> Result String { + let out = std::process::Command::new("git") + .args(args) + .current_dir(dir) + .env_remove("GIT_DIR") + .env_remove("GIT_WORK_TREE") + .env_remove("GIT_INDEX_FILE") + .env("GIT_CONFIG_GLOBAL", "/dev/null") + .env("GIT_CONFIG_SYSTEM", "/dev/null") + .env("GIT_AUTHOR_NAME", "test") + .env("GIT_AUTHOR_EMAIL", "test@test.com") + .env("GIT_COMMITTER_NAME", "test") + .env("GIT_COMMITTER_EMAIL", "test@test.com") + .output() + .expect("git command failed to execute"); + assert!( + out.status.success(), + "git {:?} failed: {}", + args, + String::from_utf8_lossy(&out.stderr) + ); + String::from_utf8_lossy(&out.stdout).trim().to_string() +} + +/// A test repository whose objects all sit in one pack: a root commit with +/// `a.c` and `b.c`, a second commit that changes `a.c`, and an annotated tag +/// `v1` on the second. Returns the commits oldest first. +#[cfg(test)] +pub(crate) fn packed_test_repo() -> (tempfile::TempDir, Vec) { + let dir = tempfile::tempdir().unwrap(); + let p = dir.path(); + test_git(p, &["init", "-q"]); + std::fs::write(p.join("a.c"), "int a(void) { return 1; }\n").unwrap(); + std::fs::write(p.join("b.c"), "int b(void) { return a(); }\n").unwrap(); + test_git(p, &["add", "a.c", "b.c"]); + test_git(p, &["commit", "-q", "-m", "one"]); + let first = test_git(p, &["rev-parse", "HEAD"]); + std::fs::write(p.join("a.c"), "int a(void) { return 2; }\n").unwrap(); + test_git(p, &["commit", "-q", "-a", "-m", "two"]); + let second = test_git(p, &["rev-parse", "HEAD"]); + test_git(p, &["tag", "-a", "-m", "v1", "v1"]); + // One pack and no loose objects, so every read goes through the pack. + test_git(p, &["repack", "-a", "-d", "-q"]); + (dir, vec![first, second]) +} + #[cfg(test)] mod tests { use super::*; @@ -413,6 +476,57 @@ mod tests { assert!(base.chars().all(|c| c.is_ascii_hexdigit())); } } + + /// A full object id resolves through the object itself: a commit to + /// itself, an annotated tag to its commit, and anything that does not + /// peel to a commit to the usual error. Short ids and names still go + /// through rev-parse. + #[test] + fn resolve_to_commit_takes_full_ids_directly() { + let (dir, commits) = packed_test_repo(); + let repo = gix::discover(dir.path()).unwrap(); + let tag = test_git(dir.path(), &["rev-parse", "v1"]); + let tree = test_git(dir.path(), &["rev-parse", "HEAD^{tree}"]); + assert_ne!(tag, commits[1]); + + let id = |spec: &str| resolve_to_commit(&repo, spec).unwrap().id.to_string(); + assert_eq!(id(&commits[0]), commits[0]); + assert_eq!(id(&tag), commits[1]); + assert_eq!(id("v1"), commits[1]); + assert_eq!(id(&commits[1][..12]), commits[1]); + assert_eq!(id(&format!("{}~1", commits[1])), commits[0]); + + let err = resolve_to_commit(&repo, &tree).unwrap_err(); + assert_eq!( + err.to_string(), + format!("'{tree}' does not resolve to a commit") + ); + assert!(resolve_to_commit(&repo, &"0".repeat(40)).is_err()); + } + + /// Resolving a full id that exists leaves the object store alone. A + /// prefix lookup refreshes the store on every call, and every thread + /// sharing that store queues on the refresh's lock. + #[test] + fn resolve_to_commit_full_id_does_not_refresh_the_store() { + let (dir, commits) = packed_test_repo(); + let repo = gix::discover(dir.path()).unwrap(); + let refreshes = || repo.objects.store_ref().metrics().num_refreshes; + resolve_to_commit(&repo, &commits[0]).unwrap(); + + let before = refreshes(); + for commit in &commits { + resolve_to_commit(&repo, commit).unwrap(); + } + assert_eq!(refreshes(), before); + + // The same commits by short id do refresh, so the count above + // would have moved had the full ids taken that path. + for commit in &commits { + resolve_to_commit(&repo, &commit[..12]).unwrap(); + } + assert!(refreshes() > before); + } } /// Get git hash for a specific file at a specific commit diff --git a/src/indexer.rs b/src/indexer.rs index 2752ea1..59dc65b 100644 --- a/src/indexer.rs +++ b/src/indexer.rs @@ -1143,25 +1143,7 @@ mod tests { /// Helper: run a git command in a directory, panic on failure. fn git(repo: &std::path::Path, args: &[&str]) -> String { - let out = std::process::Command::new("git") - .args(args) - .current_dir(repo) - // Do not inherit developer's git configuration. - .env("GIT_CONFIG_GLOBAL", "/dev/null") - .env("GIT_CONFIG_SYSTEM", "/dev/null") - .env("GIT_AUTHOR_NAME", "test") - .env("GIT_AUTHOR_EMAIL", "test@test.com") - .env("GIT_COMMITTER_NAME", "test") - .env("GIT_COMMITTER_EMAIL", "test@test.com") - .output() - .expect("git command failed to execute"); - assert!( - out.status.success(), - "git {:?} failed: {}", - args, - String::from_utf8_lossy(&out.stderr) - ); - String::from_utf8_lossy(&out.stdout).trim().to_string() + crate::git::test_git(repo, args) } /// Build a repo with merge topology and a tag on the side branch: From 4000406a5f1a634042ffe8db9df07342a3f9f5f1 Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 09:44:32 -0800 Subject: [PATCH 2/3] git: share one object store across each git pipeline's threads A `--git` run over the v7.3..v7.4-rc1 merge window peaks at 270 GiB VmRSS, and 249 GiB of that is the kernel's pack files counted once per thread. Every commit-metadata worker, range generator and tuple worker calls `gix::discover()`, and each discovered repository has an object store of its own that maps the pack and index files itself. VmRSS counts a page once per mapping, so 316 workers reading the same packs count their pages up to 316 times. The pages are shared page cache (Pss_File never exceeded 2.8 GiB), but /usr/bin/time, monitoring and the OOM killer's badness score all see the inflated number. Each pipeline now opens the repository once, converts it with `into_sync()`, and hands every thread a `to_thread_local()` handle on that one store: - `process_commits_pipeline()`: its metadata workers, whose loop moves into `commit_metadata_worker()`. `--commits` runs through it too. - `process_git_tuples_streaming()`: its generators and tuple workers. `TupleWorkerContext` carries the store instead of the repository path. - `process_git_tree_streaming()`: its tree walker and tuple workers. `process_git_range()` drops the repository it opened to list the range before the pipelines start, so it does not hold a third store. The merge window now peaks at 91.6 GiB instead of 269.6 GiB, in 690 s against 802 s, with file-backed RSS under 2.8 GiB. Indexing only its commits peaks at 9.7 GiB instead of 269.6 GiB, in 9.1 s against 17.6 s. (One run each on a 316-CPU host with other jobs running: load average 10 to 190 for these runs and the `--commits` baseline, 90 to 230 for the upstream merge-window run.) What remains is anonymous memory, mostly freed memory that glibc malloc keeps in its per-thread arenas. Rows indexed are the same in every table but `symbol_filename` and `call_edges`, which differ between runs of the unchanged code as well. The riskiest part is the state the threads now share. Their handles keep separate caches but share one slot map, read without locks, and one write lock, which an object-store refresh takes. Lookups that still refresh, such as a short id or a missing object, now queue on each other. Full commit ids stopped refreshing in the previous commit. With that change in place, indexing only the merge window's commits takes 9.1 s here against 15.1 s at the previous commit, where every worker still has a store of its own. A path that is not a repository now fails these pipelines up front with "Not in a git repository". Before, every thread logged that it could not open the repository and the run finished without reading anything. `process_lore_commits_pipeline()` keeps a store per worker; it was not measured here. Three tests hand the threads a store over a packed repository and check that the store's pack was opened and the right objects were read: - `commit_metadata_workers_read_through_the_store_they_are_handed`: the metadata workers. - `generators_and_workers_read_through_the_store_they_are_handed`: a range generator and four tuple workers. - `tree_walker_reads_through_the_store_it_is_handed`: the tree walker. --- src/git_range.rs | 140 +++++++++++++++++++++++++++++++------ src/indexer.rs | 178 ++++++++++++++++++++++++++++++++--------------- 2 files changed, 239 insertions(+), 79 deletions(-) diff --git a/src/git_range.rs b/src/git_range.rs index 8612688..eeeb847 100644 --- a/src/git_range.rs +++ b/src/git_range.rs @@ -75,7 +75,9 @@ struct TupleWorkerContext { worker_id: usize, shared_tuple_rx: Arc>>, result_tx: mpsc::SyncSender, - repo_path: PathBuf, + /// The pipeline's one object store, shared with its generators and + /// every other worker (see `stream_git_file_tuples_batch`). + repo: gix::ThreadSafeRepository, source_root: PathBuf, no_macros: bool, processed_count: Arc, @@ -223,10 +225,15 @@ fn load_git_file_to_temp( } /// Stream git file tuples to a channel from a subset of commits (producer) +/// +/// `repo` is the pipeline's one object store, shared by every generator and +/// worker. Each store maps the pack files on its own, and `VmRSS` counts a +/// mapped page once per mapping, so a store per thread counts the +/// repository's packs once per thread. #[allow(clippy::too_many_arguments)] fn stream_git_file_tuples_batch( generator_id: usize, - repo_path: PathBuf, + repo: gix::ThreadSafeRepository, commit_batch: Vec, extensions: Vec, tuple_tx: mpsc::Sender, @@ -234,8 +241,7 @@ fn stream_git_file_tuples_batch( sent_in_this_run: Arc>, scope_prefixes: Vec, ) -> Result<()> { - // Open repository ONCE per generator thread and reuse for all commits - let repo = gix::discover(&repo_path)?; + let repo = repo.to_thread_local(); let mut total_files = 0; let mut sent_files = 0; @@ -405,14 +411,7 @@ fn process_git_file_tuple_with_repo( /// Worker function that processes tuples from a shared channel and sends batches to inserters fn tuple_worker_shared(ctx: TupleWorkerContext) { - // Open repository ONCE per worker and convert to thread-safe version for sharing across rayon threads - let thread_safe_repo = match gix::discover(&ctx.repo_path) { - Ok(repo) => repo.into_sync(), - Err(e) => { - tracing::error!("Worker {} failed to open repository: {}", ctx.worker_id, e); - return; - } - }; + let thread_safe_repo = &ctx.repo; // Files per insert batch. Each batch holds full function bodies for // every file in its window (about 94 MB of bodies per 512-file batch), @@ -826,10 +825,13 @@ async fn process_git_tuples_streaming(config: StreamingConfig) -> Result Result Result, tuple_tx: mpsc::Sender, processed_files: Arc>, scope_prefixes: Vec, ) -> Result { - use crate::git::walk_tree_at_commit; - let mut sent_files = 0; let mut filtered_already_processed = 0; // Walk the tree at the commit and send tuples for matching files - walk_tree_at_commit(&repo_path, &commit_sha, |relative_path, object_id| { + let repo = repo.to_thread_local(); + walk_tree_at_commit_with_repo(&repo, &commit_sha, |relative_path, object_id| { // Skip files outside the requested scope (source subdir, --include dirs) if !path_in_scope(relative_path, &scope_prefixes) { return Ok(()); @@ -1049,6 +1050,12 @@ fn stream_tree_file_tuples( async fn process_git_tree_streaming(config: TreeStreamingConfig) -> Result { use std::sync::Mutex; + // One object store for the generator and every worker of the pipeline + // (see `stream_git_file_tuples_batch`). + let repo = gix::discover(&config.repo_path) + .map_err(|e| anyhow::anyhow!("Not in a git repository: {}", e))? + .into_sync(); + // Create progress bar for file processing let pb = ProgressBar::new_spinner(); pb.set_style( @@ -1105,7 +1112,7 @@ async fn process_git_tree_streaming(config: TreeStreamingConfig) -> Result Result Result = (0..4) + .map(|worker_id| { + let ctx = TupleWorkerContext { + worker_id, + shared_tuple_rx: shared_tuple_rx.clone(), + result_tx: result_tx.clone(), + repo: worker_repo.clone(), + source_root: dir.path().to_path_buf(), + no_macros: false, + processed_count: processed_count.clone(), + batches_sent: batches_sent.clone(), + accumulate_lock: accumulate_lock.clone(), + }; + thread::spawn(move || tuple_worker_shared(ctx)) + }) + .collect(); + drop(result_tx); + for worker in workers { + worker.join().unwrap(); + } + + // a.c at both commits and b.c: three blobs, each read once. + let files: usize = result_rx.iter().map(|batch| batch.files_processed).sum(); + assert_eq!(files, 3); + assert_eq!(processed_count.load(Ordering::Relaxed), 3); + assert_eq!(open_packs(&worker_repo), 1); + } + + /// The tree walker reads through the store it is handed, like the range + /// generators. + #[test] + fn tree_walker_reads_through_the_store_it_is_handed() { + let (dir, commits) = crate::git::packed_test_repo(); + let repo = gix::discover(dir.path()).unwrap().into_sync(); + assert_eq!(repo.objects.metrics().open_reachable_packs, 0); + + let (tuple_tx, tuple_rx) = mpsc::channel(); + let sent = stream_tree_file_tuples( + repo.clone(), + commits[1].clone(), + vec!["c".to_string()], + tuple_tx, + Arc::new(HashSet::new()), + Vec::new(), + ) + .unwrap(); + let mut paths: Vec<_> = tuple_rx.iter().map(|t| t.file_path).collect(); + paths.sort(); + assert_eq!(sent, 2); + assert_eq!(paths, [PathBuf::from("a.c"), PathBuf::from("b.c")]); + assert_eq!(repo.objects.metrics().open_reachable_packs, 1); + } } diff --git a/src/indexer.rs b/src/indexer.rs index 59dc65b..142ac1f 100644 --- a/src/indexer.rs +++ b/src/indexer.rs @@ -586,6 +586,67 @@ fn extract_symbols_from_email_body(body: &str) -> Vec { } } +/// Extract metadata for each commit SHA received from `commit_rx` until it +/// closes, sending it to `result_tx` in batches of `batch_size`. `repo` is +/// the pipeline's one object store, shared by every worker. +fn commit_metadata_worker( + worker_id: usize, + repo: gix::ThreadSafeRepository, + commit_rx: Arc>>, + result_tx: mpsc::SyncSender>, + processed: Arc, + batch_size: usize, +) { + let repo = repo.to_thread_local(); + let mut batch = Vec::new(); + + loop { + // Get next commit SHA + let commit_sha = { + match commit_rx.lock() { + Ok(rx) => rx.recv(), + Err(e) => { + tracing::error!("Worker {} failed to lock receiver: {}", worker_id, e); + break; + } + } + }; + + match commit_sha { + Ok(sha) => { + match extract_commit_metadata(&repo, &sha) { + Ok(metadata) => { + batch.push(metadata); + processed.fetch_add(1, Ordering::Relaxed); + + // Send batch when it reaches batch_size + if batch.len() >= batch_size { + if result_tx.send(batch.clone()).is_err() { + break; + } + batch.clear(); + } + } + Err(e) => { + warn!( + "Worker {} failed to extract metadata for {}: {}", + worker_id, sha, e + ); + processed.fetch_add(1, Ordering::Relaxed); + } + } + } + Err(_) => { + // Channel closed, send remaining batch + if !batch.is_empty() { + let _ = result_tx.send(batch); + } + break; + } + } + } +} + /// Process commits using a streaming pipeline pub async fn process_commits_pipeline( repo_path: &std::path::Path, @@ -599,6 +660,13 @@ pub async fn process_commits_pipeline( use indicatif::{ProgressBar, ProgressStyle}; use std::sync::Mutex; + // One object store for every worker. Each store maps the pack files on + // its own and `VmRSS` counts a mapped page once per mapping, so a store + // per worker counts the repository's packs once per worker. + let shared_repo = gix::discover(repo_path) + .map_err(|e| anyhow::anyhow!("Not in a git repository: {}", e))? + .into_sync(); + // Create progress bar let pb = ProgressBar::new(commit_shas.len() as u64); pb.set_style( @@ -640,66 +708,18 @@ pub async fn process_commits_pipeline( for worker_id in 0..num_workers { let worker_commit_rx = shared_commit_rx.clone(); let worker_result_tx = result_tx.clone(); - let worker_repo_path = repo_path.to_path_buf(); + let worker_repo = shared_repo.clone(); let worker_processed = processed_count.clone(); let handle = thread::spawn(move || { - // Open repository ONCE per worker thread, reuse for all commits - let thread_repo = match gix::discover(&worker_repo_path) { - Ok(r) => r, - Err(e) => { - tracing::error!("Worker {} failed to open repository: {}", worker_id, e); - return; - } - }; - - let mut batch = Vec::new(); - - loop { - // Get next commit SHA - let commit_sha = { - match worker_commit_rx.lock() { - Ok(rx) => rx.recv(), - Err(e) => { - tracing::error!("Worker {} failed to lock receiver: {}", worker_id, e); - break; - } - } - }; - - match commit_sha { - Ok(sha) => { - match extract_commit_metadata(&thread_repo, &sha) { - Ok(metadata) => { - batch.push(metadata); - worker_processed.fetch_add(1, Ordering::Relaxed); - - // Send batch when it reaches batch_size - if batch.len() >= batch_size { - if worker_result_tx.send(batch.clone()).is_err() { - break; - } - batch.clear(); - } - } - Err(e) => { - warn!( - "Worker {} failed to extract metadata for {}: {}", - worker_id, sha, e - ); - worker_processed.fetch_add(1, Ordering::Relaxed); - } - } - } - Err(_) => { - // Channel closed, send remaining batch - if !batch.is_empty() { - let _ = worker_result_tx.send(batch); - } - break; - } - } - } + commit_metadata_worker( + worker_id, + worker_repo, + worker_commit_rx, + worker_result_tx, + worker_processed, + batch_size, + ) }); worker_handles.push(handle); } @@ -1146,6 +1166,50 @@ mod tests { crate::git::test_git(repo, args) } + /// The metadata workers read through the store they are handed. A + /// worker that opened a store of its own would map the repository's + /// packs again, and the handed store's pack would stay unopened. + #[test] + fn commit_metadata_workers_read_through_the_store_they_are_handed() { + let (dir, commits) = crate::git::packed_test_repo(); + let repo = gix::discover(dir.path()).unwrap().into_sync(); + assert_eq!(repo.objects.metrics().open_reachable_packs, 0); + + let (commit_tx, commit_rx) = mpsc::channel(); + for commit in &commits { + commit_tx.send(commit.clone()).unwrap(); + } + drop(commit_tx); + let commit_rx = Arc::new(std::sync::Mutex::new(commit_rx)); + let (result_tx, result_rx) = mpsc::sync_channel(16); + let processed = Arc::new(AtomicUsize::new(0)); + let workers: Vec<_> = (0..4) + .map(|worker_id| { + let (repo, commit_rx, result_tx, processed) = ( + repo.clone(), + commit_rx.clone(), + result_tx.clone(), + processed.clone(), + ); + thread::spawn(move || { + commit_metadata_worker(worker_id, repo, commit_rx, result_tx, processed, 1) + }) + }) + .collect(); + drop(result_tx); + for worker in workers { + worker.join().unwrap(); + } + + let mut extracted: Vec = result_rx.iter().flatten().map(|c| c.git_sha).collect(); + extracted.sort(); + let mut expected = commits.clone(); + expected.sort(); + assert_eq!(extracted, expected); + assert_eq!(processed.load(Ordering::Relaxed), 2); + assert_eq!(repo.objects.metrics().open_reachable_packs, 1); + } + /// Build a repo with merge topology and a tag on the side branch: /// /// A -- B -- M -- D (main) From 8ad3d1f110d88eb8a548ea146df7b38c319e6865 Mon Sep 17 00:00:00 2001 From: Vlad Poenaru Date: Wed, 11 Nov 2026 10:31:10 -0800 Subject: [PATCH 3/3] git: cap tuple workers at 8 `process_git_range()` and `process_git_tree()` start one tuple worker per CPU, 316 on the measuring host. A worker that finishes a 512-file batch holds it, full function bodies included, until the depth-2 result channel takes it, so the worker count is how many parsed batches can wait ahead of the inserters. The v7.3..v7.4-rc1 merge window has 124 batches and a cold index of v7.3 about 127, both fewer than the workers, so `RESULT_CHANNEL_DEPTH` bounded nothing: in a heap profile of the merge window, parsed output waiting in workers was 11.4 GiB of a 12.9 GiB live heap. Both paths now start at most 8 tuple workers, so hosts with 8 or fewer CPUs are unaffected. Parsing parallelism does not depend on the count, because each worker parses its batch on the whole rayon pool. The merge window peaked at 83.1 GiB instead of 91.6 GiB with the previous commit alone, in 667 s against 690 s. A cold index of v7.3 with both commits peaks at 32.6 GiB against 79.3 GiB on upstream. The cap's own share of that was not isolated: an uncapped build with a shared store peaked at 33.6 GiB, within one-run noise. (One run each, load average 10 to 190 from other jobs for these runs, 90 to 230 for the upstream cold run.) Rows match as in the previous commit, except that both capped merge-window runs stored more `call_edges` rows: 3.95M and 3.90M, against 3.75M upstream and 3.81M without the cap. The per-path delete in `call_edges` (fixed in #117) removes edges of every version of a path, so how many survive depends on the order batches land in; more rows means it removed fewer versions, not that edges were lost. Most of what remains is freed memory that glibc malloc keeps in per-thread arenas. In a local build with one shared store and jemalloc as the global allocator, which this change does not add, the same cap cut the merge-window peak from 20.5 to 13.6 GiB. `tuple_workers_follow_the_cpus_up_to_the_cap` checks the count at 0, 1, 7, 8, 9 and 316 CPUs. --- src/git_range.rs | 25 ++++++++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/src/git_range.rs b/src/git_range.rs index eeeb847..97ef86b 100644 --- a/src/git_range.rs +++ b/src/git_range.rs @@ -109,6 +109,16 @@ struct StreamingConfig { /// flight at once. const RESULT_CHANNEL_DEPTH: usize = 2; +/// Most tuple workers a range or tree run starts. A worker parses its +/// batch on the whole rayon pool, so a few workers keep every CPU busy; +/// more only add finished batches waiting on the result channel. +const MAX_TUPLE_WORKERS: usize = 8; + +/// Tuple workers for a range or tree run on a host with `cpus` CPUs. +fn tuple_workers(cpus: usize) -> usize { + cpus.clamp(1, MAX_TUPLE_WORKERS) +} + /// Statistics from processing git file tuples (lightweight, no accumulated data) #[derive(Debug, Default, Clone)] struct GitTupleStats { @@ -1280,7 +1290,7 @@ pub async fn process_git_tree( } // Determine number of workers - let num_workers = num_cpus::get().max(1); + let num_workers = tuple_workers(num_cpus::get()); info!( "Starting tree streaming pipeline with {} worker threads", num_workers @@ -1497,8 +1507,7 @@ pub async fn process_git_range( } // Step 2: Determine number of workers - // Since generators are I/O bound and workers are CPU bound, use a balanced approach - let num_workers = (num_cpus::get()).max(1); + let num_workers = tuple_workers(num_cpus::get()); info!( "Starting streaming pipeline with up to 32 generator threads and {} worker threads", num_workers @@ -1918,4 +1927,14 @@ mod tests { assert_eq!(paths, [PathBuf::from("a.c"), PathBuf::from("b.c")]); assert_eq!(repo.objects.metrics().open_reachable_packs, 1); } + + #[test] + fn tuple_workers_follow_the_cpus_up_to_the_cap() { + assert_eq!(tuple_workers(0), 1); + assert_eq!(tuple_workers(1), 1); + assert_eq!(tuple_workers(7), 7); + assert_eq!(tuple_workers(MAX_TUPLE_WORKERS), 8); + assert_eq!(tuple_workers(9), 8); + assert_eq!(tuple_workers(316), 8); + } }