From 2a042a55e66ada554e5387642ddeb49b36091b77 Mon Sep 17 00:00:00 2001 From: Sergey Vasilyev Date: Mon, 5 Oct 2026 12:01:08 +0200 Subject: [PATCH] Disable garbage collection for the duration of time management in chronometers Signed-off-by: Sergey Vasilyev --- docs/tools.rst | 40 +++++++++++++++++++++++++++--- looptime/_internal/chronometers.py | 20 ++++++++++++--- tests/test_chronometers.py | 32 ++++++++++++++++++++++++ 3 files changed, 86 insertions(+), 6 deletions(-) diff --git a/docs/tools.rst b/docs/tools.rst index 88d21be..4a3e58c 100644 --- a/docs/tools.rst +++ b/docs/tools.rst @@ -11,6 +11,9 @@ to measure the duration of arbitrary code blocks in real-world time: * :class:`looptime.Chronometer` (a context manager class). * ``chronometer`` (a pytest fixture). +Pytest fixture +-------------- + It can be used as a sync or async context manager: .. code-block:: python @@ -26,6 +29,13 @@ It can be used as a sync or async context manager: await asyncio.sleep(1) assert chronometer.seconds < 0.01 # random code overhead +Asyncio event loop timing +------------------------- + +The default clock is :func:`time.perf_counter`, i.e., the real-clock time. +Other clocks can be passed as a positional argument; +it must a function with no arguments returning a ``float`` or ``int``. + Usually, the loop-time duration is not needed or can be retrieved via ``asyncio.get_running_loop().time()``. If needed, it can be measured using the provided context manager class with the event loop's clock: @@ -38,13 +48,37 @@ the provided context manager class with the event loop's clock: @pytest.mark.asyncio @pytest.mark.looptime(start=100) - async def test_me(chronometer, event_loop): - with chronometer, looptime.Chronometer(event_loop.time) as loopometer: + async def test_me(chronometer): + with chronometer, looptime.Chronometer(asyncio.get_running_loop().time) as loopometer: await asyncio.sleep(1) await asyncio.sleep(1) assert chronometer.seconds < 0.01 # random code overhead assert loopometer.seconds == 2 # precise timing, no code overhead - assert event_loop.time() == 102 + assert asyncio.get_running_loop().time() == 102 + + +Garbage collection overhead +--------------------------- + +Normally (``keep_gc=False``, the default), the chronometer disables +the automatic garbage collection for the duration of the time measurements, +in order to prevent extra time spent on the unexpected garbage collection. +With ``keep_gc=True``, the chronometer leaves the garbage collection intact. +At exit, the garbage collection is restored to the original state. +As a result, in nested chronometers, the outer one controls the gc. + +.. code-block:: python + + import gc + import looptime + + async def test_gc_disabled(): + with looptime.Chronometer() as chronometer: + assert not gc.isenabled() # expect no gc happening here + + async def test_gc_preserved(): + with looptime.Chronometer(keep_gc=True) as chronometer: + assert gc.isenabled() # expect gc possibly happening here Assertions diff --git a/looptime/_internal/chronometers.py b/looptime/_internal/chronometers.py index b2914b0..ca3ec48 100644 --- a/looptime/_internal/chronometers.py +++ b/looptime/_internal/chronometers.py @@ -1,5 +1,6 @@ from __future__ import annotations +import gc import time from collections.abc import Callable from typing import Any, Self @@ -27,11 +28,18 @@ def test_chronometer(): assert chronometer.seconds < 5.0 # 3.57s or slightly more """ - def __init__(self, clock: Callable[[], float] = time.perf_counter) -> None: + def __init__( + self, + clock: Callable[[], float] = time.perf_counter, + *, + keep_gc: bool = False, + ) -> None: super().__init__() self._clock = clock self._ts: float | None = None self._te: float | None = None + self._keep_gc = keep_gc + self.__preserved_gc: bool | None = None @property def _value(self) -> float: @@ -52,12 +60,18 @@ def __repr__(self) -> str: return f'' def __enter__(self) -> Self: - self._ts = self._clock() + if not self._keep_gc: # NB: outside of the measurement + self.__preserved_gc = gc.isenabled() + gc.disable() self._te = None + self._ts = self._clock() # NB: the innermost instruction return self def __exit__(self, *args: Any) -> None: - self._te = self._clock() + self._te = self._clock() # NB: the innermost instruction + if self.__preserved_gc: # NB: outside of the measurement + self.__preserved_gc = None + gc.enable() async def __aenter__(self) -> Self: return self.__enter__() diff --git a/tests/test_chronometers.py b/tests/test_chronometers.py index 8597a5d..bfd660b 100644 --- a/tests/test_chronometers.py +++ b/tests/test_chronometers.py @@ -1,4 +1,5 @@ import asyncio +import gc import time import pytest @@ -50,6 +51,37 @@ async def test_async_context_manager(): assert 0.1 <= chronometer.seconds < 0.11 +def test_gc_preserved_when_requested(): + gc.enable() + with looptime.Chronometer(keep_gc=True): + assert gc.isenabled() + assert gc.isenabled() + + +def test_gc_disabled_by_default(): + gc.enable() + with looptime.Chronometer(): + assert not gc.isenabled() # temporarily disabled + assert gc.isenabled() # restored + + +def test_nested_gc_remains_on_exits(): + gc.enable() + with looptime.Chronometer(): + with looptime.Chronometer(): + assert not gc.isenabled() # temporarily disabled + assert not gc.isenabled() # not restored yet + assert gc.isenabled() # restored + + +def test_gc_off_remains_unaffected(): + gc.disable() + with looptime.Chronometer(): + assert not gc.isenabled() # temporarily disabled + assert not gc.isenabled() # kept disabled, not re-enabled + gc.enable() # put it back for other tests + + @pytest.mark.asyncio @pytest.mark.looptime(start=100) async def test_readme_example(chronometer):