diff --git a/docs/API.md b/docs/API.md index 82a06f7..8eaac0e 100644 --- a/docs/API.md +++ b/docs/API.md @@ -123,7 +123,8 @@ representation. ### `get_all_signal_descriptors() -> list[SignalDescriptor]` Returns descriptors for every signal currently registered by connected -connectors. The list is empty when no connectors have completed a handshake. +connectors and bound derive units. The list is empty when no connectors have +completed a handshake and no derive units are bound. Connector membership is dynamic: a disconnected connector's signals are no longer returned. @@ -233,9 +234,9 @@ The parameter annotations must use the imported `ValueDescriptor` and `SignalDescriptor` classes shown above. The source string is executed with `exec()`, so only bind code from trusted sources. -The returned UUID identifies the bound unit. Binding invalid source or more -than one function raises `CoreMultipleFunctionsFoundError`; an invalid function -signature raises `DeriveUnitInvalidSignatureError`. +The returned UUID identifies the bound unit. Binding source that defines more +or fewer than one function raises `DeriveUnitMultipleFunctionsFoundError`; an +invalid function signature raises `DeriveUnitInvalidSignatureError`. ### `unbind_derive_unit(unit_id: UUID) -> None` @@ -252,5 +253,5 @@ Core-specific exceptions are defined in `dynalab_core.errors`. | Exception | Raised when | | --- | --- | | `CoreStateMismatchError` | `start()` is called after the core has already started or stopped. | -| `CoreMultipleFunctionsFoundError` | A derive-unit source string does not define exactly one function. | +| `DeriveUnitMultipleFunctionsFoundError` | A derive-unit source string does not define exactly one function. | | `DeriveUnitInvalidSignatureError` | A derive function's annotations or parameter order are invalid. | diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index 233c1ad..2a6b86b 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -3,25 +3,22 @@ # SPDX-License-Identifier: GPL-3.0-or-later -from collections.abc import Callable from datetime import datetime, timezone -import inspect import logging from queue import Empty, Queue import threading from threading import Lock, Thread import time -import types from typing import Literal -from uuid import UUID, uuid4 +from uuid import UUID from dynalab_core.buffer import ValueBuffer from dynalab_core.config import CoreConfig from dynalab_core.constants import CORE_VERSION -from dynalab_core.derive import DeriveUnit +from dynalab_core.derive import DeriveRegistry from dynalab_core.dlpak import DLPak -from dynalab_core.errors import CoreMultipleFunctionsFoundError, CoreStateMismatchError +from dynalab_core.errors import CoreStateMismatchError from dynalab_core.protocols.endpoint import ConnectorRegistry from dynalab_core.protocols.common import VersionDescriptor from dynalab_core.protocols.json.server import JsonServer @@ -50,14 +47,15 @@ class Core: self._live_values: dict[UUID, Value] = {} self._live_values_lock = Lock() - self._derive_units: dict[UUID, DeriveUnit] = {} - self._data_input_queue: Queue[ProtocolMessage] = Queue(524288) self._input_worker_thread = Thread( target=self._input_worker, name="input_worker_thread", daemon=True ) self._input_worker_thread.start() + self._derive_registry = DeriveRegistry( + self._data_input_queue, self._get_live_value_descriptor + ) self._connector_registry = ConnectorRegistry(self._data_input_queue) self._json_server = JsonServer(self._core_config, self._connector_registry) @@ -109,9 +107,8 @@ class Core: ) self._json_server.stop() self._connector_registry.stop() + self._derive_registry.stop() self._stop_event.set() - for unit in self._derive_units.values(): - unit.stop() self._state = "stopped" log.info( "Core stopped", @@ -129,31 +126,27 @@ class Core: self._recording_buffer.normalize() self._processing_buffer = DLPak() self._processing_buffer.set_data(self._recording_buffer) - signal_descriptors = self._connector_registry.get_all_signal_descriptors() - for unit in self._derive_units.values(): - signal_descriptors.append(unit._return_signal) + connector_signal_descriptors = ( + self._connector_registry.get_all_signal_descriptors() + ) + derive_signal_descriptors = self._derive_registry.get_all_signal_descriptors() self._processing_buffer.set_manifest( - self._recording_timestamp, signal_descriptors + self._recording_timestamp, + connector_signal_descriptors + derive_signal_descriptors, ) def get_signal_descriptor(self, signal_id: UUID) -> SignalDescriptor | None: - connector_signal = self._connector_registry.get_signal_descriptor(signal_id) - if connector_signal is None: - for unit in self._derive_units.values(): - if unit._return_signal.id == signal_id: - return unit._return_signal - else: - return None + signal = self._connector_registry.get_signal_descriptor(signal_id) + if signal is None: + signal = self._derive_registry.get_signal_descriptor(signal_id) - return connector_signal + return signal def get_all_signal_descriptors(self) -> list[SignalDescriptor]: connector_signals = self._connector_registry.get_all_signal_descriptors() - connector_signals.append( - unit._return_signal for unit in self._derive_units.values() - ) - return connector_signals + derive_signals = self._derive_registry.get_all_signal_descriptors() + return connector_signals + derive_signals def get_live_value(self, signal_id: UUID) -> float | None: with self._live_values_lock: @@ -175,32 +168,19 @@ class Core: input_signals: list[SignalDescriptor], output_signal: SignalDescriptor, ) -> UUID: - unit_id = uuid4() - - namespace = {} - exec(function, namespace) - - functions = [ - obj for obj in namespace.values() if isinstance(obj, types.FunctionType) - ] - - if len(functions) != 1: - raise CoreMultipleFunctionsFoundError - - unit = DeriveUnit( - unit_id, functions[0], input_signals, output_signal, self._data_input_queue - ) - - self._derive_units[unit_id] = unit - - return unit_id + unit = self._derive_registry.register(function, input_signals, output_signal) + return unit.uuid() def unbind_derive_unit(self, unit_id: UUID) -> None: - try: - self._derive_units[unit_id].stop() - self._derive_units.pop(unit_id, None) - except KeyError: - return + self._derive_registry.unregister(unit_id) + + def _get_live_value_descriptor(self, signal_id: UUID) -> ValueDescriptor | None: + with self._live_values_lock: + live_value = self._live_values.get(signal_id) + + if live_value is None: + return None + return live_value.to_descriptor(signal_id) def _input_worker(self) -> None: while not self._stop_event.is_set(): @@ -215,11 +195,7 @@ class Core: live_value = self._live_values.get(message.signal_id) if live_value is None: - signal_descriptor = ( - self._connector_registry.get_signal_descriptor( - message.signal_id - ) - ) + signal_descriptor = self.get_signal_descriptor(message.signal_id) live_value = Value( signal_descriptor.timeout_ms if signal_descriptor else 2000 @@ -232,26 +208,4 @@ class Core: message.timestamp, message.signal_id, message.value ) - for unit in self._derive_units.values(): - input_signals = unit.get_input_signals() - if message.signal_id in [signal.id for signal in input_signals]: - args: list[ValueDescriptor] = [] - for signal in input_signals: - if signal.id == message.signal_id: - args.append(message) - continue - - with self._live_values_lock: - live_value = self._live_values.get(signal.id) - - value_descriptor = ( - live_value.to_descriptor(signal.id) - if live_value is not None - else None - ) - if value_descriptor is None: - break - - args.append(value_descriptor) - else: - unit.put_data(args) + self._derive_registry.put_data(message) diff --git a/src/dynalab_core/derive.py b/src/dynalab_core/derive.py index d97dfad..160a463 100644 --- a/src/dynalab_core/derive.py +++ b/src/dynalab_core/derive.py @@ -5,18 +5,24 @@ from collections.abc import Callable from enum import IntEnum import inspect +import logging from queue import Empty, Queue -from threading import Thread +from threading import RLock, Thread import threading -from uuid import UUID +import types +from uuid import UUID, uuid4 from dynalab_core.errors import ( + DeriveRegistryAlreadyRegisteredError, DeriveUnitArgsMismatchError, DeriveUnitInvalidSignatureError, + DeriveUnitMultipleFunctionsFoundError, ) from dynalab_core.protocols.packets.data import ValueDescriptor from dynalab_core.protocols.packets.handshake import SignalDescriptor +log = logging.getLogger(__name__) + class ParserState(IntEnum): INIT = 0 @@ -40,8 +46,6 @@ class DeriveUnit: self._stop_event: threading.Event = threading.Event() self._stopped_event: threading.Event = threading.Event() self._input_queue: Queue[list[ValueDescriptor]] = Queue() - # TODO: Deprecate output queue - self._output_queue: Queue[ValueDescriptor] = Queue() self._return_queue: Queue[ValueDescriptor] = return_queue self._num_input_args: int = 0 @@ -85,27 +89,11 @@ class DeriveUnit: value = self._process_function(*input_args, self._return_signal) self._return_queue.put(value) - self._output_queue.put(value) self._stopped_event.set() def put_data(self, input: list[ValueDescriptor]) -> None: self._input_queue.put(input) - def get_output(self) -> ValueDescriptor | None: - try: - return self._output_queue.get_nowait() - except Empty: - return None - - def get_all_output(self) -> list[ValueDescriptor]: - output: list[ValueDescriptor] = [] - - while True: - try: - output.append(self._output_queue.get_nowait()) - except Empty: - return output - def process_offline(self, input_args: list[ValueDescriptor]) -> ValueDescriptor: if len(input_args) == self._num_input_args: return self._process_function(*input_args, self._return_signal) @@ -117,3 +105,167 @@ class DeriveUnit: def stop(self) -> None: self._stop_event.set() self._stopped_event.wait(5.0) + + def uuid(self) -> UUID: + return self._id + + def return_signal(self) -> SignalDescriptor: + return self._return_signal + + +class DeriveRegistry: + def __init__( + self, + core_input_queue: Queue, + get_live_value: Callable[[UUID], ValueDescriptor | None], + ) -> None: + self._lock = RLock() + self._units: dict[UUID, DeriveUnit] = {} + self._core_input_queue = core_input_queue + self._get_live_value = get_live_value + self._routing_queue: Queue[ValueDescriptor] = Queue() + self._routing_stop_event = threading.Event() + self._routing_stopped_event = threading.Event() + self._routing_worker_thread = Thread( + target=self._routing_worker, + name="derive_registry_routing_worker", + daemon=True, + ) + self._routing_worker_thread.start() + + def register( + self, + function: str, + input_signals: list[SignalDescriptor], + output_signal: SignalDescriptor, + unit_uuid: UUID | None = None, + ) -> DeriveUnit: + + if unit_uuid is not None: + with self._lock: + current = self.get(unit_uuid) + if current is not None: + log.warning( + "Derive Unit %s is already registered", + current._id, + extra={ + "event": "derive.registration_rejected", + "derive_uuid": str(current._id), + "reason": "duplicate_uuid", + }, + ) + raise DeriveRegistryAlreadyRegisteredError + unit_id = unit_uuid or uuid4() + + namespace = {} + exec(function, namespace) + + functions = [ + obj for obj in namespace.values() if isinstance(obj, types.FunctionType) + ] + + if len(functions) != 1: + raise DeriveUnitMultipleFunctionsFoundError + + unit = DeriveUnit( + unit_id, functions[0], input_signals, output_signal, self._core_input_queue + ) + + with self._lock: + self._units[unit_id] = unit + + return unit + + def unregister(self, unit_uuid: UUID) -> None: + with self._lock: + current = self._units.get(unit_uuid) + + if current is not None: + current.stop() + del self._units[unit_uuid] + else: + log.debug( + "Unit %s was not registered", + unit_uuid, + extra={ + "event": "derive.unregister_noop", + "derive_uuid": str(unit_uuid), + }, + ) + return + + def get(self, unit_uuid: UUID) -> DeriveUnit | None: + with self._lock: + return self._units.get(unit_uuid) + + def get_signal_descriptor(self, signal_id: UUID) -> SignalDescriptor | None: + with self._lock: + units = list(self._units.values()) + + for unit in units: + signal = unit.return_signal() + if signal.id == signal_id: + return signal + + return None + + def get_all_signal_descriptors(self) -> list[SignalDescriptor]: + with self._lock: + units = list(self._units.values()) + + signals: list[SignalDescriptor] = [] + for unit in units: + signals.append(unit.return_signal()) + + return signals + + def put_data(self, message: ValueDescriptor) -> None: + if not self._routing_stop_event.is_set(): + self._routing_queue.put_nowait(message) + + def stop(self) -> None: + self._routing_stop_event.set() + if not self._routing_worker_thread.is_alive(): + self._routing_stopped_event.set() + self._routing_stopped_event.wait(5.0) + + with self._lock: + units = list(self._units.values()) + for unit in units: + unit.stop() + self._units.clear() + + def _routing_worker(self) -> None: + try: + while not self._routing_stop_event.is_set(): + try: + message = self._routing_queue.get(timeout=0.1) + except Empty: + continue + + self._route_data(message) + finally: + self._routing_stopped_event.set() + + def _route_data(self, message: ValueDescriptor) -> None: + with self._lock: + units = list(self._units.values()) + + for unit in units: + input_signals = unit.get_input_signals() + if message.signal_id not in [signal.id for signal in input_signals]: + continue + + args: list[ValueDescriptor] = [] + for signal in input_signals: + if signal.id == message.signal_id: + args.append(message) + continue + + value_descriptor = self._get_live_value(signal.id) + if value_descriptor is None: + break + + args.append(value_descriptor) + else: + unit.put_data(args) diff --git a/src/dynalab_core/errors.py b/src/dynalab_core/errors.py index e4e098e..89c0955 100644 --- a/src/dynalab_core/errors.py +++ b/src/dynalab_core/errors.py @@ -11,10 +11,6 @@ class CoreStateMismatchError(CoreError): """DynaLab Core state error.""" -class CoreMultipleFunctionsFoundError(CoreError): - """DynaLab Core multiple functions found while trying to create derive unit.""" - - # DLPak error declarations class DLPakError(Exception): """DLPak error.""" @@ -35,3 +31,16 @@ class DeriveUnitInvalidSignatureError(DeriveUnitError): class DeriveUnitArgsMismatchError(DeriveUnitError): """DeriveUnit incorrect arguments provided error.""" + + +class DeriveUnitMultipleFunctionsFoundError(DeriveUnitError): + """DynaLab Core multiple functions found while trying to create derive unit.""" + + +# DeriveRegistry error declarations +class DeriveRegistryError(Exception): + """Generic ConnectorRegistry Error""" + + +class DeriveRegistryAlreadyRegisteredError(DeriveRegistryError): + """ConnectorRegistry endpoint already registered""" diff --git a/src/dynalab_core/protocols/endpoint.py b/src/dynalab_core/protocols/endpoint.py index fe23470..17e4026 100644 --- a/src/dynalab_core/protocols/endpoint.py +++ b/src/dynalab_core/protocols/endpoint.py @@ -386,6 +386,7 @@ class ConnectorRegistry: current = self._endpoints.get(connector_uuid) if current is endpoint: + endpoint.stop() del self._endpoints[connector_uuid] endpoint_count = len(self._endpoints) else: @@ -399,8 +400,6 @@ class ConnectorRegistry: ) return - endpoint.stop() - log.info( "Unregistered connector %s", connector_uuid, diff --git a/test/test_derive.py b/test/test_derive.py new file mode 100644 index 0000000..51a8244 --- /dev/null +++ b/test/test_derive.py @@ -0,0 +1,68 @@ +from queue import Empty, Queue +import time +from uuid import UUID, uuid4 + +from dynalab_core.derive import DeriveRegistry +from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.handshake import SignalDescriptor + + +PROCESSING_FUNCTION = """ +from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.handshake import SignalDescriptor + +def add(left: ValueDescriptor, right: ValueDescriptor, output: SignalDescriptor) -> ValueDescriptor: + return ValueDescriptor( + signal_id=output.id, + value=left.value + right.value, + timestamp=left.timestamp, + ) +""" + + +def test_derive_registry_routes_with_live_values() -> None: + output_queue: Queue[ValueDescriptor] = Queue() + left_signal = SignalDescriptor(id=uuid4(), name="Left", type="number") + right_signal = SignalDescriptor(id=uuid4(), name="Right", type="number") + output_signal = SignalDescriptor(id=uuid4(), name="Total", type="number") + live_values: dict[UUID, ValueDescriptor] = {} + registry = DeriveRegistry(output_queue, live_values.get) + + try: + unit = registry.register( + PROCESSING_FUNCTION, [left_signal, right_signal], output_signal + ) + right_value = ValueDescriptor( + signal_id=right_signal.id, value=2.0, timestamp=1 + ) + live_values[right_signal.id] = right_value + left_value = ValueDescriptor(signal_id=left_signal.id, value=3.0, timestamp=2) + + registry.put_data(left_value) + + deadline = time.monotonic() + 1.0 + while True: + try: + result = output_queue.get_nowait() + break + except Empty: + if time.monotonic() >= deadline: + raise AssertionError("Derive unit did not produce a value") + time.sleep(0.01) + + registry.unregister(unit.uuid()) + assert registry.get(unit.uuid()) is None + finally: + registry.stop() + + assert result.signal_id == output_signal.id + assert result.value == 5.0 + assert result.timestamp == left_value.timestamp + +def test_derive_registry_ignores_unknown_unit_on_unregister() -> None: + registry = DeriveRegistry(Queue(), lambda signal_id: None) + + try: + registry.unregister(uuid4()) + finally: + registry.stop()