Skip to content

Batching Queue Items by Size and Time

Writing one row per insert, sending one metric per request, or calling an API once per item wastes almost all of the work on fixed costs: a round trip, a transaction, a request signature. Batching amortises those costs, but a naive "wait until we have a thousand items" batcher makes a quiet system look broken — the last few items sit in the buffer until enough traffic arrives to flush them. The standard answer is to flush on whichever comes first: a size threshold or a deadline since the batch started. That single rule bounds latency at the deadline and gives full batches under load, and it is what every good client library does internally. This guide builds that batcher on an asyncio.Queue, measures the throughput it buys, adds a partial flush at shutdown so nothing is lost, and handles the awkward part — what to do when a batch fails and only some items are to blame.

Prerequisites

Size or deadline, whichever comes first 3 lanes over time. Size or deadline, whichever comes first busy period i1 i2 i3 full: flush quiet period i1 i2 deadline flush deadline clock started by the first item time → The clock starts with a batch's first item, never while the batcher is idle.

1. Flush on size or deadline, whichever comes first

The batcher waits for the first item without a deadline — an idle system should consume nothing — and only then starts the clock. Subsequent items are collected with a shrinking timeout until the batch is full or the deadline passes.

import asyncio
from collections.abc import Awaitable, Callable


class Batcher:
    def __init__(self, flush: Callable[[list], Awaitable[None]], max_size: int = 100,
                 max_delay: float = 0.05, queue_size: int = 10_000) -> None:
        self.flush = flush
        self.max_size = max_size
        self.max_delay = max_delay
        self.queue: asyncio.Queue = asyncio.Queue(maxsize=queue_size)
        self.batches = 0
        self.items = 0

    async def submit(self, item) -> None:
        await self.queue.put(item)                       # bounded: producers slow down when full

    async def stop(self) -> None:
        await self.queue.put(None)                       # sentinel: flush what is left, then exit

    async def run(self) -> None:
        loop = asyncio.get_running_loop()
        while True:
            first = await self.queue.get()               # no deadline while idle
            if first is None:
                return
            batch = [first]
            deadline = loop.time() + self.max_delay      # the clock starts with the first item
            while len(batch) < self.max_size:
                remaining = deadline - loop.time()
                if remaining <= 0:
                    break
                try:
                    async with asyncio.timeout(remaining):
                        item = await self.queue.get()
                except TimeoutError:
                    break                                 # deadline reached: flush what we have
                if item is None:
                    await self._flush(batch)              # stop requested mid-batch
                    return
                batch.append(item)
            await self._flush(batch)

    async def _flush(self, batch: list) -> None:
        if batch:
            await self.flush(batch)
            self.batches += 1
            self.items += len(batch)

Two properties follow directly. An item never waits longer than max_delay plus the flush duration, because its arrival either starts the clock or joins a batch whose clock is already running. And under sustained load the size threshold is reached before the deadline, so batches are full and the deadline costs nothing.

Verify: submitting a single item to a batcher with max_delay=0.05 flushes it about 50 ms later; submitting a thousand flushes immediately in full batches.

2. Measure what batching buys

Batching is only worth its complexity when the fixed cost per call dominates. Model the downstream as a fixed cost plus a small per-item cost, and measure throughput at several batch sizes.

import asyncio
import time


async def write_batch(rows: list) -> None:
    await asyncio.sleep(0.005 + 0.00002 * len(rows))     # 5 ms fixed + 20 µs per row


async def bench(max_size: int, n: int = 20_000) -> None:
    batcher = Batcher(write_batch, max_size=max_size, max_delay=0.02)
    runner = asyncio.create_task(batcher.run())
    started = time.perf_counter()
    for i in range(n):
        await batcher.submit(i)
        if i % 500 == 0:
            await asyncio.sleep(0)                        # let the batcher run
    await batcher.stop()
    await runner
    elapsed = time.perf_counter() - started
    print(f"max_size={max_size:>4}: {n / elapsed:>8.0f} rows/s in {batcher.batches} batches")


async def main() -> None:
    for size in (1, 10, 100, 1000):
        await bench(size)


asyncio.run(main())

With a 5 ms fixed cost, throughput rose from 165 rows per second unbatched to 1,635 at ten per batch, 13,573 at a hundred, and 35,152 at a thousand — the curve flattens as the per-row cost starts to dominate. Find your own knee: batches beyond it add latency and memory without adding throughput, and very large batches make failures more expensive to handle, as step 4 shows.

Verify: throughput rises steeply at first and then flattens; the batch size at the knee is the one to configure.

Throughput by batch size 4 bars comparing 1 row per call with the others. Throughput by batch size 1 row per call 165 rows/s 10 rows per call 1,635 rows/s 100 rows per call 13,573 rows/s 1000 rows per call 35,152 rows/s Measured with the step 2 benchmark: 20,000 rows, 5 ms fixed cost per call. The curve flattens once per-row cost dominates: that knee is the batch size to pick.

3. Bound latency for the last item and flush on shutdown

Two failure modes remain: an item that arrives just as a batch is flushing, and items still buffered when the process stops. The first is bounded by the deadline; the second needs an explicit flush during shutdown.

import asyncio
import time


async def main() -> None:
    flushed: list[list] = []
    latencies: list[float] = []

    async def record(batch: list) -> None:
        now = time.perf_counter()
        latencies.extend(now - submitted for _, submitted in batch)
        flushed.append([item for item, _ in batch])

    batcher = Batcher(record, max_size=50, max_delay=0.05)
    runner = asyncio.create_task(batcher.run(), name="batcher")

    for i in range(3):                                   # a trickle: three items, far apart
        await batcher.submit((f"row-{i}", time.perf_counter()))
        await asyncio.sleep(0.02)

    await batcher.submit(("row-last", time.perf_counter()))
    await batcher.stop()                                 # shutdown: flush the partial batch
    await runner

    print("batches:", flushed)
    print(f"worst latency: {max(latencies) * 1000:.0f} ms (deadline 50 ms)")


asyncio.run(main())

The trickle produced one batch of three items flushed at the deadline, and the final item was flushed by the stop sentinel rather than waiting for a deadline that would never arrive. Wire stop() into the service's shutdown sequence before the database or HTTP client is closed, as ordered in draining in-flight requests before shutdown.

Verify: the last item appears in a flushed batch, and no observed latency exceeds the deadline plus one flush duration.

4. Handle failures without losing or duplicating the batch

A failed batch is a decision point. Retrying the whole batch is simple but re-applies items that already succeeded unless the write is idempotent; splitting isolates a poison item at the cost of more calls. Choose per operation, and never drop the batch silently.

import asyncio


class PoisonRow(Exception):
    pass


async def strict_write(rows: list[int]) -> None:
    await asyncio.sleep(0.001)
    if any(row == 13 for row in rows):                   # one row the downstream rejects
        raise PoisonRow(f"batch of {len(rows)} contains an invalid row")


async def flush_with_bisect(rows: list[int], dead_letter: list[int], depth: int = 0) -> None:
    try:
        await strict_write(rows)
    except PoisonRow:
        if len(rows) == 1:
            dead_letter.append(rows[0])                  # isolated: park it and move on
            return
        middle = len(rows) // 2
        await flush_with_bisect(rows[:middle], dead_letter, depth + 1)
        await flush_with_bisect(rows[middle:], dead_letter, depth + 1)


async def main() -> None:
    dead_letter: list[int] = []
    batcher = Batcher(lambda rows: flush_with_bisect(rows, dead_letter), max_size=16, max_delay=0.01)
    runner = asyncio.create_task(batcher.run())
    for i in range(16):
        await batcher.submit(i)
    await batcher.stop()
    await runner
    print("dead-lettered rows:", dead_letter, "| batches attempted:", batcher.batches)


asyncio.run(main())

Bisecting isolated row 13 in four extra calls instead of failing sixteen rows or retrying them blindly. For writes that are not idempotent, pair retries with idempotency keys so a partially applied batch cannot double-apply, and route permanently failing rows to a dead-letter queue.

Verify: the dead-letter list contains only row 13, and every other row was written.

5. Export the numbers that tune the batcher

Four series say whether the configuration fits the traffic: batch size distribution, flush duration, queue depth, and the share of flushes triggered by the deadline rather than by size.

import asyncio
import statistics
import time


class ObservedBatcher(Batcher):
    def __init__(self, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.sizes: list[int] = []
        self.flush_seconds: list[float] = []
        self.deadline_flushes = 0

    async def _flush(self, batch: list) -> None:
        if not batch:
            return
        if len(batch) < self.max_size:
            self.deadline_flushes += 1                   # flushed early: traffic is below capacity
        started = time.perf_counter()
        await super()._flush(batch)
        self.flush_seconds.append(time.perf_counter() - started)
        self.sizes.append(len(batch))

    def report(self) -> dict:
        return {
            "batches": self.batches,
            "median_size": statistics.median(self.sizes) if self.sizes else 0,
            "deadline_flush_ratio": round(self.deadline_flushes / max(1, self.batches), 2),
            "median_flush_ms": round(statistics.median(self.flush_seconds) * 1000, 2),
            "queue_depth": self.queue.qsize(),
        }


async def main() -> None:
    batcher = ObservedBatcher(write_batch, max_size=100, max_delay=0.02)
    runner = asyncio.create_task(batcher.run())
    for i in range(500):
        await batcher.submit(i)
        if i % 100 == 0:
            await asyncio.sleep(0)
    await batcher.stop()
    await runner
    print(batcher.report())


asyncio.run(main())

Read them together. A deadline_flush_ratio near one means traffic never fills a batch — lower max_size or accept that batching is not helping. A ratio near zero with a growing queue depth means the batcher is saturated: raise max_size, add batcher instances, or slow the producers. Rising flush duration with stable sizes points at the downstream, not at the batcher.

Verify: under steady load the median size approaches max_size and the deadline ratio is low; with a trickle the ratio approaches one.

A batch write failed — now what? A decision on Why did the batch fail with 3 outcomes. A batch write failed — now what? Why did the batch fail? transient, write idempotent retry the batch with backoff transient, not idempotent retry with keys server deduplicates one bad item bisect to isolate dead-letter the item Dropping the batch is never on the list.

Verification

The batcher is configured correctly when:

  • Latency is bounded: no item waits longer than max_delay plus one flush, measured end to end.
  • Throughput sits at the knee: larger batches no longer improve rows per second meaningfully.
  • Nothing is lost at shutdown: stopping flushes the partial batch before clients are closed.
  • Failures are contained: a bad item is isolated or dead-lettered without discarding its batch.
  • Metrics guide tuning: batch sizes, deadline-flush ratio, flush duration and queue depth are exported.

Pitfalls & edge cases

  • Starting the deadline when the batcher starts. The clock must start with the first item of a batch, or a busy batcher flushes tiny batches on a fixed cadence.
  • Unbounded intake. An unbounded queue in front of a slow downstream turns a throughput problem into a memory problem; bound it and let producers wait.
  • Batches too large for the downstream. Databases and APIs have statement, payload and timeout limits; cap batch size by bytes as well as by count where payloads vary.
  • Blocking flushes. A synchronous driver call inside flush stalls the loop; offload it or use an async client.
  • Ordering assumptions. Items within a batch keep their order, but a retried or bisected batch can reorder relative to later batches. If order matters per key, batch per key or use per-key lanes.

Frequently Asked Questions

How do I batch asyncio queue items without adding latency?

Flush on whichever comes first: a maximum batch size or a deadline measured from the first item in the batch. Wait indefinitely for that first item so an idle system does nothing, then collect further items with a shrinking timeout. Latency is bounded by the deadline, and busy periods still produce full batches.

What batch size should I use?

Measure throughput against batch size for your downstream and pick the knee of the curve. With a 5 ms fixed cost per call, our benchmark rose from about 165 rows per second unbatched to 13,573 at 100 rows and 35,152 at 1,000, with diminishing returns as per-row cost takes over. Larger batches also make failures more expensive.

How do I make sure buffered items are not lost on shutdown?

Send an explicit stop sentinel through the queue so the batcher flushes the partial batch and exits, and await the batcher task during shutdown before closing database or HTTP clients. Relying on cancellation alone discards whatever is buffered.

What should happen when a batch write fails?

Decide per operation. If the write is idempotent, retry the whole batch with backoff. If one bad item fails the batch, bisect the batch to isolate the offending item and dead-letter it. Never drop the batch silently, and use idempotency keys when retrying non-idempotent writes.