From 4fa8a012b4cc9e52591c405896da6f0b79bd76d8 Mon Sep 17 00:00:00 2001 From: Chris McDonough Date: Mon, 1 Jun 2026 06:28:16 -0400 Subject: [PATCH] Use asyncio.Condition for event-driven worker wakeup Workers previously polled the queue with a fixed 1s sleep between claim attempts, adding ~500ms average latency to job pickup. Now JobRepo.job_available (an asyncio.Condition) is notified on every successful enqueue, waking idle workers immediately. The poll interval remains as a timeout fallback for stop signals and breaker state changes. --- haiku_rag_slim/haiku/rag/ingester/queue/repository.py | 6 ++++++ haiku_rag_slim/haiku/rag/ingester/workers/pool.py | 9 ++++++++- 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py index c2801e53..799c0529 100644 --- a/haiku_rag_slim/haiku/rag/ingester/queue/repository.py +++ b/haiku_rag_slim/haiku/rag/ingester/queue/repository.py @@ -66,6 +66,9 @@ class JobRepo: # JobRepo and SyncStateRepo share the same connection, callers must # pass the same lock instance so cross-repo calls also serialize. self._lock = lock or asyncio.Lock() + # Notified after a successful enqueue so workers can wake + # immediately instead of polling on a fixed sleep interval. + self.job_available = asyncio.Condition() async def enqueue( self, @@ -110,6 +113,9 @@ class JobRepo: ) as cursor: row = await cursor.fetchone() await self._conn.commit() + if row is not None: + async with self.job_available: + self.job_available.notify_all() return _row_to_job(row) if row else None async def claim_next(self, worker_id: str) -> Job | None: diff --git a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py index 122866b8..32bb6056 100644 --- a/haiku_rag_slim/haiku/rag/ingester/workers/pool.py +++ b/haiku_rag_slim/haiku/rag/ingester/workers/pool.py @@ -130,7 +130,14 @@ class WorkerPool: continue job = await self._jobs.claim_next(worker_id) if job is None: - await self._sleep_or_stop(self._poll_idle_s) + try: + async with self._jobs.job_available: + await asyncio.wait_for( + self._jobs.job_available.wait(), + timeout=self._poll_idle_s, + ) + except TimeoutError: + pass continue await self._process(job)