From 68ca2a1fb18a5eb2883c64fb52113d2abdb1ac20 Mon Sep 17 00:00:00 2001 From: Pringled Date: Wed, 7 Oct 2026 10:47:23 +0200 Subject: [PATCH 1/6] perf: Chunk files in worker processes during index builds --- src/semble/index/create.py | 137 ++++++++++++++++++++++++++++--------- tests/index/test_create.py | 41 ++++++++++- 2 files changed, 145 insertions(+), 33 deletions(-) diff --git a/src/semble/index/create.py b/src/semble/index/create.py index 9026cf1b..d3689587 100644 --- a/src/semble/index/create.py +++ b/src/semble/index/create.py @@ -1,7 +1,15 @@ import contextlib +import itertools import logging +import multiprocessing +import multiprocessing.connection +import os +import signal import sys -from collections.abc import Sequence +import threading +from collections import deque +from collections.abc import Iterator, Sequence +from concurrent.futures import Future, ProcessPoolExecutor from pathlib import Path import numpy as np @@ -28,6 +36,13 @@ logger = logging.getLogger(__name__) +# Below this many files to chunk, starting worker processes costs more than it saves. +_MIN_FILES_FOR_PROCESSES = 200 +# Chunking in a few processes keeps up with embedding in the main process. +_WORKERS = 4 +# Bounds the chunked files waiting to be embedded. +_MAX_FILES_IN_FLIGHT = 256 + def _warn_skipped_large(skipped_large: list[str]) -> None: """Warn about files skipped for exceeding the maximum indexable file size.""" @@ -42,18 +57,80 @@ def _warn_skipped_large(skipped_large: list[str]) -> None: ) +def _chunk_file(file_path: Path, indexed_path: str) -> tuple[list[Chunk], list[list[str]]] | None: + """Chunk a file and tokenize its chunks for BM25, or return None if it can't be read.""" + try: + source = read_file_text(file_path) + except OSError: + return None + file_chunks = chunk_source(source, indexed_path, detect_language(file_path)) + return file_chunks, [tokenize(enrich_for_bm25(chunk)) for chunk in file_chunks] + + +def _init_worker() -> None: + """Leave Ctrl-C to the parent, and exit this worker once the parent is gone, even if it never shut the pool down.""" + signal.signal(signal.SIGINT, signal.SIG_IGN) + parent = multiprocessing.parent_process() + assert parent is not None + + def wait_and_exit() -> None: + multiprocessing.connection.wait([parent.sentinel]) + os._exit(0) + + threading.Thread(target=wait_and_exit, daemon=True).start() + + +def _chunk_files(files: list[tuple[Path, str]]) -> Iterator[tuple[list[Chunk], list[list[str]]] | None]: + """Chunk files in order, in worker processes when there are enough of them.""" + if len(files) < _MIN_FILES_FOR_PROCESSES: + yield from itertools.starmap(_chunk_file, files) + else: + with ProcessPoolExecutor( + max_workers=_WORKERS, mp_context=multiprocessing.get_context("spawn"), initializer=_init_worker + ) as executor: + pending: deque[Future[tuple[list[Chunk], list[list[str]]] | None]] = deque() + for file_path, indexed_path in files: + pending.append(executor.submit(_chunk_file, file_path, indexed_path)) + if len(pending) == _MAX_FILES_IN_FLIGHT: + yield pending.popleft().result() + yield from (future.result() for future in pending) + + +def _stat_files( + path: Path, + extensions: Sequence[str], + display_root: Path | None, + previous_manifest: dict[str, FileManifestEntry], + skipped_large: list[str], +) -> list[tuple[Path, str, int, FileManifestEntry | None]]: + """Return each indexable file's path, indexed path, mtime and previous manifest entry.""" + files = [] + for file_path in walk_files(path, extensions): + with contextlib.suppress(OSError): + stat = file_path.stat() + file_status = get_file_status(file_path, stat) + if file_status is FileStatus.TOO_LARGE: + skipped_large.append(str(file_path)) + if file_status != FileStatus.VALID: + continue + + indexed_path = str(file_path.relative_to(display_root) if display_root else file_path) + files.append((file_path, indexed_path, stat.st_mtime_ns, previous_manifest.get(indexed_path))) + return files + + def _reindex_file( bm25_index: BM25, indexed_path: str, - file_chunks: list[Chunk], + file_tokens: list[list[str]], previous_entry: FileManifestEntry | None, ) -> None: """Replace a file's BM25 postings: remove its old slots (if any), then add its new ones.""" if previous_entry is not None: for slot in range(previous_entry.count): bm25_index.remove_document(make_chunk_id(indexed_path, slot)) - for slot, chunk in enumerate(file_chunks): - bm25_index.add_document(make_chunk_id(indexed_path, slot), tokenize(enrich_for_bm25(chunk))) + for slot, tokens in enumerate(file_tokens): + bm25_index.add_document(make_chunk_id(indexed_path, slot), tokens) def _has_same_vector_layout( @@ -102,8 +179,15 @@ def create_index_from_path( skipped_large: list[str] = [] - files = list(walk_files(path, resolved_extensions)) - for file_path in tqdm( + files = _stat_files(path, resolved_extensions, display_root, previous_manifest, skipped_large) + stale = [ + (file_path, indexed_path) + for file_path, indexed_path, mtime_ns, previous_entry in files + if previous_entry is None or previous_entry.mtime_ns != mtime_ns + ] + + chunked = _chunk_files(stale) + for _, indexed_path, mtime_ns, previous_entry in tqdm( files, desc="Indexing", unit="file", @@ -113,34 +197,23 @@ def create_index_from_path( miniters=1, # tqdm's adaptive miniters stalls the bar when fast files are followed by slow ones disable=not show_progress_bar, ): - language = detect_language(file_path) - with contextlib.suppress(OSError): - stat = file_path.stat() - file_status = get_file_status(file_path, stat) - if file_status is FileStatus.TOO_LARGE: - skipped_large.append(str(file_path)) - if file_status != FileStatus.VALID: + if previous is not None and previous_entry is not None and previous_entry.mtime_ns == mtime_ns: + file_chunks = previous.chunks[previous_entry.start : previous_entry.end] + vector_parts.append(previous.vectors[previous_entry.start : previous_entry.end]) + else: + result = next(chunked) + if result is None: continue + file_chunks, file_tokens = result + _reindex_file(bm25_index, indexed_path, file_tokens, previous_entry) - indexed_path = str(file_path.relative_to(display_root) if display_root else file_path) - mtime_ns = stat.st_mtime_ns - previous_entry = previous_manifest.get(indexed_path) - - if previous is not None and previous_entry is not None and previous_entry.mtime_ns == mtime_ns: - file_chunks = previous.chunks[previous_entry.start : previous_entry.end] - vector_parts.append(previous.vectors[previous_entry.start : previous_entry.end]) - else: - source = read_file_text(file_path) - file_chunks = chunk_source(source, indexed_path, language) - _reindex_file(bm25_index, indexed_path, file_chunks, previous_entry) - - embedding_parts.append((len(vector_parts), len(chunks), len(file_chunks))) - vector_parts.append(embed_chunks(model, file_chunks)) - - start = len(chunks) - chunks.extend(file_chunks) - chunk_ids.extend(make_chunk_id(indexed_path, slot) for slot in range(len(file_chunks))) - manifest[indexed_path] = FileManifestEntry(mtime_ns=mtime_ns, start=start, count=len(file_chunks)) + embedding_parts.append((len(vector_parts), len(chunks), len(file_chunks))) + vector_parts.append(embed_chunks(model, file_chunks)) + + start = len(chunks) + chunks.extend(file_chunks) + chunk_ids.extend(make_chunk_id(indexed_path, slot) for slot in range(len(file_chunks))) + manifest[indexed_path] = FileManifestEntry(mtime_ns=mtime_ns, start=start, count=len(file_chunks)) for indexed_path in previous_manifest.keys() - manifest.keys(): _reindex_file(bm25_index, indexed_path, [], previous_manifest[indexed_path]) diff --git a/tests/index/test_create.py b/tests/index/test_create.py index cb280a3e..0cace4d7 100644 --- a/tests/index/test_create.py +++ b/tests/index/test_create.py @@ -1,4 +1,7 @@ +import multiprocessing +import threading from pathlib import Path +from types import SimpleNamespace from typing import Any from unittest.mock import patch @@ -7,6 +10,7 @@ import pytest from semble.cache import load_previous_for_incremental +from semble.index import create from semble.index.bm25 import BM25 from semble.index.create import create_index_from_path from semble.index.index import SembleIndex @@ -21,8 +25,14 @@ def _write_files(root: Path, files: dict[str, str]) -> None: path.write_text(content) -def test_incremental_reindex_reuses_updates_and_prunes(mock_model: Any, tmp_path: Path) -> None: +@pytest.mark.parametrize("in_processes", [False, True]) +def test_incremental_reindex_reuses_updates_and_prunes( + mock_model: Any, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, in_processes: bool +) -> None: """One incremental pass reuses unchanged vectors, re-embeds changes, and keeps BM25 slots current.""" + if in_processes: + monkeypatch.setattr(create, "_MIN_FILES_FOR_PROCESSES", 0) + monkeypatch.setattr(create, "_MAX_FILES_IN_FLIGHT", 2) _write_files( tmp_path, { @@ -89,6 +99,35 @@ def test_incremental_reindex_reuses_updates_and_prunes(mock_model: Any, tmp_path assert set(bm25_after.doc_order) == expected_ids +def test_unreadable_file_is_skipped(mock_model: Any, tmp_path: Path) -> None: + """A file that can't be read is left out of the index.""" + _write_files(tmp_path, {"a.py": "def a():\n return 1\n", "b.py": "def b():\n return 2\n"}) + read_file_text = create.read_file_text + + def read_or_fail(file_path: Path) -> str: + if file_path.name == "b.py": + raise PermissionError(file_path) + return read_file_text(file_path) + + with patch.object(create, "read_file_text", read_or_fail): + _, _, _, manifest = create_index_from_path(tmp_path, mock_model, display_root=tmp_path) + assert list(manifest) == ["a.py"] + + +def test_worker_exits_with_parent(monkeypatch: pytest.MonkeyPatch) -> None: + """A chunking worker exits once its parent process is gone.""" + parent_end, worker_end = multiprocessing.Pipe() + exited = threading.Event() + monkeypatch.setattr(multiprocessing, "parent_process", lambda: SimpleNamespace(sentinel=worker_end)) + monkeypatch.setattr(create.os, "_exit", lambda _: exited.set()) + monkeypatch.setattr(create.signal, "signal", lambda *_: None) + + create._init_worker() + assert not exited.wait(0.1) + parent_end.close() + assert exited.wait(5) + + def _build_valid_cache(index_path: Path, mock_model: Any) -> dict: """Build a real, well-formed on-disk index and return its metadata dict for mutation.""" src = index_path.parent / "src" From 17a14e2557501b023690170ee783fb4a2aa4e1f7 Mon Sep 17 00:00:00 2001 From: Pringled Date: Wed, 7 Oct 2026 10:54:14 +0200 Subject: [PATCH 2/6] Add SEMBLE_INDEX_WORKERS and simplify the in-flight window --- README.md | 2 ++ src/semble/index/create.py | 29 +++++++++++++++-------------- tests/index/test_create.py | 1 - 3 files changed, 17 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index b062ce7f..1729228a 100644 --- a/README.md +++ b/README.md @@ -186,6 +186,8 @@ By default, your Semble savings statistics and any saved indexes are stored in t Files larger than 1 MB are skipped during indexing to keep index builds lean. Skipped files are reported as a warning at index time. If you work with large generated or ingested documents, you can raise (or lower) this limit with the `SEMBLE_MAX_FILE_BYTES` environment variable (in bytes). +When many files need indexing, Semble chunks them in 4 worker processes. Set `SEMBLE_INDEX_WORKERS` to change the number of workers, or to `0` to index in a single process. + On first use, Semble also downloads the embedding model from Hugging Face and caches it in the standard Hugging Face cache (`~/.cache/huggingface/` by default, or `$HF_HOME` if set); this only happens once and requires network access. Use `semble clear` to remove cached data: `semble clear index` (saved indexes), `semble clear savings` (usage stats), `semble clear orphans` (indexes for repos no longer present on disk), or `semble clear all` (everything). diff --git a/src/semble/index/create.py b/src/semble/index/create.py index d3689587..a931d069 100644 --- a/src/semble/index/create.py +++ b/src/semble/index/create.py @@ -9,7 +9,7 @@ import threading from collections import deque from collections.abc import Iterator, Sequence -from concurrent.futures import Future, ProcessPoolExecutor +from concurrent.futures import ProcessPoolExecutor from pathlib import Path import numpy as np @@ -36,13 +36,15 @@ logger = logging.getLogger(__name__) -# Below this many files to chunk, starting worker processes costs more than it saves. +# Processes that chunk files when at least _MIN_FILES_FOR_PROCESSES need chunking; 0 chunks in the main process. +_WORKERS = int(os.environ.get("SEMBLE_INDEX_WORKERS", 4)) _MIN_FILES_FOR_PROCESSES = 200 -# Chunking in a few processes keeps up with embedding in the main process. -_WORKERS = 4 -# Bounds the chunked files waiting to be embedded. +# Files handed to the workers but not yet consumed by the main process. _MAX_FILES_IN_FLIGHT = 256 +# A file's chunks and their BM25 tokens. +_ChunkedFile = tuple[list[Chunk], list[list[str]]] + def _warn_skipped_large(skipped_large: list[str]) -> None: """Warn about files skipped for exceeding the maximum indexable file size.""" @@ -57,7 +59,7 @@ def _warn_skipped_large(skipped_large: list[str]) -> None: ) -def _chunk_file(file_path: Path, indexed_path: str) -> tuple[list[Chunk], list[list[str]]] | None: +def _chunk_file(file_path: Path, indexed_path: str) -> _ChunkedFile | None: """Chunk a file and tokenize its chunks for BM25, or return None if it can't be read.""" try: source = read_file_text(file_path) @@ -80,20 +82,19 @@ def wait_and_exit() -> None: threading.Thread(target=wait_and_exit, daemon=True).start() -def _chunk_files(files: list[tuple[Path, str]]) -> Iterator[tuple[list[Chunk], list[list[str]]] | None]: +def _chunk_files(files: list[tuple[Path, str]]) -> Iterator[_ChunkedFile | None]: """Chunk files in order, in worker processes when there are enough of them.""" - if len(files) < _MIN_FILES_FOR_PROCESSES: + if _WORKERS == 0 or len(files) < _MIN_FILES_FOR_PROCESSES: yield from itertools.starmap(_chunk_file, files) else: with ProcessPoolExecutor( max_workers=_WORKERS, mp_context=multiprocessing.get_context("spawn"), initializer=_init_worker ) as executor: - pending: deque[Future[tuple[list[Chunk], list[list[str]]] | None]] = deque() - for file_path, indexed_path in files: - pending.append(executor.submit(_chunk_file, file_path, indexed_path)) - if len(pending) == _MAX_FILES_IN_FLIGHT: - yield pending.popleft().result() - yield from (future.result() for future in pending) + submitted = (executor.submit(_chunk_file, *file) for file in files) + pending = deque(itertools.islice(submitted, _MAX_FILES_IN_FLIGHT)) + while pending: + yield pending.popleft().result() + pending.extend(itertools.islice(submitted, 1)) def _stat_files( diff --git a/tests/index/test_create.py b/tests/index/test_create.py index 0cace4d7..8421e7c1 100644 --- a/tests/index/test_create.py +++ b/tests/index/test_create.py @@ -32,7 +32,6 @@ def test_incremental_reindex_reuses_updates_and_prunes( """One incremental pass reuses unchanged vectors, re-embeds changes, and keeps BM25 slots current.""" if in_processes: monkeypatch.setattr(create, "_MIN_FILES_FOR_PROCESSES", 0) - monkeypatch.setattr(create, "_MAX_FILES_IN_FLIGHT", 2) _write_files( tmp_path, { From 7ff3bb93ee543de08c0f86d0ba20906d9a968c28 Mon Sep 17 00:00:00 2001 From: Pringled Date: Wed, 7 Oct 2026 14:49:58 +0200 Subject: [PATCH 3/6] Make index workers opt-in for the Python API and shut the pool down with the build loop --- README.md | 2 +- src/semble/cli.py | 2 + src/semble/index/create.py | 77 ++++++++++++++++++++------------------ tests/index/test_create.py | 1 + 4 files changed, 44 insertions(+), 38 deletions(-) diff --git a/README.md b/README.md index 1729228a..723c359f 100644 --- a/README.md +++ b/README.md @@ -186,7 +186,7 @@ By default, your Semble savings statistics and any saved indexes are stored in t Files larger than 1 MB are skipped during indexing to keep index builds lean. Skipped files are reported as a warning at index time. If you work with large generated or ingested documents, you can raise (or lower) this limit with the `SEMBLE_MAX_FILE_BYTES` environment variable (in bytes). -When many files need indexing, Semble chunks them in 4 worker processes. Set `SEMBLE_INDEX_WORKERS` to change the number of workers, or to `0` to index in a single process. +When many files need indexing, the CLI and MCP server chunk them in 4 worker processes. Set `SEMBLE_INDEX_WORKERS` to change the number of workers, or to `0` to index in a single process. In the Python API, workers are off by default; set `SEMBLE_INDEX_WORKERS` to turn them on, and guard your script with `if __name__ == "__main__":`. On first use, Semble also downloads the embedding model from Hugging Face and caches it in the standard Hugging Face cache (`~/.cache/huggingface/` by default, or `$HF_HOME` if set); this only happens once and requires network access. diff --git a/src/semble/cli.py b/src/semble/cli.py index faa094e8..94e7123c 100644 --- a/src/semble/cli.py +++ b/src/semble/cli.py @@ -95,6 +95,8 @@ def main() -> None: for stream in (sys.stdout, sys.stderr): if isinstance(stream, io.TextIOWrapper): stream.reconfigure(errors="replace") + # Worker processes re-import __main__, so only guarded entry points like this one turn them on by default. + os.environ.setdefault("SEMBLE_INDEX_WORKERS", "4") if len(sys.argv) > 1 and sys.argv[1] in _CLI_DISPATCH_ARGS: _cli_main() else: diff --git a/src/semble/index/create.py b/src/semble/index/create.py index a931d069..4c354779 100644 --- a/src/semble/index/create.py +++ b/src/semble/index/create.py @@ -8,7 +8,7 @@ import sys import threading from collections import deque -from collections.abc import Iterator, Sequence +from collections.abc import Generator, Sequence from concurrent.futures import ProcessPoolExecutor from pathlib import Path @@ -36,8 +36,7 @@ logger = logging.getLogger(__name__) -# Processes that chunk files when at least _MIN_FILES_FOR_PROCESSES need chunking; 0 chunks in the main process. -_WORKERS = int(os.environ.get("SEMBLE_INDEX_WORKERS", 4)) +# Below this many files to chunk, SEMBLE_INDEX_WORKERS is ignored and files are chunked in the main process. _MIN_FILES_FOR_PROCESSES = 200 # Files handed to the workers but not yet consumed by the main process. _MAX_FILES_IN_FLIGHT = 256 @@ -82,19 +81,23 @@ def wait_and_exit() -> None: threading.Thread(target=wait_and_exit, daemon=True).start() -def _chunk_files(files: list[tuple[Path, str]]) -> Iterator[_ChunkedFile | None]: - """Chunk files in order, in worker processes when there are enough of them.""" - if _WORKERS == 0 or len(files) < _MIN_FILES_FOR_PROCESSES: +def _chunk_files(files: list[tuple[Path, str]]) -> Generator[_ChunkedFile | None, None, None]: + """Chunk files in order, in SEMBLE_INDEX_WORKERS worker processes when there are enough files.""" + workers = int(os.environ.get("SEMBLE_INDEX_WORKERS", 0)) + if workers == 0 or len(files) < _MIN_FILES_FOR_PROCESSES: yield from itertools.starmap(_chunk_file, files) else: - with ProcessPoolExecutor( - max_workers=_WORKERS, mp_context=multiprocessing.get_context("spawn"), initializer=_init_worker - ) as executor: + executor = ProcessPoolExecutor( + workers, mp_context=multiprocessing.get_context("spawn"), initializer=_init_worker + ) + try: submitted = (executor.submit(_chunk_file, *file) for file in files) pending = deque(itertools.islice(submitted, _MAX_FILES_IN_FLIGHT)) while pending: yield pending.popleft().result() pending.extend(itertools.islice(submitted, 1)) + finally: + executor.shutdown(cancel_futures=True) def _stat_files( @@ -187,34 +190,34 @@ def create_index_from_path( if previous_entry is None or previous_entry.mtime_ns != mtime_ns ] - chunked = _chunk_files(stale) - for _, indexed_path, mtime_ns, previous_entry in tqdm( - files, - desc="Indexing", - unit="file", - file=sys.stderr, - leave=False, - colour="green", - miniters=1, # tqdm's adaptive miniters stalls the bar when fast files are followed by slow ones - disable=not show_progress_bar, - ): - if previous is not None and previous_entry is not None and previous_entry.mtime_ns == mtime_ns: - file_chunks = previous.chunks[previous_entry.start : previous_entry.end] - vector_parts.append(previous.vectors[previous_entry.start : previous_entry.end]) - else: - result = next(chunked) - if result is None: - continue - file_chunks, file_tokens = result - _reindex_file(bm25_index, indexed_path, file_tokens, previous_entry) - - embedding_parts.append((len(vector_parts), len(chunks), len(file_chunks))) - vector_parts.append(embed_chunks(model, file_chunks)) - - start = len(chunks) - chunks.extend(file_chunks) - chunk_ids.extend(make_chunk_id(indexed_path, slot) for slot in range(len(file_chunks))) - manifest[indexed_path] = FileManifestEntry(mtime_ns=mtime_ns, start=start, count=len(file_chunks)) + with contextlib.closing(_chunk_files(stale)) as chunked: + for _, indexed_path, mtime_ns, previous_entry in tqdm( + files, + desc="Indexing", + unit="file", + file=sys.stderr, + leave=False, + colour="green", + miniters=1, # tqdm's adaptive miniters stalls the bar when fast files are followed by slow ones + disable=not show_progress_bar, + ): + if previous is not None and previous_entry is not None and previous_entry.mtime_ns == mtime_ns: + file_chunks = previous.chunks[previous_entry.start : previous_entry.end] + vector_parts.append(previous.vectors[previous_entry.start : previous_entry.end]) + else: + result = next(chunked) + if result is None: + continue + file_chunks, file_tokens = result + _reindex_file(bm25_index, indexed_path, file_tokens, previous_entry) + + embedding_parts.append((len(vector_parts), len(chunks), len(file_chunks))) + vector_parts.append(embed_chunks(model, file_chunks)) + + start = len(chunks) + chunks.extend(file_chunks) + chunk_ids.extend(make_chunk_id(indexed_path, slot) for slot in range(len(file_chunks))) + manifest[indexed_path] = FileManifestEntry(mtime_ns=mtime_ns, start=start, count=len(file_chunks)) for indexed_path in previous_manifest.keys() - manifest.keys(): _reindex_file(bm25_index, indexed_path, [], previous_manifest[indexed_path]) diff --git a/tests/index/test_create.py b/tests/index/test_create.py index 8421e7c1..15b0936f 100644 --- a/tests/index/test_create.py +++ b/tests/index/test_create.py @@ -32,6 +32,7 @@ def test_incremental_reindex_reuses_updates_and_prunes( """One incremental pass reuses unchanged vectors, re-embeds changes, and keeps BM25 slots current.""" if in_processes: monkeypatch.setattr(create, "_MIN_FILES_FOR_PROCESSES", 0) + monkeypatch.setenv("SEMBLE_INDEX_WORKERS", "2") _write_files( tmp_path, { From 22779ca3e28f7eed6d5f19ccce450ebfd2985aae Mon Sep 17 00:00:00 2001 From: Pringled Date: Sat, 10 Oct 2026 19:57:03 +0200 Subject: [PATCH 4/6] Move the index workers note to Library usage --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 723c359f..545f378e 100644 --- a/README.md +++ b/README.md @@ -186,8 +186,6 @@ By default, your Semble savings statistics and any saved indexes are stored in t Files larger than 1 MB are skipped during indexing to keep index builds lean. Skipped files are reported as a warning at index time. If you work with large generated or ingested documents, you can raise (or lower) this limit with the `SEMBLE_MAX_FILE_BYTES` environment variable (in bytes). -When many files need indexing, the CLI and MCP server chunk them in 4 worker processes. Set `SEMBLE_INDEX_WORKERS` to change the number of workers, or to `0` to index in a single process. In the Python API, workers are off by default; set `SEMBLE_INDEX_WORKERS` to turn them on, and guard your script with `if __name__ == "__main__":`. - On first use, Semble also downloads the embedding model from Hugging Face and caches it in the standard Hugging Face cache (`~/.cache/huggingface/` by default, or `$HF_HOME` if set); this only happens once and requires network access. Use `semble clear` to remove cached data: `semble clear index` (saved indexes), `semble clear savings` (usage stats), `semble clear orphans` (indexes for repos no longer present on disk), or `semble clear all` (everything). @@ -234,6 +232,8 @@ result.chunk.end_line # 150 result.chunk.content # "def save_pretrained(self, path: PathLike, ..." ``` +When many files need indexing, the CLI and MCP server chunk them in 4 worker processes (`SEMBLE_INDEX_WORKERS` sets the number, `0` turns them off). In the Python API they are off by default. To turn them on, set `SEMBLE_INDEX_WORKERS` and call Semble from inside an `if __name__ == "__main__":` block, since each worker re-imports your script. + ## MCP Server From 3eebdae4472d886c69de4e86801d230d08034c84 Mon Sep 17 00:00:00 2001 From: Pringled Date: Sat, 10 Oct 2026 21:10:22 +0200 Subject: [PATCH 5/6] Give index workers NUL as stdin on Windows so the MCP server can't deadlock them --- src/semble/index/create.py | 34 ++++++++++++++++++++++++++++++++-- 1 file changed, 32 insertions(+), 2 deletions(-) diff --git a/src/semble/index/create.py b/src/semble/index/create.py index 4c354779..ad61dc27 100644 --- a/src/semble/index/create.py +++ b/src/semble/index/create.py @@ -8,7 +8,7 @@ import sys import threading from collections import deque -from collections.abc import Generator, Sequence +from collections.abc import Generator, Iterator, Sequence from concurrent.futures import ProcessPoolExecutor from pathlib import Path @@ -81,6 +81,34 @@ def wait_and_exit() -> None: threading.Thread(target=wait_and_exit, daemon=True).start() +if sys.platform == "win32": # pragma: no cover + import ctypes + import msvcrt + + @contextlib.contextmanager + def _stdin_as_nul() -> Iterator[None]: + """Give processes started in this block NUL as stdin. + + A Windows pipe serves one operation at a time, so a child that shares a stdin this process is blocked + reading (as the MCP server is) hangs in Python's startup until the next line arrives. + """ + kernel32 = ctypes.WinDLL("kernel32") + kernel32.GetStdHandle.restype = ctypes.c_void_p + kernel32.GetStdHandle.argtypes = [ctypes.c_uint32] + kernel32.SetStdHandle.argtypes = [ctypes.c_uint32, ctypes.c_void_p] + std_input_handle = ctypes.c_uint32(-10).value + stdin = kernel32.GetStdHandle(std_input_handle) + with open(os.devnull, "rb") as nul: + kernel32.SetStdHandle(std_input_handle, msvcrt.get_osfhandle(nul.fileno())) + try: + yield + finally: + kernel32.SetStdHandle(std_input_handle, stdin) + +else: + _stdin_as_nul = contextlib.nullcontext + + def _chunk_files(files: list[tuple[Path, str]]) -> Generator[_ChunkedFile | None, None, None]: """Chunk files in order, in SEMBLE_INDEX_WORKERS worker processes when there are enough files.""" workers = int(os.environ.get("SEMBLE_INDEX_WORKERS", 0)) @@ -92,7 +120,9 @@ def _chunk_files(files: list[tuple[Path, str]]) -> Generator[_ChunkedFile | None ) try: submitted = (executor.submit(_chunk_file, *file) for file in files) - pending = deque(itertools.islice(submitted, _MAX_FILES_IN_FLIGHT)) + # Workers start on the first submits. + with _stdin_as_nul(): + pending = deque(itertools.islice(submitted, _MAX_FILES_IN_FLIGHT)) while pending: yield pending.popleft().result() pending.extend(itertools.islice(submitted, 1)) From 899ee2652858c729d59989fb1eb89dc7dd5152e1 Mon Sep 17 00:00:00 2001 From: Pringled Date: Sun, 11 Oct 2026 09:44:53 +0200 Subject: [PATCH 6/6] Drop the Windows stdin workaround now that MCP 2.x isolates stdio from child processes --- src/semble/index/create.py | 34 ++-------------------------------- 1 file changed, 2 insertions(+), 32 deletions(-) diff --git a/src/semble/index/create.py b/src/semble/index/create.py index ad61dc27..4c354779 100644 --- a/src/semble/index/create.py +++ b/src/semble/index/create.py @@ -8,7 +8,7 @@ import sys import threading from collections import deque -from collections.abc import Generator, Iterator, Sequence +from collections.abc import Generator, Sequence from concurrent.futures import ProcessPoolExecutor from pathlib import Path @@ -81,34 +81,6 @@ def wait_and_exit() -> None: threading.Thread(target=wait_and_exit, daemon=True).start() -if sys.platform == "win32": # pragma: no cover - import ctypes - import msvcrt - - @contextlib.contextmanager - def _stdin_as_nul() -> Iterator[None]: - """Give processes started in this block NUL as stdin. - - A Windows pipe serves one operation at a time, so a child that shares a stdin this process is blocked - reading (as the MCP server is) hangs in Python's startup until the next line arrives. - """ - kernel32 = ctypes.WinDLL("kernel32") - kernel32.GetStdHandle.restype = ctypes.c_void_p - kernel32.GetStdHandle.argtypes = [ctypes.c_uint32] - kernel32.SetStdHandle.argtypes = [ctypes.c_uint32, ctypes.c_void_p] - std_input_handle = ctypes.c_uint32(-10).value - stdin = kernel32.GetStdHandle(std_input_handle) - with open(os.devnull, "rb") as nul: - kernel32.SetStdHandle(std_input_handle, msvcrt.get_osfhandle(nul.fileno())) - try: - yield - finally: - kernel32.SetStdHandle(std_input_handle, stdin) - -else: - _stdin_as_nul = contextlib.nullcontext - - def _chunk_files(files: list[tuple[Path, str]]) -> Generator[_ChunkedFile | None, None, None]: """Chunk files in order, in SEMBLE_INDEX_WORKERS worker processes when there are enough files.""" workers = int(os.environ.get("SEMBLE_INDEX_WORKERS", 0)) @@ -120,9 +92,7 @@ def _chunk_files(files: list[tuple[Path, str]]) -> Generator[_ChunkedFile | None ) try: submitted = (executor.submit(_chunk_file, *file) for file in files) - # Workers start on the first submits. - with _stdin_as_nul(): - pending = deque(itertools.islice(submitted, _MAX_FILES_IN_FLIGHT)) + pending = deque(itertools.islice(submitted, _MAX_FILES_IN_FLIGHT)) while pending: yield pending.popleft().result() pending.extend(itertools.islice(submitted, 1))