From 493d5264fb99a1ec6b3083d23b87b32101ba4fe1 Mon Sep 17 00:00:00 2001 From: glopesdev Date: Fri, 21 Aug 2026 08:40:44 +0100 Subject: [PATCH] Fix reply correlation and failure reporting A pending request is keyed by address and message type rather than address alone, and each key holds every waiter, so a reply satisfies only requests of its own type. Before, a write reply would satisfy a read on the same register. A transport failure in the reader thread is reported to every waiting request and to any that follow, in place of being re-raised inside a daemon thread where it was lost, so a removed port no longer presents as a device that went quiet. An error reply raises DeviceError, which keeps the reply, in place of a RuntimeError that formatted the frame into its message and discarded it. --- docs/api/device.md | 1 + src/packages/harp-device/README.md | 4 + .../src/harp/device/client/__init__.py | 3 +- .../src/harp/device/client/_device.py | 82 +++++++--- tests/device/test_device.py | 145 +++++++++++++++++- 5 files changed, 215 insertions(+), 20 deletions(-) diff --git a/docs/api/device.md b/docs/api/device.md index 17b854d..3fd3865 100644 --- a/docs/api/device.md +++ b/docs/api/device.md @@ -3,6 +3,7 @@ --- ::: harp.device.client.Device +::: harp.device.client.DeviceError ::: harp.device.client.HarpFramer ::: harp.device.client.ITransport ::: harp.device.client.TransportError diff --git a/src/packages/harp-device/README.md b/src/packages/harp-device/README.md index 32d74f7..713ed0a 100644 --- a/src/packages/harp-device/README.md +++ b/src/packages/harp-device/README.md @@ -14,6 +14,10 @@ who = device.read(core.WhoAmI).payload # -> np.uint16 device.write(core.OperationControl, payload) # write a register ``` +## When a request fails + +An error reply raises `DeviceError`, which keeps the reply as `reply` so the frame sent by the device stays available for inspection. Pass `raise_on_error=False` to the constructor to receive such a reply as an ordinary return value instead. A transport failure raises `TransportError`, and every later request reports the same failure rather than waiting for a reply that cannot arrive. A device that never answers raises `TimeoutError` after `REPLY_TIMEOUT`, which is also what happens when `close` is called during a request. + ## Extending for a specific device A device is described by a module. Downstream, often generated, packages record the device identity as `WHO_AM_I`, declare the register classes at module level, and expand the core `REGISTER_MAP` beside them: diff --git a/src/packages/harp-device/src/harp/device/client/__init__.py b/src/packages/harp-device/src/harp/device/client/__init__.py index c2d142e..15e23bd 100644 --- a/src/packages/harp-device/src/harp/device/client/__init__.py +++ b/src/packages/harp-device/src/harp/device/client/__init__.py @@ -1,11 +1,12 @@ """Talking to a Harp device: the device itself, its transport and the framer.""" -from ._device import Device, EventHandler, Subscription +from ._device import Device, DeviceError, EventHandler, Subscription from ._framer import HarpFramer from ._transport import ITransport, TransportError __all__ = [ "Device", + "DeviceError", "EventHandler", "Subscription", "HarpFramer", diff --git a/src/packages/harp-device/src/harp/device/client/_device.py b/src/packages/harp-device/src/harp/device/client/_device.py index 1ff0333..151425d 100644 --- a/src/packages/harp-device/src/harp/device/client/_device.py +++ b/src/packages/harp-device/src/harp/device/client/_device.py @@ -31,6 +31,9 @@ _DEFAULT_MESSAGE_TYPES: frozenset[MessageType] = frozenset({MessageType.Event}) """Default filter for :meth:`Device.subscribe`: unsolicited events only.""" +_Waiter = queue.SimpleQueue["HarpMessage | Exception"] +"""The received reply or failure that ends the wait of a blocked request.""" + def _normalize_message_types(message_types: MessageTypeFilter) -> frozenset[MessageType]: if isinstance(message_types, MessageType): @@ -38,6 +41,23 @@ def _normalize_message_types(message_types: MessageTypeFilter) -> frozenset[Mess return frozenset(message_types) +class DeviceError(Exception): + """Raised when the device replies to a request with the error flag set. + + The reply is kept as :attr:`reply`, so the frame sent by the device stays available + for inspection. Construct the device with ``raise_on_error=False`` to receive such + a reply as an ordinary return value instead. + """ + + def __init__(self, reply: HarpMessage) -> None: + super().__init__( + f"Device returned error for register address {reply.address} " + f"(0x{reply.address:02x}). Payload: {reply.payload_bytes.hex()}" + ) + self.reply = reply + """The error reply, as received.""" + + class Subscription: """Handle returned by :meth:`Device.subscribe`. Cancel with :meth:`unsubscribe`, or use as a context manager to auto-cancel on exit.""" @@ -85,6 +105,13 @@ class Device(Generic[M]): Omitting ``device_module`` skips that check. The module is not otherwise consulted: registers reach :meth:`read`, :meth:`write` and :meth:`subscribe` as arguments either way, and only a subscribed register is parsed on arrival. + + A request fails in one of three ways. An error reply raises :class:`DeviceError` + carrying the reply, unless ``raise_on_error=False``. A transport failure raises + :class:`~harp.device.client.TransportError`, and every later request reports the + same failure rather than waiting for a reply that cannot arrive. A device that + never answers raises :class:`TimeoutError` after ``REPLY_TIMEOUT``, which is also + what happens when :meth:`close` is called during a request. """ REPLY_TIMEOUT: ClassVar[float] = 5.0 # seconds @@ -114,8 +141,9 @@ def __init__( self._device_module = device_module self.raise_on_error = raise_on_error self._framer = HarpFramer() - self._pending: dict[int, queue.SimpleQueue] = {} + self._pending: dict[tuple[int, MessageType], list[_Waiter]] = {} self._pending_lock = threading.Lock() + self._fault: Exception | None = None self._running = False self._thread: threading.Thread | None = None @@ -207,7 +235,7 @@ def read( # Note: ty can't correctly infer the return type, and this is a known issue: # https://github.com/astral-sh/ty/issues/623 frame = register.format(message_type=MessageType.Read, timestamp=timestamp, port=port) - msg = self._request(register.address, frame) + msg = self._request(register.address, MessageType.Read, frame) return msg.decode(register) def write( @@ -221,7 +249,7 @@ def write( frame = register.format( value, message_type=MessageType.Write, timestamp=timestamp, port=port ) - msg = self._request(register.address, frame) + msg = self._request(register.address, MessageType.Write, frame) return msg.decode(register) # ------------------------------------------------------------------ @@ -337,9 +365,10 @@ def _read_loop(self) -> None: while self._running: try: chunk = self._transport.read() - except TransportError: + except TransportError as exc: if self._running: - raise + _logger.exception("The transport failed, so the device stopped reading") + self._fault_pending(exc) break # expected while shutting down if not chunk: continue @@ -348,37 +377,54 @@ def _read_loop(self) -> None: for msg in self._framer.frames(): self._dispatch(msg) + def _fault_pending(self, exc: Exception) -> None: + """Report ``exc`` to every waiting request, and record it so a later request + reports it at once instead of waiting for a reply that cannot arrive.""" + with self._pending_lock: + self._fault = exc + waiting = [q for waiters in self._pending.values() for q in waiters] + self._pending.clear() + for q in waiting: + q.put(exc) + def _dispatch(self, msg: HarpMessage) -> None: # Fast path: correlate replies to a pending synchronous request. This is # O(1) and non-blocking, so it should never stall behind a slow subscriber. # Events are unsolicited and never correlate to requests if msg.message_type != MessageType.Event: with self._pending_lock: - q = self._pending.get(msg.address) - if q is not None: + waiting = list(self._pending.get((msg.address, msg.message_type), ())) + for q in waiting: q.put(msg) self._event_queue.put(msg) - def _request(self, address: int, frame: bytes) -> HarpMessage: - q: queue.SimpleQueue = queue.SimpleQueue() + def _request(self, address: int, message_type: MessageType, frame: bytes) -> HarpMessage: + key = (address, message_type) + q: _Waiter = queue.SimpleQueue() with self._pending_lock: - self._pending[address] = q + if self._fault is not None: + raise self._fault + self._pending.setdefault(key, []).append(q) try: self._transport.write(frame) try: - msg = q.get(timeout=self.REPLY_TIMEOUT) - if msg.has_error and self.raise_on_error: - raise RuntimeError( - f"Device returned error for register address {address} " - f"(0x{address:02x}). Payload: {msg.payload_bytes.hex()}" - ) - return msg + reply = q.get(timeout=self.REPLY_TIMEOUT) except queue.Empty as exc: raise TimeoutError( f"No reply from device for register address {address} " f"within {self.REPLY_TIMEOUT}s" ) from exc + if isinstance(reply, Exception): + raise reply + if reply.has_error and self.raise_on_error: + raise DeviceError(reply) + return reply finally: with self._pending_lock: - self._pending.pop(address, None) + waiters = self._pending.get(key) + if waiters is not None: + if q in waiters: + waiters.remove(q) + if not waiters: + del self._pending[key] diff --git a/tests/device/test_device.py b/tests/device/test_device.py index 7d782de..4350c76 100644 --- a/tests/device/test_device.py +++ b/tests/device/test_device.py @@ -1,6 +1,19 @@ +import queue +import threading import types +from collections.abc import Callable, Iterable +from concurrent.futures import ThreadPoolExecutor -from harp.device.client import Device +import numpy as np +import pytest + +from harp.device import core +from harp.device.client import Device, DeviceError, TransportError +from harp.protocol import MessageType +from tests.fixtures import make_frame_from_raw + +_U16 = 0x02 +"""Payload-type byte of a U16 payload, as the size nibble alone.""" class _NullTransport: @@ -14,6 +27,45 @@ def read(self) -> bytes: return b"" +class _ScriptedTransport: + """A transport replying with whatever ``on_write`` returns for each request. + + Frames are queued from inside ``write``, which is reached only once the request is + registered, so a reply cannot be dispatched before there is a waiter to receive it. + Setting ``failing`` makes the next read fail, as a removed port would. + """ + + def __init__(self) -> None: + self.writes: list[bytes] = [] + self.failing = False + self.on_write: Callable[[bytes], Iterable[bytes]] | None = None + self._inbox: queue.SimpleQueue[bytes] = queue.SimpleQueue() + + def open(self) -> None: ... + + def close(self) -> None: ... + + def write(self, data: bytes) -> None: + self.writes.append(data) + if self.on_write is not None: + for frame in self.on_write(data): + self._inbox.put(frame) + + def read(self) -> bytes: + if self.failing: + raise TransportError("simulated transport failure") + try: + return self._inbox.get(timeout=0.01) + except queue.Empty: + return b"" + + +class _ShortTimeoutDevice(Device[None]): + """A device that gives up on a reply quickly, so a test never waits five seconds.""" + + REPLY_TIMEOUT = 0.5 + + def _module(name: str, **attrs: object) -> types.ModuleType: mod = types.ModuleType(name) mod.DEVICE_NAME = name @@ -33,3 +85,94 @@ def test_module_is_returned_by_the_property(): module = _module("Behavior", WHO_AM_I=0, REGISTER_MAP={}) assert Device(_NullTransport(), module).module is module assert Device(_NullTransport()).module is None + + +def test_write_reply_does_not_satisfy_read(): + # A reply carries the message type of its request, so a write reply on the same + # register must not answer a pending read. + transport = _ScriptedTransport() + transport.on_write = lambda _: ( + core.WhoAmI.format(np.uint16(7), message_type=MessageType.Write), + core.WhoAmI.format(np.uint16(9), message_type=MessageType.Read), + ) + with _ShortTimeoutDevice(transport) as device: + assert int(device.read(core.WhoAmI).payload) == 9 + + +def test_event_does_not_satisfy_read(): + # Events are unsolicited, so one for the same register must not answer a read. + transport = _ScriptedTransport() + transport.on_write = lambda _: ( + core.WhoAmI.format(np.uint16(7), message_type=MessageType.Event), + core.WhoAmI.format(np.uint16(9), message_type=MessageType.Read), + ) + with _ShortTimeoutDevice(transport) as device: + assert int(device.read(core.WhoAmI).payload) == 9 + + +def test_concurrent_reads_share_one_reply(): + # The wire carries no request identifier, so two reads in flight on one register + # cannot be told apart in the reply. Both are answered by it rather than one + # replacing the waiter of the other. + transport = _ScriptedTransport() + both_in_flight = threading.Barrier(2) + + def reply_once(_: bytes) -> Iterable[bytes]: + first = both_in_flight.wait(timeout=2) == 0 + return () if first else (core.WhoAmI.format(np.uint16(9), message_type=MessageType.Read),) + + transport.on_write = reply_once + with _ShortTimeoutDevice(transport) as device, ThreadPoolExecutor(2) as pool: + replies = [pool.submit(device.read, core.WhoAmI) for _ in range(2)] + assert [int(reply.result(timeout=2).payload) for reply in replies] == [9, 9] + + +def test_transport_failure_faults_pending_read(): + transport = _ScriptedTransport() + + def fail(_: bytes) -> Iterable[bytes]: + transport.failing = True + return () + + transport.on_write = fail + with _ShortTimeoutDevice(transport) as device: + with pytest.raises(TransportError): + device.read(core.WhoAmI) + + +def test_read_after_transport_failure_raises_transport_error(): + # A failure that has already stopped the reader is reported to the next request + # rather than leaving it to time out on a reply that cannot arrive. + transport = _ScriptedTransport() + + def fail(_: bytes) -> Iterable[bytes]: + transport.failing = True + return () + + transport.on_write = fail + with _ShortTimeoutDevice(transport) as device: + with pytest.raises(TransportError): + device.read(core.WhoAmI) + with pytest.raises(TransportError): + device.read(core.WhoAmI) + assert len(transport.writes) == 1 # the second request never reached the transport + + +def test_error_reply_raises_device_error(): + frame = make_frame_from_raw(MessageType.Read | 0x08, 0, 255, _U16, b"\x07\x00") + transport = _ScriptedTransport() + transport.on_write = lambda _: (frame,) + with _ShortTimeoutDevice(transport) as device: + with pytest.raises(DeviceError) as error: + device.read(core.WhoAmI) + assert error.value.reply.bytes == frame # the frame is kept, not formatted away + + +def test_error_reply_returned_when_not_raising(): + frame = make_frame_from_raw(MessageType.Read | 0x08, 0, 255, _U16, b"\x07\x00") + transport = _ScriptedTransport() + transport.on_write = lambda _: (frame,) + with _ShortTimeoutDevice(transport, raise_on_error=False) as device: + reply = device.read(core.WhoAmI) + assert reply.has_error + assert int(reply.payload) == 7