Cap claimed-row count at max_concurren
This commit is contained in:
parent
87af5e5139
commit
6f74fe67b3
2 changed files with 11 additions and 5 deletions
|
|
@ -186,6 +186,11 @@ once `claim_timeout_s` elapses.
|
||||||
- `claim_timeout_s` must exceed the longest legitimate job duration; a
|
- `claim_timeout_s` must exceed the longest legitimate job duration; a
|
||||||
shorter value lets the reaper resurrect in-flight jobs.
|
shorter value lets the reaper resurrect in-flight jobs.
|
||||||
- `worker_count <= max_concurrent`; extras stall in the semaphore.
|
- `worker_count <= max_concurrent`; extras stall in the semaphore.
|
||||||
|
- `max_concurrent` should match downstream capacity. docling-serve
|
||||||
|
processes one task per instance, so `max_concurrent` above the number
|
||||||
|
of `providers.docling_serve.base_url` entries over-subscribes the
|
||||||
|
fleet — extra submissions queue inside docling-serve and inflate
|
||||||
|
`claimed_at` duration toward `claim_timeout_s`.
|
||||||
- `poll_idle_interval_s`: lower = faster pickup, more SQLite churn.
|
- `poll_idle_interval_s`: lower = faster pickup, more SQLite churn.
|
||||||
- `reaper_interval_s`: worst-case post-crash reclaim is
|
- `reaper_interval_s`: worst-case post-crash reclaim is
|
||||||
`claim_timeout_s + reaper_interval_s`.
|
`claim_timeout_s + reaper_interval_s`.
|
||||||
|
|
|
||||||
|
|
@ -132,12 +132,13 @@ class WorkerPool:
|
||||||
if self._breaker.is_open:
|
if self._breaker.is_open:
|
||||||
await self._sleep_or_stop(self._poll_idle_s)
|
await self._sleep_or_stop(self._poll_idle_s)
|
||||||
continue
|
continue
|
||||||
job = await self._jobs.claim_next(worker_id)
|
# Semaphore wraps claim + process: at most max_concurrent workers
|
||||||
if job is None:
|
# hold a claimed job at any one time.
|
||||||
await self._sleep_or_stop(self._poll_idle_s)
|
|
||||||
continue
|
|
||||||
|
|
||||||
async with self._semaphore:
|
async with self._semaphore:
|
||||||
|
job = await self._jobs.claim_next(worker_id)
|
||||||
|
if job is None:
|
||||||
|
await self._sleep_or_stop(self._poll_idle_s)
|
||||||
|
continue
|
||||||
await self._process(job)
|
await self._process(job)
|
||||||
|
|
||||||
async def _reaper_loop(self) -> None:
|
async def _reaper_loop(self) -> None:
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue