Backpressure and Queue Tuning for Sensor Streams

A LiDAR bursting at 20 Hz into a fusion step that takes 80 ms per frame will, without backpressure, grow its intermediate queue until the process is killed — and long before that, latency has blown past anything a controller can use. This task bounds the queue, chooses a drop-oldest policy so perception always sees the freshest scan, and sizes the buffer from measured throughput so end-to-end latency stays under the fusion budget. It is the flow-control layer of async data pipeline architecture, and it protects every consumer downstream in the sensor fusion stack.

A bounded queue with a drop-oldest policy keeps a fast producer from swamping a slow consumer:

Prerequisites #

• Python 3.10+, asyncio or threading (queue.Queue, asyncio.Queue), NumPy 1.24+.
• Input: a producer emitting timestamped sensor frames and a consumer with a measurable per-frame service time.
• Upstream stage: frames carry hardware timestamps for latency measurement, as set up in aligning LiDAR and camera timestamps in ROS.
• Output: a bounded queue with an explicit drop policy and a verified latency profile under load.

Step-by-Step #

1. Bound the queue and pick a drop-oldest policy #

Use a fixed-capacity queue and, when full, evict the head before enqueuing the new frame so the freshest data survives.

import asyncio

class DropOldestQueue:
    def __init__(self, capacity: int):
        self.q = asyncio.Queue(maxsize=capacity)
        self.dropped = 0

    async def put(self, frame):
        if self.q.full():
            _ = self.q.get_nowait()      # evict oldest
            self.dropped += 1
        self.q.put_nowait(frame)

Key parameter: capacity bounds memory and latency; evicting on full() implements drop-oldest so a stale frame never blocks a fresh one. Expected behaviour: dropped counts overload events for monitoring.

2. Size the buffer from measured throughput #

Set capacity from the producer burst and consumer service rate so a normal burst is absorbed without exceeding the latency budget.

def queue_capacity(burst_frames: int, service_hz: float, latency_budget_s: float) -> int:
    """Smallest capacity that absorbs a burst within the latency budget."""
    drain_capacity = int(service_hz * latency_budget_s)   # frames drained in budget
    return max(1, min(burst_frames, drain_capacity))

Key parameters: burst_frames is the measured producer burst; drain_capacity caps the queue so a full queue drains within latency_budget_s. Expected output: an integer capacity.

3. Instrument latency under load #

Stamp each frame on enqueue and measure age on dequeue to see real end-to-end latency, not just queue depth.

import time

async def consume(dq: DropOldestQueue, process):
    while True:
        frame = await dq.q.get()
        latency = time.monotonic() - frame.enqueued_at
        process(frame)                          # your fusion step
        _record_latency(latency)

Key parameter: latency is the true time a frame waited plus processing; track its p99, since a controller cares about the worst case, not the mean.

Verification & Acceptance Criteria #

Load-test the pipeline at the producer's peak rate and confirm latency and drop behaviour.

def assert_flow_control(latencies_ms, dropped, budget_ms, max_drop_frac):
    p99 = float(np.percentile(latencies_ms, 99))
    assert p99 <= budget_ms, f"p99 latency {p99:.1f} ms > {budget_ms}"
    assert dropped_frac(dropped) <= max_drop_frac, "dropping too much data"
    print(f"p99 latency {p99:.1f} ms, drop fraction {dropped_frac(dropped):.2%}")

Acceptance gate: p99 end-to-end latency within the fusion budget; drop fraction below the tolerated ceiling at peak load; and queue memory bounded (never grows unboundedly). A high drop fraction at nominal load means the consumer is genuinely too slow — parallelize it rather than enlarging the queue.

Common Errors & Fixes #

Memory grows until the process is killed. The queue is unbounded. Cap it with maxsize and add the drop-oldest eviction; an unbounded queue only defers the crash.

Latency climbs steadily under sustained load. The consumer is slower than the producer on average — no queue size fixes a throughput deficit. Scale the consumer with the Celery worker pattern or shed load earlier.

Perception acts on stale data. Drop-newest or a block policy preserved old frames. Switch to drop-oldest so the freshest scan is always processed first.

Drops spike in bursts even though average throughput is fine. The queue is too small to absorb the burst. Raise burst_frames in the sizing calc to the measured p99 burst, staying under the drain-capacity cap.

FAQ #

What is backpressure in a sensor pipeline? #

Backpressure is the mechanism by which a slow consumer signals a fast producer to slow down or by which the system decides what to do when it cannot keep up. A LiDAR at 10 to 20 Hz can burst faster than a heavy fusion step can process, and without backpressure the intermediate queue grows until the process runs out of memory. Backpressure bounds that queue and forces an explicit decision — block the producer, or drop data — rather than an implicit crash.

Should I drop oldest or newest under overload? #

For real-time perception, drop the oldest. A stale scan is worthless to a controller that needs the current world state, so discarding the oldest keeps the freshest data flowing and bounds latency. Dropping newest would preserve stale data and grow latency, which is exactly wrong for control. The exception is lossless batch processing, such as offline map building, where you block the producer instead of dropping anything.

How do I size the queue? #

Size it from measurement, not guesswork. Measure the producer's burst size and the consumer's service rate, then set capacity to absorb a typical burst without adding more than the latency budget allows — roughly the burst count, capped so that a full queue drained at the service rate stays under the end-to-end deadline. Too small drops under normal bursts; too large hides a genuine throughput deficit behind growing latency.

Related #

• Building Async Sensor-Fusion Pipelines with Celery — scaling the consumer this queue feeds.
• Handling Coordinate Drift in Multi-Sensor Setups — a downstream consumer sensitive to stale, dropped frames.
• Async Data Pipeline Architecture — the parent workflow on pipeline stages and worker orchestration.

Up one level: Async Data Pipeline Architecture — the parent workflow whose flow-control layer this queue tuning implements.