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.
This commit is contained in:
parent
d5e5733f67
commit
4fa8a012b4
2 changed files with 14 additions and 1 deletions
|
|
@ -66,6 +66,9 @@ class JobRepo:
|
||||||
# JobRepo and SyncStateRepo share the same connection, callers must
|
# JobRepo and SyncStateRepo share the same connection, callers must
|
||||||
# pass the same lock instance so cross-repo calls also serialize.
|
# pass the same lock instance so cross-repo calls also serialize.
|
||||||
self._lock = lock or asyncio.Lock()
|
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(
|
async def enqueue(
|
||||||
self,
|
self,
|
||||||
|
|
@ -110,6 +113,9 @@ class JobRepo:
|
||||||
) as cursor:
|
) as cursor:
|
||||||
row = await cursor.fetchone()
|
row = await cursor.fetchone()
|
||||||
await self._conn.commit()
|
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
|
return _row_to_job(row) if row else None
|
||||||
|
|
||||||
async def claim_next(self, worker_id: str) -> Job | None:
|
async def claim_next(self, worker_id: str) -> Job | None:
|
||||||
|
|
|
||||||
|
|
@ -130,7 +130,14 @@ class WorkerPool:
|
||||||
continue
|
continue
|
||||||
job = await self._jobs.claim_next(worker_id)
|
job = await self._jobs.claim_next(worker_id)
|
||||||
if job is None:
|
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
|
continue
|
||||||
await self._process(job)
|
await self._process(job)
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue