TL;DR: Each caller submits an item and awaits a future. The batcher appends to a shared queue and flushes on either of two triggers: the queue reaching the maximum size, or a timer that started when the first item arrived reaching the maximum wait. Flushing calls the handler once with the whole batch and resolves each caller's future with its result. The bug that appears in almost every first implementation is in the timer path: the flush routine cancels the pending timer task, and when the flush is invoked from inside that same timer task it cancels itself, so the handler never completes and every caller waits forever. The fix is to check whether the task being cancelled is the current one. The cost of the design is latency: a request arriving at an empty queue waits the full window, so a 50 millisecond window adds up to 50 milliseconds to a request that would otherwise have gone immediately, which is the trade against the throughput a larger batch buys.
How to approach it
Describe the two triggers and the per-caller future, since that structure is the answer. Then write it, and be honest about the timer deadlock, because it is the one tricky part and an interviewer who has implemented this will be waiting for it. Then the latency cost, which is what makes the parameters a decision rather than defaults. Close with the error path.
A strong answer
A typical situation: a serving proxy batches inference requests. Under load it works well. Under light load, requests occasionally hang forever, and the pattern is that it happens when a batch is flushed by the timer rather than by reaching its size. The timer task cancels itself partway through the flush.
The implementation:
import asyncio
class Batcher:
"""Flush when the batch is full or when max_wait elapses, whichever comes first."""
def __init__(self, handler, max_size=8, max_wait=0.05):
self.handler, self.max_size, self.max_wait = handler, max_size, max_wait
self._queue = [] # list of (item, future)
self._flush_task = None # the pending timer, if any
self._lock = asyncio.Lock()
self._running = set() # in-flight handler tasks, kept from GC
async def submit(self, item):
fut = asyncio.get_running_loop().create_future()
async with self._lock:
self._queue.append((item, fut))
if len(self._queue) >= self.max_size:
self._dispatch(self._detach_locked())
elif self._flush_task is None:
self._flush_task = asyncio.create_task(self._timer())
return await fut # the caller waits outside the lock
async def _timer(self):
await asyncio.sleep(self.max_wait)
async with self._lock:
self._dispatch(self._detach_locked())
def _detach_locked(self):
"""Take the pending batch and cancel the timer. Cheap, synchronous, and
the ONLY thing that happens under the lock."""
if not self._queue:
return None
batch, self._queue = self._queue, []
# take the reference and clear it BEFORE cancelling, and never cancel the task we
# are currently running inside: that is the self-cancellation deadlock
task, self._flush_task = self._flush_task, None
if task is not None and task is not asyncio.current_task():
task.cancel()
return batch
def _dispatch(self, batch):
"""Run the handler in its own task. The batch's fate must not depend on
whichever caller happened to fill it: if that caller is cancelled while
awaiting the handler, its peers would otherwise wait forever."""
if not batch:
return
t = asyncio.create_task(self._run(batch))
self._running.add(t)
t.add_done_callback(self._running.discard)
async def _run(self, batch):
items = [i for i, _ in batch]
try:
results = await self.handler(items)
except asyncio.CancelledError as exc:
# NOT caught by `except Exception`: CancelledError derives from
# BaseException. This is the path that used to strand a batch.
self._settle(batch, exc=exc)
raise
except Exception as exc:
self._settle(batch, exc=exc) # every caller in the batch gets the failure
return
if len(results) != len(batch):
self._settle(batch, exc=RuntimeError(
f"handler returned {len(results)} results for {len(batch)} items"))
return
for (_, fut), r in zip(batch, results):
if not fut.done():
fut.set_result(r)
@staticmethod
def _settle(batch, exc):
for _, fut in batch:
if not fut.done():
fut.set_exception(exc)
Running it:
10 submissions with max_size 4 -> [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
the last two flushed on the timer
a single submission -> flushed on the timer, returns
handler raising -> every caller in the batch receives RuntimeError
caller cancelled mid-flush -> its PEERS still resolve
a third submit during a flush -> enqueues immediately, timer starts
handler returns the wrong count -> every caller gets a RuntimeError, none hang
The last three lines are the ones worth writing a test for, and they are where a first version of this usually breaks.
The bug and its fix, stated plainly because it is the point of the question:
the first version I wrote
if self._flush_task is not None:
self._flush_task.cancel()
self._flush_task = None
what happens when the timer fires
_timer awakens, takes the lock, calls _flush_locked
_flush_locked cancels self._flush_task, which IS the currently running task
cancellation raises CancelledError at the next await, which is `await self.handler(items)`
the handler never completes, no future is resolved, and every caller in that batch waits
forever
why it hides
it only occurs when the flush is triggered by the timer, so a test that submits enough items
to fill a batch never sees it
under load, batches usually fill before the timer, so the bug appears only at low traffic
the fix
task, self._flush_task = self._flush_task, None
if task is not None and task is not asyncio.current_task():
task.cancel()
clearing the reference first also prevents a second flush from cancelling a task that has
already completed
Batching Queues and Backpressure covers the pattern; Continuous Batching is the inference-specific version where the batch is re-formed every step rather than once.
The latency the window costs:
a request arriving at an empty queue waits until either
max_size - 1 more requests arrive, or
max_wait elapses
worst case added latency = max_wait, paid by a request that arrives alone
expected added latency at arrival rate r, batch size n:
time to fill = (n - 1) / r
if (n - 1) / r < max_wait, the batch fills first and the added latency is that fill time
otherwise the timer fires and the added latency is max_wait
worked
max_size 8, max_wait 50 ms
at 200 requests/s: fill time = 7 / 200 = 35 ms, so batches fill first and add 35 ms
at 20 requests/s: fill time = 7 / 20 = 350 ms > 50 ms, so the timer fires and adds 50 ms
with an average batch of about 1 + 20 x 0.05 = 2 items
sanity: at low rates the batcher adds the full window and delivers batches of two, which may
be worse than not batching at all. The parameters must be set from the actual arrival
rate, and a batcher that helps at peak can hurt at trough
The error path:
the choice: does one failure fail the whole batch, or just its own item?
if the handler raises, it usually failed for the batch, so failing every caller is correct
if the handler returns per-item results including per-item errors, resolve each future with
its own outcome, so one bad item does not fail nine good ones
the second shape is better and requires the handler's contract to support it, which is a
design decision to make explicitly
what must not happen
a caller whose future is never resolved. Every exit path from _run must settle every
future in the batch, which is why _settle loops over the whole batch and why the
`if not fut.done()` guard is there
the three exits people miss:
CancelledError, which derives from BaseException and so slips past `except Exception`
a handler returning a different number of results than it was given, where zip()
silently drops the surplus futures and they wait forever
a batch whose flushing caller is cancelled, which is why the handler runs in its own
task rather than inside whichever submit() happened to fill the batch
The reversal condition: batching is worth it only when the handler's cost is dominated by a per-call overhead that the batch amortizes. If the handler's cost is proportional to the number of items, batching adds latency and buys nothing. For inference the per-call overhead is enormous, since a forward pass at batch 1 and batch 8 read the same weights, which is why it pays there. For a handler that is a simple database lookup per item, it usually does not, and the first question is whether the downstream cost is per call or per item.
What interviewers probe next
- "Why start the timer on the first item rather than on every item?" Because restarting it per item means a steady stream of arrivals never triggers the timer, and a batch could wait indefinitely.
- "What if the handler is slow?" Items arriving during a flush go into the new queue and start their own timer, so the batcher pipelines. Bounding the queue is a separate backpressure decision.
- "How would you add backpressure?" Cap the queue and reject or block on submit when full, since an unbounded queue converts a throughput problem into an out-of-memory failure.
- "How do you test the timer path?" Drive the event loop with a controllable clock, or submit fewer items than max_size and assert the flush happens within a tolerance of max_wait. The second is what caught the deadlock.
Common mistakes
- Cancelling the timer task from inside itself, which hangs every caller in that batch and only at low load.
- Restarting the timer on every submission, so a steady stream never flushes on time.
- Leaving a future unresolved on an error path, which hangs one caller silently.
- Choosing max_wait without reference to the arrival rate, so the batcher adds its full window and delivers batches of two.
Key takeaways
- Two triggers, one shared queue, one future per caller; the caller awaits outside the lock.
- The timer must not cancel itself: compare against the current task before cancelling, and clear the reference first.
- Added latency is max_wait in the worst case; at 20 requests per second with an 8-item batch the timer always wins and the batch averages two.
- Every exit path must resolve every future in the batch, and whether one failure fails all of them is a contract decision.