diff --git a/.basedpyright/baseline.json b/.basedpyright/baseline.json index 06a39d376..f4b52256b 100644 --- a/.basedpyright/baseline.json +++ b/.basedpyright/baseline.json @@ -1211,14 +1211,6 @@ "lineCount": 1 } }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 73, - "endColumn": 99, - "lineCount": 1 - } - }, { "code": "reportAttributeAccessIssue", "range": { @@ -6895,14 +6887,6 @@ "lineCount": 1 } }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 22, - "endColumn": 29, - "lineCount": 1 - } - }, { "code": "reportAttributeAccessIssue", "range": { @@ -6911,14 +6895,6 @@ "lineCount": 1 } }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 26, - "endColumn": 33, - "lineCount": 1 - } - }, { "code": "reportAttributeAccessIssue", "range": { @@ -6951,14 +6927,6 @@ "lineCount": 1 } }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 26, - "endColumn": 33, - "lineCount": 1 - } - }, { "code": "reportAttributeAccessIssue", "range": { @@ -6966,15 +6934,7 @@ "endColumn": 46, "lineCount": 1 } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 22, - "endColumn": 26, - "lineCount": 1 - } } ] } -} +} \ No newline at end of file diff --git a/pimm/tests/test_calls.py b/pimm/tests/test_calls.py index 09187272e..45c4699c8 100644 --- a/pimm/tests/test_calls.py +++ b/pimm/tests/test_calls.py @@ -241,7 +241,8 @@ def test_wrappers_do_not_apply_to_calls(self): caller, handler = ControlSystemCaller(Passive()), ControlSystemHandler(Passive()) with World() as world: with pytest.raises(AssertionError): - world.connect(caller, handler, emitter_wrapper=lambda e: e) + # A call takes no wrapper, so `connect` has no overload for one: the assertion is what says so. + world.connect(caller, handler, emitter_wrapper=lambda e: e) # pyright: ignore[reportArgumentType] def test_in_process(self): client, adder = Client([(1, 2), (-1, 2), (3, 4)]), Adder(defer=2) diff --git a/pimm/tests/test_world.py b/pimm/tests/test_world.py index 9c7dbe92c..f2d53e93f 100644 --- a/pimm/tests/test_world.py +++ b/pimm/tests/test_world.py @@ -9,6 +9,7 @@ import pytest +import pimm.world from pimm.core import ( ControlSystem, ControlSystemEmitter, @@ -26,7 +27,15 @@ from pimm.logging import LOG_LEVEL_ENV from pimm.shared_memory import SMCompliant from pimm.tests.testing import MockClock -from pimm.world import EventReceiver, LocalQueueEmitter, QueueEmitter, SystemClock, VirtualClock, World +from pimm.world import ( + EventReceiver, + LocalQueueEmitter, + MultiprocessReceiver, + QueueEmitter, + SystemClock, + VirtualClock, + World, +) def dummy_process(stop_reader, clock): @@ -99,6 +108,63 @@ def read_from_buffer(self, buffer: memoryview | bytes) -> None: self.value = struct.unpack('d', buffer[:8])[0] +def test_an_interrupt_taken_inside_a_send_is_what_stops_the_process(monkeypatch): + """Nothing installs a handler in a process for it: the transport it interrupts is what records it. + + A main-process control system reaches the same transports a background one does, so an interrupt that + tears a connection has to be noted where the tearing happens. + """ + monkeypatch.setattr(pimm.world, '_interrupted', False) + + def torn(data, ts): + raise KeyboardInterrupt + + with World() as world: + emitter, receiver = world.mp_pipes() + assert not isinstance(receiver, list) + emitter.emit('before', ts=1) + assert receiver.read() is not None + + monkeypatch.setattr(emitter, '_emit_queue', torn) + with pytest.raises(KeyboardInterrupt): + emitter.emit('torn', ts=2) + + assert pimm.world._interrupted, 'the interrupt went unrecorded' + assert receiver.read() is None + + +def test_a_process_that_took_an_interrupt_stops_talking_to_the_manager(monkeypatch): + """An interrupt can land inside a call to the manager, and that connection then holds half a message. + + Reading it again returns what another call asked for, so a reader takes a value off a channel it never + subscribed to. A process that has taken one neither reads nor sends after it. + """ + with World() as world: + emitter, receiver = world.mp_pipes() + assert not isinstance(receiver, list) + emitter.emit('before', ts=1) + assert receiver.read() is not None + + monkeypatch.setattr(pimm.world, '_interrupted', True) + + emitter.emit('after', ts=2) # dropped, not sent + assert receiver.read() is None + + +def test_a_queue_that_answers_with_anything_but_a_message_says_the_connection_is_torn(): + """A connection torn by an interrupt answers with what another call asked for, of whatever type that + call wanted -- a float as readily as nothing. Reading `.data` off it would hide the interrupt.""" + with World() as world: + emitter, receiver = world.mp_pipes() + assert isinstance(receiver, MultiprocessReceiver) + emitter.emit('before', ts=1) # settles the channel on the queue transport + assert receiver.read() is not None + receiver._queue.put(0.5) # what the torn connection hands back + + with pytest.raises(ConnectionError, match='tore its connection'): + receiver.read() + + class TestQueueEmitter: """Test the QueueEmitter class.""" diff --git a/pimm/world.py b/pimm/world.py index fcd47c495..daaf3c5da 100644 --- a/pimm/world.py +++ b/pimm/world.py @@ -11,6 +11,7 @@ import traceback from collections import Counter, defaultdict, deque from collections.abc import Callable, Iterator, Mapping +from contextlib import contextmanager from enum import IntEnum from multiprocessing import resource_tracker from multiprocessing.managers import ValueProxy @@ -42,6 +43,7 @@ logger = logging.getLogger(__name__) T = TypeVar('T') +U = TypeVar('U') Req = TypeVar('Req') Res = TypeVar('Res') @@ -75,6 +77,28 @@ def emit(self, data: T, ts: int = -1): pass +# Set in a process that has taken an interrupt. An interrupt can land inside a call to the manager, and +# that connection then holds half a message: the next call over it returns what another one asked for, so +# a reader takes a value from a channel it never subscribed to. Nothing may be sent or read after it. +_interrupted = False + + +@contextmanager +def _noting_interrupt() -> Iterator[None]: + """Record an interrupt taken inside the block, and let it go on. + + A connection is torn by an interrupt that lands in the middle of a call over it, so every process that + reaches a transport records its own -- there is nowhere else the tearing can happen, and no process has + to have had a handler installed for it. + """ + global _interrupted + try: + yield + except KeyboardInterrupt: + _interrupted = True + raise + + class MultiprocessEmitter(SignalEmitter[T]): """Signal emitter that transparently bridges processes. @@ -118,7 +142,8 @@ def __init__( @property def transport_mode(self) -> TransportMode: if self._mode is TransportMode.UNDECIDED: - self._mode = TransportMode(self._mode_value.value) + with _noting_interrupt(): # reading the manager is where a connection is torn + self._mode = TransportMode(self._mode_value.value) return self._mode @property @@ -188,9 +213,12 @@ def _emit_shared_memory(self, data: SMCompliant, ts: int) -> bool: return True + @_noting_interrupt() def emit(self, data: T, ts: int = -1): + if _interrupted: + return ts = ts if ts >= 0 else self._clock.now_ns() - mode = self._ensure_mode(data) + mode = self._ensure_mode(data) # itself a call to the manager, so it sits inside the guard if mode is TransportMode.SHARED_MEMORY: if not isinstance(data, SMCompliant): @@ -262,7 +290,8 @@ def __init__( @property def transport_mode(self) -> TransportMode: if self._mode is TransportMode.UNDECIDED: - self._mode = TransportMode(self._mode_value.value) + with _noting_interrupt(): # reading the manager is where a connection is torn + self._mode = TransportMode(self._mode_value.value) return self._mode @property @@ -275,6 +304,10 @@ def _read_queue(self) -> Message[T] | None: except Empty: message = None else: + if not isinstance(message, Message): + # An interrupt that lands inside a manager call leaves that connection holding half a + # message, and every read after it comes back as whatever another call asked for. + raise ConnectionError(f'the queue was read after an interrupt tore its connection: {message!r}') self._last_queue_message = Message(message.data, message.ts, True) if self._mode is TransportMode.UNDECIDED: self._mode = TransportMode.QUEUE @@ -332,8 +365,11 @@ def _read_shared_memory(self) -> Message[T] | None: self._up_value.value = False return Message(data=self._out_value, ts=self._ts_value.value, updated=updated) # instead of True + @_noting_interrupt() def read(self) -> Message[T] | None: - mode = self.transport_mode + if _interrupted: + return None + mode = self.transport_mode # itself a call to the manager, so it sits inside the guard if mode is TransportMode.SHARED_MEMORY: return self._read_shared_memory() @@ -642,13 +678,35 @@ def interleave(self, *loops: ControlLoop) -> Iterator[Command]: self._advance_to(target_ns) yield Sleep(wait_ns / 1e9) if wait_ns else Yield() + @overload + def connect(self, source: ControlSystemCaller[Req, Res], target: ControlSystemHandler[Req, Res]) -> None: ... + + @overload + def connect( + self, + source: ControlSystemEmitter[T], + target: ControlSystemReceiver[T], + *, + emitter_wrapper: Callable[[SignalEmitter[T]], SignalEmitter[T]] = ..., + ) -> None: ... + + @overload + def connect( + self, + source: ControlSystemEmitter[T], + target: ControlSystemReceiver[U], + *, + emitter_wrapper: Callable[[SignalEmitter[T]], SignalEmitter[T]] = ..., + receiver_wrapper: Callable[[SignalReceiver[T]], SignalReceiver[U]], + ) -> None: ... + def connect( self, source: ControlSystemEmitter[T] | ControlSystemCaller[Req, Res], - target: ControlSystemReceiver[T] | ControlSystemHandler[Req, Res], + target: ControlSystemReceiver[U] | ControlSystemHandler[Req, Res], *, emitter_wrapper: Callable[[SignalEmitter[T]], SignalEmitter[T]] = identity, - receiver_wrapper: Callable[[SignalReceiver[T]], SignalReceiver[T]] = identity, + receiver_wrapper: Callable[[SignalReceiver[T]], SignalReceiver[U]] = identity, ) -> None: """Declare a logical connection: an Emitter feeding a Receiver, or a Caller invoking a Handler. @@ -663,7 +721,8 @@ def connect( emitter_wrapper: Optional function to wrap the underlying SignalEmitter before binding. Defaults to identity function. receiver_wrapper: Optional function to wrap the underlying SignalReceiver - before binding. Defaults to identity function. + before binding. Defaults to identity function. It is the only way the + receiver may carry a type other than the emitter's. The wrapper functions allow for transformation or decoration of the underlying signal transport mechanisms, such as adding logging, diff --git a/positronic/tests/test_components.py b/positronic/tests/test_components.py index 2f9d0a645..5bb438da2 100644 --- a/positronic/tests/test_components.py +++ b/positronic/tests/test_components.py @@ -1,3 +1,4 @@ +import logging import pickle from collections.abc import Callable from importlib import import_module @@ -8,11 +9,24 @@ import pimm from pimm.core import ControlSystem +logger = logging.getLogger(__name__) + +_OURS = ('positronic', 'pimm') + def _optional_import(module: str, symbol: str) -> Any | None: + """``symbol``, or ``None`` where the vendor package it needs is not usable on this machine. + + A vendor that is installed but cannot load — its own native library is missing — leaves this station + without the component just as an absent package does. A project module that fails to import is a broken + build rather than a missing vendor, and is not one of those. + """ try: return getattr(import_module(module), symbol) - except ModuleNotFoundError: # pragma: no cover - optional dependency + except ImportError as e: # pragma: no cover - optional dependency + if (e.name or '').split('.')[0] in _OURS: + raise + logger.error('%s is not available: %s', module, e) return None diff --git a/positronic/wire.py b/positronic/wire.py index d49225f94..965eb2b9a 100644 --- a/positronic/wire.py +++ b/positronic/wire.py @@ -16,7 +16,7 @@ def wire( # noqa: C901 world: pimm.World, harness: pimm.ControlSystem, dataset_factory: DatasetFactory | None, - cameras: Mapping[str, pimm.SignalEmitter], + cameras: Mapping[str, pimm.ControlSystemEmitter], robot_arm: pimm.ControlSystem | None, gripper: pimm.ControlSystem | None, gui: pimm.ControlSystem | None, @@ -101,7 +101,7 @@ def wire_embodiment( *, record: bool = True, privileged: dict[str, Observation] | None = None, - done: pimm.SignalEmitter | None = None, + done: pimm.ControlSystemEmitter | None = None, ): """Wire an embodiment to the Harness for the inference path.