Prevent stacking of power requests

The power distributing actor processes one power request at a time
to prevent multiple requests for the same components from being sent
to the microgrid API concurrently. Previously, this could lead to
the request channel receiver becoming full if the power request
frequency was higher than the processing time. Even worse, the
requests could be processed late, causing unexpected behavior
for applications setting power requests. Moreover, the actor was
blocking power requests with different sets of components from being
processed if there was any existing request.

This patch ensures that the actor processes one request at a time
for different sets of components and keeps track of the latest pending
request if there is an existing request with the same set of
components being processed. The pending request will be overwritten
by the latest received request with the same set of components,
and the actor will process it once the request with the same components
is done processing.

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

Author: daniel-zullo-frequenz

src/frequenz/sdk/actor/power_distributing/power_distributing.py

@@ -10,6 +10,8 @@
 """
 
 
+import asyncio
+import logging
 from datetime import timedelta
 
 from frequenz.channels import Receiver, Sender
@@ -27,6 +29,8 @@
 from .request import Request
 from .result import Result
 
+_logger = logging.getLogger(__name__)
+
 
 class PowerDistributingActor(Actor):
     # pylint: disable=too-many-instance-attributes
@@ -97,6 +101,16 @@ def __init__(  # pylint: disable=too-many-arguments
         self._result_sender = results_sender
         self._api_power_request_timeout = api_power_request_timeout
 
+        self._processing_tasks: dict[frozenset[int], asyncio.Task[None]] = {}
+        """Track the power request tasks currently being processed."""
+
+        self._pending_requests: dict[frozenset[int], Request] = {}
+        """Track the power requests that are waiting to be processed.
+
+        Only one pending power request is kept for each set of components, the
+        latest request will overwrite the previous one.
+        """
+
         self._component_manager: ComponentManager
         if component_category == ComponentCategory.BATTERY:
             self._component_manager = BatteryManager(
@@ -135,7 +149,18 @@ async def _run(self) -> None:
         await self._component_manager.start()
 
         async for request in self._requests_receiver:
-            await self._component_manager.distribute_power(request)
+            req_id = frozenset(request.component_ids)
+
+            if req_id in self._processing_tasks:
+                if pending_request := self._pending_requests.get(req_id):
+                    _logger.warning(
+                        "Pending request: %s, overwritten with request: %s",
+                        pending_request,
+                        request,
+                    )
+                self._pending_requests[req_id] = request
+            else:
+                self._process_request(req_id, request)
 
     @override
     async def stop(self, msg: str | None = None) -> None:
@@ -146,3 +171,41 @@ async def stop(self, msg: str | None = None) -> None:
         """
         await self._component_manager.stop()
         await super().stop(msg)
+
+    def _handle_task_completion(
+        self, req_id: frozenset[int], request: Request, task: asyncio.Task[None]
+    ) -> None:
+        """Handle the completion of a power request task.
+
+        Args:
+            req_id: The id to identify the power request.
+            request: The power request that has been processed.
+            task: The task that has completed.
+        """
+        try:
+            task.result()
+        except Exception:  # pylint: disable=broad-except
+            _logger.exception("Failed power request: %s", request)
+
+        if req_id in self._pending_requests:
+            self._process_request(req_id, self._pending_requests.pop(req_id))
+        elif req_id in self._processing_tasks:
+            del self._processing_tasks[req_id]
+        else:
+            _logger.error("Request id not found in processing tasks: %s", req_id)
+
+    def _process_request(self, req_id: frozenset[int], request: Request) -> None:
+        """Process a power request.
+
+        Args:
+            req_id: The id to identify the power request.
+            request: The power request to process.
+        """
+        task = asyncio.create_task(
+            self._component_manager.distribute_power(request),
+            name=f"{type(self).__name__}:{request}",
+        )
+        task.add_done_callback(
+            lambda t: self._handle_task_completion(req_id, request, t)
+        )
+        self._processing_tasks[req_id] = task

tests/timeseries/_battery_pool/test_battery_pool_control_methods.py

@@ -213,6 +213,7 @@ async def test_case_1(
             await bounds_rx.receive(), power=1000.0, lower=-4000.0, upper=4000.0
         )
 
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 4
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, 250.0)
@@ -248,6 +249,7 @@ async def side_effect(inv_id: int, _: float) -> None:
                 result, power_distributing.Success
             ),
         )
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 4
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, 25.0) for inv_id in mocks.microgrid.battery_inverter_ids
@@ -267,6 +269,7 @@ async def side_effect(inv_id: int, _: float) -> None:
 
         # There should be an automatic retry.
         set_power.side_effect = None
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 4
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, 25.0) for inv_id in mocks.microgrid.battery_inverter_ids
@@ -318,6 +321,7 @@ async def test_case_2(self, mocks: Mocks, mocker: MockerFixture) -> None:
         self._assert_report(
             await bounds_1_rx.receive(), power=1000.0, lower=-2000.0, upper=2000.0
         )
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 2
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, 500.0)
@@ -331,6 +335,7 @@ async def test_case_2(self, mocks: Mocks, mocker: MockerFixture) -> None:
         if not latest_dist_result_2.has_value():
             bounds = await bounds_2_rx.receive()
         self._assert_report(bounds, power=1000.0, lower=-2000.0, upper=2000.0)
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 2
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, 500.0)
@@ -375,7 +380,7 @@ async def test_case_3(self, mocks: Mocks, mocker: MockerFixture) -> None:
         self._assert_report(
             await bounds_2_rx.receive(), power=-1000.0, lower=-1000.0, upper=0.0
         )
-
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 4
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, -250.0)
@@ -394,7 +399,7 @@ async def test_case_3(self, mocks: Mocks, mocker: MockerFixture) -> None:
         if not latest_dist_result_2.has_value():
             bounds = await bounds_2_rx.receive()
         self._assert_report(bounds, power=0.0, lower=-1000.0, upper=0.0)
-
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 4
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, 0.0) for inv_id in mocks.microgrid.battery_inverter_ids
@@ -428,6 +433,7 @@ async def test_case_4(self, mocks: Mocks, mocker: MockerFixture) -> None:
         self._assert_report(
             await bounds_rx.receive(), power=1000.0, lower=-4000.0, upper=4000.0
         )
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 4
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, 250.0)
@@ -453,6 +459,7 @@ async def test_case_4(self, mocks: Mocks, mocker: MockerFixture) -> None:
         self._assert_report(
             await bounds_rx.receive(), power=400.0, lower=-4000.0, upper=4000.0
         )
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 4
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, 100.0)
@@ -477,6 +484,7 @@ async def test_case_4(self, mocks: Mocks, mocker: MockerFixture) -> None:
         self._assert_report(
             await bounds_rx.receive(), power=0.0, lower=-4000.0, upper=4000.0
         )
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 4
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, 0.0) for inv_id in mocks.microgrid.battery_inverter_ids
@@ -501,6 +509,7 @@ async def test_case_4(self, mocks: Mocks, mocker: MockerFixture) -> None:
         self._assert_report(
             await bounds_rx.receive(), power=-400.0, lower=-4000.0, upper=4000.0
         )
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 4
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, -100.0)
@@ -586,7 +595,7 @@ async def test_case_5(  # pylint: disable=too-many-statements,too-many-locals
         self._assert_report(
             await bounds_1_rx.receive(), power=200.0, lower=-1000.0, upper=1500.0
         )
-
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 4
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, 50.0) for inv_id in mocks.microgrid.battery_inverter_ids
@@ -624,6 +633,7 @@ async def test_case_5(  # pylint: disable=too-many-statements,too-many-locals
             if dist_result.succeeded_power == Power.from_watts(720.0):
                 break
 
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 4
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, 720.0 / 4)
@@ -664,6 +674,7 @@ async def test_case_5(  # pylint: disable=too-many-statements,too-many-locals
             if dist_result.succeeded_power == Power.from_watts(-280.0):
                 break
 
+        await asyncio.sleep(0.0)  # Wait for the power to be distributed.
         assert set_power.call_count == 4
         assert sorted(set_power.call_args_list) == [
             mocker.call(inv_id, -280.0 / 4)