Backpressure & Bounded Queues Under Load

The Failure Mode: The Unbounded Queue

Scenario: A fast producer (an HTTP handler accepting uploads, a Kafka reader, a socket) pushes work onto a queue. A slow consumer (image resize, a DB write, a downstream API) drains it. The producer is 10× faster than the consumer.

What happens: the queue is unbounded, so every item the consumer can't keep up with just accumulates in RAM. RSS climbs steadily, GC starts thrashing, and after a few minutes the kernel OOM-killer sends SIGKILL. You didn't crash from a bug — you crashed from succeeding at accepting work you had no capacity to do.

# ❌ Anti-pattern: unbounded queue + fast producer + slow consumer = OOM
import queue, threading, time

q = queue.Queue()          # maxsize=0 → UNBOUNDED. This is the bug.

def producer():
    for i in range(10_000_000):
        q.put(fetch_item(i))     # never blocks → memory grows without limit

def consumer():
    while True:
        item = q.get()
        slow_process(item)       # 10x slower than producer → queue only grows

threading.Thread(target=consumer, daemon=True).start()
producer()   # RSS climbs → GC thrash → 💥 OOM kill. The queue ate all your RAM.

The core insight: a queue is just a buffer, and an unbounded buffer converts a throughput mismatch into a memory leak. The queue depth is your unshed backlog. If it can grow forever, so can your memory.

The Fix: Bound the Buffer, Propagate the Pressure

The fix is one word: bound the queue. Give it a maxsize. Now when the consumer falls behind and the queue fills, the producer's put() can no longer succeed instantly — it must block, drop, or reject. That stall is backpressure: the slow consumer's pain is transmitted back up the pipeline to the producer, and (if the producer is itself serving a caller) all the way back to the client.

Backpressure = the queue refusing more work until the consumer catches up. A bounded queue turns "accept infinite work, run out of RAM" into "accept exactly N in-flight items, then push back". Memory is now capped at maxsize × item_size — a number you chose, not the whim of traffic.

Three Policies for a Full Queue

Once the queue is full, you must decide what a producer does. There is no free lunch — pick deliberately.

Policy	Behavior when full	Cost	Use when
Block	Producer waits on `put()` until a slot frees	Latency rises upstream; true end-to-end backpressure	Producer can slow down safely (batch jobs, pipelines)
Drop	Discard oldest (or newest) item, keep going	Data loss — accept only if stale data is worthless	Metrics, live telemetry, video frames (freshness > completeness)
Reject / load-shed	Refuse immediately, return an error (HTTP 503)	Some requests fail fast, but the service stays alive	Request/response servers — fail some to protect the rest

Blocking is backpressure; rejecting is load-shedding; dropping is sacrifice. A server that can't push back on its clients (they'll just retry) usually rejects with 503. A pipeline whose producer you control usually blocks. Never silently grow unbounded — that's the only wrong answer.

In Threading — `queue.Queue(maxsize=N)`

queue.Queue is thread-safe and, with a maxsize, gives you blocking backpressure for free: put() blocks the producer thread when the queue is full.

import queue, threading, time

q = queue.Queue(maxsize=100)     # ✅ BOUNDED — memory capped at ~100 items

def producer():
    for i in range(1_000_000):
        q.put(build_item(i))      # BLOCKS here when full → producer is throttled
    q.put(None)                   # sentinel: tell consumer to stop

def consumer():
    while True:
        item = q.get()
        if item is None:
            break
        slow_process(item)        # ~100ms each
        q.task_done()

t = threading.Thread(target=consumer)
t.start()
producer()                        # naturally paced by the consumer's drain rate
t.join()

Prefer non-blocking policies? Use the _nowait variants to reject or drop instead of blocking:

# Reject policy (load-shed): fail fast instead of waiting
try:
    q.put_nowait(item)            # raises queue.Full immediately if no slot
except queue.Full:
    metrics.increment("shed")     # e.g. return HTTP 503 to the caller

# Drop-oldest policy: make room by discarding the stalest item
try:
    q.put_nowait(item)
except queue.Full:
    try:
        q.get_nowait()            # evict oldest
    except queue.Empty:
        pass
    q.put_nowait(item)            # then insert newest

Bounded put(timeout=...) is the pragmatic middle ground: block, but only for so long. q.put(item, timeout=0.5) waits up to 500ms, then raises queue.Full so you can shed instead of stalling a request thread forever.

In Asyncio — `asyncio.Queue(maxsize=N)` + task caps

asyncio.Queue(maxsize=N) is the async analogue: await q.put(item) suspends the producer coroutine when the queue is full instead of blocking a thread. Same backpressure, no OS threads consumed.

import asyncio

async def producer(q: asyncio.Queue):
    for i in range(1_000_000):
        await q.put(build_item(i))    # SUSPENDS when full → producer paced
    await q.put(None)

async def consumer(q: asyncio.Queue):
    while True:
        item = await q.get()
        if item is None:
            break
        await slow_process(item)
        q.task_done()

async def main():
    q = asyncio.Queue(maxsize=100)    # ✅ BOUNDED
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())

The async-specific trap: a bounded queue caps buffered items, but nothing stops you scheduling millions of coroutines.

asyncio.gather(*(handle(x) for x in
                millions))

creates a Task object per item eagerly — each Task is live memory, and you'll OOM on the tasks themselves before any queue helps. Bound the concurrency too.

# ✅ Bound total in-flight coroutines with a semaphore (this is backpressure on scheduling)
import asyncio

sem = asyncio.Semaphore(100)          # at most 100 coroutines past the gate at once

async def handle(item):
    async with sem:                   # await here IS the backpressure
        await slow_process(item)

async def main(items):
    # still creates Task objects, but the semaphore keeps only 100 actually running.
    # For truly huge streams, feed through a bounded asyncio.Queue + a fixed worker pool instead.
    await asyncio.gather(*(handle(i) for i in items))

Two knobs, both required at scale: a bounded asyncio.Queue caps buffered data, and a bounded worker pool (fixed number of consumer tasks pulling from that queue) caps live coroutines. A semaphore is the quick version of the second knob. See concurrency limits for the full pattern.

In Multiprocessing — `multiprocessing.Queue(maxsize=N)`

Across processes you use multiprocessing.Queue(maxsize=N). It gives the same blocking backpressure — put() blocks the producer process when full — but there are two nuances that bite under load.

import multiprocessing as mp
import time

def producer(q: mp.Queue):
    for i in range(1_000_000):
        q.put(build_item(i))          # BLOCKS the producer process when full
    q.put(None)

def consumer(q: mp.Queue):
    while True:
        item = q.get()
        if item is None:
            break
        slow_process(item)

if __name__ == "__main__":
    q = mp.Queue(maxsize=100)         # ✅ BOUNDED across the process boundary
    c = mp.Process(target=consumer, args=(q,))
    c.start()
    producer(q)
    c.join()

Nuance 1 — pickling: every item crosses the process boundary via pickle. Big or unpicklable objects (open sockets, DB connections, lambdas) either explode the cost of each put() or fail outright. Pass small, plain data (ids, bytes, dataclasses) — not live handles.

Nuance 2 — the feeder thread: mp.Queue.put() doesn't write to the pipe directly. It appends to an in-process buffer that a hidden feeder thread drains into the OS pipe. So put() can return "immediately" while items sit in that buffer — meaning maxsize backpressure is slightly lazy, and a process can exit before the feeder has flushed. Always drain the queue and join() before shutdown, or use q.close(); q.join_thread().

Prefer Pool or JoinableQueue in practice. For most CPU fan-out, mp.Pool(processes=N) already bounds concurrency (see concurrency limits) and hides the queue plumbing. Reach for a raw bounded mp.Queue only when you need an explicit producer/consumer topology.

Backpressure Is End-to-End or It's Nothing

A bounded queue in the middle of your service is useless if the layer in front of it happily accepts unlimited work. Backpressure has to propagate all the way to the source, or the OOM just moves one hop upstream.

Where the pressure surfaces in a real system:

HTTP server accept queue — the OS listen() backlog and your worker/connection cap are backpressure. When workers are all busy, new connections queue in the kernel; when that backlog fills, the OS refuses SYNs and clients get connection errors — the signal to back off.
Streaming (Kafka / gRPC / TCP) — don't ack/commit until you've actually processed. TCP flow control and gRPC's HTTP/2 flow-control windows are transport-level backpressure: a slow reader shrinks the window and the sender naturally throttles.
Message brokers — bounded consumer prefetch + not-committing-early lets the broker hold the backlog durably on disk instead of your process holding it in RAM.

Checklist: ① every queue has a maxsize ② a full queue blocks, drops, or sheds — never grows ③ bound in-flight tasks, not just buffered items (asyncio) ④ don't ack/commit streaming work until it's done ⑤ the accept queue / connection cap is your first backpressure valve. Cap the memory on purpose, or the kernel will cap it for you.