The ActorDispatcher now supports persistent instances of actors.

Signed-off-by: Mathias L. Baumann <[email protected]>

Author: Marenz

RELEASE_NOTES.md

@@ -19,6 +19,7 @@ This release introduces a more flexible and powerful mechanism for managing disp
 
 ## New Features
 
+* The `ActorDispatcher` now supports persistent instances of actors. This allows you to keep the actor instance around after stopping the dispatch. The actor will be started again when an according dispatch is received.
 * A new parameter `initial_fetch_timeout` has been added to `Dispatcher.new_running_state_event_receiver` and related methods. It allows you to specify a timeout for the initial fetch of dispatches. After the timeout, startup continues as normal.
 * A new feature "merge strategy" (`MergeByType`, `MergeByTypeTarget`) has been added to the `Dispatcher.new_running_state_event_receiver` method. Using it, you can automatically merge consecutive and overlapping dispatch start/stop events of the same type. E.g. dispatch `A` starting at 10:10 and ending at 10:30 and dispatch `B` starts at 10:30 until 11:00, with the feature enabled this would in total trigger one start event, one reconfigure event at 10:30 and one stop event at 11:00.
 * The SDK dependency was widened to allow versions up to (excluding) v1.0.0-rc1600.

src/frequenz/dispatch/_actor_dispatcher.py

@@ -33,6 +33,7 @@ class DispatchInfo:
     """Additional options."""
 
 
+# pylint: disable=too-many-instance-attributes
 class ActorDispatcher(BackgroundService):
     """Helper class to manage actors based on dispatches.
 
@@ -186,6 +187,7 @@ def __init__(  # pylint: disable=too-many-arguments, too-many-positional-argumen
         ],
         running_status_receiver: Receiver[Dispatch],
         dispatch_identity: Callable[[Dispatch], int] | None = None,
+        persist_instances: bool = True,
         retry_interval: timedelta | None = timedelta(seconds=60),
     ) -> None:
         """Initialize the dispatch handler.
@@ -196,16 +198,20 @@ def __init__(  # pylint: disable=too-many-arguments, too-many-positional-argumen
             running_status_receiver: The receiver for dispatch running status changes.
             dispatch_identity: A function to identify to which actor a dispatch refers.
                 By default, it uses the dispatch ID.
+            persist_instances: Whether to keep instances of actors that have
+                stopped. Not recommended when you expect a varying number of instances.
             retry_interval: The interval between retries. If `None`, retries are disabled.
         """
         super().__init__()
         self._dispatch_identity: Callable[[Dispatch], int] = (
             dispatch_identity if dispatch_identity else lambda d: d.id
         )
 
+        self._persist_instances = persist_instances
         self._dispatch_rx = running_status_receiver
         self._actor_factory = actor_factory
         self._actors: dict[int, Actor] = {}
+
         self._updates_channel = Broadcast[DispatchInfo](
             name="dispatch_updates_channel", resend_latest=True
         )
@@ -277,10 +283,18 @@ async def _stop_actor(self, stopping_dispatch: Dispatch, msg: str) -> None:
         actor: Actor | None = None
         identity = self._dispatch_identity(stopping_dispatch)
 
-        actor = self._actors.get(identity)
+        if self._persist_instances:
+            actor = self._actors.get(identity)
+        else:
+            # Remove reference to actor
+            actor = self._actors.pop(identity, None)
 
         if actor:
             await actor.stop(msg)
+
+            if not self._persist_instances:
+                del actor
+
         else:
             _logger.warning(
                 "Actor for dispatch type %r is not running", stopping_dispatch.type

src/frequenz/dispatch/_dispatcher.py

@@ -233,6 +233,7 @@ def is_managed(self, dispatch_type: str) -> bool:
         """
         return dispatch_type in self._actor_dispatchers
 
+    # pylint: disable=too-many-arguments, too-many-positional-arguments
     async def start_dispatching(
         self,
         dispatch_type: str,
@@ -241,6 +242,7 @@ async def start_dispatching(
             [DispatchInfo, Receiver[DispatchInfo]], Awaitable[Actor]
         ],
         merge_strategy: MergeStrategy | None = None,
+        persist_instances: bool = True,
         retry_interval: timedelta = timedelta(seconds=60),
         initial_fetch_timeout: timedelta = timedelta(seconds=10),
     ) -> None:
@@ -257,6 +259,8 @@ async def start_dispatching(
             dispatch_type: The type of the dispatch to manage.
             actor_factory: The factory to create actors.
             merge_strategy: The strategy to merge running intervals.
+            persist_instances: Whether to reuse actor instances. Not recommended
+                when you expect a varying number of instances.
             retry_interval: Retry interval for when actor creation fails.
             initial_fetch_timeout: Timeout for the initial fetch of dispatches.
                 After the timeout, the receiver will continue to start and
@@ -286,12 +290,15 @@ def id_identity(dispatch: Dispatch) -> int:
             dispatch_identity=(
                 id_identity if merge_strategy is None else merge_strategy.identity
             ),
+            persist_instances=persist_instances,
             retry_interval=retry_interval,
         )
 
         self._actor_dispatchers[dispatch_type] = dispatcher
         dispatcher.start()
 
+    # pylint: enable=too-many-arguments, too-many-positional-arguments
+
     async def stop_dispatching(self, dispatch_type: str) -> None:
         """Stop managing actors for a given dispatch type.
 

tests/test_mananging_actor.py

@@ -132,11 +132,17 @@ async def test_env() -> AsyncIterator[TestEnv]:
     await actors_service.stop()
 
 
+@pytest.mark.parametrize("persist_instances", [True, False])
 async def test_simple_start_stop(
     test_env: TestEnv,
     fake_time: time_machine.Coordinates,
+    persist_instances: bool,
 ) -> None:
     """Test behavior when receiving start/stop messages."""
+    # pylint: disable=protected-access
+    test_env.actors_service._persist_instances = persist_instances
+    # pylint: enable=protected-access
+
     now = _now()
     duration = timedelta(minutes=10)
     dispatch = test_env.generator.generate_dispatch()
@@ -178,7 +184,10 @@ async def test_simple_start_stop(
     await asyncio.sleep(1)
 
     # pylint: disable=protected-access
-    assert 1 not in test_env.actors_service._actors
+    if persist_instances:
+        assert 1 in test_env.actors_service._actors
+    else:
+        assert 1 not in test_env.actors_service._actors
     # pylint: enable=protected-access