From d352ed595808e9ddb04b8215557d4de5a7949ea4 Mon Sep 17 00:00:00 2001 From: sinnercode228 Date: Tue, 27 Oct 2026 20:13:26 +0700 Subject: [PATCH] Run the hashing embedder off the event loop HashingEmbedder.embed_documents was async but its body was a plain list comprehension over the sync self.embed(text), so a document with hundreds of chunks kept the asyncio loop busy for the whole embed. The ingest pipeline runs extract in a thread and awaits embed_documents right after, so the API worker was blocked on the CPU-bound half. Dispatch the inner loop to a worker thread the same way extraction does. embed_query is a single call and stays on the loop. Fixes #6 --- backend/src/docmind/embeddings/hashing.py | 6 +++++ backend/tests/test_retrieval.py | 28 +++++++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/backend/src/docmind/embeddings/hashing.py b/backend/src/docmind/embeddings/hashing.py index 5d56ccb..8be27cb 100644 --- a/backend/src/docmind/embeddings/hashing.py +++ b/backend/src/docmind/embeddings/hashing.py @@ -7,6 +7,7 @@ from __future__ import annotations +import asyncio import hashlib import itertools import math @@ -59,6 +60,11 @@ def embed(self, text: str) -> Vector: return l2_normalize(vector) async def embed_documents(self, texts: list[str]) -> list[Vector]: + # CPU-bound hashing + numpy; dispatch to a worker thread so the ingest + # loop can serve other requests while a long document is embedded. + return await asyncio.to_thread(self._embed_many, texts) + + def _embed_many(self, texts: list[str]) -> list[Vector]: return [self.embed(text) for text in texts] async def embed_query(self, text: str) -> Vector: diff --git a/backend/tests/test_retrieval.py b/backend/tests/test_retrieval.py index f1ab30f..5e4e850 100644 --- a/backend/tests/test_retrieval.py +++ b/backend/tests/test_retrieval.py @@ -1,5 +1,7 @@ from __future__ import annotations +import asyncio +import contextlib import json from pathlib import Path @@ -57,6 +59,32 @@ def test_rejects_tiny_dimension(self) -> None: with pytest.raises(ValueError, match="dim"): HashingEmbedder(8) + async def test_embed_documents_yields_to_the_event_loop(self) -> None: + # The ingest pipeline runs extract in a worker thread and then awaits + # embed_documents on the same event loop; if embedding keeps the loop + # busy, nothing else on the API worker gets to run meanwhile. + emb = HashingEmbedder(384) + texts = [(CORPUS["vac"] + " ") * 400] * 40 # long enough to be slow + + ticks = 0 + + async def ticker() -> None: + nonlocal ticks + while True: + ticks += 1 + await asyncio.sleep(0) + + ticker_task = asyncio.create_task(ticker()) + await asyncio.sleep(0) + ticks_before = ticks + await emb.embed_documents(texts) + ticks_during = ticks - ticks_before + ticker_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await ticker_task + + assert ticks_during > 5, f"event loop was blocked; ticker advanced {ticks_during} times" + async def test_openai_compatible_embedder_batches_and_sorts() -> None: calls: list[list[str]] = []