Bounding asyncio Queues for Sensor Consumers #

asyncio.Queue() with no arguments is unbounded, and that default has killed more long-running sensor consumers than any algorithmic mistake. It works perfectly in testing, where the sink is fast and the feed is a fixture, and it fails in production the first time a database failover makes the consumer slower than the producer for ninety seconds — at which point the queue grows until the kernel kills the process, taking everything in it. This guide sizes the queue deliberately, instruments it, and handles the two moments that matter: what happens when it is full, and what happens on shutdown. It is the in-process half of backpressure handling in Python IoT streams.

A Bound Converts an Invisible Failure into a Visible One #

An unbounded queue does not remove backpressure; it hides it. The producer never waits, so the consumer’s slowness never propagates anywhere it could be noticed — no metric moves, no call blocks, no error is raised. The only symptom is resident memory, and by the time anyone looks at that graph the process has already been killed and restarted, losing everything queued.

Bounding the queue makes the same overload observable in three ways at once. queue.qsize() approaches maxsize, which is a gauge you can alert on. await queue.put(...) starts taking measurable time, which shows up in the producer’s stage latency. And if you choose to drop instead of block, the drop counter increments with a reason — the observability signals that turn “the data looks wrong” into “we shed 4 200 readings between 02:10 and 02:14”.

The bound also caps the blast radius. Two thousand readings lost to a SIGKILL is an incident; four million is a reporting period.

Production-Ready Implementation #

# python 3.11 · prometheus-client==0.20.0
from __future__ import annotations

import asyncio
import contextlib
from dataclasses import dataclass

from prometheus_client import Counter, Gauge

QUEUE_DEPTH = Gauge("ingest_queue_depth", "Readings waiting in the ingest queue")
QUEUE_FULL_WAITS = Counter("ingest_queue_full_waits_total", "Times the producer had to wait")
DROPPED = Counter("ingest_records_dropped_total", "Readings shed under pressure", ["reason"])


@dataclass
class BoundedIngest:
    """A bounded producer/consumer pair with an explicit overload policy.

    maxsize is chosen from the longest sink stall you intend to absorb, not from
    available memory: a queue that can hold ten minutes of traffic simply delays
    the moment anyone notices the sink is down.
    """

    maxsize: int = 2000
    drop_when_full: bool = False          # block by default; dropping must be deliberate

    def __post_init__(self) -> None:
        self.queue: asyncio.Queue = asyncio.Queue(maxsize=self.maxsize)

    async def put(self, reading: dict) -> bool:
        """Enqueue one reading. Returns False if it was shed."""
        if self.drop_when_full:
            try:
                self.queue.put_nowait(reading)
            except asyncio.QueueFull:
                DROPPED.labels(reason="queue_full").inc()
                return False
        else:
            if self.queue.full():
                QUEUE_FULL_WAITS.inc()     # count the wait, then actually wait
            await self.queue.put(reading)
        QUEUE_DEPTH.set(self.queue.qsize())
        return True

    async def consume(self, sink, batch_size: int = 200, linger_s: float = 0.5) -> None:
        """Drain the queue in batches until cancelled, flushing partial batches."""
        batch: list[dict] = []
        while True:
            try:
                timeout = linger_s if batch else None
                reading = await asyncio.wait_for(self.queue.get(), timeout=timeout)
                batch.append(reading)
                self.queue.task_done()
            except asyncio.TimeoutError:
                pass                        # linger expired: flush what we have
            except asyncio.CancelledError:
                if batch:
                    await sink.write_batch(batch)
                raise

            if len(batch) >= batch_size or (batch and self.queue.empty()):
                await sink.write_batch(batch)
                batch = []
                QUEUE_DEPTH.set(self.queue.qsize())

Two details in consume earn their place. The linger_s timeout means a partial batch is flushed during quiet periods instead of waiting indefinitely for batch_size readings that will not arrive until morning. And the CancelledError handler flushes before re-raising, so shutdown does not discard the batch already assembled.

Shutdown ordering is the other half, and it is where most implementations lose data:

async def run(source, sink, ingest: BoundedIngest) -> None:
    """Producer, consumer and an ordered shutdown that drains rather than discards."""
    consumer = asyncio.create_task(ingest.consume(sink))
    try:
        async for reading in source:            # producer stops first, on cancellation
            await ingest.put(reading)
    finally:
        await ingest.queue.join()               # let the consumer finish the backlog
        consumer.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await consumer

Cancel the consumer before joining the queue and everything still queued is gone. The order above — stop producing, join, then cancel — is the whole difference between a clean deploy and a two-minute data gap on every restart.

Parameter Tuning Guide #

Assume roughly 400 bytes per decoded reading as a dict; measure your own with sys.getsizeof plus the payload, because a reading carrying a registry-enriched geometry is several times that.

Verification and Testing #

The two tests that matter are the full-queue behaviour and the shutdown drain.

# python 3.11 · pytest==8.2.0 · pytest-asyncio==0.23.7
import asyncio
import pytest


class SlowSink:
    def __init__(self, delay: float = 0.05):
        self.delay, self.written = delay, []

    async def write_batch(self, rows):
        await asyncio.sleep(self.delay)
        self.written.extend(rows)


@pytest.mark.asyncio
async def test_producer_blocks_rather_than_growing_the_queue():
    ingest = BoundedIngest(maxsize=10, drop_when_full=False)
    for i in range(10):
        await ingest.put({"i": i})

    with pytest.raises(asyncio.TimeoutError):          # the 11th put must not complete
        await asyncio.wait_for(ingest.put({"i": 10}), timeout=0.2)
    assert ingest.queue.qsize() == 10                  # bound held


@pytest.mark.asyncio
async def test_shutdown_drains_the_queue():
    ingest, sink = BoundedIngest(maxsize=100), SlowSink(delay=0.01)
    consumer = asyncio.create_task(ingest.consume(sink, batch_size=10, linger_s=0.05))

    for i in range(35):
        await ingest.put({"i": i})

    await ingest.queue.join()
    consumer.cancel()
    with pytest.raises(asyncio.CancelledError):
        await consumer

    assert len(sink.written) == 35                     # nothing discarded at shutdown

In production, alert on ingest_queue_depth / maxsize above 0.8 sustained for a minute. That threshold fires before the queue is full, while there is still room to act, and it will not fire on the normal sawtooth of a healthy batching consumer.

Gotchas #

asyncio.Queue() with no maxsize. The default is zero, which means unbounded. It reads like “empty”, and it is the bug this page exists for.

put_nowait in a blocking design. Mixing the two produces a queue that silently drops under load while the code around it is written as though it blocks. Choose one policy per queue and make it a constructor argument, not a call-site decision.

Forgetting task_done(). queue.join() waits for a task_done() per put, so a consumer that omits it makes shutdown hang forever — which usually gets “fixed” by removing the join, which reintroduces the data loss.

One queue shared by producers with different priorities. A backfill replay and a live feed sharing a queue means the backfill fills it and the live path blocks behind hours of history. Use separate queues with separate bounds, as the backlog replay guide describes.

FAQ #

Related #

• Backpressure Handling in Python IoT Streams — the backpressure stage this queue implements
• Pausing and Resuming Kafka Consumers for Slow Sinks — applying the same pressure one layer further upstream
• Managing Python Memory Limits for Continuous Sensor Streams — why an unbounded queue is the classic long-running-process leak