Fix resampler behaviour when resampling is to late

Previous solution used Timer, that was fired after the specific time.
Also it didn't take time difference spent on resampling into
account when sleeping.
In result the time difference between resampling period was
constantly increasing. And there was no option to catch up with
resampling.

Signed-off-by: ela-kotulska-frequenz <[email protected]>

Author: ela-kotulska-frequenz

src/frequenz/sdk/timeseries/_resampling.py

@@ -429,6 +429,30 @@ def remove_timeseries(self, source: Source) -> bool:
             return False
         return True
 
+    async def _wait_for_next_resampling_period(self) -> None:
+        """Wait for next resampling period.
+
+        If resampling period already started, then return without sleeping.
+        That would allow us to catch up with resampling.
+        Print warning if function woke up to late.
+        """
+        now = datetime.now(tz=timezone.utc)
+        if self._window_end > now:
+            sleep_for = self._window_end - now
+            await asyncio.sleep(sleep_for.total_seconds())
+
+        timer_error_s = (now - self._window_end).total_seconds()
+        if timer_error_s > (self._config.resampling_period_s / 10.0):
+            _logger.warning(
+                "The resampling task woke up too late. Resampling should have started "
+                "at %s, but it started at %s (%s seconds difference; resampling "
+                "period is %s seconds)",
+                self._window_end,
+                now,
+                timer_error_s,
+                self._config.resampling_period_s,
+            )
+
     async def resample(self, *, one_shot: bool = False) -> None:
         """Start resampling all known timeseries.
 
@@ -447,12 +471,17 @@ async def resample(self, *, one_shot: bool = False) -> None:
                 timeseries from the resampler before calling this method
                 again).
         """
-        async for timer_timestamp in self._timer:
+        while True:
+            await self._wait_for_next_resampling_period()
+
             results = await asyncio.gather(
                 *[r.resample(self._window_end) for r in self._resamplers.values()],
                 return_exceptions=True,
             )
-            self._update_window_end(timer_timestamp)
+
+            self._window_end = self._window_end + timedelta(
+                seconds=self._config.resampling_period_s
+            )
             exceptions = {
                 source: results[i]
                 for i, source in enumerate(self._resamplers)
@@ -465,27 +494,6 @@ async def resample(self, *, one_shot: bool = False) -> None:
             if one_shot:
                 break
 
-    def _update_window_end(self, timer_timestamp: datetime) -> None:
-        # We use abs() here to account for errors in the timer where the timer
-        # fires before its time even when in theory it shouldn't be possible,
-        # but we want to be resilient to timer implementation changes that
-        # could end up in this case, either because there were some time jump
-        # or some rounding error.
-        timer_error_s = abs((timer_timestamp - self._window_end).total_seconds())
-        if timer_error_s > (self._config.resampling_period_s / 10.0):
-            _logger.warning(
-                "The resampling timer fired too late. It should have fired at "
-                "%s, but it fired at %s (%s seconds difference; resampling "
-                "period is %s seconds)",
-                self._window_end,
-                timer_timestamp,
-                timer_error_s,
-                self._config.resampling_period_s,
-            )
-        self._window_end = self._window_end + timedelta(
-            seconds=self._config.resampling_period_s
-        )
-
 
 class _ResamplingHelper:
     """Keeps track of *relevant* samples to pass them to the resampling function.

tests/timeseries/test_resampling.py

@@ -341,8 +341,9 @@ async def test_timer_errors_are_logged(
     assert (
         "frequenz.sdk.timeseries._resampling",
         logging.WARNING,
-        "The resampling timer fired too late. It should have fired at 1970-01-01 00:00:04+00:00, "
-        "but it fired at 1970-01-01 00:00:04.399800+00:00 (0.3998 seconds difference; resampling "
+        "The resampling task woke up too late. Resampling should have started at "
+        "1970-01-01 00:00:04+00:00, but it started at "
+        "1970-01-01 00:00:04.399800+00:00 (0.3998 seconds difference; resampling "
         "period is 2 seconds)",
     ) in _filter_logs(caplog.record_tuples, logger_level=logging.WARNING)
     sink_mock.reset_mock()