📦 Batching & Request Coalescing

Why Batching Wins Under Load

Nearly every operation has a fixed per-call overhead that is independent of how much data it carries: a network round trip, the TLS/protocol framing, a syscall context switch, a DB transaction begin/commit, an index update, a lock acquisition. At low volume this overhead is invisible. Under load it dominates.

Batch those same rows in groups of 1,000 and you pay the toll 100 times instead of 100,000. The overhead collapses by 1000×. This is the core equation:

# total_cost ≈ num_calls * fixed_overhead  +  total_items * per_item_cost
#
# row-by-row:  100_000 * 2ms   + 100_000 * tiny  ≈ 200s   (overhead-bound)
# batched/1k:  100     * 2ms   + 100_000 * tiny  ≈ 0.2s   (data-bound)  ✅
#
# Batching converts an overhead-bound workload into a data-bound one.

The Throughput ↔ Latency Tradeoff

Batching is not free lunch. To fill a batch you must wait for items to accumulate, and that wait is added latency for the items that arrived first. The two goals fight:

• Bigger batch → higher throughput. Fewer round trips per item, less overhead, more items/sec.
• Bigger batch → worse latency. The first item in a batch of 1,000 waits for the other 999 to show up before anything is sent.

The answer is micro-batching: flush the buffer when it hits max_size OR when a max_delay timer fires — whichever comes first. Under high load the size trigger dominates (batches fill instantly → max throughput). Under low load the timer dominates (a lone item ships after, say, 5ms → bounded latency). One mechanism, self-tuning to traffic.

Example 1: Bulk DB Insert vs Row-by-Row

The most common and highest-ROI batch there is. Never loop execute() in Python when the driver offers executemany or the database offers a bulk path like COPY.

# ❌ Row-by-row: one round trip + commit per row
for row in rows:
    cur.execute("INSERT INTO events (uid, ts, kind) VALUES (%s, %s, %s)", row)
    conn.commit()

# ✅ executemany: one round trip carries the whole batch
cur.executemany(
    "INSERT INTO events (uid, ts, kind) VALUES (%s, %s, %s)",
    rows,                     # a list of tuples — sent as one batched statement
)
conn.commit()                 # one transaction for the whole batch

# ✅✅ COPY: the fastest bulk path in Postgres — bypasses the INSERT machinery
import io, csv
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerows(rows)
buf.seek(0)
cur.copy_expert("COPY events (uid, ts, kind) FROM STDIN WITH CSV", buf)
conn.commit()                 # 10-100x faster than executemany for large loads

Example 2: A Micro-Batcher (Size OR Time)

The reusable pattern: callers submit single items; a batcher accumulates them and flushes on max_size or max_delay, whichever fires first. Below, the same design in both concurrency models from the model page.

In Threading

A background flusher thread owns the buffer behind a lock; producers just append. The flusher wakes on a condition variable with a timeout equal to the remaining window, so it flushes on size (signaled early) or on time (timeout expires).

import threading, time

class MicroBatcher:
    def __init__(self, flush_fn, max_size=100, max_delay=0.05):
        self.flush_fn = flush_fn          # called with a list of items
        self.max_size = max_size
        self.max_delay = max_delay
        self._buf = []
        self._lock = threading.Lock()
        self._cond = threading.Condition(self._lock)
        self._stop = False
        self._worker = threading.Thread(target=self._run, daemon=True)
        self._worker.start()

    def submit(self, item):
        with self._cond:
            self._buf.append(item)
            if len(self._buf) >= self.max_size:
                self._cond.notify()       # size trigger → wake the flusher now

    def _run(self):
        while not self._stop:
            with self._cond:
                # wait until we have items, then only up to the time window
                while not self._buf and not self._stop:
                    self._cond.wait()
                if self._stop:
                    break
                # wait out the remaining window OR until size trigger notifies us
                self._cond.wait(timeout=self.max_delay)
                batch, self._buf = self._buf, []      # swap out under the lock
            if batch:
                self.flush_fn(batch)                  # do the slow call OUTSIDE the lock

    def close(self):
        with self._cond:
            self._stop = True
            self._cond.notify()
        self._worker.join()

# usage
def flush(items):
    print(f"flushing {len(items)} items in one call")

b = MicroBatcher(flush, max_size=100, max_delay=0.05)
for i in range(250):
    b.submit(i)
b.close()   # flushes remaining items on shutdown

In Asyncio

An asyncio.Queue replaces the lock+buffer, and asyncio.wait_for gives the time window. The flusher pulls items until it has max_size or the timeout fires — the async idiom for "size OR time, whichever first".

import asyncio

class AsyncMicroBatcher:
    def __init__(self, flush_fn, max_size=100, max_delay=0.05):
        self.flush_fn = flush_fn          # async callable, takes a list of items
        self.max_size = max_size
        self.max_delay = max_delay
        self._queue: asyncio.Queue = asyncio.Queue()
        self._task = asyncio.create_task(self._run())

    async def submit(self, item):
        await self._queue.put(item)

    async def _collect_batch(self):
        # block for the first item (no busy-wait), then greedily fill up to the window
        batch = [await self._queue.get()]
        deadline = asyncio.get_event_loop().time() + self.max_delay
        while len(batch) < self.max_size:
            remaining = deadline - asyncio.get_event_loop().time()
            if remaining <= 0:
                break                     # time trigger
            try:
                item = await asyncio.wait_for(self._queue.get(), timeout=remaining)
                batch.append(item)        # size trigger continues the loop
            except asyncio.TimeoutError:
                break                     # window elapsed → flush what we have
        return batch

    async def _run(self):
        while True:
            batch = await self._collect_batch()
            await self.flush_fn(batch)    # awaited — the loop stays free meanwhile

async def main():
    async def flush(items):
        print(f"flushing {len(items)} items")
        await asyncio.sleep(0.01)         # stand-in for a batched downstream call

    b = AsyncMicroBatcher(flush, max_size=100, max_delay=0.05)
    await asyncio.gather(*(b.submit(i) for i in range(250)))
    await asyncio.sleep(0.2)              # let the flusher drain

asyncio.run(main())

Example 3: Batching Outbound API Calls

Many APIs expose a bulk endpoint (Elasticsearch _bulk, SQS SendMessageBatch, Stripe/analytics batch ingest). One HTTP round trip carrying 500 items beats 500 round trips — fewer handshakes (see pooling), fewer rate-limit tokens burned.

import httpx, asyncio

async def send_batches(client, events, batch_size=500):
    async def send_one(chunk):
        # ONE request carries up to 500 events instead of 500 separate requests
        r = await client.post("https://api.example.com/v1/events:batch",
                              json={"events": chunk})
        r.raise_for_status()

    chunks = [events[i:i + batch_size] for i in range(0, len(events), batch_size)]
    await asyncio.gather(*(send_one(c) for c in chunks))

async def main(events):
    async with httpx.AsyncClient(timeout=10) as client:
        await send_batches(client, events)

How Big? And When NOT to Batch