Prioritizing Safety Commands in Async Instrument Queues

When a source-measure unit is sinking 3 A into a device under test, an emergency OUTP OFF cannot wait behind two hundred queued MEAS:VOLT? polls. This guide adds a preemption tier to the async command layer surveyed under Async Command Queuing Systems: safety-critical commands — E-stop, output-disable, interlock reset, *RST — jump ahead of routine traffic with a provable dispatch-latency bound, while aging keeps ordinary polls from starving. It is the safety complement to the FIFO worker in Building async command queues with asyncio for lab devices.

Where Safety-Priority Dispatch Applies

The scope is a single instrument on one non-multiplexing transport — a pyserial port, a PyVISA session, or a GPIB address — where the FIFO guarantee of an ordinary queue becomes a liability the moment an operator or an interlock demands an immediate abort. Plain FIFO ordering means a safety command enqueued during a saturated poll loop inherits the full backlog’s latency; on a 115200-baud link with a 60-deep poll queue at ~12 ms per transaction, that is over 700 ms of avoidable delay before the output actually drops. For a supply near compliance, a thermal chamber past its setpoint, or a laser controller with an open interlock, that window is the difference between a scrubbed run and damaged hardware.

Two invariants from the FIFO design still hold and constrain everything here. First, exactly one coroutine owns the transport — priority scheduling changes which command the sole writer sends next, never how many writers exist, so the one-writer-per-UART rule is preserved. Second, bytes are indivisible: once a write has begun, its matching read must complete before another command’s bytes touch the wire, so preemption operates between transactions, never mid-frame. A safety command therefore preempts the queue, not the transaction in flight. Transport line discipline — baud, flow control, latency timer — must already be settled per PySerial Configuration & Tuning before any of this scheduling matters.

Priority Levels, Aging, and the Worst-Case Latency Bound

Every command carries a discrete priority. Four levels suffice for real rigs: SAFETY (E-stop, output-disable, interlock, *RST), CONTROL (setpoint changes, ranging), QUERY (measurements a caller awaits), and POLL (background telemetry). An asyncio.PriorityQueue orders by a tuple whose first element is the priority, so SAFETY is always dequeued ahead of anything lower regardless of arrival order.

Static priority alone starves the bottom tier: under a sustained QUERY load, POLL commands never reach the head. The fix is aging — an item’s effective priority is boosted as it waits, so a long-waiting POLL eventually competes with fresh QUERY traffic. The effective key applied at insertion combines the base level with an age discount bounded by a floor so a poll can never age above the safety tier:

$$p_{\text{eff}}(c)=\max (p_{\text{floor}},p_{\text{base}}(c)-\lfloor \frac{t_{\text{now}}-t_{\text{enq}}(c)}{{\tau }_{\text{age}}}\rfloor )$$

where p_base is the numeric level (lower is more urgent), τ_age is the seconds-per-priority-step, and p_floor is one step above SAFETY so no amount of waiting lets routine traffic impersonate an E-stop. This makes starvation impossible: a POLL enqueued at level 3 with τ_age = 2 s reaches level 1 within 4 s, guaranteeing bounded wait.

The dispatch latency of a safety command is what must be provable. Because preemption happens only at a transaction boundary, the worst case is one in-flight transaction plus every already-queued item of equal-or-higher priority ahead of it:

$$L_{\text{safety}}\le t_{\text{current\_txn}}+\sum _{i\in Q,p_i\le p_{\text{safety}}}(t_{\text{write},i}+\delta +t_{\text{read},i})$$

In the common case no other SAFETY command is queued, so the sum is empty and the bound collapses to L_safety ≤ t_current_txn — the residual time of whatever single transaction is mid-flight. That residual is itself capped by that command’s timeout budget, so with a 200 ms poll timeout the safety command dispatches in at most ~200 ms plus its own service time, independent of backlog depth. Contrast the ~700 ms FIFO figure above: the bound no longer scales with queue length, which is the entire point.

A safety command inserts at the head ahead of aged polls, but the sole transport worker finishes the indivisible in-flight transaction before dispatching it — the bound is the residual transaction, not the backlog.

A SafetyAwareDispatcher with Aging and a Critical-Section Guard

The dispatcher wraps an asyncio.PriorityQueue. Each item is a (effective_priority, seq, envelope) tuple; seq is a monotonic tie-breaker so equal-priority commands stay FIFO and the envelope itself is never compared. A single worker owns the transport and holds a critical-section guard for the duration of each write-read pair, so a preempting safety command is queued for immediate dispatch but never splices its bytes into a transaction already on the wire.

Copy
from __future__ import annotations

import asyncio
import itertools
import logging
import time
from dataclasses import dataclass, field
from enum import IntEnum
from typing import Awaitable, Callable, Optional

logger = logging.getLogger(__name__)

ReadFn = Callable[[], Awaitable[bytes]]
WriteFn = Callable[[bytes], Awaitable[None]]


class Priority(IntEnum):
    """Lower value dispatches first. SAFETY can never be reached by aging."""

    SAFETY = 0
    CONTROL = 1
    QUERY = 2
    POLL = 3


AGE_FLOOR = Priority.CONTROL  # aged commands stop one step above SAFETY


@dataclass(order=True)
class _Item:
    eff_priority: int
    seq: int
    envelope: "Envelope" = field(compare=False)


@dataclass
class Envelope:
    cmd_id: str
    payload: bytes
    base_priority: Priority
    timeout: float
    enqueued_at: float = field(default_factory=time.monotonic)
    future: Optional["asyncio.Future[bytes]"] = None

    def effective_priority(self, now: float, tau_age: float) -> int:
        """Age-boosted key, floored so routine traffic never reaches SAFETY."""
        if self.base_priority == Priority.SAFETY:
            return int(Priority.SAFETY)
        steps = int((now - self.enqueued_at) // tau_age) if tau_age > 0 else 0
        return max(int(AGE_FLOOR), int(self.base_priority) - steps)


class SafetyAwareDispatcher:
    """Priority command queue with aging and non-preemptible transactions.

    One worker owns the transport. Safety commands preempt the queue head but
    never interrupt the transaction already in flight; the aging pass keeps
    low-priority polls from starving under sustained higher-priority load.
    """

    def __init__(
        self,
        read_callback: ReadFn,
        write_callback: WriteFn,
        *,
        max_depth: int = 256,
        tau_age: float = 2.0,
        inter_command_delay: float = 0.0,
    ) -> None:
        self._read = read_callback
        self._write = write_callback
        self._pq: "asyncio.PriorityQueue[_Item]" = asyncio.PriorityQueue(maxsize=max_depth)
        self._seq = itertools.count()
        self._tau_age = tau_age
        self._delay = inter_command_delay
        self._txn_lock = asyncio.Lock()  # the non-preemptible critical section
        self._worker: Optional["asyncio.Task[None]"] = None
        self._stop = asyncio.Event()

    async def start(self) -> None:
        if self._worker and not self._worker.done():
            raise RuntimeError("Dispatcher already running")
        self._stop.clear()
        self._worker = asyncio.create_task(self._run(), name="safety_dispatcher")

    async def submit(
        self,
        cmd_id: str,
        payload: bytes,
        priority: Priority,
        *,
        timeout: float = 5.0,
    ) -> "asyncio.Future[bytes]":
        """Enqueue a command keyed by priority; returns a future for its reply."""
        loop = asyncio.get_running_loop()
        env = Envelope(cmd_id, payload, priority, timeout, future=loop.create_future())
        now = time.monotonic()
        item = _Item(env.effective_priority(now, self._tau_age), next(self._seq), env)
        # SAFETY is dropped in even if the queue is otherwise full.
        if priority == Priority.SAFETY and self._pq.full():
            self._pq._queue.append(item)  # bounded bypass for E-stop only
            self._pq._queue.sort()
        else:
            self._pq.put_nowait(item)
        assert env.future is not None
        return env.future

    async def abort_low_priority(self, *, below: Priority = Priority.CONTROL) -> int:
        """Flush queued commands below a threshold, cancelling their futures.

        Called after a safety trip so the backlog does not resume driving a
        hardware state the E-stop just invalidated. Returns the count flushed.
        """
        kept: list[_Item] = []
        flushed = 0
        while not self._pq.empty():
            item = self._pq.get_nowait()
            if item.envelope.base_priority >= below and item.envelope.base_priority != Priority.SAFETY:
                fut = item.envelope.future
                if fut is not None and not fut.done():
                    fut.cancel()
                flushed += 1
            else:
                kept.append(item)
        for item in kept:
            self._pq.put_nowait(item)
        logger.warning("Flushed %d low-priority commands after safety event", flushed)
        return flushed

    def _reage(self) -> None:
        """Recompute effective priorities so waiting polls climb toward the head."""
        now = time.monotonic()
        items: list[_Item] = []
        while not self._pq.empty():
            it = self._pq.get_nowait()
            it.eff_priority = it.envelope.effective_priority(now, self._tau_age)
            items.append(it)
        for it in items:
            self._pq.put_nowait(it)

    async def _run(self) -> None:
        while not self._stop.is_set():
            self._reage()
            try:
                item = await asyncio.wait_for(self._pq.get(), timeout=0.05)
            except asyncio.TimeoutError:
                continue
            env = item.envelope
            # The critical section: one indivisible write→read transaction.
            async with self._txn_lock:
                try:
                    await self._write(env.payload)
                    if self._delay > 0:
                        await asyncio.sleep(self._delay)
                    raw = await asyncio.wait_for(self._read(), timeout=env.timeout)
                    if env.future is not None and not env.future.done():
                        env.future.set_result(raw)
                except Exception as exc:  # never let one command kill the worker
                    logger.error("Command %s failed: %r", env.cmd_id, exc)
                    if env.future is not None and not env.future.done():
                        env.future.set_exception(exc)
                finally:
                    self._pq.task_done()

The _txn_lock is the mechanism that makes a transaction non-preemptible: a safety command can be inserted at the head at any instant, but the worker only ever picks up the next item after releasing the lock, so the byte stream of the in-flight command is never interleaved. Aging runs once per loop iteration via _reage, which is cheap for the queue depths (tens to low hundreds) these transports sustain. Safety commands bypass the maxsize bound, because refusing an E-stop under backpressure is the one failure mode that is never acceptable.

Validating the Latency Bound Under Saturated Poll Load

The guarantee to prove is that a safety command dispatches within one transaction residual even when the queue is saturated. Flood the dispatcher with polls, submit one SAFETY command, and assert its measured dispatch latency stays under a single transaction budget:

Copy
import asyncio
import time

async def test_safety_preempts_saturated_polls() -> None:
    async def slow_write(_: bytes) -> None:
        await asyncio.sleep(0.002)

    async def slow_read() -> bytes:
        await asyncio.sleep(0.010)  # ~12 ms per transaction
        return b"OK\n"

    d = SafetyAwareDispatcher(slow_read, slow_write, tau_age=2.0)
    await d.start()

    for i in range(200):  # saturate with routine polls
        await d.submit(f"poll{i}", b"MEAS?\n", Priority.POLL, timeout=0.2)

    t0 = time.monotonic()
    fut = await d.submit("estop", b"OUTP OFF\n", Priority.SAFETY, timeout=0.2)
    await fut
    latency = time.monotonic() - t0

    # One in-flight txn (~12 ms) + own service (~12 ms), NOT 200 * 12 ms.
    assert latency < 0.050, f"safety dispatch too slow: {latency*1000:.1f} ms"

asyncio.run(test_safety_preempts_saturated_polls())

On a live rig the same claim is confirmed physically: fire the safety command mid-run and timestamp the instrument’s own reaction — a source-measure unit’s OUTP? reads 0, an interlock LED clears, or a scope trigger stops — within the one-transaction window. Log the effective priority of every dequeued item; a monotonic climb of POLL items toward the head over seconds is the aging pass working, and a POLL that never advances signals τ_age is too large for the observed queue turnover. Confirm the abort path by asserting that abort_low_priority() cancels the pending poll futures while leaving any queued CONTROL command intact.

Failure Modes Specific to Safety Preemption

Priority inversion via a held port lock. If a low-priority command holds the transport while blocked on a long read, a newly arrived safety command cannot dispatch until that read returns or times out — the classic inversion, here bounded by the low-priority command’s own timeout. The defense is a short timeout on POLL/QUERY commands (tens of ms) so t_current_txn in the latency bound stays small; never give a background poll a multi-second read budget on a bus that must accept an E-stop. Diagnose by logging transaction start/end and watching for a safety dispatch stalled behind a single long-running lower-priority read.

Starvation of low priority without aging. Disable aging (tau_age=0) and a sustained QUERY stream leaves POLL telemetry permanently at the tail; the queue depth for level 3 grows without bound while callers see stale sensor values. This is silent — nothing errors — so it must be caught by monitoring per-priority dwell time. The aging term in p_eff is the fix; verify a floored climb so aging can lift a poll toward CONTROL but the AGE_FLOOR clamp keeps it below SAFETY.

Safety command queued behind a long block read. A binary-capture command doing a multi-kilobyte readuntil holds the critical section far longer than a short SCPI query, inflating t_current_txn for any safety command that arrives during it. Chunk long reads or route large transfers to a separate transport so the safety-carrying bus only ever runs short transactions; large binary payloads belong under Binary & ASCII Format Parsing with bounded read sizes. If a long read is unavoidable, cap it with the same wait_for budget that Timeout Handling & Retry Logic applies elsewhere.

Duplicate E-stop flooding. An operator mashing an abort button, or a chattering interlock GPIO, can enqueue dozens of identical SAFETY commands, each bypassing the bound and each re-transmitting OUTP OFF — harmless to the output but wasteful and log-noisy, and capable of crowding out a distinct safety command like an interlock reset. Deduplicate by tracking an in-flight set of safety cmd_ids and coalescing repeats within a short debounce window. Treat a storm of E-stops as itself a signal worth surfacing through Error Code Categorization rather than swallowing silently.

Related

• Async Command Queuing Systems — the scheduling and state-machine foundations this preemption tier extends.
• Building async command queues with asyncio for lab devices — the FIFO single-worker queue that this adds priority and preemption to.
• Timeout Handling & Retry Logic — the per-command budgets that bound t_current_txn in the latency proof.
• Error Code Categorization — classifying interlock trips and E-stop storms distinctly from transient faults.
• PySerial Configuration & Tuning — the transport settings that keep the residual-transaction window short.

← Back to Async Command Queuing Systems

References

• Python asyncio.PriorityQueue documentation — ordering, put_nowait, and maxsize semantics.
• Python asyncio.Lock reference — the primitive guarding the non-preemptible critical section.