From 4255908b08fc0d37df4a281c35577a492ca79dcc Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Wed, 21 Oct 2026 14:10:34 +0000 Subject: [PATCH 1/3] Move ASH handling to an isolated subinterpreter --- bellows/ash.py | 76 +++- bellows/config/__init__.py | 2 + bellows/ezsp/__init__.py | 16 +- bellows/thread.py | 123 ------- bellows/types/named.py | 15 +- bellows/uart.py | 615 +++++++++++++++++++++++++++++--- bellows/zigbee/application.py | 2 +- bellows/zigbee/util.py | 30 -- tests/test_ezsp.py | 1 + tests/test_thread.py | 213 ------------ tests/test_uart.py | 636 ++++++++++++++++++++++++++++------ 11 files changed, 1167 insertions(+), 562 deletions(-) delete mode 100644 bellows/thread.py delete mode 100644 tests/test_thread.py diff --git a/bellows/ash.py b/bellows/ash.py index 48b492caa..4203b03fa 100644 --- a/bellows/ash.py +++ b/bellows/ash.py @@ -4,7 +4,7 @@ import asyncio from asyncio import timeout as asyncio_timeout import binascii -from collections.abc import Coroutine +from collections.abc import Coroutine, Sequence import contextlib import dataclasses import enum @@ -13,10 +13,7 @@ import time import typing -from zigpy.types import BaseDataclassMixin - -import bellows.types as t -from bellows.zigbee.util import run_length_debug +from zigpy.types import BaseDataclassMixin, enum8 _LOGGER = logging.getLogger(__name__) @@ -63,6 +60,48 @@ class Reserved(enum.IntEnum): ACK_TIMEOUTS = 5 +class NcpResetCode(enum8): + # Reset and Error Codes for NCP + RESET_UNKNOWN_REASON = 0x00 + RESET_EXTERNAL = 0x01 + RESET_POWER_ON = 0x02 + RESET_WATCHDOG = 0x03 + RESET_ASSERT = 0x06 + RESET_BOOTLOADER = 0x09 + RESET_SOFTWARE = 0x0B + ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT = 0x51 + ERROR_UNKNOWN_EM3XX_ERROR = 0x80 + + +def run_length_debug( + items: Sequence[str], + joiner: str, + prefix: str = "", + suffix: str = "", + default: str = "", +) -> str: + """Create a run-length encoded debug string from a sequence of strings.""" + counts = [] + unique_items = [] + + for item in items: + if unique_items and unique_items[-1] == item: + counts[-1] += 1 + else: + counts.append(1) + unique_items.append(item) + + result = joiner.join( + f"{count}*{item}" if count > 1 else item + for item, count in zip(unique_items, counts) + ) + + if not result: + return default + + return prefix + result + suffix + + def generate_random_sequence(length: int) -> bytes: output = bytearray() rand = 0x42 @@ -121,7 +160,8 @@ def __repr__(self) -> str: class NcpFailure(AshException): - def __init__(self, code: t.NcpResetCode) -> None: + def __init__(self, code: NcpResetCode) -> None: + super().__init__(code) self.code = code def __repr__(self) -> str: @@ -135,8 +175,8 @@ def __eq__(self, other: object) -> bool | NotImplemented: class AshFrame(abc.ABC, BaseDataclassMixin): - MASK: t.uint8_t - MASK_VALUE: t.uint8_t + MASK: int + MASK_VALUE: int @classmethod def from_bytes(cls, data: bytes) -> AshFrame: @@ -292,8 +332,8 @@ class RStackFrame(AshFrame): MASK = 0b11111111 MASK_VALUE = 0b11000001 - version: t.uint8_t - reset_code: t.NcpResetCode + version: int + reset_code: NcpResetCode @classmethod def from_bytes(cls, data: bytes) -> RStackFrame: @@ -307,7 +347,7 @@ def from_bytes(cls, data: bytes) -> RStackFrame: if version != 0x02: raise ParsingError(f"Invalid version for RSTACK frame: {data!r}") - reset_code = t.NcpResetCode(data[1]) + reset_code = NcpResetCode(data[1]) return cls( version=version, @@ -323,8 +363,8 @@ class ErrorFrame(AshFrame): MASK = 0b11111111 MASK_VALUE = 0b11000010 - version: t.uint8_t - reset_code: t.NcpResetCode + version: int + reset_code: NcpResetCode # We do not want to inherit from `RStackFrame` from_bytes = classmethod(RStackFrame.from_bytes.__func__) @@ -364,7 +404,7 @@ def __init__(self, ezsp_protocol) -> None: self._rx_seq: int = 0 self._t_rx_ack = T_RX_ACK_INIT - self._ncp_reset_code: t.NcpResetCode | None = None + self._ncp_reset_code: NcpResetCode | None = None self._ncp_state: NcpState = NcpState.CONNECTED self._in_reject_condition: bool = False @@ -605,7 +645,7 @@ def error_frame_received(self, frame: ErrorFrame) -> None: # Cancel all pending requests self._enter_failed_state(self._ncp_reset_code) - def _enter_failed_state(self, reset_code: t.NcpResetCode) -> None: + def _enter_failed_state(self, reset_code: NcpResetCode) -> None: self._ncp_state = NcpState.FAILED self._cancel_pending_data_frames(NcpFailure(code=reset_code)) self._ezsp_protocol.error_received(reset_code) @@ -658,7 +698,7 @@ async def _send_data_frame(self, frame: AshFrame) -> None: "NCP is in a failed state, not sending: %r", frame ) raise NcpFailure( - t.NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT + NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT ) if frm_num is None: @@ -694,7 +734,7 @@ async def _send_data_frame(self, frame: AshFrame) -> None: if attempt >= ACK_TIMEOUTS - 1: self._enter_failed_state( - t.NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT + NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT ) raise except NcpFailure: @@ -715,7 +755,7 @@ async def _send_data_frame(self, frame: AshFrame) -> None: if attempt >= ACK_TIMEOUTS - 1: self._enter_failed_state( - t.NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT + NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT ) raise else: diff --git a/bellows/config/__init__.py b/bellows/config/__init__.py index 74d91d9c5..61a284dda 100644 --- a/bellows/config/__init__.py +++ b/bellows/config/__init__.py @@ -21,6 +21,8 @@ CONF_BELLOWS_CONFIG = "bellows_config" CONF_MANUAL_SOURCE_ROUTING = "manual_source_routing" +# Runs ASH in its own interpreter. The key predates subinterpreters and is kept for +# compatibility with existing configuration CONF_USE_THREAD = "use_thread" CONF_EZSP_CONFIG = "ezsp_config" CONF_EZSP_POLICIES = "ezsp_policies" diff --git a/bellows/ezsp/__init__.py b/bellows/ezsp/__init__.py index a21138292..807477c79 100644 --- a/bellows/ezsp/__init__.py +++ b/bellows/ezsp/__init__.py @@ -162,9 +162,21 @@ async def startup_reset(self) -> None: await self.disconnect() raise - async def connect(self, *, use_thread: bool = True) -> None: + async def connect( + self, *, isolated: bool = True, use_thread: bool | None = None + ) -> None: + # Backwards compatibility + if use_thread is not None: + isolated = use_thread + assert self._gw is None - self._gw = await bellows.uart.connect(self._config, self, use_thread=use_thread) + self._gw = await bellows.uart.connect( + self, + path=self._config[conf.CONF_DEVICE_PATH], + baudrate=self._config[conf.CONF_DEVICE_BAUDRATE], + flow_control=self._config[zigpy.config.CONF_DEVICE_FLOW_CONTROL], + isolated=isolated, + ) await self.startup_reset() async def reset(self): diff --git a/bellows/thread.py b/bellows/thread.py deleted file mode 100644 index a10b58d93..000000000 --- a/bellows/thread.py +++ /dev/null @@ -1,123 +0,0 @@ -import asyncio -from concurrent.futures import ThreadPoolExecutor -import contextlib -import functools -import logging - -LOGGER = logging.getLogger(__name__) - - -class EventLoopThread: - """Run a parallel event loop in a separate thread.""" - - def __init__(self): - self.loop = None - self.thread_complete = None - - def run_coroutine_threadsafe(self, coroutine): - current_loop = asyncio.get_event_loop() - future = asyncio.run_coroutine_threadsafe(coroutine, self.loop) - return asyncio.wrap_future(future, loop=current_loop) - - def _thread_main(self, init_task): - self.loop = asyncio.new_event_loop() - asyncio.set_event_loop(self.loop) - - try: - self.loop.run_until_complete(init_task) - self.loop.run_forever() - finally: - # Publish `None` before closing: `self.loop` then never names a closed loop, and - # this thread no longer writes `self.loop` after a concurrent `start()` may have - # replaced it. - loop, self.loop = self.loop, None - loop.close() - - async def start(self): - current_loop = asyncio.get_event_loop() - if self.loop is not None and not self.loop.is_closed(): - return - - executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix=__name__) - - thread_started_future = current_loop.create_future() - - async def init_task(): - current_loop.call_soon_threadsafe(thread_started_future.set_result, None) - - # Use current loop so current loop has a reference to the long-running thread - # as one of its tasks - thread_complete = current_loop.run_in_executor( - executor, self._thread_main, init_task() - ) - self.thread_complete = thread_complete - current_loop.call_soon(executor.shutdown, False) - await thread_started_future - return thread_complete - - def force_stop(self): - loop = self.loop - if loop is None or loop.is_closed(): - return - - def cancel_tasks_and_stop_loop(): - tasks = asyncio.all_tasks(loop=loop) - - for task in tasks: - loop.call_soon_threadsafe(task.cancel) - - gather = asyncio.gather(*tasks, return_exceptions=True) - gather.add_done_callback(lambda _: loop.call_soon_threadsafe(loop.stop)) - - # The worker thread may close the loop after our is_closed() check. - with contextlib.suppress(RuntimeError): - loop.call_soon_threadsafe(cancel_tasks_and_stop_loop) - - -class ThreadsafeProxy: - """Proxy class which enforces threadsafe non-blocking calls - This class can be used to wrap an object to ensure any calls - using that object's methods are done on a particular event loop - """ - - def __init__(self, obj, obj_loop): - self._obj = obj - self._obj_loop = obj_loop - - def __getattr__(self, name): - func = getattr(self._obj, name) - if not callable(func): - raise TypeError( - "Can only use ThreadsafeProxy with callable attributes: {}.{}".format( - self._obj.__class__.__name__, name - ) - ) - - def func_wrapper(*args, **kwargs): - loop = self._obj_loop - curr_loop = asyncio.get_running_loop() - call = functools.partial(func, *args, **kwargs) - if loop == curr_loop: - return call() - if loop.is_closed(): - # Disconnected - LOGGER.warning("Attempted to use a closed event loop") - return - if asyncio.iscoroutinefunction(func): - future = asyncio.run_coroutine_threadsafe(call(), loop) - return asyncio.wrap_future(future, loop=curr_loop) - else: - - def check_result_wrapper(): - result = call() - if result is not None: - raise TypeError( - ( - "ThreadsafeProxy can only wrap functions with no return" - "value \nUse an async method to return values: {}.{}" - ).format(self._obj.__class__.__name__, name) - ) - - loop.call_soon_threadsafe(check_result_wrapper) - - return func_wrapper diff --git a/bellows/types/named.py b/bellows/types/named.py index aad9c4f0e..ed093a279 100644 --- a/bellows/types/named.py +++ b/bellows/types/named.py @@ -5,6 +5,8 @@ import zigpy.types as ztypes import zigpy.zdo.types as zdo_t +from bellows.ash import NcpResetCode # noqa: F401 + from . import basic LOGGER = logging.getLogger(__name__) @@ -16,19 +18,6 @@ BroadcastAddress = ztypes.BroadcastAddress -class NcpResetCode(basic.enum8): - # Reset and Error Codes for NCP - RESET_UNKNOWN_REASON = 0x00 - RESET_EXTERNAL = 0x01 - RESET_POWER_ON = 0x02 - RESET_WATCHDOG = 0x03 - RESET_ASSERT = 0x06 - RESET_BOOTLOADER = 0x09 - RESET_SOFTWARE = 0x0B - ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT = 0x51 - ERROR_UNKNOWN_EM3XX_ERROR = 0x80 - - class ExtendedPanId(ztypes.ExtendedPanId): """Extended PAN ID.""" diff --git a/bellows/uart.py b/bellows/uart.py index 9849e0a6a..598d2fb3b 100644 --- a/bellows/uart.py +++ b/bellows/uart.py @@ -1,28 +1,58 @@ +from __future__ import annotations + import asyncio from asyncio import timeout as asyncio_timeout +from concurrent.futures import ThreadPoolExecutor +import dataclasses +from io import BytesIO +import itertools import logging +from logging import LogRecord +from logging.handlers import QueueHandler +import pickle +import socket +import struct +import sys -import zigpy.config import zigpy.serial +import zigpy.types + +from bellows.ash import AshProtocol, NcpFailure, NcpResetCode -from bellows.ash import AshProtocol -from bellows.thread import EventLoopThread, ThreadsafeProxy -import bellows.types as t +try: + from concurrent import interpreters +except ImportError: + interpreters = None LOGGER = logging.getLogger(__name__) RESET_TIMEOUT = 2.5 +# The worker interpreter copies these loggers' effective levels from the application +# interpreter when it starts, and forwards every record that passes them +WORKER_LOGGERS = ( + "", + "asyncio", + "bellows.ash", + "bellows.uart", + "zigpy.serial", + "serialx", +) + class Gateway(zigpy.serial.SerialProtocol): - def __init__(self, api, connection_done_future=None): + def __init__(self, api): super().__init__() self._api = api self._reset_future = None self._startup_reset_future = None - self._connection_done_future = connection_done_future + + def _ensure_connected(self) -> None: + if self._transport is None: + raise ConnectionResetError("Connection has been closed") async def send_data(self, data: bytes) -> None: + self._ensure_connected() await self._transport.send_data(data) def data_received(self, data): @@ -31,7 +61,7 @@ def data_received(self, data): # We intentionally do not call `SerialProtocol.data_received` self._api.frame_received(data) - def reset_received(self, code: t.NcpResetCode) -> None: + def reset_received(self, code: NcpResetCode) -> None: """Reset acknowledgement frame receive handler""" LOGGER.debug("Received reset: %r", code) @@ -43,7 +73,7 @@ def reset_received(self, code: t.NcpResetCode) -> None: self._api.enter_failed_state(code) LOGGER.warning("Received an unexpected reset: %r", code) - def error_received(self, code: t.NcpResetCode) -> None: + def error_received(self, code: NcpResetCode) -> None: """Error frame receive handler.""" if self._reset_future is not None or self._startup_reset_future is not None: LOGGER.debug("Ignoring spurious error during reset: %r", code) @@ -52,6 +82,7 @@ def error_received(self, code: t.NcpResetCode) -> None: async def wait_for_startup_reset(self) -> None: """Wait for the first reset frame on startup.""" + self._ensure_connected() assert self._startup_reset_future is None self._startup_reset_future = asyncio.get_running_loop().create_future() @@ -71,20 +102,12 @@ def connection_lost(self, exc): LOGGER.debug("Connection lost: %r", exc) reason = exc or ConnectionResetError("Remote server closed connection") - # XXX: The startup reset future must be resolved with an error *before* the - # "connection done" future is completed: the secondary thread has an attached - # callback to stop itself, which will cause the a future to propagate a - # `CancelledError` into the active event loop, breaking everything! - if self._startup_reset_future: - self._startup_reset_future.set_exception(reason) - - if self._connection_done_future: - self._connection_done_future.set_result(exc) - self._connection_done_future = None + # A future may already be cancelled by a caller whose task has not yet resumed + for future in (self._startup_reset_future, self._reset_future): + if future is not None and not future.done(): + future.set_exception(reason) - if self._reset_future: - self._reset_future.set_exception(reason) - self._reset_future = None + self._reset_future = None self._api.connection_lost(exc) @@ -97,52 +120,540 @@ async def reset(self): ) return await self._reset_future + self._ensure_connected() self._transport.send_reset() - self._reset_future = asyncio.get_event_loop().create_future() + self._reset_future = asyncio.get_running_loop().create_future() self._reset_future.add_done_callback(self._reset_cleanup) async with asyncio_timeout(RESET_TIMEOUT): return await self._reset_future -async def _connect(config, api): - loop = asyncio.get_event_loop() +# Messages between the application interpreter and the worker interpreter. Every message +# is a pickled dataclass behind a 32-bit length prefix. - connection_done_future = loop.create_future() - gateway = Gateway(api, connection_done_future) - protocol = AshProtocol(gateway) +# Application -> worker. Requests carry an id that their reply repeats. +@dataclasses.dataclass(frozen=True) +class SendData: + id: int + data: bytes - flow_control = config[zigpy.config.CONF_DEVICE_FLOW_CONTROL] - transport, _ = await zigpy.serial.create_serial_connection( - loop, - lambda: protocol, - url=config[zigpy.config.CONF_DEVICE_PATH], - baudrate=config[zigpy.config.CONF_DEVICE_BAUDRATE], - xonxoff=flow_control in (None, "software"), - rtscts=flow_control == "hardware", - ) +@dataclasses.dataclass(frozen=True) +class Reset: + id: int + + +@dataclasses.dataclass(frozen=True) +class WaitForStartupReset: + id: int + + +@dataclasses.dataclass(frozen=True) +class Cancel: + id: int + + +# Worker -> application +@dataclasses.dataclass(frozen=True) +class Connected: + pass + + +@dataclasses.dataclass(frozen=True) +class ConnectFailed: + exc: Exception + + +@dataclasses.dataclass(frozen=True) +class Done: + id: int + result: object + + +@dataclasses.dataclass(frozen=True) +class Failed: + id: int + exc: Exception + + +@dataclasses.dataclass(frozen=True) +class FrameReceived: + data: bytes + + +@dataclasses.dataclass(frozen=True) +class FailedState: + code: NcpResetCode + + +@dataclasses.dataclass(frozen=True) +class ConnectionLost: + exc: Exception | None + + +@dataclasses.dataclass(frozen=True) +class Log: + record: LogRecord - await gateway.wait_until_connected() - thread_safe_protocol = ThreadsafeProxy(gateway, loop) - return thread_safe_protocol, connection_done_future +# Everything that may cross the boundary. Pickle refuses to send or receive any other +# class, so the protocol's vocabulary is this list and not whatever an object graph holds. +_ALLOWED_CLASSES = frozenset( + { + SendData, + Reset, + WaitForStartupReset, + Cancel, + Connected, + ConnectFailed, + Done, + Failed, + FrameReceived, + FailedState, + ConnectionLost, + Log, + LogRecord, + NcpResetCode, + # `NcpResetCode` pickles its value as this + zigpy.types.uint8_t, + # The exceptions ASH, the gateway and the serial transport raise + NcpFailure, + AssertionError, + RuntimeError, + TimeoutError, + OSError, + BlockingIOError, + BrokenPipeError, + ConnectionAbortedError, + ConnectionError, + ConnectionRefusedError, + ConnectionResetError, + FileNotFoundError, + IsADirectoryError, + NotADirectoryError, + PermissionError, + } +) +_ALLOWED_GLOBALS = {(cls.__module__, cls.__qualname__): cls for cls in _ALLOWED_CLASSES} -async def connect(config, api, use_thread=True): - if use_thread: - api = ThreadsafeProxy(api, asyncio.get_event_loop()) - thread = EventLoopThread() - await thread.start() +class _Pickler(pickle.Pickler): + def reducer_override(self, obj): + # Called for every instance and class in the graph, not for primitives + cls = obj if isinstance(obj, type) else type(obj) + + if cls not in _ALLOWED_CLASSES: + raise pickle.PicklingError( + f"{cls!r} may not cross the interpreter boundary" + ) + + return NotImplemented + + +class _Unpickler(pickle.Unpickler): + def find_class(self, module, name): + if (module, name) not in _ALLOWED_GLOBALS: + raise pickle.UnpicklingError( + f"{module}.{name} may not cross the interpreter boundary" + ) + + return _ALLOWED_GLOBALS[module, name] + + +def _dumps(message) -> bytes: + buffer = BytesIO() + _Pickler(buffer).dump(message) + return buffer.getvalue() + + +def _portable(exc: BaseException | None) -> Exception | None: + """The exception itself if the protocol can carry it, otherwise one describing it.""" + try: + _dumps(exc) + except pickle.PicklingError: + return RuntimeError(f"{type(exc).__qualname__}: {exc}") + + return exc + + +def _write(writer: asyncio.StreamWriter, message) -> None: + payload = _dumps(message) + writer.write(struct.pack(" LogRecord: + # The message is formatted and `args` is cleared; `()` is what a record without + # arguments normally carries + record = super().prepare(record) + record.args = () + return record + + def enqueue(self, record: LogRecord) -> None: + # Records also come from executor threads, the writer is only used on the loop + self._loop.call_soon_threadsafe(_write, self._writer, Log(record)) + + +class _Worker: + """The worker interpreter's side: owns the `Gateway` and serves requests for it.""" + + def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter): + self._reader = reader + self._writer = writer + self._gateway = None + self._requests: dict[int, asyncio.Task] = {} + + async def run(self, path: str, baudrate: int, flow_control: str | None) -> None: + connect = asyncio.ensure_future(self._connect(path, baudrate, flow_control)) + try: - protocol, connection_done = await thread.run_coroutine_threadsafe( - _connect(config, api) + # Until the application closes its end, which it may do while connecting + await self._read_messages() + finally: + connect.cancel() + + for task in self._requests.values(): + task.cancel() + + await asyncio.gather( + connect, *self._requests.values(), return_exceptions=True ) - except Exception: - thread.force_stop() + + if self._gateway is not None: + await self._gateway.disconnect() + + async def _connect(self, path: str, baudrate: int, flow_control: str | None): + try: + self._gateway = await _connect( + self, path=path, baudrate=baudrate, flow_control=flow_control + ) + except Exception as exc: + _write(self._writer, ConnectFailed(_portable(exc))) + else: + _write(self._writer, Connected()) + + async def _read_messages(self) -> None: + while (message := await _read(self._reader)) is not None: + match message: + case SendData(id=id, data=data): + self._start(id, self._gateway.send_data(data)) + case Reset(id=id): + self._start(id, self._gateway.reset()) + case WaitForStartupReset(id=id): + self._start(id, self._gateway.wait_for_startup_reset()) + case Cancel(id=id): + # The request may have finished just before the cancel arrived + if id in self._requests: + self._requests[id].cancel() + case _: + raise TypeError(f"Unexpected message: {message!r}") + + def _start(self, id: int, coro) -> None: + task = asyncio.ensure_future(coro) + self._requests[id] = task + task.add_done_callback(lambda task: self._finish(id, task)) + + def _finish(self, id: int, task: asyncio.Task) -> None: + del self._requests[id] + + if task.cancelled(): + # The application already gave up on it + return + + if task.exception() is not None: + _write(self._writer, Failed(id, _portable(task.exception()))) + else: + _write(self._writer, Done(id, task.result())) + + # Called by the `Gateway` + def frame_received(self, data: bytes) -> None: + _write(self._writer, FrameReceived(data)) + + def enter_failed_state(self, code: NcpResetCode) -> None: + _write(self._writer, FailedState(code)) + + def connection_lost(self, exc: Exception | None) -> None: + _write(self._writer, ConnectionLost(_portable(exc))) + + +async def _run_worker( + sock: socket.socket, + path: str, + baudrate: int, + flow_control: str | None, + log_levels: tuple[tuple[str, int], ...], +) -> None: + loop = asyncio.get_running_loop() + reader, writer = await asyncio.open_connection(sock=sock) + + for name, level in log_levels: + logging.getLogger(name).setLevel(level) + + handler = _ForwardingHandler(loop, writer) + logging.getLogger().addHandler(handler) + + try: + await _Worker(reader, writer).run(path, baudrate, flow_control) + finally: + logging.getLogger().removeHandler(handler) + writer.close() + await writer.wait_closed() + + +def _worker_main(fd: int, *args) -> None: + """Entry point in the worker interpreter.""" + asyncio.run(_run_worker(socket.socket(fileno=fd), *args)) + + +def _launch_worker(loop: asyncio.AbstractEventLoop, *args): + """Start the worker interpreter. Returns it and a future that resolves when it exits.""" + interpreter = interpreters.create() + + def call(): + try: + interpreter.call(_worker_main, *args) + except interpreters.ExecutionFailed as exc: + LOGGER.error("Worker interpreter crashed:\n%s", exc.excinfo.errdisplay) + + executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix=__name__) + # The thread only hosts the worker interpreter, which has its own GIL + future = loop.run_in_executor(executor, call) + executor.shutdown(wait=False) + + return interpreter, future + + +class IsolatedGateway: + """Runs a `Gateway` on an event loop in its own interpreter and bridges the two. + + The application interpreter's garbage collector stops every thread sharing its GIL, + while ASH acknowledgements have a deadline. The worker interpreter has its own GIL. + + The worker owns the gateway and exits only once the application closes its end of + the socket, so every request that passed the `_closed` check reached it before the + close did. + """ + + def __init__(self, api): + self._api = api + self._loop = asyncio.get_running_loop() + self._interpreter = None + self._worker_complete = None + self._writer = None + self._messages = None + self._connected = self._loop.create_future() + self._requests: dict[int, asyncio.Future] = {} + self._ids = itertools.count() + self._closed = False + self._lost_exc = None + + async def connect( + self, *, path: str, baudrate: int, flow_control: str | None + ) -> None: + ours, theirs = socket.socketpair() + log_levels = tuple( + (name, logging.getLogger(name).getEffectiveLevel()) + for name in WORKER_LOGGERS + ) + + self._interpreter, self._worker_complete = _launch_worker( + self._loop, theirs.detach(), path, baudrate, flow_control, log_levels + ) + + reader, self._writer = await asyncio.open_connection(sock=ours) + self._messages = asyncio.ensure_future(self._read_messages(reader)) + + try: + await asyncio.shield(self._connected) + except BaseException: + # Wait for the worker to exit so the port is released before the caller sees + # the failure and tries to reconnect + self._connected.cancel() + self._close() + await self._finish() raise - connection_done.add_done_callback(lambda _: thread.force_stop()) - else: - protocol, _ = await _connect(config, api) - return protocol + + async def _read_messages(self, reader: asyncio.StreamReader) -> None: + while (message := await _read(reader)) is not None: + match message: + case Connected(): + if not self._connected.done(): + self._connected.set_result(None) + case ConnectFailed(exc=exc): + if not self._connected.done(): + self._connected.set_exception(exc) + case Done(id=id, result=result): + if id in self._requests: + self._requests.pop(id).set_result(result) + case Failed(id=id, exc=exc): + if id in self._requests: + self._requests.pop(id).set_exception(exc) + case FrameReceived(data=data): + self._api.frame_received(data) + case FailedState(code=code): + self._api.enter_failed_state(code) + case ConnectionLost(exc=exc): + self._connection_lost(exc) + case Log(record=record): + logging.getLogger(record.name).handle(record) + case _: + raise TypeError(f"Unexpected message: {message!r}") + + # The worker has exited, which is expected only after we closed our end + crashed = not self._closed + self._closed = True + + if not self._connected.done(): + self._connected.set_exception( + ConnectionResetError("Worker exited before connecting") + ) + elif crashed: + self._connection_lost(ConnectionResetError("Worker exited unexpectedly")) + + # Requests the worker cancelled while shutting down, instead of answering + for future in self._requests.values(): + future.set_exception(self._disconnected_error()) + + self._requests.clear() + + def _connection_lost(self, exc: Exception | None) -> None: + # The transport has already closed its file descriptor by the time it reports this + self._lost_exc = exc + self._close() + self._api.connection_lost(exc) + + def _close(self) -> None: + if self._closed: + return + + self._closed = True + self._writer.write_eof() + + def _disconnected_error(self) -> ConnectionResetError: + # Chain the transport's own error, since every call made after the loss fails here + exc = ConnectionResetError("Gateway has been disconnected") + exc.__cause__ = self._lost_exc + return exc + + async def _request(self, message_type, *args): + if self._closed: + raise self._disconnected_error() + + id = next(self._ids) + future = self._loop.create_future() + self._requests[id] = future + _write(self._writer, message_type(id, *args)) + + try: + return await future + except asyncio.CancelledError: + # Our caller gave up; a reply arriving later is dropped + del self._requests[id] + + if not self._closed: + _write(self._writer, Cancel(id)) + + raise + + async def send_data(self, data: bytes) -> None: + await self._request(SendData, data) + + async def reset(self): + return await self._request(Reset) + + async def wait_for_startup_reset(self) -> None: + await self._request(WaitForStartupReset) + + async def disconnect(self) -> None: + self._close() + await self._finish() + + async def _finish(self) -> None: + await self._worker_complete + await self._messages + self._writer.close() + + if self._interpreter is not None: + self._interpreter.close() + self._interpreter = None + + +async def _connect(api, *, path: str, baudrate: int, flow_control: str | None): + loop = asyncio.get_running_loop() + + gateway = Gateway(api) + protocol = AshProtocol(gateway) + + transport, _ = await zigpy.serial.create_serial_connection( + loop, + lambda: protocol, + url=path, + baudrate=baudrate, + xonxoff=flow_control in (None, "software"), + rtscts=flow_control == "hardware", + ) + + # `connection_made` arrives on a later loop iteration, so the port is open before we + # are connected and must be closed if we are cancelled in between + try: + await gateway.wait_until_connected() + except BaseException: + transport.close() + await transport.wait_closed() + raise + + return gateway + + +async def connect( + api, + *, + path: str, + baudrate: int, + flow_control: str | None, + isolated: bool = True, + use_thread: bool | None = None, +): + # Backwards compatibility + if use_thread is not None: + isolated = use_thread + + if isolated and interpreters is None: + LOGGER.warning( + "Python %s has no subinterpreters, running ASH on the application loop", + sys.version.split()[0], + ) + isolated = False + + if not isolated: + return await _connect( + api, path=path, baudrate=baudrate, flow_control=flow_control + ) + + gateway = IsolatedGateway(api) + await gateway.connect(path=path, baudrate=baudrate, flow_control=flow_control) + + return gateway diff --git a/bellows/zigbee/application.py b/bellows/zigbee/application.py index 11b3eb0f2..e89530615 100644 --- a/bellows/zigbee/application.py +++ b/bellows/zigbee/application.py @@ -187,7 +187,7 @@ async def connect(self) -> None: self._ezsp = bellows.ezsp.EZSP(self.config[zigpy.config.CONF_DEVICE], self) try: - await self._ezsp.connect(use_thread=self.config[CONF_USE_THREAD]) + await self._ezsp.connect(isolated=self.config[CONF_USE_THREAD]) # Writing config is required here because network info can't be loaded await self._ezsp.write_config(self.config[CONF_EZSP_CONFIG]) diff --git a/bellows/zigbee/util.py b/bellows/zigbee/util.py index 12e38079f..0b7079d93 100644 --- a/bellows/zigbee/util.py +++ b/bellows/zigbee/util.py @@ -1,4 +1,3 @@ -from collections.abc import Sequence import logging import math @@ -130,32 +129,3 @@ def map_energy_to_rssi(lqi: float) -> float: x_0=RSSI_MIN + 0.45 * (RSSI_MAX - RSSI_MIN), k=0.13, ) - - -def run_length_debug( - items: Sequence[str], - joiner: str, - prefix: str = "", - suffix: str = "", - default: str = "", -) -> str: - """Create a run-length encoded debug string from a sequence of strings.""" - counts = [] - unique_items = [] - - for item in items: - if unique_items and unique_items[-1] == item: - counts[-1] += 1 - else: - counts.append(1) - unique_items.append(item) - - result = joiner.join( - f"{count}*{item}" if count > 1 else item - for item, count in zip(unique_items, counts) - ) - - if not result: - return default - - return prefix + result + suffix diff --git a/tests/test_ezsp.py b/tests/test_ezsp.py index bccf17b39..4f4060136 100644 --- a/tests/test_ezsp.py +++ b/tests/test_ezsp.py @@ -24,6 +24,7 @@ DEVICE_CONFIG = { zigpy.config.CONF_DEVICE_PATH: "/dev/null", zigpy.config.CONF_DEVICE_BAUDRATE: 115200, + zigpy.config.CONF_DEVICE_FLOW_CONTROL: None, } diff --git a/tests/test_thread.py b/tests/test_thread.py deleted file mode 100644 index 235507f7e..000000000 --- a/tests/test_thread.py +++ /dev/null @@ -1,213 +0,0 @@ -import asyncio -from asyncio import timeout as asyncio_timeout -import threading -from unittest import mock - -import pytest - -from bellows.thread import EventLoopThread, ThreadsafeProxy - - -async def test_thread_start(monkeypatch): - current_loop = asyncio.get_event_loop() - loopmock = mock.MagicMock() - - monkeypatch.setattr(asyncio, "new_event_loop", lambda: loopmock) - monkeypatch.setattr(asyncio, "set_event_loop", lambda loop: None) - - def mockrun(task): - future = asyncio.run_coroutine_threadsafe(task, loop=current_loop) - return future.result(1) - - loopmock.run_until_complete.side_effect = mockrun - thread = EventLoopThread() - thread_complete = await thread.start() - await thread_complete - - assert loopmock.run_until_complete.call_count == 1 - assert loopmock.run_forever.call_count == 1 - assert loopmock.close.call_count == 1 - - -async def test_thread_clears_loop_before_closing(monkeypatch): - """`self.loop` is cleared before the loop is closed, so it never names a closed loop.""" - thread = EventLoopThread() - real_new_event_loop = asyncio.new_event_loop - loop_at_close = mock.sentinel.close_not_called - - def new_event_loop(): - loop = real_new_event_loop() - real_close = loop.close - - def close(): - nonlocal loop_at_close - loop_at_close = thread.loop - real_close() - - loop.close = close - return loop - - monkeypatch.setattr(asyncio, "new_event_loop", new_event_loop) - - thread_complete = await thread.start() - thread.force_stop() - async with asyncio_timeout(1): - await thread_complete - - assert loop_at_close is None - assert thread.loop is None - - [t.join(1) for t in threading.enumerate() if "bellows" in t.name] - - -class ExceptionCollector: - def __init__(self): - self.exceptions = [] - - def __call__(self, thread_loop, context): - exc = context.get("exception") or Exception(context["message"]) - self.exceptions.append(exc) - - -@pytest.fixture -async def thread(): - thread = EventLoopThread() - await thread.start() - thread.loop.call_soon_threadsafe( - thread.loop.set_exception_handler, ExceptionCollector() - ) - yield thread - thread.force_stop() - if thread.thread_complete is not None: - async with asyncio_timeout(1): - await thread.thread_complete - [t.join(1) for t in threading.enumerate() if "bellows" in t.name] - threads = [t for t in threading.enumerate() if "bellows" in t.name] - assert len(threads) == 0 - - -async def yield_other_thread(thread): - await thread.run_coroutine_threadsafe(asyncio.sleep(0)) - - exception_collector = thread.loop.get_exception_handler() - if exception_collector.exceptions: - raise exception_collector.exceptions[0] - - -async def test_thread_loop(thread): - async def test_coroutine(): - return mock.sentinel.result - - future = asyncio.run_coroutine_threadsafe(test_coroutine(), loop=thread.loop) - result = await asyncio.wrap_future(future, loop=asyncio.get_event_loop()) - assert result is mock.sentinel.result - - -async def test_thread_double_start(thread): - previous_loop = thread.loop - await thread.start() - threads = [t for t in threading.enumerate() if "bellows" in t.name] - assert len(threads) == 1 - assert thread.loop is previous_loop - - -async def test_thread_already_stopped(thread): - thread.force_stop() - thread.force_stop() - - -async def test_thread_run_coroutine_threadsafe(thread): - inner_loop = None - - async def test_coroutine(): - nonlocal inner_loop - inner_loop = asyncio.get_event_loop() - return mock.sentinel.result - - result = await thread.run_coroutine_threadsafe(test_coroutine()) - assert result is mock.sentinel.result - assert inner_loop is thread.loop - - -async def test_proxy_callback(thread): - obj = mock.MagicMock() - proxy = ThreadsafeProxy(obj, thread.loop) - obj.test.return_value = None - proxy.test() - await yield_other_thread(thread) - assert obj.test.call_count == 1 - - -async def test_proxy_async(thread): - obj = mock.MagicMock() - proxy = ThreadsafeProxy(obj, thread.loop) - call_count = 0 - - async def magic(): - nonlocal thread, call_count - assert asyncio.get_event_loop() == thread.loop - call_count += 1 - return mock.sentinel.result - - obj.test = magic - result = await proxy.test() - - assert call_count == 1 - assert result == mock.sentinel.result - - -async def test_proxy_bad_function(thread): - obj = mock.MagicMock() - proxy = ThreadsafeProxy(obj, thread.loop) - obj.test.return_value = mock.sentinel.value - - with pytest.raises(TypeError): - proxy.test() - await yield_other_thread(thread) - - -async def test_proxy_not_function(): - loop = asyncio.get_event_loop() - obj = mock.MagicMock() - proxy = ThreadsafeProxy(obj, loop) - obj.test = mock.sentinel.value - with pytest.raises(TypeError): - proxy.test - - -async def test_proxy_no_thread(): - loop = asyncio.get_event_loop() - obj = mock.MagicMock() - proxy = ThreadsafeProxy(obj, loop) - proxy.test() - assert obj.test.call_count == 1 - - -async def test_proxy_loop_closed(): - loop = asyncio.new_event_loop() - obj = mock.MagicMock() - proxy = ThreadsafeProxy(obj, loop) - loop.close() - proxy.test() - assert obj.test.call_count == 0 - - -async def test_thread_task_cancellation_after_stop(thread): - loop = asyncio.get_event_loop() - obj = mock.MagicMock() - - async def wait_forever(): - return await thread.loop.create_future() - - obj.wait_forever = wait_forever - - # Stop the thread while we're waiting - loop.call_later(0.1, thread.force_stop) - - proxy = ThreadsafeProxy(obj, thread.loop) - - # The cancellation should propagate to the outer event loop - with pytest.raises(asyncio.CancelledError): - # This will stall forever without the patch - async with asyncio_timeout(1): - await proxy.wait_forever() diff --git a/tests/test_uart.py b/tests/test_uart.py index 1bfda7c20..836e95f90 100644 --- a/tests/test_uart.py +++ b/tests/test_uart.py @@ -1,14 +1,31 @@ import asyncio +from asyncio import timeout as asyncio_timeout +import logging +import pickle +import socket +import struct import threading +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, call, patch, sentinel import pytest -import zigpy.config as conf import zigpy.serial from bellows import uart +from bellows.ash import AshProtocol, NcpFailure, NcpResetCode, Reserved, RStackFrame import bellows.types as t +DEVICE = {"path": "/dev/serial", "baudrate": 115200, "flow_control": None} + + +def bellows_threads(): + return [t for t in threading.enumerate() if "bellows" in t.name] + + +def assert_no_threads(): + [t.join(1) for t in bellows_threads()] + assert bellows_threads() == [] + @pytest.mark.parametrize( ("flow_control", "xonxoff", "rtscts"), @@ -18,136 +35,522 @@ ("hardware", False, True), ], ) -async def test_connect(flow_control, xonxoff, rtscts, monkeypatch): +async def test_connect(flow_control, xonxoff, rtscts): appmock = MagicMock() transport = MagicMock() - connect_kwargs = {} + kwargs = {} - async def mockconnect(loop, protocol_factory, **kwargs): - connect_kwargs.update(kwargs) + async def mockconnect(loop, protocol_factory, **kw): + kwargs.update(kw) protocol = protocol_factory() loop.call_soon(protocol.connection_made, transport) return None, protocol - monkeypatch.setattr(zigpy.serial, "create_serial_connection", mockconnect) - gw = await uart.connect( - conf.SCHEMA_DEVICE( - { - conf.CONF_DEVICE_PATH: "/dev/serial", - conf.CONF_DEVICE_BAUDRATE: 115200, - conf.CONF_DEVICE_FLOW_CONTROL: flow_control, - } - ), - appmock, - use_thread=False, - ) - - assert connect_kwargs["xonxoff"] is xonxoff - assert connect_kwargs["rtscts"] is rtscts - - threads = [t for t in threading.enumerate() if "bellows" in t.name] - assert len(threads) == 0 + with patch("zigpy.serial.create_serial_connection", mockconnect): + gw = await uart.connect( + appmock, + path="/dev/serial", + baudrate=115200, + flow_control=flow_control, + isolated=False, + ) + + assert isinstance(gw, uart.Gateway) + assert kwargs == { + "url": "/dev/serial", + "baudrate": 115200, + "xonxoff": xonxoff, + "rtscts": rtscts, + } gw.close() -async def test_connect_threaded(monkeypatch): - appmock = MagicMock() +class MockSerial: + """Stands in for `zigpy.serial.create_serial_connection` and remembers the worker side.""" + + def __init__(self): + self.transport = MagicMock() + self.transport.close.side_effect = self.on_transport_close + self.protocol = None + + async def __call__(self, loop, protocol_factory, **kwargs): + self.protocol = protocol_factory() + loop.call_soon(self.protocol.connection_made, self.transport) + return self.transport, self.protocol + + def on_transport_close(self): + self.protocol.connection_lost(None) + + def lose_connection(self, exc): + asyncio.get_running_loop().call_soon(self.protocol.connection_lost, exc) + + +@pytest.fixture +def serial(): + serial = MockSerial() + + with patch("zigpy.serial.create_serial_connection", serial): + yield serial + + +@pytest.fixture +def app(): + return MagicMock() + + +@pytest.fixture +def gateways(): + """Every `Gateway` the worker creates, since the worker does not hand them out.""" + gateways = [] + + class RecordingGateway(uart.Gateway): + def __init__(self, api): + super().__init__(api) + gateways.append(self) + + with patch.object(uart, "Gateway", RecordingGateway): + yield gateways + + +async def in_process_worker(sock, path, baudrate, flow_control, log_levels): + reader, writer = await asyncio.open_connection(sock=sock) + + try: + await uart._Worker(reader, writer).run(path, baudrate, flow_control) + finally: + writer.close() + await writer.wait_closed() + + +@pytest.fixture +def workers(): + """Runs workers on the test loop instead of in a subinterpreter, so mocks reach them.""" + workers = SimpleNamespace(tasks=[], worker=in_process_worker) + + def launch(loop, fd, *args): + task = asyncio.ensure_future(workers.worker(socket.socket(fileno=fd), *args)) + workers.tasks.append(task) + return None, task + + # The protocol does not need subinterpreters, so these tests run on every Python + with ( + patch.object(uart, "_launch_worker", launch), + patch.object(uart, "interpreters", MagicMock()), + ): + yield workers + + +@pytest.fixture +async def isolated_gw(workers, gateways, serial, app): + gw = await uart.connect(app, **DEVICE) + assert isinstance(gw, uart.IsolatedGateway) + + yield gw + + async with asyncio_timeout(1): + await gw.disconnect() + + assert workers.tasks[0].done() + + +async def test_connect_isolated(isolated_gw, workers, serial, app): + async with asyncio_timeout(1): + await isolated_gw.disconnect() + + assert serial.transport.close.call_count == 1 + assert app.connection_lost.mock_calls == [call(None)] + assert workers.tasks[0].done() + + # A second disconnect is a no-op + async with asyncio_timeout(1): + await isolated_gw.disconnect() + + +async def test_connect_isolated_failure(workers): + mockconnect = AsyncMock(side_effect=OSError) + + with ( + patch("zigpy.serial.create_serial_connection", mockconnect), + pytest.raises(OSError), + ): + await uart.connect(MagicMock(), **DEVICE) + + assert workers.tasks[0].done() + + +async def test_connect_isolated_cancelled(workers): + async def mockconnect(loop, protocol_factory, **kwargs): + await asyncio.get_running_loop().create_future() + + with patch("zigpy.serial.create_serial_connection", mockconnect): + task = asyncio.create_task(uart.connect(MagicMock(), **DEVICE)) + await asyncio.sleep(0.1) + task.cancel() + + with pytest.raises(asyncio.CancelledError): + await task + + assert workers.tasks[0].done() + + +async def test_connect_isolated_cancelled_after_port_opened(workers): transport = MagicMock() + transport.wait_closed = AsyncMock() async def mockconnect(loop, protocol_factory, **kwargs): protocol = protocol_factory() - loop.call_soon(protocol.connection_made, transport) - return None, protocol + loop.call_later(1, protocol.connection_made, transport) + return transport, protocol - monkeypatch.setattr(zigpy.serial, "create_serial_connection", mockconnect) + with patch("zigpy.serial.create_serial_connection", mockconnect): + task = asyncio.create_task(uart.connect(MagicMock(), **DEVICE)) + await asyncio.sleep(0.1) + task.cancel() - def on_transport_close(): - gw.connection_lost(None) + with pytest.raises(asyncio.CancelledError): + await task - transport.close.side_effect = on_transport_close - gw = await uart.connect( - conf.SCHEMA_DEVICE( - {conf.CONF_DEVICE_PATH: "/dev/serial", conf.CONF_DEVICE_BAUDRATE: 115200} - ), - appmock, - ) - - # Need to close to release thread - gw.close() + assert transport.close.call_count == 1 + assert transport.wait_closed.mock_calls == [call()] + assert workers.tasks[0].done() - # Ensure all threads are cleaned up - [t.join(1) for t in threading.enumerate() if "bellows" in t.name] - threads = [t for t in threading.enumerate() if "bellows" in t.name] - assert len(threads) == 0 +async def test_connect_isolated_cancelled_after_connected(workers, serial, app): + """Cancelled once the worker has connected, before `Connected` has been read here.""" + connected = asyncio.get_running_loop().create_future() + real_connect = uart._connect -async def test_connect_threaded_failure(monkeypatch): - appmock = MagicMock() - transport = MagicMock() + async def _connect(api, **kwargs): + gateway = await real_connect(api, **kwargs) + connected.set_result(None) + return gateway - mockconnect = AsyncMock() - mockconnect.side_effect = OSError + with patch.object(uart, "_connect", _connect): + task = asyncio.create_task(uart.connect(app, **DEVICE)) + await connected + task.cancel() - monkeypatch.setattr(zigpy.serial, "create_serial_connection", mockconnect) + with pytest.raises(asyncio.CancelledError): + await task - def on_transport_close(): - gw.connection_lost(None) + assert serial.transport.close.call_count == 1 + assert workers.tasks[0].done() - transport.close.side_effect = on_transport_close - with pytest.raises(OSError): - gw = await uart.connect( - conf.SCHEMA_DEVICE( - { - conf.CONF_DEVICE_PATH: "/dev/serial", - conf.CONF_DEVICE_BAUDRATE: 115200, - } - ), - appmock, - ) - # Ensure all threads are cleaned up - [t.join(1) for t in threading.enumerate() if "bellows" in t.name] - threads = [t for t in threading.enumerate() if "bellows" in t.name] - assert len(threads) == 0 +async def test_isolated_calls_run_on_worker(isolated_gw, gateways, serial, app): + sent = [] + async def send_data(data): + sent.append(data) -async def test_connect_threaded_failure_cancellation_propagation(monkeypatch): - appmock = MagicMock() + # `serial.protocol` is the `AshProtocol`, the gateway's transport + serial.protocol.send_data = send_data - async def mock_connect(loop, protocol_factory, *args, **kwargs): - protocol = protocol_factory() - transport = AsyncMock() + await isolated_gw.send_data(b"data") + assert sent == [b"data"] - protocol.connection_made(transport) + gateways[0].data_received(b"frame") + gateways[0].error_received(NcpResetCode.RESET_SOFTWARE) - return transport, protocol + # Messages arrive in order, so the events have been handled once this reply has + await isolated_gw.send_data(b"data") - with patch("bellows.uart.zigpy.serial.create_serial_connection", mock_connect): - gw = await uart.connect( - conf.SCHEMA_DEVICE( - { - conf.CONF_DEVICE_PATH: "/dev/serial", - conf.CONF_DEVICE_BAUDRATE: 115200, - } - ), - appmock, - use_thread=True, + assert app.frame_received.mock_calls == [call(b"frame")] + assert app.enter_failed_state.mock_calls == [call(NcpResetCode.RESET_SOFTWARE)] + + +async def test_isolated_reset(isolated_gw, gateways, serial): + def send_reset(): + asyncio.get_running_loop().call_soon( + gateways[0].reset_received, NcpResetCode.RESET_SOFTWARE ) - # Begin waiting for the startup reset - wait_for_reset = gw.wait_for_startup_reset() + serial.protocol.send_reset = send_reset + + async with asyncio_timeout(1): + assert await isolated_gw.reset() is True + + +async def test_isolated_request_failure(isolated_gw, serial): + code = NcpResetCode.ERROR_EXCEEDED_MAXIMUM_ACK_TIMEOUT_COUNT + serial.protocol.send_data = AsyncMock(side_effect=NcpFailure(code)) + + with pytest.raises(NcpFailure) as exc_info: + await isolated_gw.send_data(b"data") + + assert exc_info.value.code == code + + serial.protocol.send_data = AsyncMock(side_effect=ValueError("odd")) + + with pytest.raises(RuntimeError, match="ValueError: odd"): + await isolated_gw.send_data(b"data") + + +async def test_isolated_connection_lost(isolated_gw, serial, app): + wait_for_reset = asyncio.ensure_future(isolated_gw.wait_for_startup_reset()) + await asyncio.sleep(0.1) + serial.lose_connection(RuntimeError("boom")) + + # The failure propagates to the application loop + with pytest.raises(RuntimeError, match="boom"): + async with asyncio_timeout(1): + await wait_for_reset - # But lose connection halfway through - asyncio.get_running_loop().call_later(0.1, gw.connection_lost, RuntimeError()) + async with asyncio_timeout(1): + await isolated_gw.disconnect() - # Cancellation should propagate to the outer loop - with pytest.raises(RuntimeError): - await wait_for_reset + assert len(app.connection_lost.mock_calls) == 1 + (exc,) = app.connection_lost.mock_calls[0].args + assert isinstance(exc, RuntimeError) and exc.args == ("boom",) - # Ensure all threads are cleaned up - [t.join(1) for t in threading.enumerate() if "bellows" in t.name] - threads = [t for t in threading.enumerate() if "bellows" in t.name] - assert len(threads) == 0 + # Later calls fail with the transport's error as their cause + with pytest.raises(ConnectionResetError) as exc_info: + await isolated_gw.send_data(b"data") + + assert isinstance(exc_info.value.__cause__, RuntimeError) + assert exc_info.value.__cause__.args == ("boom",) + + +async def test_isolated_call_after_disconnect(isolated_gw): + async with asyncio_timeout(1): + await isolated_gw.disconnect() + + with pytest.raises(ConnectionResetError) as exc_info: + await isolated_gw.reset() + + assert exc_info.value.__cause__ is None + + +async def test_isolated_call_after_connection_lost_on_worker(isolated_gw, serial, app): + """A call queued before the loss is delivered here fails instead of hanging.""" + lost = asyncio.get_running_loop().create_future() + app.connection_lost.side_effect = lambda exc: lost.set_result(exc) + + # Lose the connection on the worker and queue a call behind it, before the application + # loop has been told + serial.lose_connection(None) + send = asyncio.ensure_future(isolated_gw.send_data(b"data")) + + with pytest.raises(ConnectionResetError): + async with asyncio_timeout(1): + await send + + async with asyncio_timeout(1): + await lost + + +async def test_isolated_stuck_task_on_shutdown(isolated_gw, serial): + async def wait_forever(data): + await asyncio.get_running_loop().create_future() + + serial.protocol.send_data = wait_forever + + send = asyncio.ensure_future(isolated_gw.send_data(b"data")) + await asyncio.sleep(0.1) + serial.lose_connection(RuntimeError("boom")) + + # Cancelled by the worker during shutdown, so the caller sees a connection error + with pytest.raises(ConnectionResetError) as exc_info: + async with asyncio_timeout(1): + await send + + assert isinstance(exc_info.value.__cause__, RuntimeError) + + +async def test_isolated_caller_cancellation(isolated_gw, serial): + cancelled = asyncio.get_running_loop().create_future() + + async def wait_forever(data): + try: + await asyncio.get_running_loop().create_future() + except asyncio.CancelledError: + cancelled.set_result(None) + raise + + serial.protocol.send_data = wait_forever + + with pytest.raises(TimeoutError): + async with asyncio_timeout(0.1): + await isolated_gw.send_data(b"data") + + # The cancellation reached the worker + async with asyncio_timeout(1): + await cancelled + + +def _frame(payload: bytes) -> bytes: + return struct.pack("", + "Received reset: ", + ] + + +@needs_interpreters +async def test_interpreter_connect_failure(app): + with pytest.raises(FileNotFoundError): + await uart.connect( + app, path="/nonexistent/serial/port", baudrate=115200, flow_control=None + ) + + assert_no_threads() @pytest.fixture @@ -169,12 +572,37 @@ async def test_reset_timeout(gw, monkeypatch): async def test_reset_old(gw): - future = asyncio.get_event_loop().create_future() + future = asyncio.get_running_loop().create_future() future.set_result(sentinel.result) gw._reset_future = future ret = await gw.reset() assert ret == sentinel.result - gw._transport.write.assert_not_called() + assert len(gw._transport.write.mock_calls) == 0 + + +async def test_disconnected(gw): + gw._transport = None + + with pytest.raises(ConnectionResetError): + await gw.send_data(b"data") + + with pytest.raises(ConnectionResetError): + await gw.reset() + + with pytest.raises(ConnectionResetError): + await gw.wait_for_startup_reset() + + +async def test_connection_lost_cancelled_startup_reset(gw): + task = asyncio.create_task(gw.wait_for_startup_reset()) + await asyncio.sleep(0) + + # The future is cancelled synchronously, the task only resumes on a later step + task.cancel() + gw.connection_lost(RuntimeError()) + + with pytest.raises(asyncio.CancelledError): + await task def test_connection_lost_exc(gw): @@ -201,27 +629,15 @@ def on_transport_close(): gw.connection_lost(None) transport.close.side_effect = on_transport_close - gw = await uart.connect( - conf.SCHEMA_DEVICE( - {conf.CONF_DEVICE_PATH: "/dev/serial", conf.CONF_DEVICE_BAUDRATE: 115200} - ), - app, - use_thread=False, # required until #484 is merged - ) + gw = await uart.connect(app, **DEVICE, isolated=False) asyncio.get_running_loop().call_later(0.1, gw.connection_lost, ValueError()) with pytest.raises(ValueError): await gw.reset() - # Need to close to release thread gw.close() - # Ensure all threads are cleaned up - [t.join(1) for t in threading.enumerate() if "bellows" in t.name] - threads = [t for t in threading.enumerate() if "bellows" in t.name] - assert len(threads) == 0 - @pytest.mark.parametrize( "reset_code", From afd006b3b9e0ef4510a783592af8cfb4095e0d68 Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Wed, 21 Oct 2026 14:42:27 +0000 Subject: [PATCH 2/3] test: `multiprocessing`? --- bellows/uart.py | 82 ++++++++++++++++++++-------------------------- tests/test_uart.py | 26 ++++----------- 2 files changed, 42 insertions(+), 66 deletions(-) diff --git a/bellows/uart.py b/bellows/uart.py index 598d2fb3b..f0ea8ed6b 100644 --- a/bellows/uart.py +++ b/bellows/uart.py @@ -9,26 +9,21 @@ import logging from logging import LogRecord from logging.handlers import QueueHandler +import multiprocessing import pickle import socket import struct -import sys import zigpy.serial import zigpy.types from bellows.ash import AshProtocol, NcpFailure, NcpResetCode -try: - from concurrent import interpreters -except ImportError: - interpreters = None - LOGGER = logging.getLogger(__name__) RESET_TIMEOUT = 2.5 -# The worker interpreter copies these loggers' effective levels from the application -# interpreter when it starts, and forwards every record that passes them +# The worker process copies these loggers' effective levels from the application when it +# starts, and forwards every record that passes them WORKER_LOGGERS = ( "", "asyncio", @@ -129,8 +124,8 @@ async def reset(self): return await self._reset_future -# Messages between the application interpreter and the worker interpreter. Every message -# is a pickled dataclass behind a 32-bit length prefix. +# Messages between the application and the worker process. Every message is a pickled +# dataclass behind a 32-bit length prefix. # Application -> worker. Requests carry an id that their reply repeats. @@ -245,9 +240,7 @@ def reducer_override(self, obj): cls = obj if isinstance(obj, type) else type(obj) if cls not in _ALLOWED_CLASSES: - raise pickle.PicklingError( - f"{cls!r} may not cross the interpreter boundary" - ) + raise pickle.PicklingError(f"{cls!r} may not cross the process boundary") return NotImplemented @@ -256,7 +249,7 @@ class _Unpickler(pickle.Unpickler): def find_class(self, module, name): if (module, name) not in _ALLOWED_GLOBALS: raise pickle.UnpicklingError( - f"{module}.{name} may not cross the interpreter boundary" + f"{module}.{name} may not cross the process boundary" ) return _ALLOWED_GLOBALS[module, name] @@ -298,7 +291,7 @@ async def _read(reader: asyncio.StreamReader): class _ForwardingHandler(QueueHandler): - """Ships the worker's log records to the application interpreter.""" + """Ships the worker's log records to the application process.""" def __init__(self, loop: asyncio.AbstractEventLoop, writer: asyncio.StreamWriter): super().__init__(queue=None) @@ -318,7 +311,7 @@ def enqueue(self, record: LogRecord) -> None: class _Worker: - """The worker interpreter's side: owns the `Gateway` and serves requests for it.""" + """The worker process's side: owns the `Gateway` and serves requests for it.""" def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter): self._reader = reader @@ -423,34 +416,33 @@ async def _run_worker( await writer.wait_closed() -def _worker_main(fd: int, *args) -> None: - """Entry point in the worker interpreter.""" - asyncio.run(_run_worker(socket.socket(fileno=fd), *args)) +def _worker_main(sock: socket.socket, *args) -> None: + """Entry point in the worker process.""" + asyncio.run(_run_worker(sock, *args)) -def _launch_worker(loop: asyncio.AbstractEventLoop, *args): - """Start the worker interpreter. Returns it and a future that resolves when it exits.""" - interpreter = interpreters.create() - - def call(): - try: - interpreter.call(_worker_main, *args) - except interpreters.ExecutionFailed as exc: - LOGGER.error("Worker interpreter crashed:\n%s", exc.excinfo.errdisplay) +def _launch_worker(loop: asyncio.AbstractEventLoop, sock: socket.socket, *args): + """Start the worker process. Returns it and a future that resolves when it exits.""" + # `fork` would copy the application's threads and locks. The socket is duplicated + # into the child, nothing else is inherited. + process = multiprocessing.get_context("spawn").Process( + target=_worker_main, args=(sock, *args), name=__name__, daemon=True + ) + process.start() + sock.close() executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix=__name__) - # The thread only hosts the worker interpreter, which has its own GIL - future = loop.run_in_executor(executor, call) + future = loop.run_in_executor(executor, process.join) executor.shutdown(wait=False) - return interpreter, future + return process, future class IsolatedGateway: - """Runs a `Gateway` on an event loop in its own interpreter and bridges the two. + """Runs a `Gateway` on an event loop in its own process and bridges the two. - The application interpreter's garbage collector stops every thread sharing its GIL, - while ASH acknowledgements have a deadline. The worker interpreter has its own GIL. + The application's garbage collector stops every thread sharing its GIL, while ASH + acknowledgements have a deadline. The worker process has its own. The worker owns the gateway and exits only once the application closes its end of the socket, so every request that passed the `_closed` check reached it before the @@ -460,7 +452,7 @@ class IsolatedGateway: def __init__(self, api): self._api = api self._loop = asyncio.get_running_loop() - self._interpreter = None + self._process = None self._worker_complete = None self._writer = None self._messages = None @@ -479,8 +471,8 @@ async def connect( for name in WORKER_LOGGERS ) - self._interpreter, self._worker_complete = _launch_worker( - self._loop, theirs.detach(), path, baudrate, flow_control, log_levels + self._process, self._worker_complete = _launch_worker( + self._loop, theirs, path, baudrate, flow_control, log_levels ) reader, self._writer = await asyncio.open_connection(sock=ours) @@ -596,9 +588,12 @@ async def _finish(self) -> None: await self._messages self._writer.close() - if self._interpreter is not None: - self._interpreter.close() - self._interpreter = None + if self._process is not None: + if self._process.exitcode: + LOGGER.error("Worker process exited with %s", self._process.exitcode) + + self._process.close() + self._process = None async def _connect(api, *, path: str, baudrate: int, flow_control: str | None): @@ -641,13 +636,6 @@ async def connect( if use_thread is not None: isolated = use_thread - if isolated and interpreters is None: - LOGGER.warning( - "Python %s has no subinterpreters, running ASH on the application loop", - sys.version.split()[0], - ) - isolated = False - if not isolated: return await _connect( api, path=path, baudrate=baudrate, flow_control=flow_control diff --git a/tests/test_uart.py b/tests/test_uart.py index 836e95f90..3d22f7204 100644 --- a/tests/test_uart.py +++ b/tests/test_uart.py @@ -2,7 +2,6 @@ from asyncio import timeout as asyncio_timeout import logging import pickle -import socket import struct import threading from types import SimpleNamespace @@ -124,19 +123,15 @@ async def in_process_worker(sock, path, baudrate, flow_control, log_levels): @pytest.fixture def workers(): - """Runs workers on the test loop instead of in a subinterpreter, so mocks reach them.""" + """Runs workers on the test loop instead of in a subprocess, so mocks reach them.""" workers = SimpleNamespace(tasks=[], worker=in_process_worker) - def launch(loop, fd, *args): - task = asyncio.ensure_future(workers.worker(socket.socket(fileno=fd), *args)) + def launch(loop, sock, *args): + task = asyncio.ensure_future(workers.worker(sock, *args)) workers.tasks.append(task) return None, task - # The protocol does not need subinterpreters, so these tests run on every Python - with ( - patch.object(uart, "_launch_worker", launch), - patch.object(uart, "interpreters", MagicMock()), - ): + with patch.object(uart, "_launch_worker", launch): yield workers @@ -475,14 +470,9 @@ def factory(): server.close() -needs_interpreters = pytest.mark.skipif( - uart.interpreters is None, reason="Subinterpreters need Python 3.14" -) - - @pytest.fixture async def real_gw(ncp, app, caplog): - """A gateway in a real subinterpreter, always disconnected, even when the test fails.""" + """A gateway in a real subprocess, always disconnected, even when the test fails.""" ncps, path = ncp # The worker copies this level when it starts @@ -497,8 +487,7 @@ async def real_gw(ncp, app, caplog): await gw.disconnect() -@needs_interpreters -async def test_interpreter_roundtrip(real_gw, ncp, app, caplog): +async def test_process_roundtrip(real_gw, ncp, app, caplog): gw = real_gw ncps, _ = ncp frames = [] @@ -543,8 +532,7 @@ async def test_interpreter_roundtrip(real_gw, ncp, app, caplog): ] -@needs_interpreters -async def test_interpreter_connect_failure(app): +async def test_process_connect_failure(app): with pytest.raises(FileNotFoundError): await uart.connect( app, path="/nonexistent/serial/port", baudrate=115200, flow_control=None From 18b2c232ea7b6a19b39d2c950482dd9935944e7f Mon Sep 17 00:00:00 2001 From: puddly <32534428+puddly@users.noreply.github.com> Date: Wed, 21 Oct 2026 14:49:10 +0000 Subject: [PATCH 3/3] Try `os.nice(-10)` to help the ASH worker out --- bellows/uart.py | 14 ++++++++++++++ tests/test_uart.py | 17 +++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/bellows/uart.py b/bellows/uart.py index f0ea8ed6b..6fedf87f8 100644 --- a/bellows/uart.py +++ b/bellows/uart.py @@ -10,6 +10,7 @@ from logging import LogRecord from logging.handlers import QueueHandler import multiprocessing +import os import pickle import socket import struct @@ -22,6 +23,10 @@ LOGGER = logging.getLogger(__name__) RESET_TIMEOUT = 2.5 +# Favors the worker when the application's collector or anything else saturates the +# CPUs, so an ACK is not late for lack of a time slice. Needs `CAP_SYS_NICE` or root. +WORKER_NICE = -10 + # The worker process copies these loggers' effective levels from the application when it # starts, and forwards every record that passes them WORKER_LOGGERS = ( @@ -416,8 +421,17 @@ async def _run_worker( await writer.wait_closed() +def _raise_priority() -> None: + try: + os.nice(WORKER_NICE) + except PermissionError: + # Unprivileged containers refuse it; the worker then runs at normal priority + LOGGER.debug("Not permitted to raise the worker's priority") + + def _worker_main(sock: socket.socket, *args) -> None: """Entry point in the worker process.""" + _raise_priority() asyncio.run(_run_worker(sock, *args)) diff --git a/tests/test_uart.py b/tests/test_uart.py index 3d22f7204..ca03e4c35 100644 --- a/tests/test_uart.py +++ b/tests/test_uart.py @@ -434,6 +434,23 @@ async def worker(sock, *args): await gw.disconnect() +def test_raise_priority(): + with patch("bellows.uart.os.nice") as nice: + uart._raise_priority() + + assert nice.mock_calls == [call(uart.WORKER_NICE)] + + +def test_raise_priority_not_permitted(caplog): + with ( + patch("bellows.uart.os.nice", side_effect=PermissionError), + caplog.at_level(logging.DEBUG, logger="bellows.uart"), + ): + uart._raise_priority() + + assert "Not permitted" in caplog.text + + class FakeNcp(AshProtocol): """The NCP end of the link: ACKs data like the host side does, answers RST."""