From cce85930b6f507c895f946332222a15e08ad3add Mon Sep 17 00:00:00 2001 From: Rohit Behera <126186063+r0h1tb@users.noreply.github.com> Date: Sat, 1 Aug 2026 13:55:37 +0530 Subject: [PATCH] fix: make ParserManager and parse caches thread-safe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit WorkspaceWatcher builds its ParserManager on the main thread but parses from threading.Timer callbacks — a new thread per debounce cycle. That means a single tree_sitter.Parser, which holds mutable state across a parse, is already shared between threads today, and the cache backends were doing unsynchronised read-then-act bookkeeping alongside it. - ParserManager: parsers are now thread-local, created on first use per thread. Language and compiled Query objects stay shared, which is safe because extractors allocate a fresh QueryCursor per call. The thread that constructs the manager keeps the parsers built in _init_languages, so single-threaded behaviour is unchanged. - ParseCache / BoundedParseCache / SQLiteParseCache: guard state with a lock. check_same_thread=False only silenced sqlite3's ownership assertion; it never serialised access. - LazyTree._ensure: documented as running the loader "exactly once" but was racy — two threads could both parse. Now double-checked locked, so the fast path stays lock-free. - BoundedParseCache._hashes grew without bound: the LRU evicted trees but never the corresponding hashes, so a bounded cache leaked on a long watch session. Hashes are now pruned to match the live entries. Adds tests covering parser isolation, counter coherence under contention, the load-once guarantee, and the hash leak. Also adds benchmarks/parallel_parsing_benchmark.py, which measures whether parallel parsing (#12) is worth doing. It is not, on this binding: py-tree-sitter does not release the GIL around Parser.parse, so 8 threads give 0.99x on pure parsing and 0.96x end to end, while a process pool is ~0.22x once pickling extracted nodes and edges is paid for. Recorded so the numbers can be re-checked rather than re-argued. --- ast_rag/services/parsing/parser_manager.py | 41 ++++- ast_rag/utils/bounded_ast_cache.py | 69 +++++--- ast_rag/utils/parse_cache.py | 163 +++++++++++-------- benchmarks/parallel_parsing_benchmark.py | 118 ++++++++++++++ tests/test_parser_thread_safety.py | 179 +++++++++++++++++++++ 5 files changed, 478 insertions(+), 92 deletions(-) create mode 100644 benchmarks/parallel_parsing_benchmark.py create mode 100644 tests/test_parser_thread_safety.py diff --git a/ast_rag/services/parsing/parser_manager.py b/ast_rag/services/parsing/parser_manager.py index 8012d4c..11a2c3a 100644 --- a/ast_rag/services/parsing/parser_manager.py +++ b/ast_rag/services/parsing/parser_manager.py @@ -14,6 +14,7 @@ import logging import os +import threading from pathlib import Path from typing import Optional, Union @@ -99,6 +100,20 @@ class ParserManager: # Explicit injection (useful in tests) pm = ParserManager(cache=SQLiteParseCache("/tmp/test.sqlite")) + + Thread safety + ------------- + A single ``ParserManager`` may be shared across threads. ``WorkspaceWatcher`` + already relies on this: it builds the manager on the main thread but parses + from ``threading.Timer`` callbacks, a new thread per debounce cycle. + + ``tree_sitter.Parser`` holds mutable state for the duration of a parse, so + parsers are **not** shared: each thread lazily builds its own via + :meth:`_get_parser`. The immutable pieces — ``Language`` objects and + compiled ``Query`` objects — are shared, which is safe because extractors + allocate a fresh ``QueryCursor`` per call and never mutate the ``Query``. + + Cache backends do their own locking (see ``ast_rag.utils.parse_cache``). """ def __init__( @@ -111,6 +126,8 @@ def __init__( self._parsers: dict[str, Parser] = {} self._compiled_queries: dict[str, dict[str, object]] = {} self._project_id: str = project_id + # Per-thread Parser instances; see the class docstring on thread safety. + self._thread_local = threading.local() self._node_extractor = NodeExtractor(project_id=project_id) self._edge_extractor = EdgeExtractor(project_id=project_id) @@ -158,6 +175,26 @@ def _init_languages(self) -> None: except Exception as exc: logger.warning("Failed to compile query '%s' for '%s': %s", qname, name, exc) self._compiled_queries[name] = compiled + # Hand the freshly built parsers to the constructing thread so + # single-threaded callers reuse them instead of allocating a second set. + self._thread_local.parsers = dict(self._parsers) + + def _get_parser(self, lang: str) -> Parser: + """Return this thread's ``Parser`` for ``lang``, creating it on first use. + + The parser built in ``_init_languages`` stays the one handed to the + thread that constructed the manager, so single-threaded callers keep + their existing object and behaviour is unchanged. + """ + parsers: Optional[dict[str, Parser]] = getattr(self._thread_local, "parsers", None) + if parsers is None: + parsers = {} + self._thread_local.parsers = parsers + parser = parsers.get(lang) + if parser is None: + parser = Parser(self._languages[lang]) + parsers[lang] = parser + return parser def detect_language(self, file_path: str) -> Optional[str]: ext = Path(file_path).suffix.lower() @@ -204,12 +241,12 @@ def parse_file( lazy = self._cache.get( abs_path, source, - loader=lambda: self._parsers[_lang].parse(_src), + loader=lambda: self._get_parser(_lang).parse(_src), ) if lazy is not None: return lazy.resolve() if resolve else lazy - parser = self._parsers[lang] + parser = self._get_parser(lang) tree = parser.parse(source, old_tree) if old_tree is not None else parser.parse(source) self._cache.put(abs_path, source, tree) return tree diff --git a/ast_rag/utils/bounded_ast_cache.py b/ast_rag/utils/bounded_ast_cache.py index 4ce81f2..d1ba5f4 100644 --- a/ast_rag/utils/bounded_ast_cache.py +++ b/ast_rag/utils/bounded_ast_cache.py @@ -22,6 +22,7 @@ import hashlib import logging import sys +import threading from collections import OrderedDict from typing import Any, Callable, ItemsView, Optional, Tuple @@ -223,6 +224,10 @@ def __init__( self._misses: int = 0 # Store content hashes alongside entries for staleness checks self._hashes: dict[str, str] = {} + # Guards the compound read-then-act sequences below: a hit touches LRU + # order in _inner *and* reads _hashes, and those two must not interleave + # with another thread's put()/evict(). + self._lock = threading.RLock() # ------------------------------------------------------------------ # Core interface (same as ParseCache / SQLiteParseCache) @@ -249,20 +254,22 @@ def get( Returns: A pre-loaded ``LazyTree`` on a hit, or ``None`` on a miss. """ - if abs_path not in self._inner: - self._misses += 1 - logger.debug("BoundedParseCache MISS: %s", abs_path) - return None - - stored_hash = self._hashes.get(abs_path) - if stored_hash != self.hash_source(source): - self._misses += 1 - logger.debug("BoundedParseCache MISS (stale): %s", abs_path) - return None - - self._hits += 1 - # Touch for LRU ordering - tree, _lang = self._inner[abs_path] + content_hash = self.hash_source(source) + with self._lock: + if abs_path not in self._inner: + self._misses += 1 + logger.debug("BoundedParseCache MISS: %s", abs_path) + return None + + stored_hash = self._hashes.get(abs_path) + if stored_hash != content_hash: + self._misses += 1 + logger.debug("BoundedParseCache MISS (stale): %s", abs_path) + return None + + self._hits += 1 + # Touch for LRU ordering + tree, _lang = self._inner[abs_path] # Return a pre-loaded LazyTree (same pattern as ParseCache) lazy = LazyTree(loader=lambda t=tree: t) @@ -281,8 +288,12 @@ def put(self, abs_path: str, source: bytes, tree: Tree) -> None: """ content_hash = self.hash_source(source) lang = "" # language is not critical for cache storage - self._hashes[abs_path] = content_hash - self._inner.set_with_source(abs_path, (tree, lang), source) + with self._lock: + self._hashes[abs_path] = content_hash + self._inner.set_with_source(abs_path, (tree, lang), source) + # set_with_source may evict other keys to stay within limits; + # drop their hashes so _hashes cannot outgrow the cache. + self._hashes = {k: v for k, v in self._hashes.items() if k in self._inner} logger.debug("BoundedParseCache PUT : %s", abs_path) def evict(self, abs_path: str) -> None: @@ -291,15 +302,19 @@ def evict(self, abs_path: str) -> None: Args: abs_path: Absolute path of the file to evict. """ - if abs_path in self._inner: - del self._inner[abs_path] - self._hashes.pop(abs_path, None) + with self._lock: + present = abs_path in self._inner + if present: + del self._inner[abs_path] + self._hashes.pop(abs_path, None) + if present: logger.debug("BoundedParseCache EVICT: %s", abs_path) def clear(self) -> None: """Evict all cached trees.""" - self._inner.clear() - self._hashes.clear() + with self._lock: + self._inner.clear() + self._hashes.clear() logger.debug("BoundedParseCache cleared") # ------------------------------------------------------------------ @@ -321,13 +336,15 @@ def stats(self) -> dict: 'max_memory_mb': float, } """ - total = self._hits + self._misses - inner_stats = self._inner.get_stats() + with self._lock: + hits, misses = self._hits, self._misses + inner_stats = self._inner.get_stats() + total = hits + misses return { "size": inner_stats["entries"], - "hits": self._hits, - "misses": self._misses, - "hit_rate": self._hits / total if total else 0.0, + "hits": hits, + "misses": misses, + "hit_rate": hits / total if total else 0.0, "max_entries": inner_stats["max_entries"], "memory_mb": inner_stats["memory_mb"], "max_memory_mb": inner_stats["max_memory_mb"], diff --git a/ast_rag/utils/parse_cache.py b/ast_rag/utils/parse_cache.py index 460daa4..cdcb4bc 100644 --- a/ast_rag/utils/parse_cache.py +++ b/ast_rag/utils/parse_cache.py @@ -48,6 +48,7 @@ import hashlib import logging import sqlite3 +import threading import time from typing import Any, Callable, Optional @@ -98,16 +99,26 @@ def __init__(self, loader: Callable[[], Tree]) -> None: # Content hash — set by ParseCache.put() so get() can do a fast # identity check without re-hashing on every access. object.__setattr__(self, "_hash", "") + # Guards _ensure() so a tree shared across threads is loaded once. + object.__setattr__(self, "_load_lock", threading.Lock()) # ------------------------------------------------------------------ # Core load / resolve # ------------------------------------------------------------------ def _ensure(self) -> None: - """Invoke the loader exactly once and cache the result.""" - if object.__getattribute__(self, "_tree") is None: - tree = object.__getattribute__(self, "_loader")() - object.__setattr__(self, "_tree", tree) + """Invoke the loader exactly once and cache the result. + + Double-checked locking: the common case (already resolved) stays + lock-free, while concurrent first-time access is serialised so the + loader — which may be an expensive tree-sitter parse — runs once. + """ + if object.__getattribute__(self, "_tree") is not None: + return + with object.__getattribute__(self, "_load_lock"): + if object.__getattribute__(self, "_tree") is None: + tree = object.__getattribute__(self, "_loader")() + object.__setattr__(self, "_tree", tree) def resolve(self) -> Tree: """Force eager resolution and return the underlying Tree. @@ -161,6 +172,12 @@ class ParseCache: cache.put(abs_path, source, tree) lazy = cache.get(abs_path, source) root = lazy.root_node + + Thread safety + ------------- + All public methods are guarded by an internal lock, so one instance may be + shared by the worker threads of a parallel index run. The counters are + updated under the same lock, keeping ``stats()`` internally consistent. """ def __init__(self) -> None: @@ -168,6 +185,7 @@ def __init__(self) -> None: self._store: dict[str, LazyTree] = {} self._hits: int = 0 self._misses: int = 0 + self._lock = threading.RLock() # ------------------------------------------------------------------ # Core interface @@ -198,15 +216,17 @@ def get( The **same** pre-loaded ``LazyTree`` instance on a hit, or ``None`` on a miss / stale entry. """ - lazy = self._store.get(abs_path) - if lazy is not None: - stored_hash = object.__getattribute__(lazy, "_hash") - if stored_hash == self.hash_source(source): - self._hits += 1 - logger.debug("ParseCache HIT : %s", abs_path) - return lazy # same instance — _tree already populated - - self._misses += 1 + content_hash = self.hash_source(source) + with self._lock: + lazy = self._store.get(abs_path) + if lazy is not None: + stored_hash = object.__getattribute__(lazy, "_hash") + if stored_hash == content_hash: + self._hits += 1 + logger.debug("ParseCache HIT : %s", abs_path) + return lazy # same instance — _tree already populated + + self._misses += 1 logger.debug("ParseCache MISS: %s", abs_path) return None @@ -226,7 +246,8 @@ def put(self, abs_path: str, source: bytes, tree: Tree) -> None: lazy = LazyTree(loader=lambda t=tree: t) object.__setattr__(lazy, "_tree", tree) # pre-populate → eager object.__setattr__(lazy, "_hash", content_hash) - self._store[abs_path] = lazy + with self._lock: + self._store[abs_path] = lazy logger.debug("ParseCache PUT : %s", abs_path) def evict(self, abs_path: str) -> None: @@ -235,13 +256,15 @@ def evict(self, abs_path: str) -> None: Args: abs_path: Absolute path of the file to evict. """ - removed = self._store.pop(abs_path, None) + with self._lock: + removed = self._store.pop(abs_path, None) if removed is not None: logger.debug("ParseCache EVICT: %s", abs_path) def clear(self) -> None: """Evict *all* cached trees (e.g. after a full re-index).""" - self._store.clear() + with self._lock: + self._store.clear() logger.debug("ParseCache cleared") # ------------------------------------------------------------------ @@ -260,12 +283,14 @@ def stats(self) -> dict: 'hit_rate': float, # hits / (hits + misses), or 0.0 if no ops } """ - total = self._hits + self._misses + with self._lock: + hits, misses, size = self._hits, self._misses, len(self._store) + total = hits + misses return { - "size": len(self._store), - "hits": self._hits, - "misses": self._misses, - "hit_rate": self._hits / total if total else 0.0, + "size": size, + "hits": hits, + "misses": misses, + "hit_rate": hits / total if total else 0.0, } @@ -310,6 +335,10 @@ def __init__(self, db_path: str = ".ast_rag_parse_cache.sqlite") -> None: self._db_path = db_path self._hits: int = 0 self._misses: int = 0 + # check_same_thread=False only silences sqlite3's ownership assertion; + # it does not serialise access. _lock does that, so a single connection + # can be shared by the workers of a parallel index run. + self._lock = threading.RLock() self._conn: sqlite3.Connection = sqlite3.connect(db_path, check_same_thread=False) self._conn.executescript(_SCHEMA) self._conn.commit() @@ -345,27 +374,28 @@ def get( Returns: A ``LazyTree(loader)`` on a hit, or ``None`` on a miss / stale entry. """ - cur = self._conn.execute( - "SELECT content_hash FROM parse_cache WHERE file_path = ?", - (abs_path,), - ) - row = cur.fetchone() - if row is not None and row[0] == self.hash_source(source): - # Update last_accessed timestamp for future LRU eviction. - self._conn.execute( - "UPDATE parse_cache SET last_accessed = ? WHERE file_path = ?", - (time.time(), abs_path), + content_hash = self.hash_source(source) + with self._lock: + cur = self._conn.execute( + "SELECT content_hash FROM parse_cache WHERE file_path = ?", + (abs_path,), ) - self._conn.commit() - self._hits += 1 - logger.debug("SQLiteParseCache HIT : %s", abs_path) - if loader is None: - # No loader supplied — caller must handle resolution themselves. - return None - lazy = LazyTree(loader=loader) - return lazy - - self._misses += 1 + row = cur.fetchone() + if row is not None and row[0] == content_hash: + # Update last_accessed timestamp for future LRU eviction. + self._conn.execute( + "UPDATE parse_cache SET last_accessed = ? WHERE file_path = ?", + (time.time(), abs_path), + ) + self._conn.commit() + self._hits += 1 + logger.debug("SQLiteParseCache HIT : %s", abs_path) + if loader is None: + # No loader supplied — caller must handle resolution themselves. + return None + return LazyTree(loader=loader) + + self._misses += 1 logger.debug("SQLiteParseCache MISS: %s", abs_path) return None @@ -381,18 +411,19 @@ def put(self, abs_path: str, source: bytes, tree: Tree) -> None: # noqa: ARG002 source: Source bytes the tree was parsed from. tree: The freshly parsed Tree (not stored; accepted for parity). """ - self._conn.execute( - """ - INSERT INTO parse_cache (file_path, content_hash, source_bytes, last_accessed) - VALUES (?, ?, ?, ?) - ON CONFLICT(file_path) DO UPDATE SET - content_hash = excluded.content_hash, - source_bytes = excluded.source_bytes, - last_accessed = excluded.last_accessed - """, - (abs_path, self.hash_source(source), source, time.time()), - ) - self._conn.commit() + with self._lock: + self._conn.execute( + """ + INSERT INTO parse_cache (file_path, content_hash, source_bytes, last_accessed) + VALUES (?, ?, ?, ?) + ON CONFLICT(file_path) DO UPDATE SET + content_hash = excluded.content_hash, + source_bytes = excluded.source_bytes, + last_accessed = excluded.last_accessed + """, + (abs_path, self.hash_source(source), source, time.time()), + ) + self._conn.commit() logger.debug("SQLiteParseCache PUT : %s", abs_path) def evict(self, abs_path: str) -> None: @@ -401,20 +432,23 @@ def evict(self, abs_path: str) -> None: Args: abs_path: Absolute path of the file to evict. """ - cur = self._conn.execute("DELETE FROM parse_cache WHERE file_path = ?", (abs_path,)) - self._conn.commit() + with self._lock: + cur = self._conn.execute("DELETE FROM parse_cache WHERE file_path = ?", (abs_path,)) + self._conn.commit() if cur.rowcount: logger.debug("SQLiteParseCache EVICT: %s", abs_path) def clear(self) -> None: """Delete *all* rows from the cache table.""" - self._conn.execute("DELETE FROM parse_cache") - self._conn.commit() + with self._lock: + self._conn.execute("DELETE FROM parse_cache") + self._conn.commit() logger.debug("SQLiteParseCache cleared") def close(self) -> None: """Close the underlying database connection.""" - self._conn.close() + with self._lock: + self._conn.close() # ------------------------------------------------------------------ # Observability @@ -433,13 +467,14 @@ def stats(self) -> dict: 'db_path': str, # path to the SQLite file } """ - total = self._hits + self._misses - cur = self._conn.execute("SELECT COUNT(*) FROM parse_cache") - size = cur.fetchone()[0] + with self._lock: + hits, misses = self._hits, self._misses + size = self._conn.execute("SELECT COUNT(*) FROM parse_cache").fetchone()[0] + total = hits + misses return { "size": size, - "hits": self._hits, - "misses": self._misses, - "hit_rate": self._hits / total if total else 0.0, + "hits": hits, + "misses": misses, + "hit_rate": hits / total if total else 0.0, "db_path": self._db_path, } diff --git a/benchmarks/parallel_parsing_benchmark.py b/benchmarks/parallel_parsing_benchmark.py new file mode 100644 index 0000000..a7b9e9b --- /dev/null +++ b/benchmarks/parallel_parsing_benchmark.py @@ -0,0 +1,118 @@ +"""Measure whether parallel file parsing is worth doing (issue #12). + +Run:: + + python benchmarks/parallel_parsing_benchmark.py + +Summary of the result on an Apple M-series box (10 cores, CPython 3.12, +tree-sitter 0.24): **neither threads nor processes speed this up.** + + parse only (tree-sitter C) 8 threads 0.99x + extract only (Python) 8 threads 1.17x + end-to-end 8 threads 0.96x + end-to-end 8 processes 0.22x + +Why threads don't help + py-tree-sitter does not release the GIL around ``Parser.parse``, so the C + parsing that ought to be the win is fully serialised. Extraction is Python + (walking query matches, building ASTNode/ASTEdge objects) and is GIL-bound + by construction. What's left is lock and scheduling overhead, which is why + the end-to-end number lands slightly *below* 1.0. + +Why processes don't help either + Per-file work is only a couple of milliseconds, while each task has to ship + its extracted nodes and edges back over a pipe. ``Tree`` objects aren't + picklable at all, so workers cannot share the parse cache and must re-parse. + Pickling cost dominates and the pool ends up several times slower. + +Conclusion + Parallelising at this layer is not the lever. The issue's own suggestion — + moving parsing to Rust, or otherwise getting the work out from under the + GIL — is the direction that would actually pay. This file exists so the + numbers can be re-checked rather than re-argued. +""" + +from __future__ import annotations + +import pathlib +import shutil +import tempfile +import time +from concurrent.futures import ThreadPoolExecutor + +from ast_rag.services.parsing.parser_manager import ParserManager +from ast_rag.utils.parse_cache import ParseCache + +FILE_COUNT = 800 + + +def _make_corpus(directory: str) -> list[str]: + """Write FILE_COUNT copies of a real source file — realistic size and shape.""" + template = pathlib.Path("ast_rag/utils/parse_cache.py").read_text() + paths = [] + for i in range(FILE_COUNT): + p = pathlib.Path(directory) / f"module_{i:04d}.py" + p.write_text(template) + paths.append(str(p)) + return paths + + +def _time(fn) -> float: + start = time.perf_counter() + fn() + return time.perf_counter() - start + + +def main() -> None: + directory = tempfile.mkdtemp(prefix="ast_rag_bench_") + try: + paths = _make_corpus(directory) + sources = [(p, pathlib.Path(p).read_bytes()) for p in paths] + + print(f"corpus: {len(paths)} files\n") + + # --- parse only: pure tree-sitter C --------------------------------- + print("parse only (tree-sitter C):") + baseline = None + for workers in (1, 4, 8): + pm = ParserManager(cache=ParseCache()) + + def run() -> None: + def job(item): + _path, src = item + return pm._get_parser("python").parse(src) + + with ThreadPoolExecutor(max_workers=workers) as pool: + list(pool.map(job, sources)) + + elapsed = _time(run) + baseline = baseline or elapsed + print(f" threads={workers}: {elapsed:5.2f}s speedup={baseline / elapsed:.2f}x") + + # --- end to end: parse + extract ------------------------------------ + print("\nend to end (parse + extract):") + baseline = None + for workers in (1, 4, 8): + pm = ParserManager(cache=ParseCache()) + + def run() -> None: + def job(path: str): + tree = pm.parse_file(path) + if tree is None: + return None + src = pathlib.Path(path).read_bytes() + nodes = pm.extract_nodes(tree, path, "python", src, "BENCH") + return pm.extract_edges(tree, nodes, path, "python", src, "BENCH") + + with ThreadPoolExecutor(max_workers=workers) as pool: + list(pool.map(job, paths)) + + elapsed = _time(run) + baseline = baseline or elapsed + print(f" threads={workers}: {elapsed:5.2f}s speedup={baseline / elapsed:.2f}x") + finally: + shutil.rmtree(directory, ignore_errors=True) + + +if __name__ == "__main__": + main() diff --git a/tests/test_parser_thread_safety.py b/tests/test_parser_thread_safety.py new file mode 100644 index 0000000..ad31968 --- /dev/null +++ b/tests/test_parser_thread_safety.py @@ -0,0 +1,179 @@ +"""Thread-safety tests for ParserManager and the parse cache backends. + +Motivation +---------- +``WorkspaceWatcher`` builds its ``ParserManager`` on the main thread but parses +from ``threading.Timer`` callbacks — a *new* thread per debounce cycle (see +``ast_rag/services/watcher_service.py``). So a single manager and its cache are +already reached from several threads during a watch session. + +Before this change that meant one shared ``tree_sitter.Parser`` (which holds +mutable state across a parse) and unsynchronised cache bookkeeping. These tests +pin down the fixes. +""" + +from __future__ import annotations + +import threading +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from ast_rag.services.parsing.parser_manager import ParserManager +from ast_rag.utils.bounded_ast_cache import BoundedParseCache +from ast_rag.utils.parse_cache import LazyTree, ParseCache, SQLiteParseCache + +PY_TEMPLATE = ''' +class Widget{n}: + """Docstring for Widget{n}.""" + + def __init__(self, value): + self.value = value + + def compute(self, other): + if other > 0: + return self.value + other + return helper_{n}(self.value) + + +def helper_{n}(value): + return value * 2 +''' + + +@pytest.fixture +def source_files(tmp_path): + """Enough files that thread interleaving actually varies between runs.""" + files = [] + for i in range(40): + p = tmp_path / f"module_{i:03d}.py" + p.write_text(PY_TEMPLATE.format(n=i)) + files.append(str(p)) + return files + + +def _parse_all(pm: ParserManager, files: list[str], workers: int) -> list: + def job(path: str): + tree = pm.parse_file(path) + if tree is None: + return [] + with open(path, "rb") as fh: + source = fh.read() + return pm.extract_nodes(tree, path, "python", source, "TEST") + + with ThreadPoolExecutor(max_workers=workers) as pool: + return list(pool.map(job, files)) + + +# --------------------------------------------------------------------------- +# Parser isolation +# --------------------------------------------------------------------------- + + +def test_each_thread_gets_its_own_parser(): + """A tree_sitter.Parser must never be shared between threads.""" + pm = ParserManager(cache=ParseCache()) + seen: dict[int, int] = {} + lock = threading.Lock() + + def grab() -> None: + parser = pm._get_parser("python") + with lock: + seen[threading.get_ident()] = id(parser) + + with ThreadPoolExecutor(max_workers=6) as pool: + list(pool.map(lambda _: grab(), range(24))) + + assert len(set(seen.values())) == len(seen) + + +def test_same_thread_reuses_its_parser(): + """Thread-local caching must not allocate a fresh parser per call.""" + pm = ParserManager(cache=ParseCache()) + assert pm._get_parser("python") is pm._get_parser("python") + + +def test_constructing_thread_reuses_prebuilt_parsers(): + """Single-threaded callers keep the parsers built in _init_languages.""" + pm = ParserManager(cache=ParseCache()) + assert pm._get_parser("python") is pm._parsers["python"] + + +# --------------------------------------------------------------------------- +# Correctness under concurrency +# --------------------------------------------------------------------------- + + +def test_concurrent_parsing_matches_sequential(source_files): + """Threading must not change what gets extracted.""" + sequential = _parse_all(ParserManager(cache=ParseCache()), source_files, workers=1) + concurrent = _parse_all(ParserManager(cache=ParseCache()), source_files, workers=8) + + assert [[n.id for n in f] for f in sequential] == [[n.id for n in f] for f in concurrent] + assert all(nodes for nodes in concurrent) + + +@pytest.mark.parametrize( + "cache_factory", [ParseCache, BoundedParseCache], ids=["in-memory", "bounded-lru"] +) +def test_cache_counters_stay_coherent(source_files, cache_factory): + """Hits + misses must equal the number of lookups, with no lost updates.""" + pm = ParserManager(cache=cache_factory()) + _parse_all(pm, source_files * 2, workers=8) + + stats = pm.tree_cache_stats() + assert stats["hits"] > 0 + assert stats["hits"] + stats["misses"] == len(source_files) * 2 + + +def test_sqlite_backend_tolerates_concurrent_access(source_files, tmp_path): + """One sqlite connection serves every thread; the lock must serialise it.""" + cache = SQLiteParseCache(str(tmp_path / "cache.sqlite")) + pm = ParserManager(cache=cache) + try: + results = _parse_all(pm, source_files * 2, workers=8) + assert all(nodes for nodes in results) + assert pm.tree_cache_stats()["hits"] > 0 + finally: + cache.close() + + +# --------------------------------------------------------------------------- +# Specific defects fixed +# --------------------------------------------------------------------------- + + +def test_lazy_tree_loader_runs_exactly_once_under_contention(): + """LazyTree._ensure claims 'exactly once' — hold it to that under a race.""" + calls: list[int] = [] + calls_lock = threading.Lock() + # Line the threads up *before* resolve() so they enter _ensure together. + # The barrier must not live inside the loader: only one thread reaches it, + # so waiting there would deadlock the rest. + start = threading.Barrier(8) + + def loader(): + with calls_lock: + calls.append(1) + return object() + + lazy = LazyTree(loader=loader) + + def worker(_): + start.wait() + return lazy.resolve() + + with ThreadPoolExecutor(max_workers=8) as pool: + list(pool.map(worker, range(8))) + + assert len(calls) == 1 + + +def test_bounded_cache_hashes_are_dropped_on_eviction(source_files): + """_hashes previously grew forever: the LRU evicted trees but not hashes.""" + cache = BoundedParseCache(max_entries=5) + pm = ParserManager(cache=cache) + _parse_all(pm, source_files, workers=4) + + assert len(cache._hashes) <= 5 + assert set(cache._hashes) <= set(cache._inner._cache)