Implement BackgroundService and new Actor class (#564)

`BackgroundService` is a new abstract base class can be used to write
other classes that runs one or more tasks in the background. It provides
a consistent API to start and stop these services and also takes care of
the handling of the background tasks. It can also work as an `async`
context manager, giving the service a deterministic lifetime and
guaranteed cleanup.

The new `Actor` class brings quite a few new improvements over the old
`@actor` decorator. These are the main differences:

* It doesn't start automatically, `start()` needs to be called to start
an actor.
* The method to implement the main logic was renamed from `run()` to
`_run()`, as it is not intended to be run externally.
* Actors can have an optional `name` (useful for debugging/logging
purposes) and `loop` (if the actor should run in a loop different from
the currently running loop).
* The actor will only be restarted if an unhandled `Exception` is raised
by `_run()`. It will not be restarted if the `_run()` method finishes
normally. If an unhandled `BaseException` is raised instead, it will be
re-raised. For normal cancellation the `_run()` method should handle
`asyncio.CancelledError` if the cancellation shouldn't be propagated
(this is the same as with the decorator).
* The `_stop()` method is public (`stop()`) and will `cancel()` and
`await` for the task to finish, catching the `asyncio.CancelledError`.
* The `join()` method is renamed to `wait()`, but they can also be
awaited directly ( `await actor`).
* For deterministic cleanup, actors can now be used as `async` context
managers.

The base actors (`ConfigManagingActor`,
`ComponentMetricsResamplingActor`, `DataSourcingActor`,
`PowerDistributingActor`) now inherit from the new `Actor` class, as
well as the `MovingWindow`.

Fixes #240, fixes #45, fixes #196.

Author: llucax

RELEASE_NOTES.md

@@ -2,29 +2,110 @@
 
 ## Summary
 
-<!-- Here goes a general summary of what this release is about -->
+This release replaces the `@actor` decorator with a new `Actor` class.
 
 ## Upgrading
 
+
 - The `frequenz.sdk.power` package contained the power distribution algorithm, which is for internal use in the sdk, and is no longer part of the public API.
 
 - `PowerDistributingActor`'s result type `OutOfBound` has been renamed to `OutOfBounds`, and its member variable `bound` has been renamed to `bounds`.
 
+- The `@actor` decorator was replaced by the new `Actor` class. The main differences between the new class and the old decorator are:
+
+  * It doesn't start automatically, `start()` needs to be called to start an actor (using the `frequenz.sdk.actor.run()` function is recommended).
+  * The method to implement the main logic was renamed from `run()` to `_run()`, as it is not intended to be run externally.
+  * Actors can have an optional `name` (useful for debugging/logging purposes).
+  * The actor will only be restarted if an unhandled `Exception` is raised by `_run()`. It will not be restarted if the `_run()` method finishes normally. If an unhandled `BaseException` is raised instead, it will be re-raised. For normal cancellation the `_run()` method should handle `asyncio.CancelledError` if the cancellation shouldn't be propagated (this is the same as with the decorator).
+  * The `_stop()` method is public (`stop()`) and will `cancel()` and `await` for the task to finish, catching the `asyncio.CancelledError`.
+  * The `join()` method is renamed to `wait()`, but they can also be awaited directly ( `await actor`).
+  * For deterministic cleanup, actors can now be used as `async` context managers.
+
+  Most actors can be migrated following these steps:
+
+  1. Remove the decorator
+  2. Add `Actor` as a base class
+  3. Rename `run()` to `_run()`
+  4. Forward the `name` argument (optional but recommended)
+
+  For example, this old actor:
+
+  ```python
+  from frequenz.sdk.actor import actor
+
+  @actor
+  class TheActor:
+      def __init__(self, actor_args) -> None:
+          # init code
+
+      def run(self) -> None:
+          # run code
+  ```
+
+  Can be migrated as:
+
+  ```python
+  import asyncio
+  from frequenz.sdk.actor import Actor
+
+  class TheActor(Actor):
+      def __init__(self, actor_args,
+          *,
+          name: str | None = None,
+      ) -> None:
+          super().__init__(name=name)
+          # init code
+
+      def _run(self) -> None:
+          # run code
+  ```
+
+  Then you can instantiate all your actors first and then run them using:
+
+  ```python
+  from frequenz.sdk.actor import run
+  # Init code
+  actor = TheActor()
+  other_actor = OtherActor()
+  # more setup
+  await run(actor, other_actor)  # Start and await for all the actors
+  ```
+
+- The `MovingWindow` is now a `BackgroundService`, so it needs to be started manually with `await window.start()`. It is recommended to use it as an `async` context manager if possible though:
+
+    ```python
+    async with MovingWindow(...) as window:
+        # The moving windows is started here
+        use(window)
+    # The moving window is stopped here
+    ```
+
+- The base actors (`ConfigManagingActor`, `ComponentMetricsResamplingActor`, `DataSourcingActor`, `PowerDistributingActor`) now inherit from the new `Actor` class, if you are using them directly, you need to start them manually with `await actor.start()` and you might need to do some other adjustments.
+
 ## New Features
 
 - DFS for compentent graph
 
+- `BackgroundService`: This new abstract base class can be used to write other classes that runs one or more tasks in the background. It provides a consistent API to start and stop these services and also takes care of the handling of the background tasks. It can also work as an `async` context manager, giving the service a deterministic lifetime and guaranteed cleanup.
+
+  All classes spawning tasks that are expected to run for an indeterminate amount of time are likely good candidates to use this as a base class.
+
+- `Actor`: This new class inherits from `BackgroundService` and it replaces the `@actor` decorator.
+
 ## Bug Fixes
 
 - Fixes a bug in the ring buffer updating the end timestamp of gaps when they are outdated.
-- Properly handles PV configurations with no or only some meters before the PV
-  component.
-  So far we only had configurations like this: Meter -> Inverter -> PV. However
-  the scenario with Inverter -> PV is also possible and now handled correctly.
+
+- Properly handles PV configurations with no or only some meters before the PV component.
+
+  So far we only had configurations like this: `Meter -> Inverter -> PV`. However the scenario with `Inverter -> PV` is also possible and now handled correctly.
+
 - Fix `consumer_power()` not working certain configurations.
-  In microgrids without consumers and no main meter, the formula
-  would never return any values.
-- Fix `pv_power` not working in setups with 2 grid meters by using a new
-  reliable function to search for components in the components graph
+
+  In microgrids without consumers and no main meter, the formula would never return any values.
+
+- Fix `pv_power` not working in setups with 2 grid meters by using a new reliable function to search for components in the components graph
+
 - Fix `consumer_power` similar to `pv_power`
+
 - Zero value requests received by the `PowerDistributingActor` will now always be accepted, even when there are non-zero exclusion bounds.

benchmarks/power_distribution/power_distributor.py

@@ -108,19 +108,16 @@ async def run_test(  # pylint: disable=too-many-locals
     power_request_channel = Broadcast[Request]("power-request")
     battery_status_channel = Broadcast[BatteryStatus]("battery-status")
     channel_registry = ChannelRegistry(name="power_distributor")
-    distributor = PowerDistributingActor(
+    async with PowerDistributingActor(
         channel_registry=channel_registry,
         requests_receiver=power_request_channel.new_receiver(),
         battery_status_sender=battery_status_channel.new_sender(),
-    )
-
-    tasks: List[Coroutine[Any, Any, List[Result]]] = []
-    tasks.append(send_requests(batteries, num_requests))
-
-    result = await asyncio.gather(*tasks)
-    exec_time = timeit.default_timer() - start
+    ):
+        tasks: List[Coroutine[Any, Any, List[Result]]] = []
+        tasks.append(send_requests(batteries, num_requests))
 
-    await distributor._stop()  # type: ignore # pylint: disable=no-member, protected-access
+        result = await asyncio.gather(*tasks)
+        exec_time = timeit.default_timer() - start
 
     summary = parse_result(result)
     summary["num_requests"] = num_requests

benchmarks/timeseries/benchmark_datasourcing.py

@@ -113,23 +113,22 @@ async def consume(channel: Receiver[Any]) -> None:
             await request_sender.send(request)
             consume_tasks.append(asyncio.create_task(consume(recv_channel)))
 
-    DataSourcingActor(request_receiver, channel_registry)
-
-    await asyncio.gather(*consume_tasks)
-
-    time_taken = perf_counter() - start_time
-
-    await mock_grid.cleanup()
-
-    print(f"Samples Sent: {samples_sent}, time taken: {time_taken}")
-    print(f"Samples per second: {samples_sent / time_taken}")
-    print(
-        "Expected samples: "
-        f"{num_expected_messages}, missing: {num_expected_messages - samples_sent}"
-    )
-    print(
-        f"Missing per EVC: {(num_expected_messages - samples_sent) / num_ev_chargers}"
-    )
+    async with DataSourcingActor(request_receiver, channel_registry):
+        await asyncio.gather(*consume_tasks)
+
+        time_taken = perf_counter() - start_time
+
+        await mock_grid.cleanup()
+
+        print(f"Samples Sent: {samples_sent}, time taken: {time_taken}")
+        print(f"Samples per second: {samples_sent / time_taken}")
+        print(
+            "Expected samples: "
+            f"{num_expected_messages}, missing: {num_expected_messages - samples_sent}"
+        )
+        print(
+            f"Missing per EVC: {(num_expected_messages - samples_sent) / num_ev_chargers}"
+        )
 
 
 def parse_args() -> Tuple[int, int, bool]:

benchmarks/timeseries/periodic_feature_extractor.py

@@ -12,6 +12,8 @@
 from __future__ import annotations
 
 import asyncio
+import collections.abc
+import contextlib
 import logging
 from datetime import datetime, timedelta, timezone
 from functools import partial
@@ -27,19 +29,23 @@
 from frequenz.sdk.timeseries._quantities import Quantity
 
 
-async def init_feature_extractor(period: int) -> PeriodicFeatureExtractor:
+@contextlib.asynccontextmanager
+async def init_feature_extractor(
+    period: int,
+) -> collections.abc.AsyncIterator[PeriodicFeatureExtractor]:
     """Initialize the PeriodicFeatureExtractor class."""
     # We only need the moving window to initialize the PeriodicFeatureExtractor class.
     lm_chan = Broadcast[Sample[Quantity]]("lm_net_power")
-    moving_window = MovingWindow(
+    async with MovingWindow(
         timedelta(seconds=1), lm_chan.new_receiver(), timedelta(seconds=1)
-    )
-
-    await lm_chan.new_sender().send(Sample(datetime.now(tz=timezone.utc), Quantity(0)))
+    ) as moving_window:
+        await lm_chan.new_sender().send(
+            Sample(datetime.now(tz=timezone.utc), Quantity(0))
+        )
 
-    # Initialize the PeriodicFeatureExtractor class with a period of period seconds.
-    # This works since the sampling period is set to 1 second.
-    return PeriodicFeatureExtractor(moving_window, timedelta(seconds=period))
+        # Initialize the PeriodicFeatureExtractor class with a period of period seconds.
+        # This works since the sampling period is set to 1 second.
+        yield PeriodicFeatureExtractor(moving_window, timedelta(seconds=period))
 
 
 def _calculate_avg_window(
@@ -211,22 +217,22 @@ async def main() -> None:
 
     # create a random ndarray with 29 days -5 seconds of data
     days_29_s = 29 * DAY_S
-    feature_extractor = await init_feature_extractor(10)
-    data = rng.standard_normal(days_29_s)
-    run_benchmark(data, 4, feature_extractor)
-
-    days_29_s = 29 * DAY_S + 3
-    data = rng.standard_normal(days_29_s)
-    run_benchmark(data, 4, feature_extractor)
-
-    # create a random ndarray with 29 days +5 seconds of data
-    data = rng.standard_normal(29 * DAY_S + 5)
-
-    feature_extractor = await init_feature_extractor(7 * DAY_S)
-    # TEST one day window and 6 days distance. COPY (Case 3)
-    run_benchmark(data, DAY_S, feature_extractor)
-    # benchmark one day window and 6 days distance. NO COPY (Case 1)
-    run_benchmark(data[: 28 * DAY_S], DAY_S, feature_extractor)
+    async with init_feature_extractor(10) as feature_extractor:
+        data = rng.standard_normal(days_29_s)
+        run_benchmark(data, 4, feature_extractor)
+
+        days_29_s = 29 * DAY_S + 3
+        data = rng.standard_normal(days_29_s)
+        run_benchmark(data, 4, feature_extractor)
+
+        # create a random ndarray with 29 days +5 seconds of data
+        data = rng.standard_normal(29 * DAY_S + 5)
+
+    async with init_feature_extractor(7 * DAY_S) as feature_extractor:
+        # TEST one day window and 6 days distance. COPY (Case 3)
+        run_benchmark(data, DAY_S, feature_extractor)
+        # benchmark one day window and 6 days distance. NO COPY (Case 1)
+        run_benchmark(data[: 28 * DAY_S], DAY_S, feature_extractor)
 
 
 logging.basicConfig(level=logging.DEBUG)

src/frequenz/sdk/actor/__init__.py

@@ -4,20 +4,22 @@
 """A base class for creating simple composable actors."""
 
 from ..timeseries._resampling import ResamplerConfig
+from ._actor import Actor
+from ._background_service import BackgroundService
 from ._channel_registry import ChannelRegistry
 from ._config_managing import ConfigManagingActor
 from ._data_sourcing import ComponentMetricRequest, DataSourcingActor
-from ._decorator import actor
 from ._resampling import ComponentMetricsResamplingActor
 from ._run_utils import run
 
 __all__ = [
+    "Actor",
+    "BackgroundService",
     "ChannelRegistry",
     "ComponentMetricRequest",
     "ComponentMetricsResamplingActor",
     "ConfigManagingActor",
     "DataSourcingActor",
     "ResamplerConfig",
-    "actor",
     "run",
 ]