Backpressure between the GNSS reader and the spatial worker

The reader cannot stop. A UART delivering NMEA has a FIFO of 16 to 64 bytes and no flow control, so a reader that pauses loses sentences, and a lost sentence is a lost fix that no retry recovers. The spatial worker, meanwhile, can absolutely fall behind — a zone update triples its per-fix cost and it does. Between those two facts sits a queue, and how that queue behaves when it fills is the whole subject of this page. It belongs to async execution for spatial workloads in the Local Spatial Processing Patterns guide.

Why classic backpressure does not apply #

In a network service, backpressure means telling the producer to slow down, and the producer obliges. Here the producer is a satellite constellation via a serial port, and it does not take instruction. The three options available are therefore different from the textbook ones.

Shed at the queue. Accept the fix, decide it will not be processed, and drop it deliberately with a counter. The reader keeps draining the UART, the queue stays bounded, and the loss is recorded.

Degrade the work. Reduce the per-fix cost so the worker catches up: skip the exact predicate and use the envelope result, widen the prefilter, or drop to a coarser zone set. The fix is still processed, less precisely.

Sample at the source. Reconfigure the receiver to a lower fix rate. This is the only option that reduces the input rather than discarding it, and it is the slowest to take effect — a receiver reconfiguration takes seconds — so it is a sustained-overload response rather than a burst response.

The implementation #

# ingest.py — reader, bounded queue and load-shedding controller.
# The reader task never awaits anything but the serial read; the queue is the
# only place a decision is made. Threading: one reader task, one controller
# task, N worker processes behind an executor.
import asyncio
import time

QUEUE_MAX = 64                 # ≈64 s of 1 Hz fixes
DEGRADE_AT = 0.70              # fraction full
SUSTAINED_S = 30.0             # before touching the receiver


class IngestQueue:
    """A bounded queue that sheds rather than blocking, and reports why."""

    __slots__ = ("q", "dropped", "degraded", "high_since")

    def __init__(self):
        self.q: asyncio.Queue = asyncio.Queue(maxsize=QUEUE_MAX)
        self.dropped = 0
        self.degraded = 0
        self.high_since = None

    def offer(self, fix) -> bool:
        """Non-blocking. Returns False when the fix was shed.
        Called from the reader, which must never await on a full queue."""
        try:
            self.q.put_nowait(fix)
            return True
        except asyncio.QueueFull:
            self.dropped += 1
            return False

    def pressure(self) -> float:
        return self.q.qsize() / QUEUE_MAX

    def mode(self) -> str:
        """Full pipeline, degraded, or overloaded — evaluated per fix by the
        worker so the response is immediate rather than waiting for a tick."""
        p = self.pressure()
        now = time.monotonic()
        if p >= DEGRADE_AT:
            if self.high_since is None:
                self.high_since = now
            if now - self.high_since >= SUSTAINED_S:
                return "overloaded"
            return "degraded"
        self.high_since = None
        return "full"


async def reader_task(serial, parser, queue: IngestQueue):
    """Drains the UART unconditionally. The only place a fix is created and
    the only task that must never be delayed."""
    while True:
        line = await serial.readline()          # never gated on the queue
        fix = parser.parse(line)
        if fix is None:
            continue
        if not queue.offer(fix):
            # Shed. Nothing else to do here: logging per drop would itself
            # become the bottleneck, so the counter is the record.
            pass


async def worker_task(queue: IngestQueue, pipeline, receiver):
    """Consumes fixes, adapting the work to the queue's pressure."""
    while True:
        fix = await queue.q.get()
        mode = queue.mode()
        if mode == "full":
            await pipeline.process(fix, exact=True)
        elif mode == "degraded":
            queue.degraded += 1
            await pipeline.process(fix, exact=False)      # envelope verdict only
        else:
            queue.degraded += 1
            await pipeline.process(fix, exact=False)
            # Sustained: reduce the input rather than the output.
            await receiver.set_rate_hz(0.2)
            queue.high_since = None                       # give it time to settle
        queue.q.task_done()

The controller evaluates the mode per fix rather than on a timer. That matters because the pressure changes on the timescale of a burst — a fix that arrives while the queue is 90% full should be treated differently from one that arrived a second earlier at 40%, and a one-second polling controller would treat them the same.

Constraint validation #

Gotchas and edge cases #

• A queue that never fills is not proof of health. It may mean the reader is not reading. Export the enqueue rate as well as the depth: a flat zero depth with a zero enqueue rate is a dead receiver, and the two look identical if only depth is watched.
• Degraded verdicts must be labelled. An envelope-only containment result is a maybe, not a yes. Emitting it as an ordinary verdict pollutes the event stream with false positives that nobody can later separate out. Carry a precision: envelope flag on the result.
• Do not reconfigure the receiver on a burst. A cold start, a batch of buffered sentences after a reconnect, or a one-off zone rebuild all produce brief spikes. The sustained-condition timer is what stops the device from permanently lowering its fix rate because of a thirty-second event.
• Shedding oldest is usually better than shedding newest. A full queue means the worker is behind; the freshest fix is the one worth processing. asyncio.Queue sheds newest by refusing the put — replacing it with a deque that pops the oldest on overflow is a few lines and usually the better policy for position data.
• The queue is not the spool. A shed fix is gone; a spooled fix is delayed. If the deployment cannot tolerate losing fixes at all, the queue must feed the spool before the worker, so overload delays processing rather than discarding input — at the cost of a spool that grows during the overload.

What to measure #

Three numbers describe this system completely: enqueue rate, queue depth as a fraction, and the split between full, degraded and shed outcomes. Together they distinguish every failure this component has. A rising depth with a constant enqueue rate is a worker that got slower — usually a zone-layer change. A falling enqueue rate with an empty queue is a receiver problem. A shed counter that is anything other than zero means the degradation ladder did not act fast enough, which is a tuning question about the thresholds rather than about the pipeline.

Alert on the shed counter, not on the depth. Depth crossing a threshold is the system working as designed; a shed fix is the system having run out of options, and it is the only one of the three that represents data that no longer exists.

Sizing the queue #

The queue length is a latency budget, not a memory decision. Each entry is a few dozen bytes, so memory is irrelevant at any sane size; what matters is how stale the oldest entry is allowed to be.

At 1 Hz, a 64-entry queue that is full holds a fix from 64 seconds ago. If the worker’s output drives a geofence decision, acting on a 64-second-old position means the asset may be a kilometre away from where the verdict claims. That is the real constraint: the queue should be no longer than the time after which a verdict stops being useful.

For real-time interlocks that is a handful of seconds. For asset tracking with a five-minute reporting cadence it can be minutes. For a duty-cycled node processing a burst after a wake, the queue can be as long as the burst, because none of it is real-time by definition.

Deriving it that way usually produces a shorter queue than intuition suggests, and a shorter queue is better: it sheds sooner, which is a visible, counted event, instead of quietly delivering stale verdicts that nobody can distinguish from fresh ones.

One further consequence is worth stating: because the queue length is a latency budget, it should be derived per consumer rather than per device. A gateway that drives both an interlock and a reporting stream needs two queues, not one — a three-entry queue in front of the interlock path and a ninety-entry queue in front of the reporting path, both fed by the same reader. Sharing one queue forces the looser consumer’s tolerance onto the stricter one, and the interlock silently starts acting on positions that are a minute old whenever the reporting worker falls behind.

Related #

• Async Execution for Spatial Workloads — the pool this queue feeds and its sizing.
• Asyncio cancellation and timeouts on low-RAM targets — bounding the work a single fix can consume.
• On-Device Geometry Filtering — the exact-versus-envelope distinction degraded mode relies on.