[bugfix]: drain queued jobs greedily when batching delay is 0 (#1453)
With the default batching_delay_ms=0, _collect_batch called wait_for(timeout=0) which timed out before draining the queue, so every dispatched batch had size 1 and dynamic batching silently no-oped. When delay is 0, drain already-queued jobs non-blocking (until empty, max_size, or an incompatible job) instead of waiting.
This commit is contained in:
@@ -87,13 +87,24 @@ class VideoBatchScheduler:
|
||||
deadline = first.enqueue_time + delay_s
|
||||
|
||||
while len(batch) < max_size:
|
||||
timeout = deadline - time.perf_counter()
|
||||
if timeout <= 0 and delay_s > 0:
|
||||
break
|
||||
try:
|
||||
candidate = await asyncio.wait_for(self._get_next_job(), timeout=max(0.0, timeout))
|
||||
except TimeoutError:
|
||||
break
|
||||
if delay_s > 0:
|
||||
timeout = deadline - time.perf_counter()
|
||||
if timeout <= 0:
|
||||
break
|
||||
try:
|
||||
candidate = await asyncio.wait_for(self._get_next_job(), timeout=timeout)
|
||||
except TimeoutError:
|
||||
break
|
||||
else:
|
||||
# delay=0 means "don't wait": greedily drain whatever is
|
||||
# already queued so max_size still coalesces.
|
||||
if self._pending:
|
||||
candidate = self._pending.popleft()
|
||||
else:
|
||||
try:
|
||||
candidate = self._queue.get_nowait()
|
||||
except asyncio.QueueEmpty:
|
||||
break
|
||||
if candidate is None:
|
||||
await self._queue.put(None)
|
||||
break
|
||||
|
||||
Reference in New Issue
Block a user