From 522b9f7e479a4153e0c2d64d88a2b34bc94bd22f Mon Sep 17 00:00:00 2001 From: SolitaryThinker Date: Sun, 5 Jul 2026 13:45:21 -0700 Subject: [PATCH] [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. --- fastvideo/entrypoints/openai/batching.py | 25 +++++++++++++++++------- 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/fastvideo/entrypoints/openai/batching.py b/fastvideo/entrypoints/openai/batching.py index a506aecba..bd4d1b70e 100644 --- a/fastvideo/entrypoints/openai/batching.py +++ b/fastvideo/entrypoints/openai/batching.py @@ -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