Disable garbage collection for the duration of time measurement in chronometers

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

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'<Chronometer: {self.seconds}s ({status})>'
 
     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__()

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):