Timer: Add support for rearming timer with new interval

RELEASE_NOTES.md

@@ -1,5 +1,11 @@
 # Frequenz channels Release Notes
 
+## New Features
+
+- `Timer.reset()` now supports setting the interval and will restart the timer with the new interval.
+
 ## Bug Fixes
 
 - `FileWatcher`: Fixed `ready()` method to return False when an error occurs. Before this fix, `select()` (and other code using `ready()`) never detected the `FileWatcher` was stopped and the `select()` loop was continuously waking up to inform the receiver was ready.
+
+- `Timer.stop()` and `Timer.reset()` now immediately stop the timer if it is running. Before this fix, the timer would continue to run until the next interval.

src/frequenz/channels/timer.py

@@ -523,6 +523,8 @@ def __init__(  # pylint: disable=too-many-arguments
         See the documentation of `MissedTickPolicy` for details.
         """
 
+        self._reset_event = asyncio.Event()
+
         self._loop: asyncio.AbstractEventLoop = (
             loop if loop is not None else asyncio.get_running_loop()
         )
@@ -584,7 +586,12 @@ def is_running(self) -> bool:
         """Whether the timer is running."""
         return not self._stopped
 
-    def reset(self, *, start_delay: timedelta = timedelta(0)) -> None:
+    def reset(
+        self,
+        *,
+        interval: timedelta | None = None,
+        start_delay: timedelta = timedelta(0),
+    ) -> None:
         """Reset the timer to start timing from now (plus an optional delay).
 
         If the timer was stopped, or not started yet, it will be started.
@@ -593,6 +600,8 @@ def reset(self, *, start_delay: timedelta = timedelta(0)) -> None:
         more details.
 
         Args:
+            interval: The new interval between ticks. If `None`, the current
+                interval is kept.
             start_delay: The delay before the timer should start. This has microseconds
                 resolution, anything smaller than a microsecond means no delay.
 
@@ -604,8 +613,16 @@ def reset(self, *, start_delay: timedelta = timedelta(0)) -> None:
 
         if start_delay_ms < 0:
             raise ValueError(f"`start_delay` can't be negative, got {start_delay}")
-        self._stopped = False
+
+        if interval is not None:
+            self._interval = _to_microseconds(interval)
+
         self._next_tick_time = self._now() + start_delay_ms + self._interval
+
+        if self.is_running:
+            self._reset_event.set()
+
+        self._stopped = False
         self._current_drift = None
 
     def stop(self) -> None:
@@ -621,6 +638,7 @@ def stop(self) -> None:
         self._stopped = True
         # We need to make sure it's not None, otherwise `ready()` will start it
         self._next_tick_time = self._now()
+        self._reset_event.set()
 
     # We need a noqa here because the docs have a Raises section but the documented
     # exceptions are raised indirectly.
@@ -664,7 +682,15 @@ async def ready(self) -> bool:  # noqa: DOC502
         # could be reset while we are sleeping, in which case we need to recalculate
         # the time to the next tick and try again.
         while time_to_next_tick > 0:
-            await asyncio.sleep(time_to_next_tick / 1_000_000)
+            await next(
+                asyncio.as_completed(
+                    [
+                        asyncio.sleep(time_to_next_tick / 1_000_000),
+                        self._reset_event.wait(),
+                    ]
+                )
+            )
+            self._reset_event.clear()
             now = self._now()
             time_to_next_tick = self._next_tick_time - now
 

tests/test_timer.py

@@ -558,3 +558,60 @@ async def test_timer_skip_missed_and_drift(
     drift = await timer.receive()
     assert event_loop.time() == pytest.approx(interval * 14 + tolerance * 3 + 0.001)
     assert drift == pytest.approx(timedelta(seconds=0.0))
+
+
+async def test_timer_reset_with_new_interval(
+    event_loop: async_solipsism.EventLoop,  # pylint: disable=redefined-outer-name
+) -> None:
+    """Test resetting the timer with a new interval."""
+    initial_interval = timedelta(seconds=1.0)
+    new_interval = timedelta(seconds=2.0)
+    timer = Timer(initial_interval, TriggerAllMissed())
+
+    # Wait for the first tick
+    drift = await timer.receive()
+    assert drift == timedelta(seconds=0.0)
+    assert event_loop.time() == pytest.approx(1.0)
+
+    # Reset the timer with a new interval
+    timer.reset(interval=new_interval)
+
+    # The next tick should occur after the new interval
+    drift = await timer.receive()
+    assert drift == timedelta(seconds=0.0)
+    assert event_loop.time() == pytest.approx(3.0)
+
+    # Ensure the timer continues with the new interval
+    drift = await timer.receive()
+    assert drift == timedelta(seconds=0.0)
+    assert event_loop.time() == pytest.approx(5.0)
+
+
+async def test_timer_immediate_interruption_on_reset() -> None:
+    """Test that the timer is interrupted immediately upon reset."""
+    timer1 = Timer(timedelta(seconds=5.0), TriggerAllMissed())
+    timer2 = Timer(timedelta(seconds=1.0), TriggerAllMissed())
+    timer3 = Timer(timedelta(seconds=4.0), TriggerAllMissed())
+
+    timer_trigger_order = []
+
+    async def reset_timer1() -> None:
+        await timer2.receive()
+        timer_trigger_order.append(2)
+        timer1.reset(interval=timedelta(seconds=1.0))
+        timer2.stop()
+
+    async def receive_timer2() -> None:
+        await timer1.receive()
+        timer_trigger_order.append(1)
+
+    async def receive_timer3() -> None:
+        await timer3.receive()
+        timer_trigger_order.append(3)
+
+    task1 = asyncio.create_task(reset_timer1())
+    task2 = asyncio.create_task(receive_timer2())
+    task3 = asyncio.create_task(receive_timer3())
+
+    await asyncio.wait([task1, task2, task3])
+    assert timer_trigger_order == [2, 1, 3]