From 39ae5900741cf19c8fd0854051cb211018a243b5 Mon Sep 17 00:00:00 2001 From: Minsu Lee Date: Wed, 14 Oct 2026 21:40:40 +0900 Subject: [PATCH 1/2] perf(index): parallelize per-file read, hash, chunk, and tokenize (#115) Port of upstream semble #295. Per-file work that touches no shared state runs on a local rayon pool in batches of 256 files; BM25 mutation and manifest assembly stay serial in walk order, so the index is identical for any worker count. CSP_INDEX_WORKERS / CreateIndexOptions::index_workers select the worker count (1 = serial on the calling thread). Resolved tree-sitter grammars are cached per language so the language pack's global lock is not hit per file, and parsing is serialized per grammar because some external scanners keep process-global state. --- Cargo.lock | 1 + Cargo.toml | 1 + README.ko.md | 2 + README.md | 2 + crates/csp/Cargo.toml | 2 + crates/csp/benches/core.rs | 58 ++++- crates/csp/src/chunking/core.rs | 86 ++++++- crates/csp/src/indexing/create.rs | 291 +++++++++++++++++++----- crates/csp/src/indexing/create/tests.rs | 138 +++++++++++ crates/csp/src/indexing/index.rs | 1 + docs/src/content/docs/cli.mdx | 2 + 11 files changed, 512 insertions(+), 72 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a17e1cd..fcf6b2d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -343,6 +343,7 @@ dependencies = [ "ignore", "indexmap", "model2vec-rs", + "rayon", "regex", "rmcp", "schemars", diff --git a/Cargo.toml b/Cargo.toml index 55665c0..fec17c7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,6 +39,7 @@ fancy-regex = "0.16" # Phase 1 — boosting definition patterns (lookbehind/lo indexmap = "2" # Phase 1 — insertion-ordered score maps (Map parity) sha2 = "0.10" # Phase 3 — content-hash cache keys (sha256, parity with node:crypto) tempfile = "3" # Phase 3/4 — temp dirs (git clone checkout) + tests +rayon = "1.12" # parallel per-file read/hash/chunk/tokenize during indexing (#115) # napi-rs native SDK (separate distribution channel — the @pleaseai/csp-sdk npm # package; the CLI/MCP launcher under npm/ stays a standalone-binary shim). Only # crates/csp-node opts in; the core/CLI crates do not depend on napi. diff --git a/README.ko.md b/README.ko.md index 419000d..1ac2fef 100644 --- a/README.ko.md +++ b/README.ko.md @@ -396,6 +396,8 @@ csp find-related src/auth.ts 42 ./my-project 인덱스를 가볍게 유지하기 위해 1 MB보다 큰 파일은 인덱싱에서 제외됩니다. 제외된 파일은 인덱싱 시점에 stderr 경고로 앞쪽 몇 개의 경로와 함께 보고됩니다. 생성된 대용량 문서나 수집 문서를 다뤄야 한다면 `CSP_MAX_FILE_BYTES` 환경 변수(바이트 단위)로 이 한도를 올리거나 내릴 수 있습니다. 값이 잘못됐거나 0 이하이면 경고 후 기본값 1 MB로 돌아갑니다. +인덱싱 중 파일 읽기, 해시 계산, 청킹, 토큰화는 병렬로 처리되며, 기본적으로 CPU 하나당 워커 하나를 씁니다. 워커 수를 바꾸려면 `CSP_INDEX_WORKERS`에 양의 정수를 지정하세요. `CSP_INDEX_WORKERS=1`로 두면 순차적으로 인덱싱합니다. 워커 수와 상관없이 만들어지는 인덱스는 같습니다. +
토큰 절약량 보기 diff --git a/README.md b/README.md index 8788e9c..4fb0a2d 100644 --- a/README.md +++ b/README.md @@ -396,6 +396,8 @@ When you run `csp search` or `csp find-related` **without** `--index`, `csp` aut Files larger than 1 MB are skipped during indexing to keep index builds lean. Skipped files are reported as a warning on stderr at index time, naming the first few paths. If you work with large generated or ingested documents, you can raise (or lower) this limit with the `CSP_MAX_FILE_BYTES` environment variable (in bytes); a malformed or non-positive value warns and falls back to the 1 MB default. +Indexing reads, hashes, chunks, and tokenizes files in parallel, one worker per CPU by default. Set `CSP_INDEX_WORKERS` to a positive integer to change the worker count; `CSP_INDEX_WORKERS=1` indexes serially. The resulting index is identical for any worker count. +
Savings diff --git a/crates/csp/Cargo.toml b/crates/csp/Cargo.toml index 64e6e97..1254870 100644 --- a/crates/csp/Cargo.toml +++ b/crates/csp/Cargo.toml @@ -30,6 +30,8 @@ indexmap = { workspace = true } ignore = { workspace = true } sha2 = { workspace = true } tempfile = { workspace = true } +# Parallel per-file indexing work (read, hash, chunk, tokenize) — semble #295. +rayon = { workspace = true } # Phase 3 — real Model2Vec dense embeddings (official MinishLab Rust port). model2vec-rs = { workspace = true } # Phase 3 — tree-sitter AST chunking. Full upstream-parity grammar coverage via diff --git a/crates/csp/benches/core.rs b/crates/csp/benches/core.rs index 8bf466e..2f9a658 100644 --- a/crates/csp/benches/core.rs +++ b/crates/csp/benches/core.rs @@ -2,14 +2,19 @@ //! //! Covers the pure, allocation-heavy functions on the indexing and ranking //! critical path: identifier tokenization (run over every indexed file) and the -//! ranking penalty / top-k reranking pipeline (run on every search query). +//! ranking penalty / top-k reranking pipeline (run on every search query), plus +//! a cold index build comparing serial and parallel per-file work. //! //! Uses the `codspeed-criterion-compat` shim (aliased as `criterion`), so the //! same file runs under vanilla Criterion locally and under CodSpeed's //! instrumentation in CI. +use std::path::Path; + use criterion::{black_box, criterion_group, criterion_main, Criterion}; +use csp::indexing::create::{create_index_from_path, CreateIndexOptions}; +use csp::indexing::dense::make_stub_model; use csp::ranking::penalties::{file_path_penalty, rerank_top_k}; use csp::ranking::Scores; use csp::tokens::{split_identifier, tokenize}; @@ -129,11 +134,62 @@ fn bench_rerank_top_k(c: &mut Criterion) { group.finish(); } +/// Write a generated ~300-file polyglot tree (TypeScript, Python, Rust, Go), +/// each file a handful of functions so chunking and tokenizing dominate. +fn write_index_tree(root: &Path) { + for i in 0..300 { + let (dir, ext) = [("web", "ts"), ("svc", "py"), ("core", "rs"), ("cmd", "go")][i % 4]; + let body: String = (0..12) + .map(|f| match ext { + "ts" => format!( + "export function handleRequest{i}_{f}(userId: string, retryCount: number) {{\n const requestQueue = buildQueue(userId)\n return requestQueue.map((item) => item.value + retryCount)\n}}\n\n" + ), + "py" => format!( + "def compute_value_{i}_{f}(user_id, retry_count):\n request_queue = build_queue(user_id)\n return [item.value + retry_count for item in request_queue]\n\n\n" + ), + "rs" => format!( + "pub fn parse_item_{i}_{f}(user_id: u32, retry_count: u32) -> Vec {{\n let request_queue = build_queue(user_id);\n request_queue.iter().map(|item| item + retry_count).collect()\n}}\n\n" + ), + _ => format!( + "func RenderView{i}_{f}(userID int, retryCount int) []int {{\n\tqueue := buildQueue(userID)\n\tfor i := range queue {{\n\t\tqueue[i] += retryCount\n\t}}\n\treturn queue\n}}\n\n" + ), + }) + .collect(); + let header = if ext == "go" { "package main\n\n" } else { "" }; + let path = root.join(dir).join(format!("mod_{i}.{ext}")); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(path, format!("{header}{body}")).unwrap(); + } +} + +fn bench_index_build(c: &mut Criterion) { + let dir = tempfile::tempdir().unwrap(); + write_index_tree(dir.path()); + let model = make_stub_model(16); + + let mut group = c.benchmark_group("index_build"); + group.sample_size(10); + // `0` is rayon's default (one worker per CPU), pinned rather than left to + // `CSP_INDEX_WORKERS` so the comparison is stable. + for (name, workers) in [("300_files_serial", 1), ("300_files_parallel", 0)] { + let options = CreateIndexOptions { + display_root: Some(dir.path().to_path_buf()), + index_workers: Some(workers), + ..CreateIndexOptions::new(&model) + }; + group.bench_function(name, |b| { + b.iter(|| black_box(create_index_from_path(dir.path(), &options, None).unwrap())) + }); + } + group.finish(); +} + criterion_group!( benches, bench_tokenize, bench_split_identifier, bench_file_path_penalty, bench_rerank_top_k, + bench_index_build, ); criterion_main!(benches); diff --git a/crates/csp/src/chunking/core.rs b/crates/csp/src/chunking/core.rs index 267e6e0..378d932 100644 --- a/crates/csp/src/chunking/core.rs +++ b/crates/csp/src/chunking/core.rs @@ -11,8 +11,8 @@ //! back to [`chunk_lines`], exactly the upstream behavior when the language pack //! has no parser for the language. -use std::collections::HashSet; -use std::sync::{LazyLock, Mutex}; +use std::collections::{HashMap, HashSet}; +use std::sync::{Arc, LazyLock, Mutex, MutexGuard}; use tree_sitter::{Language, Parser}; @@ -24,6 +24,62 @@ pub const MIN_CHUNK_SIZE: usize = 50; static WARNED_LANGUAGES: LazyLock>> = LazyLock::new(|| Mutex::new(HashSet::new())); +/// A resolved grammar plus the lock that serializes parsing with it. +#[derive(Clone)] +struct Grammar { + language: Language, + /// Some grammars' external scanners keep process-global mutable state (the + /// language pack serializes all parsing for that reason; upstream sidesteps + /// it with worker processes). One lock per grammar keeps a scanner from + /// ever running concurrently with itself while different languages still + /// parse in parallel. Shared by every alias that resolves to the grammar. + parse_lock: Arc>, +} + +/// Grammars resolved so far, keyed by the requested language name. +/// [`tree_sitter_language_pack::get_language`] takes a global lock on every +/// call, so caching keeps parallel indexing from contending on it per file. +/// Failures are not cached, so a transient download error is retried on the +/// next call, as before. +static GRAMMARS: LazyLock>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +/// Parse locks keyed by the grammar's own name, so aliases of one grammar +/// (e.g. `shell` and `bash`) share a lock. +static PARSE_LOCKS: LazyLock>>>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +/// Lock `mutex`, recovering the data from a poisoned lock: every value guarded +/// here stays consistent even if a holder panicked. +fn lock(mutex: &Mutex) -> MutexGuard<'_, T> { + mutex + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} + +/// The cached [`Grammar`] for `language`, resolving it on first use. The cache +/// lock is not held across the resolve (which may download a parser), so a slow +/// fetch for one language does not stall the others; two threads racing on the +/// same language both resolve it and the first insert wins. +fn grammar_for(language: &str) -> Option { + if let Some(cached) = lock(&GRAMMARS).get(language) { + return Some(cached.clone()); + } + let resolved = resolve_language(language)?; + let lock_key = resolved.name().unwrap_or(language).to_string(); + let parse_lock = lock(&PARSE_LOCKS).entry(lock_key).or_default().clone(); + let grammar = Grammar { + language: resolved, + parse_lock, + }; + Some( + lock(&GRAMMARS) + .entry(language.to_string()) + .or_insert(grammar) + .clone(), + ) +} + /// Resolve a semble language name (the values in /// [`crate::indexing::files`]'s `EXTENSION_TO_LANGUAGE`) to a tree-sitter grammar /// from the language pack, or `None` when the pack has no grammar for it. @@ -34,8 +90,15 @@ static WARNED_LANGUAGES: LazyLock>> = /// language degrades to `None` → line chunking, with a one-time stderr warning /// per language so the cause is visible without spamming a polyglot index. Use /// [`is_supported_language`] (a metadata-only check) when you only need to know -/// whether a grammar *exists* without triggering a download. +/// whether a grammar *exists* without triggering a download. A resolved +/// grammar is cached per language for the rest of the process. pub fn language_for(language: &str) -> Option { + grammar_for(language).map(|grammar| grammar.language) +} + +/// Uncached [`language_for`]: one language-pack lookup, warning once per +/// language on failure. +fn resolve_language(language: &str) -> Option { match tree_sitter_language_pack::get_language(language) { Ok(lang) => Some(lang), Err(err) => { @@ -275,12 +338,17 @@ pub fn chunk(text: &str, language: &str, desired_length: usize) -> Option { pub model: &'a Model, @@ -44,6 +61,11 @@ pub struct CreateIndexOptions<'a> { /// warning. `None` resolves `CSP_MAX_FILE_BYTES` (default 1 MB) via /// [`get_max_file_bytes`]. pub max_file_bytes: Option, + /// Worker threads for the per-file read / hash / chunk / tokenize phase. + /// `Some(1)` runs serially on the calling thread; `Some(0)` uses rayon's + /// default. `None` resolves `CSP_INDEX_WORKERS` via [`get_index_workers`]. + /// The output is identical for every worker count. + pub index_workers: Option, } impl<'a> CreateIndexOptions<'a> { @@ -60,10 +82,40 @@ impl<'a> CreateIndexOptions<'a> { content: None, display_root: None, max_file_bytes: None, + index_workers: None, } } } +/// Parse a raw [`INDEX_WORKERS_ENV`] value: a positive integer, else `None` +/// (unset, malformed, and zero all fall back to rayon's default). +fn parse_index_workers(raw: Option<&str>) -> Option { + raw?.trim().parse::().ok().filter(|&n| n > 0) +} + +/// The indexing worker count from [`INDEX_WORKERS_ENV`], read once per process, +/// or `None` for rayon's default. +pub fn get_index_workers() -> Option { + static INDEX_WORKERS: OnceLock> = OnceLock::new(); + *INDEX_WORKERS.get_or_init(|| { + let raw = std::env::var_os(INDEX_WORKERS_ENV); + parse_index_workers(raw.as_deref().map(|s| s.to_string_lossy()).as_deref()) + }) +} + +/// The pool for the parallel phase, or `None` to run it serially on the +/// calling thread (one worker, or a pool that failed to start). A local pool +/// leaves the global rayon configuration to the embedding application. +fn index_pool(workers: Option) -> Option { + if workers == Some(1) { + return None; + } + rayon::ThreadPoolBuilder::new() + .num_threads(workers.unwrap_or(0)) + .build() + .ok() +} + /// Result of [`create_index_from_path`]. #[derive(Debug)] pub struct CreateIndexResult { @@ -120,11 +172,11 @@ pub(crate) fn skipped_large_warning( } /// Replace a file's BM25 postings: remove its old slots (if any), then add its -/// new ones. +/// new ones from each chunk's precomputed BM25 tokens. fn reindex_file( bm25_index: &mut Bm25Index, indexed_path: &str, - file_chunks: &[Chunk], + chunk_tokens: &[Vec], previous_entry: Option<&FileManifestEntry>, ) -> Result<(), String> { if let Some(entry) = previous_entry { @@ -132,15 +184,100 @@ fn reindex_file( bm25_index.remove_document(&make_chunk_id(indexed_path, slot)); } } - for (slot, chunk) in file_chunks.iter().enumerate() { - bm25_index.add_document( - &make_chunk_id(indexed_path, slot), - &tokenize(&enrich_for_bm25(chunk)), - )?; + for (slot, tokens) in chunk_tokens.iter().enumerate() { + bm25_index.add_document(&make_chunk_id(indexed_path, slot), tokens)?; } Ok(()) } +/// A file's chunks plus each chunk's BM25 tokens. +struct FreshChunks { + chunks: Vec, + tokens: Vec>, +} + +/// Chunk a file and tokenize every chunk for BM25. Lossy UTF-8 decode (invalid +/// bytes → U+FFFD): only an IO error skips a file, never an encoding error. +fn chunk_and_tokenize(bytes: &[u8], indexed_path: &str, language: Option<&str>) -> FreshChunks { + let source = String::from_utf8_lossy(bytes); + let chunks = chunk_source(&source, indexed_path, language); + let tokens = chunks + .iter() + .map(|chunk| tokenize(&enrich_for_bm25(chunk))) + .collect(); + FreshChunks { chunks, tokens } +} + +/// The outcome of the parallel per-file phase. +enum Prepared { + /// Metadata or read failed: skipped silently. + Unreadable, + /// Over `max_file_bytes`; carries the display path for the warning. + TooLarge(String), + Ready(PreparedFile), +} + +struct PreparedFile { + indexed_path: String, + language: Option<&'static str>, + hash: String, + payload: Payload, +} + +enum Payload { + /// The hash matches the previous index and its manifest range is in + /// bounds, so its rows will be reused. The bytes are dropped; the serial + /// phase re-reads the file in the unexpected case that reuse still fails. + Unchanged, + /// New or changed: already chunked and tokenized. + Fresh(FreshChunks), +} + +/// What [`prepare_file`] needs to predict reuse: the previous manifest and how +/// many chunk + vector rows it can still hand out. +struct PreviousLayout<'a> { + files: &'a FileManifest, + rows: usize, +} + +/// Read, hash, and — unless the previous index has the same content — chunk +/// and tokenize one file. Touches no shared state, so it runs on any thread. +fn prepare_file( + file_path: &Path, + display_root: Option<&Path>, + max_file_bytes: u64, + previous: &PreviousLayout, +) -> Prepared { + let size = match std::fs::metadata(file_path) { + Ok(meta) => meta.len(), + Err(_) => return Prepared::Unreadable, + }; + if size > max_file_bytes { + return Prepared::TooLarge(display_path(file_path, display_root)); + } + let Ok(bytes) = std::fs::read(file_path) else { + return Prepared::Unreadable; + }; + let hash = sha256_hex(&bytes); + let indexed_path = display_path(file_path, display_root); + let language = detect_language(&file_path.to_string_lossy()); + let unchanged = previous + .files + .get(&indexed_path) + .is_some_and(|entry| entry.hash == hash && entry.end() <= previous.rows); + let payload = if unchanged { + Payload::Unchanged + } else { + Payload::Fresh(chunk_and_tokenize(&bytes, &indexed_path, language)) + }; + Prepared::Ready(PreparedFile { + indexed_path, + language, + hash, + payload, + }) +} + /// Split a previous index into the parts the rebuild consumes: its BM25 index /// is mutated in place, and its chunk / vector rows are wrapped in `Option` so /// unchanged files can move them out without copying. @@ -258,64 +395,94 @@ pub fn create_index_from_path( let mut files = FileManifest::new(); let mut skipped_large: Vec = Vec::new(); - for file_path in walk_files(path, &ext_refs, &[]) { - let language = detect_language(&file_path.to_string_lossy()); - let size = match std::fs::metadata(&file_path) { - Ok(meta) => meta.len(), - Err(_) => continue, - }; - if size > max_file_bytes { - skipped_large.push(display_path(&file_path, options.display_root.as_deref())); - continue; - } - let Ok(bytes) = std::fs::read(&file_path) else { - continue; + let workers = options.index_workers.or_else(get_index_workers); + let pool = index_pool(workers); + let display_root = options.display_root.as_deref(); + let previous_layout = PreviousLayout { + files: &previous_files, + rows: previous_chunks.len().min(previous_vectors.len()), + }; + let prepare = |file_path: &PathBuf| { + prepare_file(file_path, display_root, max_file_bytes, &previous_layout) + }; + + for batch in walk_files(path, &ext_refs, &[]).chunks(INDEX_BATCH_SIZE) { + // Phase 1 (parallel, order-preserving): read, hash, chunk, tokenize. + let prepared: Vec = match &pool { + Some(pool) => pool.install(|| batch.par_iter().map(prepare).collect()), + None => batch.iter().map(prepare).collect(), }; - let hash = sha256_hex(&bytes); - let indexed_path = display_path(&file_path, options.display_root.as_deref()); - // `to_string_lossy` is not injective: on Unix, file names that differ - // only in invalid UTF-8 bytes collapse to the same display path, and the - // BM25 chunk ids derived from it would then collide and abort the whole - // build. Keep the first such file and skip the rest, as a valid UTF-8 - // tree can never hit this. - if files.contains_key(&indexed_path) { - eprintln!( - "csp: skipping {}: its display path collides with an already indexed file \ - (non-UTF-8 file name)", - escape_control(&file_path.display().to_string()) - ); - continue; - } - let previous_entry = previous_files.get(&indexed_path); - - let reused = take_previous_rows( - previous_entry, - &hash, - &mut previous_chunks, - &mut previous_vectors, - ); - let start = chunks.len(); - let file_chunks = match reused { - Some((file_chunks, file_vectors)) => { - vectors.extend(file_vectors.into_iter().map(Some)); - file_chunks - } - None => { - // Lossy UTF-8 decode (invalid bytes → U+FFFD): only an IO error - // skips a file, never an encoding error. - let source = String::from_utf8_lossy(&bytes).into_owned(); - let file_chunks = chunk_source(&source, &indexed_path, language); - reindex_file(&mut bm25_index, &indexed_path, &file_chunks, previous_entry)?; - fresh_rows.extend(start..start + file_chunks.len()); - vectors.extend(std::iter::repeat_n(None, file_chunks.len())); - file_chunks + + // Phase 2 (serial, walk order): everything that mutates the indexes. + for (file_path, item) in batch.iter().zip(prepared) { + let file = match item { + Prepared::Unreadable => continue, + Prepared::TooLarge(display) => { + skipped_large.push(display); + continue; + } + Prepared::Ready(file) => file, + }; + let PreparedFile { + indexed_path, + language, + hash, + payload, + } = file; + // `to_string_lossy` is not injective: on Unix, file names that + // differ only in invalid UTF-8 bytes collapse to the same display + // path, and the BM25 chunk ids derived from it would then collide + // and abort the whole build. Keep the first such file and skip the + // rest, as a valid UTF-8 tree can never hit this. + if files.contains_key(&indexed_path) { + eprintln!( + "csp: skipping {}: its display path collides with an already indexed file \ + (non-UTF-8 file name)", + escape_control(&file_path.display().to_string()) + ); + continue; } - }; + let previous_entry = previous_files.get(&indexed_path); - let count = file_chunks.len(); - chunk_ids.extend((0..count).map(|slot| make_chunk_id(&indexed_path, slot))); - chunks.extend(file_chunks); - files.insert(indexed_path, FileManifestEntry { hash, start, count }); + let reused = take_previous_rows( + previous_entry, + &hash, + &mut previous_chunks, + &mut previous_vectors, + ); + let start = chunks.len(); + let file_chunks = match reused { + Some((file_chunks, file_vectors)) => { + vectors.extend(file_vectors.into_iter().map(Some)); + file_chunks + } + None => { + let fresh = match payload { + Payload::Fresh(fresh) => fresh, + // Reuse was predicted but the rows were gone: chunk + // the file here after all, skipping it if it vanished. + Payload::Unchanged => match std::fs::read(file_path) { + Ok(bytes) => chunk_and_tokenize(&bytes, &indexed_path, language), + Err(_) => continue, + }, + }; + reindex_file( + &mut bm25_index, + &indexed_path, + &fresh.tokens, + previous_entry, + )?; + fresh_rows.extend(start..start + fresh.chunks.len()); + vectors.extend(std::iter::repeat_n(None, fresh.chunks.len())); + fresh.chunks + } + }; + + let count = file_chunks.len(); + chunk_ids.extend((0..count).map(|slot| make_chunk_id(&indexed_path, slot))); + chunks.extend(file_chunks); + files.insert(indexed_path, FileManifestEntry { hash, start, count }); + } } // Files that vanished since the previous index: drop their postings. diff --git a/crates/csp/src/indexing/create/tests.rs b/crates/csp/src/indexing/create/tests.rs index c4f9e60..d52e59c 100644 --- a/crates/csp/src/indexing/create/tests.rs +++ b/crates/csp/src/indexing/create/tests.rs @@ -52,6 +52,7 @@ fn respects_extensions_override() { content: Some(vec![ContentType::Docs]), display_root: Some(dir.path().to_path_buf()), max_file_bytes: None, + index_workers: None, }; let result = create_index_from_path(dir.path(), &options, None).unwrap(); assert_eq!(result.chunks.len(), 1); @@ -335,3 +336,140 @@ fn colliding_lossy_paths_skip_the_duplicate_instead_of_aborting() { assert_eq!(result.bm25_index.corpus_size(), result.chunks.len()); into_previous(result); } + +// --- parallel indexing (upstream #295): output is independent of workers --- + +/// A polyglot tree with more files than one parallel batch, so batch +/// boundaries are exercised too. +fn write_polyglot_tree(root: &Path) { + for i in 0..(INDEX_BATCH_SIZE + 44) { + let (rel, content) = match i % 4 { + 0 => ( + format!("ts/mod_{i}.ts"), + format!( + "export function handler{i}(input: string) {{\n return input + '{i}'\n}}\n" + ), + ), + 1 => ( + format!("py/mod_{i}.py"), + format!("def compute_value_{i}(x):\n return x * {i}\n"), + ), + 2 => ( + format!("rs/mod_{i}.rs"), + format!("pub fn parse_item_{i}(v: u32) -> u32 {{\n v + {i}\n}}\n"), + ), + _ => ( + format!("go/mod_{i}.go"), + format!("package main\n\nfunc RenderView{i}() int {{\n\treturn {i}\n}}\n"), + ), + }; + write_files(root, &[(rel.as_str(), content.as_str())]); + } +} + +fn build_with_workers( + root: &Path, + model: &Model, + workers: usize, + previous: Option, +) -> CreateIndexResult { + let options = CreateIndexOptions { + index_workers: Some(workers), + ..opts(model, Some(root.to_path_buf())) + }; + create_index_from_path(root, &options, previous).unwrap() +} + +/// Assert two builds are identical: chunks (order + content), vectors, +/// manifest, BM25 doc order, the persisted `bm25.json`, and query scores. +fn assert_same_index(serial: &CreateIndexResult, parallel: &CreateIndexResult) { + assert_eq!(serial.chunks, parallel.chunks); + assert_eq!( + serial.semantic_index.vectors, + parallel.semantic_index.vectors + ); + assert_eq!(serial.files, parallel.files); + assert_eq!( + serial.bm25_index.doc_order(), + parallel.bm25_index.doc_order() + ); + let (a, b) = (tempdir().unwrap(), tempdir().unwrap()); + serial.bm25_index.save(a.path()).unwrap(); + parallel.bm25_index.save(b.path()).unwrap(); + assert_eq!( + std::fs::read(a.path().join("bm25.json")).unwrap(), + std::fs::read(b.path().join("bm25.json")).unwrap() + ); + for query in [ + "handler7 input", + "compute_value", + "parse_item u32", + "RenderView", + ] { + assert_eq!( + serial.bm25_index.get_scores(&tokenize(query), None), + parallel.bm25_index.get_scores(&tokenize(query), None), + "{query}" + ); + } +} + +#[test] +fn parallel_cold_build_matches_serial() { + let dir = tempdir().unwrap(); + let root = dir.path().to_path_buf(); + write_polyglot_tree(&root); + let model = make_stub_model(4); + + let serial = build_with_workers(&root, &model, 1, None); + let parallel = build_with_workers(&root, &model, 4, None); + assert_eq!(serial.files.len(), INDEX_BATCH_SIZE + 44); + assert_same_index(&serial, ¶llel); +} + +#[test] +fn parallel_incremental_rebuild_matches_serial() { + let dir = tempdir().unwrap(); + let root = dir.path().to_path_buf(); + write_polyglot_tree(&root); + let model = make_stub_model(4); + let seed = build_with_workers(&root, &model, 1, None); + let seed_previous = || { + PreviousIndex::try_new( + seed.chunks.clone(), + seed.semantic_index.vectors.clone(), + seed.files.clone(), + seed.bm25_index.clone(), + ) + .unwrap() + }; + + write_files( + &root, + &[ + ("py/mod_5.py", "def compute_value_5(x):\n return x - 5\n"), + ("rs/added.rs", "pub fn freshly_added() -> u8 {\n 1\n}\n"), + ], + ); + std::fs::remove_file(root.join("go/mod_7.go")).unwrap(); + + let serial = build_with_workers(&root, &model, 1, Some(seed_previous())); + let parallel = build_with_workers(&root, &model, 4, Some(seed_previous())); + assert!(serial + .files + .contains_key(Path::new("rs").join("added.rs").to_str().unwrap())); + assert!(!serial + .files + .contains_key(Path::new("go").join("mod_7.go").to_str().unwrap())); + assert_same_index(&serial, ¶llel); +} + +#[test] +fn parse_index_workers_accepts_only_positive_integers() { + assert_eq!(parse_index_workers(None), None); + assert_eq!(parse_index_workers(Some(" 3 ")), Some(3)); + assert_eq!(parse_index_workers(Some("1")), Some(1)); + assert_eq!(parse_index_workers(Some("0")), None); + assert_eq!(parse_index_workers(Some("-2")), None); + assert_eq!(parse_index_workers(Some("many")), None); +} diff --git a/crates/csp/src/indexing/index.rs b/crates/csp/src/indexing/index.rs index 61d1d28..4adee3f 100644 --- a/crates/csp/src/indexing/index.rs +++ b/crates/csp/src/indexing/index.rs @@ -149,6 +149,7 @@ impl CspIndex { content: Some(content.clone()), display_root: Some(path.to_path_buf()), max_file_bytes: None, + index_workers: None, }, previous, )?; diff --git a/docs/src/content/docs/cli.mdx b/docs/src/content/docs/cli.mdx index b7102e0..6368744 100644 --- a/docs/src/content/docs/cli.mdx +++ b/docs/src/content/docs/cli.mdx @@ -54,6 +54,8 @@ Passing `--index ` uses that exact path instead and bypasses the auto-cach Files larger than 1 MB are skipped during indexing to keep index builds lean. Skipped files are reported as a warning on stderr at index time, naming the first few paths. If you work with large generated or ingested documents, you can raise (or lower) this limit with the `CSP_MAX_FILE_BYTES` environment variable (in bytes); a malformed or non-positive value warns and falls back to the 1 MB default. +Indexing reads, hashes, chunks, and tokenizes files in parallel, one worker per CPU by default. Set `CSP_INDEX_WORKERS` to a positive integer to change the worker count; `CSP_INDEX_WORKERS=1` indexes serially. The resulting index is identical for any worker count. + ### Ignore rules The file walker respects `.gitignore` and a repo-local `.cspignore`. Directories such as `.git`, `node_modules`, `dist`, `build`, `.next`, and `.csp` are ignored by default. From 6bc0b3dfe4dd44cc167a2ec7a7070198186d03db Mon Sep 17 00:00:00 2001 From: Minsu Lee Date: Thu, 15 Oct 2026 01:34:06 +0900 Subject: [PATCH 2/2] test(index): pin the bench size limit and cover concurrent first grammar use --- crates/csp/benches/core.rs | 7 +++++-- crates/csp/src/chunking/core.rs | 28 ++++++++++++++++++++++++++++ 2 files changed, 33 insertions(+), 2 deletions(-) diff --git a/crates/csp/benches/core.rs b/crates/csp/benches/core.rs index 2f9a658..dfd2eb0 100644 --- a/crates/csp/benches/core.rs +++ b/crates/csp/benches/core.rs @@ -15,6 +15,7 @@ use criterion::{black_box, criterion_group, criterion_main, Criterion}; use csp::indexing::create::{create_index_from_path, CreateIndexOptions}; use csp::indexing::dense::make_stub_model; +use csp::indexing::files::DEFAULT_MAX_FILE_BYTES; use csp::ranking::penalties::{file_path_penalty, rerank_top_k}; use csp::ranking::Scores; use csp::tokens::{split_identifier, tokenize}; @@ -169,12 +170,14 @@ fn bench_index_build(c: &mut Criterion) { let mut group = c.benchmark_group("index_build"); group.sample_size(10); - // `0` is rayon's default (one worker per CPU), pinned rather than left to - // `CSP_INDEX_WORKERS` so the comparison is stable. + // `0` is rayon's default (one worker per CPU). Workers and the size limit + // are pinned rather than left to `CSP_INDEX_WORKERS` / `CSP_MAX_FILE_BYTES` + // so the comparison is stable. for (name, workers) in [("300_files_serial", 1), ("300_files_parallel", 0)] { let options = CreateIndexOptions { display_root: Some(dir.path().to_path_buf()), index_workers: Some(workers), + max_file_bytes: Some(DEFAULT_MAX_FILE_BYTES), ..CreateIndexOptions::new(&model) }; group.bench_function(name, |b| { diff --git a/crates/csp/src/chunking/core.rs b/crates/csp/src/chunking/core.rs index 378d932..67636a1 100644 --- a/crates/csp/src/chunking/core.rs +++ b/crates/csp/src/chunking/core.rs @@ -602,6 +602,34 @@ mod tests { ); } + #[test] + #[ignore = "downloads a tree-sitter grammar from GitHub releases"] + fn concurrent_first_use_parses_with_the_grammar() { + // Many threads hit an uncached grammar at the same moment. Go is resolved + // by no other test here, so the process-wide cache starts cold for it. + // Every thread must get the same AST split, which line chunking would + // not produce for this source. + let src = "func alpha() {\n\tx := 1\n\tprintln(x)\n}\n\nfunc beta() {\n\ty := 2\n\tprintln(y)\n}\n"; + let start = std::sync::Barrier::new(8); + let results: Vec<_> = std::thread::scope(|scope| { + let handles: Vec<_> = (0..8) + .map(|_| { + scope.spawn(|| { + start.wait(); + chunk(src, "go", 40) + }) + }) + .collect(); + handles.into_iter().map(|h| h.join().unwrap()).collect() + }); + let first = results[0].clone().expect("go is supported → Some"); + assert!(first.len() >= 2, "expected an AST split: {first:?}"); + assert_ne!(first, chunk_lines(src, 40), "fell back to line chunking"); + for r in &results { + assert_eq!(r.as_ref(), Some(&first)); + } + } + #[test] #[ignore = "downloads a tree-sitter grammar from GitHub releases"] fn chunk_byte_to_char_handles_multibyte() {