perf(index): parallelize per-file read, hash, chunk, and tokenize (#115)

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<Chunk,number> 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.

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`로 두면 순차적으로 인덱싱합니다. 워커 수와 상관없이 만들어지는 인덱스는 같습니다.
+
 <details>
 <summary>토큰 절약량 보기</summary>
 

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.
+
 <details>
 <summary>Savings</summary>
 

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

crates/csp/benches/core.rs

@@ -2,14 +2,20 @@
 //!
 //! 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::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};
@@ -129,11 +135,64 @@ 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<u32> {{\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). 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| {
+            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);

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<Mutex<HashSet<String>>> =
     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<Mutex<()>>,
+}
+
+/// 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<Mutex<HashMap<String, Grammar>>> =
+    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<Mutex<HashMap<String, Arc<Mutex<()>>>>> =
+    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<T>(mutex: &Mutex<T>) -> 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<Grammar> {
+    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<Mutex<HashSet<String>>> =
 /// 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<Language> {
+    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<Language> {
     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<Vec<Ch
     if text.trim().is_empty() {
         return Some(Vec::new());
     }
-    let lang = language_for(language)?;
-    let mut parser = Parser::new();
-    if parser.set_language(&lang).is_err() {
-        return None;
-    }
-    let tree = parser.parse(text.as_bytes(), None)?;
+    let grammar = grammar_for(language)?;
+    let tree = {
+        // The scanner is created in `set_language`, used by `parse`, and
+        // destroyed when the parser drops, so all three stay under the lock.
+        let _guard = lock(&grammar.parse_lock);
+        let mut parser = Parser::new();
+        if parser.set_language(&grammar.language).is_err() {
+            return None;
+        }
+        parser.parse(text.as_bytes(), None)?
+    };
     let byte_boundaries = merge_node(&TsNode(tree.root_node()), desired_length);
     Some(
         byte_boundaries
@@ -534,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() {