Add initial buffer length to the resampler

This commit makes the resampler use a proper ring buffer instead of
using an unbound buffer that only gets clear up when a resampling is
done (which could easily end up in memory issues if the input sampling
rate is much higher than the resampling rate).

This also improves the performance of the resampling buy an average of
20% (on my local machine, your millage might vary), even when the
current implementation now needs to copy the buffer when passed to the
resampling function.

Here are the results of running benchmarks/timeseries/resampling.py for
the old implementation, the new implementation (without doing the
bisection and copying) and the new complete implementation with the
current fixable inefficiencies:

         OLD                         NEW                      NEW WITH BISECT

resamples=10 samples=10     resamples=10 samples=10     resamples=10 samples=10
0.0008896420185919851       0.00062773801619187         0.0008012260077521205
resamples=10 samples=100    resamples=10 samples=100    resamples=10 samples=100
0.007817161997081712        0.005761806009104475        0.006307974021183327
resamples=10 samples=1000   resamples=10 samples=1000   resamples=10 samples=1000
0.07768873398890719         0.05851042701397091         0.0604277040110901
resamples=100 samples=10    resamples=100 samples=10    resamples=100 samples=10
0.008742492995224893        0.0062527229893021286       0.00808265499654226
resamples=100 samples=100   resamples=100 samples=100   resamples=100 samples=100
0.07808284999919124         0.057997508003609255        0.0624066719901748
resamples=100 samples=1000  resamples=100 samples=1000  resamples=100 samples=1000
0.782658567011822           0.5870920980232768          0.6098103950207587
resamples=1000 samples=10   resamples=1000 samples=10   resamples=1000 samples=10
0.08764891701866873         0.062448524025967345        0.07815460601705126
resamples=1000 samples=100  resamples=1000 samples=100  resamples=1000 samples=100
0.78426024899818            0.5858371119829826          0.6357307220168877
resamples=1000 samples=1000 resamples=1000 samples=1000 resamples=1000 samples=1000
7.513815971993608           5.984694316983223           6.42200836900156
Average improvement:        35.3%                       19.7%

This commit sadly introduces some nasty hack to be able to bisect the
buffer, we need to make Sample ordered by timestamp because bisect
doesn't support using a key extraction function until Python 3.10.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

benchmarks/timeseries/resampling.py

@@ -25,6 +25,7 @@ def _benchmark_resampling_helper(resamples: int, samples: int) -> None:
             resampling_period_s=1.0,
             max_data_age_in_periods=3.0,
             resampling_function=nop,
+            initial_buffer_len=samples * 3,
         )
     )
     now = datetime.now(timezone.utc)

src/frequenz/sdk/timeseries/_base_types.py

@@ -3,12 +3,16 @@
 
 """Timeseries basic types."""
 
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Optional
 
 
-@dataclass(frozen=True)
+# Ordering by timestamp is a bit arbitrary, and it is not always what might be
+# wanted. We are using this order now because usually we need to do binary
+# searches on sequences of samples, and the Python `bisect` module doesn't
+# support providing a key until Python 3.10.
+@dataclass(frozen=True, order=True)
 class Sample:
     """A measurement taken at a particular point in time.
 
@@ -17,5 +21,8 @@ class Sample:
     coherent view on a group of component metrics for a particular timestamp.
     """
 
-    timestamp: datetime
-    value: Optional[float] = None
+    timestamp: datetime = field(compare=True)
+    """The time when this sample was generated."""
+
+    value: Optional[float] = field(compare=False, default=None)
+    """The value of this sample."""

src/frequenz/sdk/timeseries/_resampling.py

@@ -6,8 +6,10 @@
 from __future__ import annotations
 
 import asyncio
+import itertools
 import logging
 import math
+from bisect import bisect
 from collections import deque
 from dataclasses import dataclass
 from datetime import datetime, timedelta
@@ -21,6 +23,15 @@
 _logger = logging.Logger(__name__)
 
 
+DEFAULT_BUFFER_LEN_INIT = 16
+"""Default initial buffer length.
+
+Buffers will be created initially with this length, but they could grow or
+shrink depending on the source characteristics, like sampling rate, to make
+sure all the requested past sampling periods can be stored.
+"""
+
+
 Source = AsyncIterator[Sample]
 """A source for a timeseries.
 
@@ -113,6 +124,14 @@ class ResamplerConfig:
     value.
     """
 
+    initial_buffer_len: int = DEFAULT_BUFFER_LEN_INIT
+    """The initial length of the resampling buffer.
+
+    The buffer could grow or shrink depending on the source characteristics,
+    like sampling rate, to make sure all the requested past sampling periods
+    can be stored.
+    """
+
 
 class SourceStoppedError(RuntimeError):
     """A timeseries stopped producing samples."""
@@ -306,7 +325,7 @@ def __init__(self, config: ResamplerConfig) -> None:
             config: The configuration for the resampler.
         """
         self._config = config
-        self._buffer: deque[Sample] = deque()
+        self._buffer: deque[Sample] = deque(maxlen=config.initial_buffer_len)
 
     def add_sample(self, sample: Sample) -> None:
         """Add a new sample to the internal buffer.
@@ -316,30 +335,6 @@ def add_sample(self, sample: Sample) -> None:
         """
         self._buffer.append(sample)
 
-    def _remove_outdated_samples(self, threshold: datetime) -> None:
-        """Remove samples that are older than the provided time threshold.
-
-        It is assumed that items in the buffer are in a sorted order (ascending order
-        by timestamp).
-
-        The removal works by traversing the buffer starting from the oldest sample
-        (smallest timestamp) and comparing sample's timestamp with the threshold.
-        If the sample's threshold is smaller than `threshold`, it means that the
-        sample is outdated and it is removed from the buffer. This continues until
-        the first sample that is with timestamp greater or equal to `threshold` is
-        encountered, then buffer is considered up to date.
-
-        Args:
-            threshold: samples whose timestamp is older than the threshold are
-                considered outdated and should be remove from the buffer
-        """
-        while self._buffer:
-            sample: Sample = self._buffer[0]
-            if sample.timestamp > threshold:
-                return
-
-            self._buffer.popleft()
-
     def resample(self, timestamp: datetime) -> Sample:
         """Generate a new sample based on all the current *relevant* samples.
 
@@ -352,18 +347,23 @@ def resample(self, timestamp: datetime) -> Sample:
                 If there are no *relevant* samples, then the new sample will
                 have `None` as `value`.
         """
-        threshold = timestamp - timedelta(
-            seconds=self._config.max_data_age_in_periods
-            * self._config.resampling_period_s
+        conf = self._config
+        minimum_relevant_timestamp = timestamp - timedelta(
+            seconds=conf.resampling_period_s * conf.max_data_age_in_periods
         )
-        self._remove_outdated_samples(threshold=threshold)
-
+        # We need to pass a dummy Sample to bisect because it only support
+        # specifying a key extraction function in Python 3.10, so we need to
+        # compare samples at the moment.
+        cut_index = bisect(self._buffer, Sample(minimum_relevant_timestamp, None))
+        # pylint: disable=fixme
+        # FIXME: This is far from efficient, but we don't want to start new
+        # ring buffer implementation here that uses a list to overcome the
+        # deque limitation of not being able to get slices
+        relevant_samples = list(itertools.islice(self._buffer, cut_index, None))
         value = (
-            None
-            if not self._buffer
-            else self._config.resampling_function(
-                self._buffer, self._config.resampling_period_s
-            )
+            conf.resampling_function(relevant_samples, conf.resampling_period_s)
+            if relevant_samples
+            else None
         )
         return Sample(timestamp, value)