Add a drop_old_proposals method in PowerManager algorithms

In the matryoshka algorithm, this method is implemented to delete
proposals that haven't been replaced/updated in the last
`max_proposal_age`.

Signed-off-by: Sahas Subramanian <[email protected]>

Author: shsms

src/frequenz/sdk/actor/_power_managing/_base_classes.py

@@ -285,3 +285,13 @@ def get_status(
         Returns:
             The bounds for the components.
         """
+
+    @abc.abstractmethod
+    def drop_old_proposals(self, loop_time: float) -> None:
+        """Drop old proposals.
+
+        This method is called periodically by the power manager.
+
+        Args:
+            loop_time: The current loop time.
+        """

src/frequenz/sdk/actor/_power_managing/_matryoshka.py

@@ -21,6 +21,7 @@
 
 import logging
 import typing
+from datetime import timedelta
 
 from typing_extensions import override
 
@@ -40,8 +41,9 @@
 class Matryoshka(BaseAlgorithm):
     """The matryoshka algorithm."""
 
-    def __init__(self) -> None:
+    def __init__(self, max_proposal_age: timedelta) -> None:
         """Create a new instance of the matryoshka algorithm."""
+        self._max_proposal_age_sec = max_proposal_age.total_seconds()
         self._component_buckets: dict[frozenset[int], SortedSet[Proposal]] = {}
         self._target_power: dict[frozenset[int], Power] = {}
 
@@ -267,3 +269,21 @@ def get_status(
             _exclusion_bounds=system_bounds.exclusion_bounds,
             distribution_result=distribution_result,
         )
+
+    @override
+    def drop_old_proposals(self, loop_time: float) -> None:
+        """Drop old proposals.
+
+        This will remove all proposals that have not been updated for longer than
+        `max_proposal_age`.
+
+        Args:
+            loop_time: The current loop time.
+        """
+        for bucket in self._component_buckets.values():
+            to_delete = []
+            for proposal in bucket:
+                if (loop_time - proposal.creation_time) > self._max_proposal_age_sec:
+                    to_delete.append(proposal)
+            for proposal in to_delete:
+                bucket.delete(proposal)

src/frequenz/sdk/actor/_power_managing/_power_managing_actor.py

@@ -75,7 +75,9 @@ def __init__(  # pylint: disable=too-many-arguments
         self._subscriptions: dict[frozenset[int], dict[int, Sender[_Report]]] = {}
         self._distribution_results: dict[frozenset[int], power_distributing.Result] = {}
 
-        self._algorithm: BaseAlgorithm = Matryoshka()
+        self._algorithm: BaseAlgorithm = Matryoshka(
+            max_proposal_age=timedelta(seconds=60.0)
+        )
 
         super().__init__()
 

tests/actor/_power_managing/test_matryoshka.py

@@ -4,7 +4,7 @@
 """Tests for the Matryoshka power manager algorithm."""
 
 import asyncio
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 from frequenz.sdk import timeseries
 from frequenz.sdk.actor._power_managing import Proposal
@@ -23,7 +23,7 @@ def __init__(
         """Create a new instance of the stateful tester."""
         self._call_count = 0
         self._batteries = batteries
-        self._algorithm = Matryoshka()
+        self._algorithm = Matryoshka(max_proposal_age=timedelta(seconds=60.0))
         self._system_bounds = system_bounds
 
     def tgt_power(  # pylint: disable=too-many-arguments