Add run() method to simplify awaiting of actors completion

The manual join() of actors was causing issues with linting
tools such as pylint and mypy due to the indirection created
by the actor decorator. The new run() function simplifies and
improves the synchronization of actors on the client side.

Signed-off-by: Daniel Zullo <[email protected]>

Author: daniel-zullo-frequenz

RELEASE_NOTES.md

@@ -11,6 +11,7 @@
 ## New Features
 
 * A new class `SerializableRingbuffer` is now available, extending the `OrderedRingBuffer` class with the ability to load & dump the data to disk.
+* Add the `run(*actors)` function for running and synchronizing the execution of actors. This new function simplifies the way actors are managed on the client side, allowing for a cleaner and more streamlined approach. Users/apps can now run actors simply by calling run(actor1, actor2, actor3...) without the need to manually call join() and deal with linting errors.
 
 ## Bug Fixes
 

examples/power_distribution.py

@@ -18,14 +18,13 @@
 
 from frequenz.channels import Bidirectional, Broadcast, Receiver, Sender
 
-from frequenz.sdk import microgrid
+from frequenz.sdk import actor, microgrid
 from frequenz.sdk.actor import (
     ChannelRegistry,
     ComponentMetricRequest,
     ComponentMetricsResamplingActor,
     DataSourcingActor,
     ResamplerConfig,
-    actor,
 )
 from frequenz.sdk.actor.power_distributing import (
     PowerDistributingActor,
@@ -42,7 +41,7 @@
 PORT = 61060
 
 
-@actor
+@actor.actor
 class DecisionMakingActor:
     """Actor that receives set receives power for given batteries."""
 
@@ -112,7 +111,7 @@ async def run(self) -> None:
                 _logger.info("Set power with %d succeed.", power_to_set)
 
 
-@actor
+@actor.actor
 class DataCollectingActor:
     """Actor that makes decisions about how much to charge/discharge batteries."""
 
@@ -227,10 +226,7 @@ async def run() -> None:
         active_power_data=await logical_meter.grid_power(),
     )
 
-    # pylint: disable=no-member
-    await service_actor.join()  # type: ignore[attr-defined]
-    await client_actor.join()  # type: ignore[attr-defined]
-    await power_distributor.join()  # type: ignore[attr-defined]
+    await actor.run(service_actor, client_actor, power_distributor)
 
 
 asyncio.run(run())

src/frequenz/sdk/actor/__init__.py

@@ -9,6 +9,7 @@
 from ._data_sourcing import ComponentMetricRequest, DataSourcingActor
 from ._decorator import actor
 from ._resampling import ComponentMetricsResamplingActor
+from ._run_utils import run
 
 __all__ = [
     "ChannelRegistry",
@@ -18,4 +19,5 @@
     "DataSourcingActor",
     "ResamplerConfig",
     "actor",
+    "run",
 ]

src/frequenz/sdk/actor/_run_utils.py

@@ -0,0 +1,27 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Utility functions to run and synchronize the execution of actors."""
+
+
+import asyncio
+from typing import Any
+
+from ._decorator import BaseActor
+
+
+async def run(*actors: Any) -> None:
+    """Await the completion of all actors.
+
+    Args:
+        actors: the actors to be awaited.
+
+    Raises:
+        AssertionError: if any of the actors is not an instance of BaseActor.
+    """
+    # Check that each actor is an instance of BaseActor at runtime,
+    # due to the indirection created by the actor decorator.
+    for actor in actors:
+        assert isinstance(actor, BaseActor), f"{actor} is not an instance of BaseActor"
+
+    await asyncio.gather(*(actor.join() for actor in actors))

tests/actor/test_decorator.py

@@ -5,7 +5,7 @@
 from frequenz.channels import Broadcast, Receiver, Sender
 from frequenz.channels.util import Select
 
-from frequenz.sdk.actor import actor
+from frequenz.sdk.actor import actor, run
 
 
 @actor
@@ -111,5 +111,4 @@ async def test_actor_does_not_restart() -> None:
     )
 
     await channel.new_sender().send(1)
-    # pylint: disable=no-member
-    await _faulty_actor.join()  # type: ignore
+    await run(_faulty_actor)