Building Async Command Queues with asyncio for Lab Devices

An asyncio command queue serializes traffic to a lab instrument that shares a single physical bus, decoupling non-blocking command submission from strictly ordered I/O execution. This guide builds a single-worker queue that drains commands in FIFO order, applies a per-command timeout budget, bounds retries, and guarantees that one malformed response or timeout never corrupts the command that follows it. It is the concrete implementation behind the paradigms surveyed on the parent Async Command Queuing Systems page.

Where a Single-Worker asyncio Queue Applies

The scope here is narrow: one instrument reachable over one command-response transport that cannot multiplex — a pyserial port, a PyVISA session, or a GPIB address — where every command must be written, then its reply read back, before the next command may go out. This is the near-universal shape of SCPI-style hardware: source-measure units, oscilloscopes, temperature controllers, and syringe pumps all speak a request/reply protocol with no native pipelining. As established across the Serial, USB, and GPIB Communication Workflows domain, three constraints dictate the architecture: hardware receive buffers are finite and non-preemptive, the command-response handshake is strictly sequential, and firmware may silently drop or reorder frames under bus contention.

Two assumptions hold throughout. First, exactly one coroutine owns the transport — concurrent writers are the single most common source of interleaved, unparseable replies, so the queue makes the worker a singleton by construction. Second, callers submit from within a running event loop and await a Future for the result, rather than blocking on the port directly. If your driver is a synchronous pyserial object, its blocking read/write calls must be pushed to a thread with asyncio.to_thread so they never stall the event loop; native-async transports attach directly. Baud, parity, and latency-timer configuration that govern how bytes actually arrive belong to PySerial Configuration & Tuning and must be settled before the queue is started.

Serialized Execution Model and Worst-Case Latency Bound

The queue’s correctness rests on a single invariant: at most one command is in flight on the transport at any instant, and commands leave the queue in the exact order they were enqueued. Submission is non-blocking — submit() only appends an envelope and returns a Future — while the worker coroutine imposes total order at the I/O boundary. This separation is what lets an experiment sequencer fire off a batch of commands concurrently without ever risking a reordered or spliced reply.

Determinism extends to timing. Because the worker never overlaps commands, the wall-clock time until a command enqueued at depth k completes is bounded by the sum of the service times of every command ahead of it plus its own retry budget:

$$T_k\le \sum _{i=1}^k(r_i+1)(t_{\text{write}}+\delta +t_{\text{read},i})$$

where r_i is the retry ceiling for command i, δ is the fixed inter_command_delay inserted after each write to respect firmware turnaround, and t_read,i is capped by that command’s timeout budget. This bound is what makes queue latency predictable enough to reason about in a real assay: a slow query at the head of the queue delays everything behind it by a computable amount rather than an unbounded one. Retries are counted, not randomized — reproducible execution traces matter more here than the thundering-herd avoidance that motivates jittered backoff in distributed systems. When a command does time out, the retry delay itself should follow the deterministic curve documented in Timeout Handling & Retry Logic; the queue owns ordering and lifecycle, while that page owns how long to wait before the next attempt.

A Production asyncio Device Queue

The implementation below is a single self-contained module. Each command is wrapped in a CommandEnvelope carrying its identity, payload, timeout, retry ceiling, and the asyncio.Future that resolves the caller. The AsyncDeviceQueue owns one worker task that drains the queue, enforces the timeout budget with asyncio.wait_for, applies bounded retries, and — critically — never lets an exception escape the execution boundary: every failure is captured, logged, and routed to the originating Future.

Copy
from __future__ import annotations

import asyncio
import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Awaitable, Callable, Optional

logger = logging.getLogger(__name__)

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


class CommandStatus(Enum):
    PENDING = "pending"
    EXECUTING = "executing"
    COMPLETED = "completed"
    FAILED = "failed"
    TIMEOUT = "timeout"


@dataclass
class CommandEnvelope:
    """A single command plus the state needed to serialize and retry it.

    The ``response_future`` is created by the queue on submission so the
    envelope is never bound to an event loop at import time.
    """

    cmd_id: str
    payload: bytes
    timeout: float
    max_retries: int = 1
    status: CommandStatus = CommandStatus.PENDING
    response_future: Optional["asyncio.Future[bytes]"] = None
    _attempts: int = field(default=0, init=False)

    @property
    def can_retry(self) -> bool:
        return self._attempts <= self.max_retries

    def record_attempt(self) -> None:
        self._attempts += 1


class AsyncDeviceQueue:
    """FIFO command queue with a single serialized I/O worker.

    One worker coroutine owns the transport, guaranteeing that commands are
    written and read back in submission order. Callers ``submit()`` a payload
    and ``await`` the returned future for the raw response bytes.
    """

    def __init__(
        self,
        read_callback: ReadFn,
        write_callback: WriteFn,
        *,
        max_queue_size: int = 128,
        default_timeout: float = 5.0,
        inter_command_delay: float = 0.0,
    ) -> None:
        self._read = read_callback
        self._write = write_callback
        self.default_timeout = default_timeout
        self.inter_command_delay = inter_command_delay
        self.queue: "asyncio.Queue[CommandEnvelope]" = asyncio.Queue(maxsize=max_queue_size)
        self._worker_task: Optional["asyncio.Task[None]"] = None
        self._shutdown = asyncio.Event()
        self._log = logging.getLogger(f"{__name__}.AsyncDeviceQueue")

    async def start(self) -> None:
        """Launch the single worker task; refuses to start a second one."""
        if self._worker_task and not self._worker_task.done():
            raise RuntimeError("Queue worker is already running")
        self._shutdown.clear()
        self._worker_task = asyncio.create_task(self._worker_loop(), name="device_queue_worker")

    async def stop(self, *, drain: bool = True) -> None:
        """Signal shutdown and await the worker.

        With ``drain=True`` in-flight and queued commands finish or time out
        before the transport is released; with ``drain=False`` the worker is
        cancelled after the current command.
        """
        self._shutdown.set()
        if not self._worker_task:
            return
        if not drain:
            self._worker_task.cancel()
        try:
            await self._worker_task
        except asyncio.CancelledError:
            pass

    async def submit(
        self,
        cmd_id: str,
        payload: bytes,
        *,
        timeout: Optional[float] = None,
        max_retries: int = 1,
    ) -> "asyncio.Future[bytes]":
        """Enqueue a command and return a future that resolves to its reply.

        Raises ``asyncio.QueueFull`` immediately if the bounded queue is at
        capacity, surfacing backpressure to the caller instead of hiding it.
        """
        if self._shutdown.is_set():
            raise RuntimeError("Queue is shutting down; refusing new commands")
        loop = asyncio.get_running_loop()
        envelope = CommandEnvelope(
            cmd_id=cmd_id,
            payload=payload,
            timeout=timeout if timeout is not None else self.default_timeout,
            max_retries=max_retries,
            response_future=loop.create_future(),
        )
        self.queue.put_nowait(envelope)  # raises QueueFull rather than blocking
        assert envelope.response_future is not None
        return envelope.response_future

    async def _worker_loop(self) -> None:
        self._log.info("Queue worker started")
        while not self._shutdown.is_set():
            try:
                envelope = await asyncio.wait_for(self.queue.get(), timeout=0.1)
            except asyncio.TimeoutError:
                continue  # poll the shutdown flag on an empty queue
            except asyncio.CancelledError:
                break
            try:
                await self._execute(envelope)
            except Exception:  # a bug here must not kill the loop silently
                self._log.critical("Worker crashed on %s", envelope.cmd_id, exc_info=True)
                if not envelope.response_future.done():
                    envelope.response_future.set_exception(
                        RuntimeError(f"Worker crashed while executing {envelope.cmd_id}")
                    )
            finally:
                self.queue.task_done()
        self._log.info("Queue worker terminated")

    async def _execute(self, envelope: CommandEnvelope) -> None:
        while envelope.can_retry:
            envelope.record_attempt()
            envelope.status = CommandStatus.EXECUTING
            try:
                await self._write(envelope.payload)
                if self.inter_command_delay > 0:
                    await asyncio.sleep(self.inter_command_delay)
                raw = await asyncio.wait_for(self._read(), timeout=envelope.timeout)
                envelope.status = CommandStatus.COMPLETED
                if not envelope.response_future.done():
                    envelope.response_future.set_result(raw)
                return
            except asyncio.TimeoutError:
                envelope.status = CommandStatus.TIMEOUT
                self._log.warning(
                    "Timeout on %s (attempt %d/%d)",
                    envelope.cmd_id, envelope._attempts, envelope.max_retries + 1,
                )
                if not envelope.can_retry and not envelope.response_future.done():
                    envelope.response_future.set_exception(
                        asyncio.TimeoutError(
                            f"{envelope.cmd_id} timed out after {envelope._attempts} attempts"
                        )
                    )
                    return
            except Exception as exc:  # transport-level fault: do not retry blindly
                envelope.status = CommandStatus.FAILED
                if not envelope.response_future.done():
                    envelope.response_future.set_exception(exc)
                return

The worker uses a short queue.get() timeout so it stays responsive to the shutdown flag instead of blocking forever on an empty queue. Deferring Future creation to submit() avoids event-loop binding issues at import time and keeps submission safe to call from any coroutine on the loop. Note the deliberate asymmetry in the except clauses: an asyncio.TimeoutError is retryable and consumes the retry budget, whereas any other exception is treated as a transport fault and fails the command immediately — retrying a SerialException on a yanked USB cable only wedges the loop harder.

Verifying Deterministic Draining Against Live Hardware

Correct behavior is observable, not assumed. The following harness swaps in a scripted transport so ordering, retry, and timeout semantics can be proven before any real instrument is attached, then the same assertions apply against live hardware:

Copy
import asyncio

async def _demo() -> None:
    replies = asyncio.Queue()
    for r in (b"OK1\n", b"OK2\n", b"OK3\n"):
        replies.put_nowait(r)

    async def fake_write(_: bytes) -> None: ...
    async def fake_read() -> bytes: return await replies.get()

    q = AsyncDeviceQueue(fake_read, fake_write, inter_command_delay=0.01)
    await q.start()
    futures = [await q.submit(f"c{i}", f"MEAS{i}?\n".encode()) for i in range(3)]
    results = await asyncio.gather(*futures)
    assert results == [b"OK1\n", b"OK2\n", b"OK3\n"]  # FIFO order preserved
    await q.stop(drain=True)

asyncio.run(_demo())

Against real hardware, three signals confirm the queue is healthy. First, ordering: log every cmd_id at the point of set_result and confirm the completion order matches the submission order exactly — any inversion means a second writer is touching the transport. Second, backpressure: instrument the queue with counters for commands_submitted, commands_completed, commands_failed, and sample queue.qsize(); a qsize that climbs monotonically means the instrument cannot keep up with the submission rate and you are heading for QueueFull. Third, physical completion: for SCPI instruments, follow any actuation command with an *OPC? query — because the queue guarantees serialized execution, *OPC? returns 1 only after every preceding command has physically completed, giving you a hardware-side proof that the queue drained in order. Enabling PYTHONASYNCIODEBUG=1 surfaces any coroutine that blocks the loop longer than 100 ms, which is the fingerprint of a synchronous driver call that should have been wrapped in asyncio.to_thread.

Failure Modes Unique to asyncio Command Queues

These are the failures specific to a single-worker async queue, distinct from the generic transport errors catalogued under Error Code Categorization.

The most damaging of these is the orphaned future. If the worker task dies and no one is awaiting self._worker_task, callers block forever on futures that will never resolve. The guard clause in _worker_loop closes the common cases by failing the in-flight future, but you should still attach a done-callback to the worker task itself so a dead worker raises an alert rather than degrading into a silent hang. Treat a crashed worker as a framing/lifecycle fault, not a transient error: reset the transport and call start() fresh rather than reusing a queue whose ordering guarantee may already be compromised.

Integration and Related Guidance

Upstream, an experiment sequencer should only ever await queue.submit(...) and await the returned future — never poll queue.empty() or inspect worker state from outside the loop. Downstream, the raw bytes a future resolves to are unvalidated: instrument replies carrying a non-zero status byte are protocol errors, not transport failures, and must pass through Error Code Categorization before any recovery logic fires. When the payload is binary telemetry rather than a short SCPI reply, hand it to Checksum & CRC Validation before parsing. For labs coordinating several instruments through a shared session layer, allocate transports through a VISA Resource Manager and give each instrument its own AsyncDeviceQueue, since the single-worker guarantee is per-transport, not global.

Related

• Async Command Queuing Systems — the paradigm survey (polling vs event-driven vs transactional) this implementation sits under.
• Timeout Handling & Retry Logic — how long to wait before the next attempt, including deterministic exponential backoff.
• Error Code Categorization — classifying the status bytes a resolved future returns, e.g. categorizing SCPI error codes.
• PySerial Configuration & Tuning — baud, flow control, and latency-timer settings the transport callbacks depend on.
• VISA Resource Manager Setup — allocating one transport per instrument for multi-vendor labs.

← Back to Async Command Queuing Systems

References

• Python asyncio queues documentation — Queue, put_nowait, and QueueFull semantics.
• Python asyncio.wait_for reference — timeout enforcement and cancellation behavior.