Drop empty Matryoshka buckets (#1155)

If buckets remain for ever in the power manager's Matryoshka instances,
proposals with overlapping component IDs will never get accepted. But if
empty buckets go away as soon as they become empty, then it becomes
possible to propose requests for components without conflicting with old
unused component ID buckets.

Author: shsms

RELEASE_NOTES.md

@@ -14,4 +14,4 @@
 
 ## Bug Fixes
 
-<!-- Here goes notable bug fixes that are worth a special mention or explanation -->
+- Fixed a bug that was preventing power proposals to go through if there once existed some proposals with overlapping component IDs, even if the old proposals have expired.

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

@@ -199,7 +199,15 @@ def calculate_target_power(
             bucket = self._component_buckets.setdefault(component_ids, set())
             if proposal in bucket:
                 bucket.remove(proposal)
-            bucket.add(proposal)
+            if (
+                proposal.preferred_power is not None
+                or proposal.bounds.lower is not None
+                or proposal.bounds.upper is not None
+            ):
+                bucket.add(proposal)
+            elif not bucket:
+                del self._component_buckets[component_ids]
+                _ = self._target_power.pop(component_ids, None)
 
         # If there has not been any proposal for the given components, don't calculate a
         # target power and just return `None`.
@@ -292,10 +300,17 @@ def drop_old_proposals(self, loop_time: float) -> None:
         Args:
             loop_time: The current loop time.
         """
-        for bucket in self._component_buckets.values():
-            to_delete = []
-            for proposal in bucket:
+        buckets_to_delete: list[frozenset[int]] = []
+        for component_ids, proposals in self._component_buckets.items():
+            to_delete: list[Proposal] = []
+            for proposal in proposals:
                 if (loop_time - proposal.creation_time) > self._max_proposal_age_sec:
                     to_delete.append(proposal)
             for proposal in to_delete:
-                bucket.remove(proposal)
+                proposals.remove(proposal)
+            if not proposals:
+                buckets_to_delete.append(component_ids)
+
+        for component_ids in buckets_to_delete:
+            del self._component_buckets[component_ids]
+            _ = self._target_power.pop(component_ids, None)

tests/actor/_power_managing/test_matryoshka.py

@@ -4,8 +4,10 @@
 """Tests for the Matryoshka power manager algorithm."""
 
 import asyncio
+import re
 from datetime import datetime, timedelta, timezone
 
+import pytest
 from frequenz.quantities import Power
 
 from frequenz.sdk import timeseries
@@ -36,13 +38,14 @@ def tgt_power(  # pylint: disable=too-many-arguments,too-many-positional-argumen
         expected: float | None,
         creation_time: float | None = None,
         must_send: bool = False,
+        batteries: frozenset[int] | None = None,
     ) -> None:
         """Test the target power calculation."""
         self._call_count += 1
         tgt_power = self.algorithm.calculate_target_power(
-            self._batteries,
+            self._batteries if batteries is None else batteries,
             Proposal(
-                component_ids=self._batteries,
+                component_ids=self._batteries if batteries is None else batteries,
                 source_id=f"actor-{priority}",
                 preferred_power=None if power is None else Power.from_watts(power),
                 bounds=timeseries.Bounds(
@@ -368,6 +371,7 @@ async def test_matryoshka_drop_old_proposals() -> None:
     With inclusion bounds, and exclusion bounds -30.0 to 30.0.
     """
     batteries = frozenset({2, 5})
+    overlapping_batteries = frozenset({5, 8})
 
     system_bounds = _base_types.SystemBounds(
         timestamp=datetime.now(tz=timezone.utc),
@@ -424,3 +428,115 @@ async def test_matryoshka_drop_old_proposals() -> None:
     tester.tgt_power(
         priority=1, power=20.0, bounds=(20.0, 50.0), expected=25.0, must_send=True
     )
+
+    # When all proposals are too old, they are dropped, and the buckets are dropped as
+    # well.  After that, sending a request for a different but overlapping bucket will
+    # succeed.  And it will fail until then.
+    with pytest.raises(
+        NotImplementedError,
+        match=re.escape(
+            "PowerManagingActor: component IDs frozenset({8, 5}) are already "
+            + "part of another bucket.  Overlapping buckets are not yet supported."
+        ),
+    ):
+        tester.tgt_power(
+            priority=1,
+            power=25.0,
+            bounds=(25.0, 50.0),
+            expected=25.0,
+            must_send=True,
+            batteries=overlapping_batteries,
+        )
+
+    tester.tgt_power(
+        priority=1,
+        power=25.0,
+        bounds=(25.0, 50.0),
+        creation_time=now - 70.0,
+        expected=25.0,
+        must_send=True,
+    )
+    tester.tgt_power(
+        priority=2,
+        power=25.0,
+        bounds=(25.0, 50.0),
+        creation_time=now - 70.0,
+        expected=25.0,
+        must_send=True,
+    )
+    tester.tgt_power(
+        priority=3,
+        power=25.0,
+        bounds=(25.0, 50.0),
+        creation_time=now - 70.0,
+        expected=25.0,
+        must_send=True,
+    )
+
+    tester.algorithm.drop_old_proposals(now)
+
+    tester.tgt_power(
+        priority=1,
+        power=25.0,
+        bounds=(25.0, 50.0),
+        expected=25.0,
+        must_send=True,
+        batteries=overlapping_batteries,
+    )
+
+
+async def test_matryoshka_none_proposals() -> None:
+    """Tests for the power managing actor.
+
+    When a `None` proposal is received, is source id should be dropped from the bucket.
+    Then if the bucket becomes empty, it should be dropped as well.
+    """
+    batteries = frozenset({2, 5})
+    overlapping_batteries = frozenset({5, 8})
+
+    system_bounds = _base_types.SystemBounds(
+        timestamp=datetime.now(tz=timezone.utc),
+        inclusion_bounds=timeseries.Bounds(
+            lower=Power.from_watts(-200.0), upper=Power.from_watts(200.0)
+        ),
+        exclusion_bounds=timeseries.Bounds(lower=Power.zero(), upper=Power.zero()),
+    )
+
+    def ensure_overlapping_bucket_request_fails() -> None:
+        with pytest.raises(
+            NotImplementedError,
+            match=re.escape(
+                "PowerManagingActor: component IDs frozenset({8, 5}) are already "
+                + "part of another bucket.  Overlapping buckets are not yet supported."
+            ),
+        ):
+            tester.tgt_power(
+                priority=1,
+                power=None,
+                bounds=(20.0, 50.0),
+                expected=None,
+                must_send=True,
+                batteries=overlapping_batteries,
+            )
+
+    tester = StatefulTester(batteries, system_bounds)
+
+    tester.tgt_power(priority=3, power=22.0, bounds=(22.0, 30.0), expected=22.0)
+    tester.tgt_power(priority=2, power=25.0, bounds=(25.0, 50.0), expected=25.0)
+    tester.tgt_power(priority=1, power=20.0, bounds=(20.0, 50.0), expected=None)
+
+    ensure_overlapping_bucket_request_fails()
+    tester.tgt_power(priority=1, power=None, bounds=(None, None), expected=None)
+    ensure_overlapping_bucket_request_fails()
+    tester.tgt_power(priority=3, power=None, bounds=(None, None), expected=None)
+    ensure_overlapping_bucket_request_fails()
+    tester.tgt_power(priority=2, power=None, bounds=(None, None), expected=None)
+
+    # Overlapping battery bucket is dropped.
+    tester.tgt_power(
+        priority=1,
+        power=20.0,
+        bounds=(20.0, 50.0),
+        expected=20.0,
+        batteries=overlapping_batteries,
+    )