DataPipeline: Fix resampling/ds/power actors not started (#603)

Author: Marenz

RELEASE_NOTES.md

@@ -71,7 +71,7 @@ This release replaces the `@actor` decorator with a new `Actor` class.
   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:
+- The `MovingWindow` is now a `BackgroundService`, so it needs to be started manually with `window.start()`. It is recommended to use it as an `async` context manager if possible though:
 
     ```python
     async with MovingWindow(...) as window:
@@ -80,7 +80,7 @@ This release replaces the `@actor` decorator with a new `Actor` class.
     # 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.
+- 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 `actor.start()` and you might need to do some other adjustments.
 
 - The `BatteryPool.power_distribution_results` method has been enhanced to provide power distribution results in the form of `Power` objects, replacing the previous use of `float` values.
 

src/frequenz/sdk/actor/_actor.py

@@ -149,7 +149,7 @@ async def _run(self) -> None:
         This is mostly used for testing purposes and shouldn't be set in production.
     """
 
-    async def start(self) -> None:
+    def start(self) -> None:
         """Start this actor.
 
         If this actor is already running, this method does nothing.
@@ -172,7 +172,7 @@ async def _run_loop(self) -> None:
                 the maximum number of restarts.
             BaseException: If this actor's `_run()` raises any other `BaseException`.
         """
-        _logger.info("Actor %s: Starting...", self)
+        _logger.info("Actor %s: Started.", self)
         n_restarts = 0
         while True:
             try:

src/frequenz/sdk/actor/_background_service.py

@@ -46,7 +46,7 @@ def __init__(self, resolution_s: float, *, name: str | None = None) -> None:
                 super().__init__(name=name)
                 self._resolution_s = resolution_s
 
-            async def start(self) -> None:
+            def start(self) -> None:
                 self._tasks.add(asyncio.create_task(self._tick()))
 
             async def _tick(self) -> None:
@@ -61,7 +61,7 @@ async def main() -> None:
 
             # Manual start/stop (only use if necessary, as cleanup is more complicated)
             clock = Clock(resolution_s=1)
-            await clock.start()
+            clock.start()
             await asyncio.sleep(5)
             await clock.stop()
 
@@ -80,7 +80,7 @@ def __init__(self, *, name: str | None = None) -> None:
         self._tasks: set[asyncio.Task[Any]] = set()
 
     @abc.abstractmethod
-    async def start(self) -> None:
+    def start(self) -> None:
         """Start this background service."""
 
     @property
@@ -166,7 +166,7 @@ async def __aenter__(self) -> Self:
         Returns:
             This background service.
         """
-        await self.start()
+        self.start()
         return self
 
     async def __aexit__(

src/frequenz/sdk/actor/_run_utils.py

@@ -19,26 +19,16 @@ async def run(*actors: Actor) -> None:
         actors: the actors to be awaited.
     """
     _logger.info("Starting %s actor(s)...", len(actors))
-    await _wait_tasks(
-        set(asyncio.create_task(a.start(), name=str(a)) for a in actors),
-        "starting",
-        "started",
-    )
 
-    # Wait until all actors are done
-    await _wait_tasks(
-        set(asyncio.create_task(a.wait(), name=str(a)) for a in actors),
-        "running",
-        "finished",
-    )
-
-    _logger.info("All %s actor(s) finished.", len(actors))
+    for actor in actors:
+        if actor.is_running:
+            _logger.info("Actor %s: Already running, skipping start.", actor)
+        else:
+            _logger.info("Actor %s: Starting...", actor)
+            actor.start()
 
-
-async def _wait_tasks(
-    tasks: set[asyncio.Task[None]], error_str: str, success_str: str
-) -> None:
-    pending_tasks = tasks
+    # Wait until all actors are done
+    pending_tasks = set(asyncio.create_task(a.wait(), name=str(a)) for a in actors)
     while pending_tasks:
         done_tasks, pending_tasks = await asyncio.wait(
             pending_tasks, return_when=asyncio.FIRST_COMPLETED
@@ -49,19 +39,14 @@ async def _wait_tasks(
             # Cancellation needs to be checked first, otherwise the other methods
             # could raise a CancelledError
             if task.cancelled():
-                _logger.info(
-                    "Actor %s: Cancelled while %s.",
-                    task.get_name(),
-                    error_str,
-                )
+                _logger.info("Actor %s: Cancelled while running.", task.get_name())
             elif exception := task.exception():
                 _logger.error(
-                    "Actor %s: Raised an exception while %s.",
+                    "Actor %s: Raised an exception while running.",
                     task.get_name(),
-                    error_str,
                     exc_info=exception,
                 )
             else:
-                _logger.info(
-                    "Actor %s: %s normally.", task.get_name(), success_str.capitalize()
-                )
+                _logger.info("Actor %s: Finished normally.", task.get_name())
+
+    _logger.info("All %s actor(s) finished.", len(actors))

src/frequenz/sdk/microgrid/_data_pipeline.py

@@ -215,6 +215,7 @@ def _start_power_distributing_actor(self) -> None:
             channel_registry=self._channel_registry,
             battery_status_sender=self._battery_status_channel.new_sender(),
         )
+        self._power_distributing_actor.start()
 
     def _data_sourcing_request_sender(self) -> Sender[ComponentMetricRequest]:
         """Return a Sender for sending requests to the data sourcing actor.
@@ -237,6 +238,7 @@ def _data_sourcing_request_sender(self) -> Sender[ComponentMetricRequest]:
                 registry=self._channel_registry,
             )
             self._data_sourcing_actor = _ActorInfo(actor, channel)
+            self._data_sourcing_actor.actor.start()
         return self._data_sourcing_actor.channel.new_sender()
 
     def _resampling_request_sender(self) -> Sender[ComponentMetricRequest]:
@@ -262,17 +264,9 @@ def _resampling_request_sender(self) -> Sender[ComponentMetricRequest]:
                 config=self._resampler_config,
             )
             self._resampling_actor = _ActorInfo(actor, channel)
+            self._resampling_actor.actor.start()
         return self._resampling_actor.channel.new_sender()
 
-    async def _start(self) -> None:
-        """Start the data pipeline actors."""
-        if self._data_sourcing_actor:
-            await self._data_sourcing_actor.actor.start()
-        if self._resampling_actor:
-            await self._resampling_actor.actor.start()
-        # The power distributing actor is started lazily when the first battery pool is
-        # created.
-
     async def _stop(self) -> None:
         """Stop the data pipeline actors."""
         if self._data_sourcing_actor:
@@ -300,7 +294,6 @@ async def initialize(resampler_config: ResamplerConfig) -> None:
     if _DATA_PIPELINE is not None:
         raise RuntimeError("DataPipeline is already initialized.")
     _DATA_PIPELINE = _DataPipeline(resampler_config)
-    await _DATA_PIPELINE._start()  # pylint: disable=protected-access
 
 
 def logical_meter() -> LogicalMeter:

src/frequenz/sdk/timeseries/_moving_window.py

@@ -190,7 +190,7 @@ def __init__(  # pylint: disable=too-many-arguments
             align_to=align_to,
         )
 
-    async def start(self) -> None:
+    def start(self) -> None:
         """Start the MovingWindow.
 
         This method starts the MovingWindow tasks.

tests/actor/test_actor.py

@@ -175,7 +175,7 @@ async def test_basic_actor(caplog: pytest.LogCaptureFixture) -> None:
         original_tasks = set(actor.tasks)
 
         # Start is a no-op if already started
-        await actor.start()
+        actor.start()
         assert actor.is_running is True
         assert original_tasks == set(actor.tasks)
 
@@ -192,7 +192,7 @@ async def test_basic_actor(caplog: pytest.LogCaptureFixture) -> None:
     assert actor.is_running is False
     assert BaseTestActor.restart_count == 0
     assert caplog.record_tuples == [
-        (*ACTOR_INFO, "Actor EchoActor[EchoActor]: Starting..."),
+        (*ACTOR_INFO, "Actor EchoActor[EchoActor]: Started."),
         (*ACTOR_INFO, "Actor EchoActor[EchoActor]: Cancelled."),
     ]
 
@@ -222,12 +222,14 @@ async def test_restart_on_unhandled_exception(
                 await channel.new_sender().send(i)
 
             await run(actor)
+            await actor.wait()
 
     assert actor.is_running is False
     assert BaseTestActor.restart_count == restart_limit
     expected_log = [
         (*RUN_INFO, "Starting 1 actor(s)..."),
-        (*ACTOR_INFO, "Actor RaiseExceptionActor[test]: Starting..."),
+        (*RUN_INFO, "Actor RaiseExceptionActor[test]: Starting..."),
+        (*ACTOR_INFO, "Actor RaiseExceptionActor[test]: Started."),
     ]
     for i in range(restart_limit):
         expected_log.extend(
@@ -253,10 +255,6 @@ async def test_restart_on_unhandled_exception(
                 "Actor RaiseExceptionActor[test]: Maximum restarts attempted "
                 f"({restart_limit}/{restart_limit}), bailing out...",
             ),
-            (
-                *RUN_INFO,
-                "Actor RaiseExceptionActor[test]: Started normally.",
-            ),
             (
                 *RUN_ERROR,
                 "Actor RaiseExceptionActor[test]: Raised an exception while running.",
@@ -286,10 +284,10 @@ async def test_does_not_restart_on_normal_exit(
     assert BaseTestActor.restart_count == 0
     assert caplog.record_tuples == [
         (*RUN_INFO, "Starting 1 actor(s)..."),
-        (*ACTOR_INFO, "Actor NopActor[test]: Starting..."),
+        (*RUN_INFO, "Actor NopActor[test]: Starting..."),
+        (*ACTOR_INFO, "Actor NopActor[test]: Started."),
         (*ACTOR_INFO, "Actor NopActor[test]: _run() returned without error."),
         (*ACTOR_INFO, "Actor NopActor[test]: Stopped."),
-        (*RUN_INFO, "Actor NopActor[test]: Started normally."),
         (*RUN_INFO, "Actor NopActor[test]: Finished normally."),
         (*RUN_INFO, "All 1 actor(s) finished."),
     ]
@@ -319,9 +317,9 @@ async def test_does_not_restart_on_base_exception(
     assert BaseTestActor.restart_count == 0
     assert caplog.record_tuples == [
         (*RUN_INFO, "Starting 1 actor(s)..."),
-        (*ACTOR_INFO, "Actor RaiseBaseExceptionActor[test]: Starting..."),
+        (*RUN_INFO, "Actor RaiseBaseExceptionActor[test]: Starting..."),
+        (*ACTOR_INFO, "Actor RaiseBaseExceptionActor[test]: Started."),
         (*ACTOR_ERROR, "Actor RaiseBaseExceptionActor[test]: Raised a BaseException."),
-        (*RUN_INFO, "Actor RaiseBaseExceptionActor[test]: Started normally."),
         (
             *RUN_ERROR,
             "Actor RaiseBaseExceptionActor[test]: Raised an exception while running.",
@@ -373,8 +371,8 @@ async def cancel_actor() -> None:
     assert BaseTestActor.restart_count == 0
     assert caplog.record_tuples == [
         (*RUN_INFO, "Starting 1 actor(s)..."),
-        (*ACTOR_INFO, "Actor EchoActor[EchoActor]: Starting..."),
-        (*RUN_INFO, "Actor EchoActor[EchoActor]: Started normally."),
+        (*RUN_INFO, "Actor EchoActor[EchoActor]: Starting..."),
+        (*ACTOR_INFO, "Actor EchoActor[EchoActor]: Started."),
         (*ACTOR_INFO, "Actor EchoActor[EchoActor]: Cancelled."),
         (*RUN_ERROR, "Actor EchoActor[EchoActor]: Raised an exception while running."),
         (*RUN_INFO, "All 1 actor(s) finished."),

tests/actor/test_background_service.py

@@ -36,7 +36,7 @@ def __init__(
         self._sleep = sleep
         self._exc = exc
 
-    async def start(self) -> None:
+    def start(self) -> None:
         """Start this service."""
 
         async def nop() -> None:
@@ -76,7 +76,7 @@ async def test_start_await() -> None:
     await fake_service.stop()
     assert fake_service.is_running is False
 
-    await fake_service.start()
+    fake_service.start()
     assert fake_service.is_running is True
 
     # Should stop immediately
@@ -96,7 +96,7 @@ async def test_start_stop() -> None:
     await fake_service.stop()
     assert fake_service.is_running is False
 
-    await fake_service.start()
+    fake_service.start()
     assert fake_service.is_running is True
 
     await asyncio.sleep(1.0)
@@ -119,7 +119,7 @@ async def test_start_and_crash(
     assert fake_service.name == "test"
     assert fake_service.is_running is False
 
-    await fake_service.start()
+    fake_service.start()
     with pytest.raises(BaseExceptionGroup) as exc_info:
         match method:
             case "await":
@@ -146,7 +146,8 @@ async def test_async_context_manager() -> None:
     async with FakeService(name="test", sleep=1.0) as fake_service:
         assert fake_service.is_running is True
         # Is a no-op if the service is running
-        await fake_service.start()
+        fake_service.start()
+        await asyncio.sleep(0)
         assert fake_service.is_running is True
 
     assert fake_service.is_running is False

tests/microgrid/test_datapipeline.py

@@ -0,0 +1,75 @@
+# License: MIT
+# Copyright © 2023 Frequenz Energy-as-a-Service GmbH
+
+"""Basic tests for the DataPipeline."""
+
+import asyncio
+from datetime import timedelta
+from typing import Iterator
+
+import async_solipsism
+import pytest
+from pytest_mock import MockerFixture
+
+from frequenz.sdk.microgrid._data_pipeline import _DataPipeline
+from frequenz.sdk.microgrid.client import Connection
+from frequenz.sdk.microgrid.component import Component, ComponentCategory, InverterType
+from frequenz.sdk.timeseries._resampling import ResamplerConfig
+
+from ..utils.mock_microgrid_client import MockMicrogridClient
+
+
+@pytest.fixture
+def event_loop() -> Iterator[async_solipsism.EventLoop]:
+    """Replace the loop with one that doesn't interact with the outside world."""
+    loop = async_solipsism.EventLoop()
+    asyncio.set_event_loop(loop)  # Set the loop as default
+    yield loop
+    loop.close()
+
+
+async def test_actors_started(mocker: MockerFixture) -> None:
+    """Test that the datasourcing, resampling and power distributing actors are started."""
+
+    datapipeline = _DataPipeline(
+        resampler_config=ResamplerConfig(resampling_period=timedelta(seconds=1))
+    )
+    await asyncio.sleep(1)
+
+    # pylint: disable=protected-access
+    assert datapipeline._data_sourcing_actor is None
+    assert datapipeline._resampling_actor is None
+    assert datapipeline._power_distributing_actor is None
+
+    datapipeline.logical_meter()
+
+    assert datapipeline._data_sourcing_actor is not None
+    assert datapipeline._data_sourcing_actor.actor is not None
+    await asyncio.sleep(1)
+    assert datapipeline._data_sourcing_actor.actor.is_running
+
+    assert datapipeline._resampling_actor is not None
+    assert datapipeline._resampling_actor.actor is not None
+    assert datapipeline._resampling_actor.actor.is_running
+
+    assert datapipeline._power_distributing_actor is None
+
+    mock_client = MockMicrogridClient(
+        set(
+            [
+                Component(1, ComponentCategory.GRID),
+                Component(4, ComponentCategory.INVERTER, InverterType.BATTERY),
+                Component(15, ComponentCategory.BATTERY),
+            ]
+        ),
+        connections=set([Connection(1, 4), Connection(4, 15)]),
+    )
+    mock_client.initialize(mocker)
+
+    datapipeline.battery_pool()
+
+    assert datapipeline._power_distributing_actor is not None
+    await asyncio.sleep(1)
+    assert datapipeline._power_distributing_actor.is_running
+
+    await datapipeline._stop()