Moved derive units to registry architectur

DeriveUnits are now owned by a DeriveRegistry which has its own routing
thread to handle routing values to the different DeriveUnits

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. |

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)

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)

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"""

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,

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()