diff --git a/mode/services.py b/mode/services.py index e09376a..bca65e0 100644 --- a/mode/services.py +++ b/mode/services.py @@ -1053,6 +1053,7 @@ async def itertimer( interval: Seconds, *, max_drift_correction: float = 0.1, + max_drift: Optional[float] = None, sleep: Optional[Callable[..., Awaitable]] = None, clock: ClockArg = perf_counter, name: str = "", @@ -1070,6 +1071,13 @@ async def itertimer( Will sleep the full `interval` seconds before returning from first iteration. + Arguments: + max_drift: Log a warning when the timer's wakeup drifts by more + than this many seconds. Defaults to `~mode.timers.Timer`'s + usual ``min(interval * 30%, 1.2s)`` heuristic; pass an + explicit value to raise (or lower) the threshold for timers + where the default is too sensitive. + Examples: ```python @@ -1086,6 +1094,7 @@ async def itertimer( interval, name=name, max_drift_correction=max_drift_correction, + max_drift=max_drift, clock=clock, sleep=sleepfun, ): diff --git a/mode/threads.py b/mode/threads.py index 1cff5fe..972abf8 100644 --- a/mode/threads.py +++ b/mode/threads.py @@ -88,6 +88,14 @@ class ServiceThread(Service): #: underlying thread to be fully started. wait_for_thread: bool = True + #: Drift threshold (in seconds) for the internal keepalive timer's + #: "woke up too late/early" warning. ``None`` uses the same default + #: heuristic as any other `Service.itertimer`. The keepalive timer + #: runs every second purely to keep the thread's own event loop + #: scheduled, so overrides here only affect how noisy that specific + #: log is, not the thread's actual behavior. + keepalive_max_drift: Optional[float] = None + _thread: Optional["WorkerThread"] = None _thread_started: Event _thread_running: Optional[asyncio.Future] = None @@ -264,7 +272,9 @@ async def _serve(self) -> None: @Service.task async def _thread_keepalive(self) -> None: async for _sleep_time in self.itertimer( - 1.0, name=f"_thread_keepalive-{self.label}" + 1.0, + max_drift=self.keepalive_max_drift, + name=f"_thread_keepalive-{self.label}", ): # pragma: no cover # The consumer thread will have a separate event loop, # and so we use this trick to make sure our loop is diff --git a/mode/timers.py b/mode/timers.py index 1b6afdb..101dd06 100644 --- a/mode/timers.py +++ b/mode/timers.py @@ -4,7 +4,7 @@ from collections.abc import AsyncIterator, Awaitable, Iterator from itertools import count from time import perf_counter -from typing import Callable +from typing import Callable, Optional from .utils.logging import get_logger from .utils.times import Seconds, want_seconds @@ -39,6 +39,7 @@ def __init__( interval: Seconds, *, max_drift_correction: float = 0.1, + max_drift: Optional[float] = None, name: str = "", clock: ClockArg = perf_counter, sleep: SleepArg = asyncio.sleep, @@ -50,8 +51,16 @@ def __init__( self.sleep: SleepArg = sleep interval_s = self.interval_s = want_seconds(interval) - # Log when drift exceeds this number - self.max_drift = min(interval_s * MAX_DRIFT_PERCENT, MAX_DRIFT_CEILING) + # Log when drift exceeds this number. + # Callers that find the default threshold too sensitive (e.g. it + # logs on drift that isn't actionable for them) can pass an + # explicit `max_drift` to raise -- or lower -- it. + if max_drift is not None: + self.max_drift = max_drift + else: + self.max_drift = min( + interval_s * MAX_DRIFT_PERCENT, MAX_DRIFT_CEILING + ) if interval_s > self.max_drift_correction: self.min_interval_s = interval_s - self.max_drift_correction diff --git a/tests/functional/test_timers.py b/tests/functional/test_timers.py index f4f7772..3969a5f 100644 --- a/tests/functional/test_timers.py +++ b/tests/functional/test_timers.py @@ -21,6 +21,70 @@ async def test_Timer_real_run(): i += 1 +def test_Timer_max_drift__defaults_to_heuristic_when_omitted(): + timer = Timer(1.0, name="test") + assert timer.max_drift == pytest.approx(0.30) # min(1.0 * 0.30, 1.2) + + +def test_Timer_max_drift__uses_explicit_value_when_given(): + timer = Timer(1.0, name="test", max_drift=5.0) + assert timer.max_drift == 5.0 + + +@pytest.mark.asyncio +async def test_Timer_max_drift__can_be_changed_at_runtime(): + # `max_drift` is read fresh from the instance on every `tick()`, so it + # can be retuned live -- after the timer has already started iterating, + # not only at construction. This drives the same physical ~0.5s drift on + # three consecutive iterations and changes *only* `timer.max_drift` + # between them: the identical drift is silent, then warns, then is + # silent again, proving the new threshold takes effect immediately. + clock = Mock() + clock.side_effect = [ + 9.0, # __init__ epoch + 10.0, + 11.5, # iter 1 (first tick, no drift check): slept 1.5s + 12.0, + 13.5, # iter 2: drift -0.5s + 14.0, + 15.5, # iter 3: drift -0.5s + 16.0, + 17.5, # iter 4: drift -0.5s + ] + sleep = AsyncMock() + timer = Timer(1.0, name="test", clock=clock, sleep=sleep, max_drift=1.0) + it = timer.__aiter__() + + with patch("mode.timers.logger") as logger: + await it.__anext__() # iter 1: first tick, establishes timing + await it.__anext__() # iter 2: 0.5s drift < 1.0 threshold -> silent + logger.info.assert_not_called() + assert timer.drifting == 0 + + # Lower the threshold live; the next identical drift must now warn. + timer.max_drift = 0.1 + await it.__anext__() # iter 3: 0.5s drift >= 0.1 threshold -> warns + logger.info.assert_called_once_with( + "Timer %s woke up too late, with a drift of +%r " + "runtime=%r sleeptime=%r", + "test", + ANY, + ANY, + ANY, + ) + assert timer.drifting == 1 + assert timer.drifting_late == 1 + + # Raise it back live; the same drift falls silent again. + timer.max_drift = 1.0 + await it.__anext__() # iter 4: 0.5s drift < 1.0 threshold -> silent + logger.info.assert_called_once() # still just the one warning + assert timer.drifting == 1 + assert timer.drifting_late == 1 + + await it.aclose() + + class Interval(NamedTuple): interval: float wakeup_time: float @@ -229,6 +293,98 @@ class test_Timer_30s_five_second_skew(test_Timer): skew = 5.0 +class test_Timer_explicit_max_drift_suppresses_warning(test_Timer): + # Same 1s interval / 0.3s skew as the base class, which normally logs + # a warning (0.3s drift >= the default ~0.3 threshold for a 1s + # interval). An explicit `max_drift` higher than the induced drift + # must silence it. + + @pytest.fixture + def timer(self, *, clock, sleep) -> Timer: + return Timer( + self.interval, + name="test", + clock=clock, + sleep=sleep, + max_drift=self.skew + 0.1, + ) + + @pytest.mark.asyncio + async def test_too_early(self, *, clock, timer, first_interval): + interval = self.interval + skew = self.skew + intervals = [ + first_interval, + (None, None), + (None, None), + (interval - skew, None), + (None, interval + skew), + (None, None), + ] + async with self.assert_timer(timer, clock, intervals) as logger: + logger.info.assert_not_called() + assert not timer.drifting + assert not timer.drifting_early + + @pytest.mark.asyncio + async def test_too_late(self, *, clock, timer, first_interval): + interval = self.interval + skew = self.skew + intervals = [ + first_interval, + (None, None), + (None, None), + (interval + skew, None), + (None, interval + skew), + (None, None), + ] + async with self.assert_timer(timer, clock, intervals) as logger: + logger.info.assert_not_called() + assert not timer.drifting + assert not timer.drifting_late + + +class test_Timer_explicit_max_drift_lowers_threshold(test_Timer): + # A small skew that would normally stay silent (well under the + # default ~0.3 threshold for a 1s interval) must warn once `max_drift` + # is explicitly set below it. + skew = 0.05 + + @pytest.fixture + def timer(self, *, clock, sleep) -> Timer: + return Timer( + self.interval, + name="test", + clock=clock, + sleep=sleep, + max_drift=0.01, + ) + + @pytest.mark.asyncio + async def test_too_late(self, *, clock, timer, first_interval): + interval = self.interval + skew = self.skew + intervals = [ + first_interval, + (None, None), + (None, None), + (interval + skew, None), + (None, interval + skew), + (None, None), + ] + async with self.assert_timer(timer, clock, intervals) as logger: + logger.info.assert_called_once_with( + "Timer %s woke up too late, with a drift " + "of +%r runtime=%r sleeptime=%r", + "test", + ANY, + ANY, + ANY, + ) + assert timer.drifting == 1 + assert timer.drifting_late == 1 + + class test_Timer_30s_five_second_skew_late_epoch(test_Timer): epoch = 300000.0 interval = 30.0 diff --git a/tests/unit/test_threads.py b/tests/unit/test_threads.py index caf385d..659c43e 100644 --- a/tests/unit/test_threads.py +++ b/tests/unit/test_threads.py @@ -196,6 +196,23 @@ async def timer(interval, **kwargs): await thread._thread_keepalive(thread) + @pytest.mark.asyncio + async def test__thread_keepalive__forwards_configured_max_drift( + self, *, thread + ): + thread.keepalive_max_drift = 5.0 + calls = [] + + async def timer(interval, **kwargs): + calls.append(kwargs) + yield interval + + thread.itertimer = timer + + await thread._thread_keepalive(thread) + + assert calls[0]["max_drift"] == 5.0 + @pytest.mark.asyncio async def test_serve(self, *, thread): self.mock_for_serve(thread)