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/git_range.rs b/src/git_range.rs index 8612688..97ef86b 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, @@ -107,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 { @@ -223,10 +235,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 +251,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 +421,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 +835,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 +1060,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 +1122,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); + } + + #[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); + } } diff --git a/src/indexer.rs b/src/indexer.rs index 3d22039..6c4f787 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); } @@ -1211,25 +1231,51 @@ 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) + } + + /// 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: