From 0420830be6d0d84145826cc6121653e7297ec7cb Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Fri, 18 Sep 2026 10:20:34 +0000 Subject: [PATCH 1/2] Add a dedup history preserving `catch_warnings` CPython deduplicates warnings (the "default", "module" and "once" actions) through per-module `__warningregistry__` dictionaries, and invalidates all of them whenever the filters change, including on every enter and exit of a `warnings.catch_warnings` block, even when the filters are restored to exactly what they were. So the documented way of suppressing a warning locally (`with catch_warnings(): simplefilter(...)`) makes every warning in the program repeat on every call, including warnings emitted by unrelated callers. This is python/cpython#73858, open since 2017, and it affects the C implementation on 3.11 to 3.15, with and without `-X context_aware_warnings`. The new `frequenz.core.warnings.catch_warnings` is a drop-in replacement that snapshots the registries of all loaded modules on entry and brings them back to a valid state on exit. The catch is that a registry is only valid when stamped with the current filters version, an internal counter that is not exposed, so it is read by emitting a warning of a private category into a private registry, from inside the block, where the "ignore" filter the probe needs is discarded on exit. The enter and exit bumps are then arithmetic, and are verified once at import time: if the interpreter doesn't bump the counter as expected the class degrades to a plain `warnings.catch_warnings`. Correctness takes precedence over deduplication: a history that was already invalid on entry (a filter changed outside any block) stays invalid, history recorded inside the block is only kept when the block added nothing but "ignore" filters through the `action` argument (an "ignore" can only remove warnings, so what was shown is still valid outside), and nothing is restored if the filters after the block differ from the ones before it, which can happen with context-aware warnings when another thread changes them. Nested blocks report their bumps to the enclosing one so they compose. Every race considered degrades to the stdlib behaviour of repeating a warning, never to suppressing one. The workaround suggested in the CPython issue (saving and restoring the current module's own registry and reading the version through a leaked "ignore" filter) was not adopted: it only covers the module defining the block, and it leaves a filter behind on every exit. The cost is a scan of `sys.modules` on entry and exit (tens of microseconds with a couple of hundred modules), documented in the class. Signed-off-by: Leandro Lucarella --- RELEASE_NOTES.md | 2 +- src/frequenz/core/warnings.py | 479 ++++++++++++++++++++++++++++++++++ tests/test_warnings.py | 408 +++++++++++++++++++++++++++++ 3 files changed, 888 insertions(+), 1 deletion(-) create mode 100644 src/frequenz/core/warnings.py create mode 100644 tests/test_warnings.py diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 67f33e6..f1254fb 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -10,7 +10,7 @@ ## New Features - +- A new `frequenz.core.warnings` module with a `catch_warnings` drop-in replacement for `warnings.catch_warnings` that preserves the warnings deduplication history, so warnings already shown before a block are not shown again after it. This works around [python/cpython#73858](https://github.com/python/cpython/issues/73858), where merely entering and leaving a `catch_warnings` block makes every warning in the program repeat. It relies on CPython internals (verified on 3.11 to 3.15) and falls back to the standard behaviour on interpreters that don't behave as expected, see the module documentation for the details and limitations. ## Bug Fixes diff --git a/src/frequenz/core/warnings.py b/src/frequenz/core/warnings.py new file mode 100644 index 0000000..2047fbc --- /dev/null +++ b/src/frequenz/core/warnings.py @@ -0,0 +1,479 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Warnings utilities. + +This module provides [`catch_warnings`][.catch_warnings], a drop-in replacement for +[`warnings.catch_warnings`][] that works around a long-standing CPython issue +([python/cpython#73858](https://github.com/python/cpython/issues/73858)) where merely +entering and leaving a `catch_warnings` block makes every warning in the program be +shown again. + +The standard library deduplicates warnings (the `"default"`, `"module"` and `"once"` +filter actions) using a per-module `__warningregistry__` dictionary, and invalidates +every registry whenever the filters change, including on every enter and exit of a +[`warnings.catch_warnings`][] block, even if the filters are restored to exactly what +they were. So code like this, which is the documented way of suppressing a warning +locally: + +```python +import warnings + + +def legacy_convert(value: str) -> int: + warnings.warn("legacy_convert() is deprecated", DeprecationWarning) + return int(value) + + +def convert(value: str) -> int: + with warnings.catch_warnings(action="ignore", category=DeprecationWarning): + return legacy_convert(value) +``` + +makes any warning emitted by the *caller* (or by any other code in the program) be +shown again on every call, regardless of the filters in place. See +[python/cpython#73858](https://github.com/python/cpython/issues/73858) for the +details. + +[`catch_warnings`][.catch_warnings] restores the registries of all loaded modules on +exit, so the deduplication history is preserved across the block whenever it is safe to +do so. +""" + +import sys +import threading +import warnings +from types import ModuleType, TracebackType +from typing import Any, Final, Generic, Literal, TypeAlias, TypeVar, cast, overload + +__all__ = ["Action", "catch_warnings"] + +_W_co = TypeVar("_W_co", bound="list[warnings.WarningMessage] | None", covariant=True) +"""What entering the block returns: the recorded warnings or `None`.""" + + +Action: TypeAlias = Literal[ + "default", "error", "ignore", "always", "all", "module", "once" +] +"""The possible actions for a warning filter (see [`warnings.simplefilter`][]).""" + +_VERSION_KEY: Final[str] = "version" +"""The key used by CPython to store the filters version in warning registries.""" + +_Registry: TypeAlias = dict[Any, Any] +"""A module `__warningregistry__` dictionary.""" + + +class _RegistryProbeWarning(Warning): + """Private warning category used only to read the internal filters version.""" + + +def _registries() -> list[_Registry]: + """Collect the warning registries of all loaded modules. + + Returns: + The `__warningregistry__` dictionaries found in `sys.modules`. + """ + registries: list[_Registry] = [] + for module in list(sys.modules.values()): + namespace = getattr(module, "__dict__", None) + if not isinstance(namespace, dict): + continue + registry = namespace.get("__warningregistry__") + if isinstance(registry, dict): + registries.append(registry) + return registries + + +def _current_filters(module: Any) -> list[Any]: + """Return a copy of the warning filters in effect. + + Args: + module: The `warnings` module in use. + + Returns: + The filters in effect, honouring context-aware warnings when available. + """ + get_filters = getattr(module, "_get_filters", None) + return list(get_filters() if get_filters is not None else module.filters) + + +def _probe_version(module: Any) -> int | None: + """Read the internal filters version. + + This adds an `"ignore"` filter for a private category and emits a warning of that + category with a private registry: CPython stamps the current filters version into + the registry before consulting the filters. It must be called inside a + [`warnings.catch_warnings`][] block, so the added filter is discarded on exit. + + Args: + module: The `warnings` module in use. + + Returns: + The filters version after adding the filter, or `None` if the interpreter + doesn't behave like CPython. + """ + registry: _Registry = {} + module.simplefilter("ignore", _RegistryProbeWarning) + module.warn_explicit( + "filters version probe", + _RegistryProbeWarning, + __file__, + 0, + module=__name__, + registry=registry, + ) + version = registry.get(_VERSION_KEY) + return version if isinstance(version, int) else None + + +def _check_bump_counts() -> bool: + """Check that the interpreter bumps the filters version as this module expects. + + The workaround assumes that adding a filter, entering a + [`warnings.catch_warnings`][] block and leaving one each bump the version exactly + once. This measures them with nested blocks. + + Returns: + Whether the interpreter behaves as expected. + """ + with warnings.catch_warnings(): + first = _probe_version(warnings) + second = _probe_version(warnings) + with warnings.catch_warnings(): + inner = _probe_version(warnings) + after = _probe_version(warnings) + if first is None or second is None or inner is None or after is None: + return False + filter_bump = second - first + enter_bump = inner - second - filter_bump + exit_bump = after - inner - filter_bump + return (filter_bump, enter_bump, exit_bump) == (1, 1, 1) + + +_SUPPORTED: Final[bool] = _check_bump_counts() +"""Whether the registry preservation is active for this interpreter.""" + +_local = threading.local() +"""Per-thread stack of the active `catch_warnings` blocks, for nesting.""" + + +def _stack() -> list["catch_warnings[Any]"]: + """Get the per-thread stack of active blocks. + + Returns: + The stack, innermost block last. + """ + stack: list[catch_warnings[Any]] | None = getattr(_local, "stack", None) + if stack is None: + stack = _local.stack = [] + return stack + + +class catch_warnings( # pylint: disable=invalid-name + warnings.catch_warnings, Generic[_W_co] +): + """A `catch_warnings` block that preserves the warnings deduplication history. + + This is a drop-in replacement for [`warnings.catch_warnings`][] that takes the + same arguments and behaves the same way, except that on exit it also brings the + per-module `__warningregistry__` dictionaries back to a valid state (with some + **extra runtime costs**), so warnings that were already shown before the block are + not shown again just because a block was entered and left. See the module + documentation for the background. + + The internals are probed once when this module is imported, and if they don't + behave as expected the extra work is skipped, so on an unexpected interpreter the + class degrades to a plain [`warnings.catch_warnings`][], it never suppresses a + warning the standard library would show. + + What is preserved: + + * The history of every module loaded at the time the block is entered, so warnings + emitted before the block (by any module, not just the one using the block) are + not repeated after it. + * Warnings emitted *inside* the block, as long as the filters were only changed + through the `action="ignore"` argument or not at all. With any other change to + the filters inside the block (like a `simplefilter()` call, even one adding an + `"ignore"` filter) the history inside the block is discarded on exit, as the + standard library does, because the warnings could have been shown under filters + that no longer apply. + * Nested blocks compose: an inner block that could preserve its history doesn't + prevent the outer one from preserving its own. + + Tip: Use the `action` argument, not `simplefilter()` + To ignore warnings inside a block, pass `action="ignore"` (and + `category`) instead of calling [`warnings.simplefilter`][] inside it. + Both fix the repetition of warnings shown *before* the block, but only + the `action` form also keeps deduplicating warnings shown *inside* it + (like an unrelated warning emitted by the wrapped code), because it is + the only case where the block can be sure the filters were only + narrowed. + + What is not, on purpose: + + * If the filters in effect after the block differ from the ones in effect when it + was entered (for example, another thread changed them, which can persist with + `-X context_aware_warnings`), nothing is restored, as the history may no longer + be valid. + * If the filters were changed *outside* any block since a module last emitted a + warning, that module's history was already invalid when the block was entered + and it stays discarded, so a `"error"` or `"always"` filter set in the meantime + is honoured. + * Registries not reachable through `sys.modules` (like the globals of code run + with `exec()`) and warnings emitted with an explicit `registry` argument are not + handled. + + Warning: Cost + Entering and leaving a block iterates over `sys.modules` (twice per block when + the history can be preserved) and copies every warning registry found, which + takes tens of microseconds with a couple of hundred modules loaded and scales + linearly with them, compared to well under a microsecond for the standard + library version. This is fine for occasional use, but it is not free in tight + loops. + + Warning: Concurrency + The restoration is not atomic with respect to other threads emitting warnings or + changing filters, and there is no locking against them. Every race considered + degrades to the standard library behaviour (a warning that was already shown is + shown once more), never to suppressing a warning that should be shown, but this + was only verified with the GIL, not on free-threaded builds. + + Example: + ```python + import warnings + + from frequenz.core.warnings import catch_warnings + + + def legacy_convert(value: str) -> int: + warnings.warn("legacy_convert() is deprecated", DeprecationWarning) + return int(value) + + + def convert(value: str) -> int: + with catch_warnings(action="ignore", category=DeprecationWarning): + return legacy_convert(value) + + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("default") + for _ in range(10): + warnings.warn("shown only once", UserWarning) + convert("1") + + assert len(caught) == 1 + ``` + + Warning: Relies on CPython implementation details + This works by inspecting the undocumented `__warningregistry__` dictionaries and + the internal *filters version* counter used by CPython's `warnings` module (both + the C and the pure Python implementations). The behaviour was verified on CPython + 3.11 to 3.15, with and without `-X context_aware_warnings`. + """ + + _module: ModuleType + """The `warnings` module in use (set by the base class).""" + + @overload + def __init__( # noqa: D107 # pylint: disable=too-many-arguments + self: "catch_warnings[None]", + *, + record: Literal[False] = False, + module: ModuleType | None = None, + action: Action | None = None, + category: type[Warning] | tuple[type[Warning], ...] = Warning, + lineno: int = 0, + append: bool = False, + ) -> None: ... + + @overload + def __init__( # noqa: D107 # pylint: disable=too-many-arguments + self: "catch_warnings[list[warnings.WarningMessage]]", + *, + record: Literal[True], + module: ModuleType | None = None, + action: Action | None = None, + category: type[Warning] | tuple[type[Warning], ...] = Warning, + lineno: int = 0, + append: bool = False, + ) -> None: ... + + @overload + def __init__( # noqa: D107 # pylint: disable=too-many-arguments + self, + *, + record: bool, + module: ModuleType | None = None, + action: Action | None = None, + category: type[Warning] | tuple[type[Warning], ...] = Warning, + lineno: int = 0, + append: bool = False, + ) -> None: ... + + def __init__( # pylint: disable=too-many-arguments + self, + *, + record: bool = False, + module: ModuleType | None = None, + action: Action | None = None, + category: type[Warning] | tuple[type[Warning], ...] = Warning, + lineno: int = 0, + append: bool = False, + ) -> None: + """Initialize this instance. + + The arguments are the same as for [`warnings.catch_warnings`][]. + + Args: + record: Whether to record warnings instead of showing them. + module: An alternative `warnings` module (only for testing `warnings`). + action: If given, a filter with this action is added on entering, as if + [`warnings.simplefilter`][] was called with the remaining arguments. + category: The warning category for `action`. + lineno: The line number for `action`. + append: Whether the filter for `action` is appended instead of inserted. + """ + super().__init__(record=record, module=module) + self._own_filter: tuple[Any, ...] | None = ( + None if action is None else (action, category, lineno, append) + ) + self._snapshots: dict[int, tuple[_Registry, _Registry, bool]] = {} + self._entry_filters: list[Any] = [] + self._enter_version: int | None = None + self._harmless_bumps: int = 0 + + def _only_ignores(self) -> bool: + """Tell whether this block itself only adds `"ignore"` filters. + + Returns: + Whether the history recorded inside the block is valid outside it. + """ + return self._own_filter is None or self._own_filter[0] == "ignore" + + def __enter__(self) -> _W_co: + """Enter the block. + + Returns: + The list of recorded warnings if `record` is true, `None` otherwise. + """ + if not _SUPPORTED: + log = cast(_W_co, super().__enter__()) + if self._own_filter is not None: + self._module.simplefilter(*self._own_filter) + return log + + snapshots = [(registry, registry.copy()) for registry in _registries()] + self._entry_filters = _current_filters(self._module) + log = cast(_W_co, super().__enter__()) # Bumps the version once. + version = _probe_version(self._module) # And this once more. + self._enter_version = version + self._harmless_bumps = 0 + if version is not None: + # Only registries stamped with the version in effect before entering + # hold a valid history; older ones were already invalidated by a filter + # change and must stay that way. + entry_version = version - 2 + self._snapshots = { + id(registry): ( + registry, + saved, + saved.get(_VERSION_KEY) == entry_version, + ) + for registry, saved in snapshots + } + if self._own_filter is not None: + self._module.simplefilter(*self._own_filter) + if version is not None and self._only_ignores(): + # Only "ignore" filters were added, so what was already shown is still + # valid inside the block too: keep deduplicating in here. + inner_version = version + (0 if self._own_filter is None else 1) + for registry, _, valid in self._snapshots.values(): + if valid: + registry[_VERSION_KEY] = inner_version + _stack().append(self) + return log + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> None: + """Exit the block, restoring the filters and the warning registries.""" + if not _SUPPORTED: + super().__exit__(exc_type, exc_val, exc_tb) + return + + stack = _stack() + if stack and stack[-1] is self: + stack.pop() + try: + version = _probe_version(self._module) # Bumps the version once. + finally: + super().__exit__(exc_type, exc_val, exc_tb) # And this once more. + snapshots, self._snapshots = self._snapshots, {} + if version is None or self._enter_version is None: + return + if _current_filters(self._module) != self._entry_filters: + # The filters changed for good while we were inside (possible with + # context-aware warnings and threads): the history may not be valid. + return + + final_version = version + 1 + own_bumps = 0 if self._own_filter is None else 1 + unexplained_bumps = ( + version - self._enter_version - own_bumps - 1 - self._harmless_bumps + ) + if unexplained_bumps == 0 and self._only_ignores(): + self._keep_history(snapshots, final_version) + if stack: + stack[-1]._harmless_bumps += final_version - (self._enter_version - 2) + else: + self._restore_history(snapshots, final_version) + + @staticmethod + def _keep_history( + snapshots: dict[int, tuple[_Registry, _Registry, bool]], version: int + ) -> None: + """Keep the history recorded inside the block and re-validate everything. + + Everything recorded inside is valid under the restored filters: keep it, add + back what was valid at entry, and stamp every registry with the new version. + + Args: + snapshots: The registries and their copies taken on entry. + version: The filters version in effect after leaving the block. + """ + for registry in _registries(): + snapshot = snapshots.get(id(registry)) + if snapshot is None: + registry[_VERSION_KEY] = version + continue + _, saved, valid = snapshot + if valid: + registry.update(saved) + registry[_VERSION_KEY] = version + elif registry.get(_VERSION_KEY) != saved.get(_VERSION_KEY): + # Stale at entry but cleared and refilled inside. + registry[_VERSION_KEY] = version + + @staticmethod + def _restore_history( + snapshots: dict[int, tuple[_Registry, _Registry, bool]], version: int + ) -> None: + """Bring the registries that were valid at entry back to that state. + + The filters changed inside the block in a way we can't vouch for, so whatever + was recorded inside is dropped. + + Args: + snapshots: The registries and their copies taken on entry. + version: The filters version in effect after leaving the block. + """ + for registry, saved, valid in snapshots.values(): + if not valid: + continue + registry.clear() + registry.update(saved) + registry[_VERSION_KEY] = version diff --git a/tests/test_warnings.py b/tests/test_warnings.py new file mode 100644 index 0000000..78fc88f --- /dev/null +++ b/tests/test_warnings.py @@ -0,0 +1,408 @@ +# License: MIT +# Copyright © 2026 Frequenz Energy-as-a-Service GmbH + +"""Tests for the warnings module. + +The scenarios compare the standard library's `catch_warnings` (which is expected to +repeat warnings, see https://github.com/python/cpython/issues/73858) with ours, and +check that ours never suppresses a warning the standard library would show. +""" + +import sys +import threading +import warnings +from collections.abc import Callable, Iterator +from types import FunctionType, ModuleType +from typing import Any + +import pytest + +from frequenz.core import warnings as core_warnings +from frequenz.core.warnings import Action, catch_warnings + +_MODULE_SOURCE = """ +import warnings + +# Messages include the module name so the "once" action, which deduplicates by +# message and category only, doesn't see repeated messages across tests. +DEFAULT_MESSAGE = f"warned in {__name__}" + +def warn(message=DEFAULT_MESSAGE, category=UserWarning): + warnings.warn(message, category, stacklevel=1) + +def warn_then_block(block, message=DEFAULT_MESSAGE, category=UserWarning): + warnings.warn(message, category, stacklevel=1) + with block(): + pass + +def block_then_warn(block, message=DEFAULT_MESSAGE, category=UserWarning): + with block(): + pass + warnings.warn(message, category, stacklevel=1) +""" + +Block = Callable[[], Any] + + +@pytest.fixture(name="make_module") +def make_module_fixture() -> Iterator[Callable[[str], ModuleType]]: + """Create fresh modules, each with its own (initially missing) registry.""" + created: list[str] = [] + + def make(name: str) -> ModuleType: + module = ModuleType(name) + module.__file__ = f"<{name}>" + # pylint: disable-next=exec-used + exec(compile(_MODULE_SOURCE, f"<{name}>", "exec"), module.__dict__) + sys.modules[name] = module + created.append(name) + return module + + yield make + for name in created: + sys.modules.pop(name, None) + + +def _run(action: Action, function: Callable[[], Any], times: int = 10) -> int: + """Call `function` repeatedly under `action` and count the warnings shown.""" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter(action) + for _ in range(times): + function() + return len(caught) + + +def test_interpreter_is_supported() -> None: + """Test that the CPython internals behave as the workaround expects.""" + assert sys.implementation.name == "cpython" + assert core_warnings._SUPPORTED # pylint: disable=protected-access + assert core_warnings._check_bump_counts() # pylint: disable=protected-access + + +@pytest.mark.parametrize("action", ["default", "module", "once"]) +def test_stdlib_repeats_warnings( + make_module: Callable[[str], ModuleType], action: Action +) -> None: + """Test the standard library behaviour we are working around (control).""" + if action == "once" and isinstance(warnings.warn_explicit, FunctionType): + pytest.skip("The pure Python implementation doesn't repeat 'once' warnings") + module = make_module("control") + assert _run(action, lambda: module.warn_then_block(warnings.catch_warnings)) == 10 + + +@pytest.mark.parametrize("action", ["default", "module", "once"]) +def test_warning_before_block_is_shown_once( + make_module: Callable[[str], ModuleType], action: Action +) -> None: + """Test that a block after a warning doesn't make it repeat.""" + module = make_module("same_site") + assert _run(action, lambda: module.warn_then_block(catch_warnings)) == 1 + + +def test_warning_after_block_is_shown_once( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that a block before a warning doesn't make it repeat.""" + module = make_module("block_first") + assert _run("default", lambda: module.block_then_warn(catch_warnings)) == 1 + + +def test_other_module_history_is_preserved( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that a block in one module preserves the history of another one.""" + warner = make_module("warner") + blocker = make_module("blocker") + + def step() -> None: + warner.warn("from warner") + blocker.block_then_warn(catch_warnings, "from blocker") + + assert _run("default", step) == 2 + + +@pytest.mark.parametrize("action, expected", [("always", 10), ("ignore", 0)]) +def test_always_and_ignore_are_unchanged( + make_module: Callable[[str], ModuleType], action: Action, expected: int +) -> None: + """Test that actions not using the registry behave as usual.""" + module = make_module(f"unchanged_{action}") + assert _run(action, lambda: module.warn_then_block(catch_warnings)) == expected + + +def test_error_action_keeps_raising( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that a warning turned into an error keeps raising after blocks.""" + module = make_module("error") + with warnings.catch_warnings(): + warnings.simplefilter("error") + for _ in range(3): + with pytest.raises(UserWarning): + module.warn_then_block(catch_warnings) + + +@pytest.mark.parametrize("action", ["error", "always"]) +def test_filter_change_outside_block_is_honoured( + make_module: Callable[[str], ModuleType], action: Action +) -> None: + """Test that a history invalidated before the block is not revived. + + A warning is shown, then the filters are changed outside any block (the registry + is stale but not yet cleared), then a block is entered and left: the new filter + must apply, the block must not re-validate the stale history. + """ + module = make_module(f"stale_{action}") + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("default") + module.warn() + warnings.simplefilter(action) + with catch_warnings(): + pass + if action == "error": + with pytest.raises(UserWarning): + module.warn() + else: + module.warn() + module.warn() + assert len(caught) == 3 + + +@pytest.mark.parametrize("how", ["simplefilter", "action"]) +def test_history_recorded_under_other_filters_is_dropped( + make_module: Callable[[str], ModuleType], how: str +) -> None: + """Test that a warning shown inside under "default" still raises outside.""" + module = make_module(f"inner_{how}") + with warnings.catch_warnings(): + warnings.simplefilter("error") + if how == "simplefilter": + with catch_warnings(record=True) as inner: + warnings.simplefilter("default") + module.warn() + else: + with catch_warnings(record=True, action="default") as inner: + module.warn() + assert len(inner) == 1 + with pytest.raises(UserWarning): + module.warn() + + +def test_ignore_by_simplefilter_inside_block( + make_module: Callable[[str], ModuleType], +) -> None: + """Test the documented suppression pattern, with the filter added by hand.""" + module = make_module("simplefilter_ignore") + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("default") + for _ in range(5): + with catch_warnings(): + warnings.simplefilter("ignore") + module.warn() + module.warn() + assert len(caught) == 1 + + +def test_ignore_by_action_inside_block( + make_module: Callable[[str], ModuleType], +) -> None: + """Test the documented suppression pattern, with the `action` argument.""" + module = make_module("action_ignore") + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("default") + for _ in range(5): + with catch_warnings(record=True, action="ignore") as inner: + module.warn("inside") + module.warn("outside") + assert not inner + assert len(caught) == 1 + + +def test_unrelated_warning_inside_ignore_block_is_shown_once( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that history recorded inside a block adding only ignores is kept.""" + module = make_module("unrelated") + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("default") + for _ in range(5): + with catch_warnings(action="ignore", category=DeprecationWarning): + module.warn("deprecated", DeprecationWarning) + module.warn("unrelated", RuntimeWarning) + module.warn("outside") + assert [str(w.message) for w in caught] == ["unrelated", "outside"] + + +def test_nested_blocks_compose(make_module: Callable[[str], ModuleType]) -> None: + """Test that an inner block preserving its history lets the outer do it too.""" + module = make_module("nested") + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("default") + for _ in range(5): + with catch_warnings(): + with catch_warnings(action="ignore", category=DeprecationWarning): + module.warn("deprecated", DeprecationWarning) + module.warn("middle") + module.warn("outside") + assert [str(w.message) for w in caught] == ["middle", "outside"] + + +def test_nested_block_changing_filters_drops_inner_history_only( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that an inner block changing filters by hand behaves like stdlib. + + The outer block can't tell what "middle" was shown under, so it repeats like it + does with the standard library, but the history from before the outer block is + still preserved. + """ + module = make_module("nested_manual") + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("default") + for _ in range(5): + with catch_warnings(): + with catch_warnings(): + warnings.simplefilter("ignore") + module.warn("ignored") + module.warn("middle") + module.warn("outside") + messages = [str(w.message) for w in caught] + assert messages.count("middle") == 5 + assert messages.count("outside") == 1 + assert messages.count("ignored") == 0 + + +def test_exception_inside_block(make_module: Callable[[str], ModuleType]) -> None: + """Test that exceptions propagate and the history is still preserved.""" + module = make_module("exception") + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("default") + for _ in range(5): + with pytest.raises(KeyError): + with catch_warnings(): + raise KeyError("boom") + module.warn() + assert len(caught) == 1 + + +def test_module_first_warning_inside_block( + make_module: Callable[[str], ModuleType], +) -> None: + """Test a module whose registry is created inside a block.""" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("default") + for i in range(5): + with catch_warnings(): + if i == 0: + module = make_module("late") + module.warn("inside") + module.warn("outside") + assert [str(w.message) for w in caught] == ["inside", "outside"] + + +def test_categories_and_messages_are_independent( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that distinct warnings are each shown once.""" + module = make_module("categories") + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("default") + for _ in range(3): + with catch_warnings(): + pass + module.warn("a", DeprecationWarning) + module.warn("a", RuntimeWarning) + module.warn("b", RuntimeWarning) + assert len(caught) == 3 + + +def test_filters_are_restored_without_leftovers() -> None: + """Test that the probe filter doesn't leak out of the block.""" + with warnings.catch_warnings(): + warnings.simplefilter("default") + before = list(warnings.filters) + with catch_warnings(): + warnings.simplefilter("ignore") + assert list(warnings.filters) == before + assert not any("Probe" in repr(item) for item in before) + + +def test_record_returns_the_log() -> None: + """Test that the `record` argument works as in the standard library.""" + with catch_warnings(record=True) as caught: + warnings.simplefilter("always") + warnings.warn("recorded", UserWarning) + assert [str(w.message) for w in caught] == ["recorded"] + with catch_warnings() as nothing: + pass + assert nothing is None + + +def test_filter_changed_by_another_thread_is_honoured( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that a filter change that persists after the block is not undone. + + Without context-aware warnings the standard library discards the other thread's + change on exit, so the history is valid and the warning is deduplicated. With + them the change persists, so the history must not be restored and the warning + must raise. Both outcomes are checked against the standard library's own. + """ + module = make_module("threads") + outcomes: list[tuple[str, bool]] = [] + for block in (warnings.catch_warnings, catch_warnings): + module.__dict__.pop("__warningregistry__", None) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("default") + module.warn() + with block(): + thread = threading.Thread(target=lambda: warnings.simplefilter("error")) + thread.start() + thread.join() + try: + module.warn() + outcomes.append(("shown" if len(caught) > 1 else "deduped", False)) + except UserWarning: + outcomes.append(("raised", True)) + assert len(outcomes) == 2 + stdlib, ours = outcomes[0], outcomes[1] + assert ours[1] == stdlib[1], "we must raise exactly when the stdlib raises" + if not stdlib[1]: + assert ours[0] == "deduped" + + +def test_unsupported_interpreter_falls_back( + monkeypatch: pytest.MonkeyPatch, make_module: Callable[[str], ModuleType] +) -> None: + """Test that without support the class is a plain `catch_warnings`.""" + monkeypatch.setattr(core_warnings, "_SUPPORTED", False) + module = make_module("fallback") + assert _run("default", lambda: module.warn_then_block(catch_warnings)) == 10 + with catch_warnings(record=True, action="ignore") as caught: + module.warn() + assert not caught + + +def test_concurrent_blocks_do_not_crash( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that overlapping blocks in several threads don't raise.""" + module = make_module("concurrent") + errors: list[BaseException] = [] + + def worker() -> None: + try: + for _ in range(200): + with catch_warnings(action="ignore"): + module.warn() + except BaseException as error: # pylint: disable=broad-exception-caught + errors.append(error) + + with warnings.catch_warnings(): + warnings.simplefilter("ignore") + threads = [threading.Thread(target=worker) for _ in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert not errors From 27adcf796adeb7e0199d74a26191e9f284149545 Mon Sep 17 00:00:00 2001 From: Leandro Lucarella Date: Fri, 18 Sep 2026 11:40:24 +0000 Subject: [PATCH 2/2] Add a fast path for `action="ignore"` MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Field testing with the dispatch client showed the registry repair costing ~130 µs per block with ~370 modules loaded, which turned a 2.8 s test suite into a 12 s one: the converters silencing a deprecation are called per message, so scanning `sys.modules` on every call is not acceptable there, and it is the whole reason this class exists. The registries only go stale because the filters version moves, and it only moves through `simplefilter()`, `filterwarnings()` and `catch_warnings`; the lookup itself reads the live filters list on every warning (`warnings.filters`, or the context's `_filters` with context-aware warnings, both through `_get_filters()` on 3.14+). So when the block only adds an `"ignore"` filter and doesn't record, insert the filter tuple into that list directly and take it out again by identity on exit, restoring `showwarning` as the standard library does. Nothing ever becomes stale, so there is nothing to scan or repair, and the block costs about twice the standard one instead of fifty times. This is only sound for `"ignore"`: the registry is consulted before the filters, and an ignore can only remove warnings, so the history recorded on either side of the block is valid on the other. If the filters were changed by hand inside such a block they are restored on exit and the version bumped, like the standard library would (`simplefilter()` even replaces our equal entry with its own, so the comparison is against a copy taken on entry, not by looking for our tuple). Every other use keeps going through the standard library plus the registry repair, and the cost note says which is which. Signed-off-by: Leandro Lucarella --- src/frequenz/core/warnings.py | 108 ++++++++++++++++++++++++++++++---- tests/test_warnings.py | 95 ++++++++++++++++++++++++++++++ 2 files changed, 193 insertions(+), 10 deletions(-) diff --git a/src/frequenz/core/warnings.py b/src/frequenz/core/warnings.py index 2047fbc..372a09e 100644 --- a/src/frequenz/core/warnings.py +++ b/src/frequenz/core/warnings.py @@ -85,6 +85,35 @@ def _registries() -> list[_Registry]: return registries +def _live_filters(module: Any) -> list[Any] | None: + """Return the list of warning filters in effect, the live object. + + Args: + module: The `warnings` module in use. + + Returns: + The list CPython consults on each warning, honouring context-aware warnings + when available, or `None` if it can't be found. + """ + get_filters = getattr(module, "_get_filters", None) + filters = get_filters() if get_filters is not None else module.filters + return filters if isinstance(filters, list) else None + + +def _mutated(module: Any) -> None: + """Bump the internal filters version. + + Args: + module: The `warnings` module in use. + """ + mutated = getattr(module, "_filters_mutated", None) + if mutated is not None: + mutated() + return + with module.catch_warnings(): # Entering and leaving bumps it. + pass + + def _current_filters(module: Any) -> list[Any]: """Return a copy of the warning filters in effect. @@ -94,8 +123,7 @@ def _current_filters(module: Any) -> list[Any]: Returns: The filters in effect, honouring context-aware warnings when available. """ - get_filters = getattr(module, "_get_filters", None) - return list(get_filters() if get_filters is not None else module.filters) + return list(_live_filters(module) or []) def _probe_version(module: Any) -> int | None: @@ -170,7 +198,7 @@ def _stack() -> list["catch_warnings[Any]"]: return stack -class catch_warnings( # pylint: disable=invalid-name +class catch_warnings( # pylint: disable=invalid-name,too-many-instance-attributes warnings.catch_warnings, Generic[_W_co] ): """A `catch_warnings` block that preserves the warnings deduplication history. @@ -225,12 +253,18 @@ class degrades to a plain [`warnings.catch_warnings`][], it never suppresses a handled. Warning: Cost - Entering and leaving a block iterates over `sys.modules` (twice per block when - the history can be preserved) and copies every warning registry found, which - takes tens of microseconds with a couple of hundred modules loaded and scales - linearly with them, compared to well under a microsecond for the standard - library version. This is fine for occasional use, but it is not free in tight - loops. + With `action="ignore"` and no `record`, the block doesn't go through the + standard library at all: the filter is added to the live filters list and + removed on exit, which leaves the internal filters version untouched, so no + registry ever becomes stale and there is nothing to restore. This costs about as + much as the standard library block. + + Any other use (another `action`, `record=True`, or no `action` at all) must go + through the standard library and then repair the registries, which iterates over + `sys.modules` twice per block and copies every warning registry found: tens of + microseconds with a few hundred modules loaded, linear in their number, compared + to well under a microsecond for the standard library version. Fine for + occasional use, but not free in tight loops. Warning: Concurrency The restoration is not atomic with respect to other threads emitting warnings or @@ -272,9 +306,12 @@ def convert(value: str) -> int: 3.11 to 3.15, with and without `-X context_aware_warnings`. """ - _module: ModuleType + _module: Any """The `warnings` module in use (set by the base class).""" + _record: bool + """Whether warnings are recorded (set by the base class).""" + @overload def __init__( # noqa: D107 # pylint: disable=too-many-arguments self: "catch_warnings[None]", @@ -342,6 +379,10 @@ def __init__( # pylint: disable=too-many-arguments self._entry_filters: list[Any] = [] self._enter_version: int | None = None self._harmless_bumps: int = 0 + self._fast_item: tuple[Any, ...] | None = None + self._fast_saved: list[Any] = [] + self._fast_showwarning: Any = None + self._fast_entered: bool = False def _only_ignores(self) -> bool: """Tell whether this block itself only adds `"ignore"` filters. @@ -356,6 +397,9 @@ def __enter__(self) -> _W_co: Returns: The list of recorded warnings if `record` is true, `None` otherwise. + + Raises: + RuntimeError: If this instance was already entered. """ if not _SUPPORTED: log = cast(_W_co, super().__enter__()) @@ -363,6 +407,26 @@ def __enter__(self) -> _W_co: self._module.simplefilter(*self._own_filter) return log + if self._own_filter is not None and self._own_filter[0] == "ignore": + filters = _live_filters(self._module) if not self._record else None + if filters is not None: + # Ignoring only removes warnings, so history recorded on either + # side of the block is valid on the other: add the filter without + # going through the standard library (which would bump the filters + # version and invalidate every registry) and take it out on exit. + if self._fast_entered: + raise RuntimeError(f"Cannot enter {self!r} twice") + self._fast_entered = True + _, category, lineno, append = self._own_filter + self._fast_item = ("ignore", None, category, None, lineno) + self._fast_saved = filters[:] + self._fast_showwarning = self._module.showwarning + if append: + filters.append(self._fast_item) + else: + filters.insert(0, self._fast_item) + return cast(_W_co, None) + snapshots = [(registry, registry.copy()) for registry in _registries()] self._entry_filters = _current_filters(self._module) log = cast(_W_co, super().__enter__()) # Bumps the version once. @@ -405,6 +469,10 @@ def __exit__( super().__exit__(exc_type, exc_val, exc_tb) return + if self._fast_item is not None: + self._exit_fast() + return + stack = _stack() if stack and stack[-1] is self: stack.pop() @@ -432,6 +500,26 @@ def __exit__( else: self._restore_history(snapshots, final_version) + def _exit_fast(self) -> None: + """Leave a block that only added an `"ignore"` filter to the live list.""" + item, self._fast_item = self._fast_item, None + filters = _live_filters(self._module) + if filters is not None: + # By identity: somebody may have inserted an equal filter meanwhile. + for index, existing in enumerate(filters): + if existing is item: + del filters[index] + break + if filters != self._fast_saved: + # The filters were changed by hand inside the block: restore them + # as the standard library would, bumping the version as it does, + # since the registries were filled under other filters. + filters[:] = self._fast_saved + _mutated(self._module) + self._fast_saved = [] + self._module.showwarning = self._fast_showwarning + self._fast_showwarning = None + @staticmethod def _keep_history( snapshots: dict[int, tuple[_Registry, _Registry, bool]], version: int diff --git a/tests/test_warnings.py b/tests/test_warnings.py index 78fc88f..2e5b71c 100644 --- a/tests/test_warnings.py +++ b/tests/test_warnings.py @@ -406,3 +406,98 @@ def worker() -> None: for thread in threads: thread.join() assert not errors + + +def _live_filters() -> list[Any]: + """Get the live filters list (not `warnings.filters` in context-aware mode).""" + filters = core_warnings._live_filters(warnings) # pylint: disable=W0212 + assert filters is not None + return filters + + +def test_ignore_fast_path_does_not_bump_the_version( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that an `action="ignore"` block leaves the filters version alone.""" + module = make_module("fast") + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("default") + module.warn("before") + registry = module.__dict__["__warningregistry__"] + stamped = registry["version"] + with catch_warnings(action="ignore", category=DeprecationWarning) as log: + module.warn("ignored", DeprecationWarning) + module.warn("inside") + assert log is None + assert registry["version"] == stamped + module.warn("before") + module.warn("inside") + module.warn("ignored", DeprecationWarning) + assert [str(w.message) for w in caught] == ["before", "inside", "ignored"] + + +def test_ignore_fast_path_restores_filters_changed_by_hand( + make_module: Callable[[str], ModuleType], +) -> None: + """Test that filters changed inside the block are restored, as in stdlib. + + Also covers `simplefilter()` replacing our own (equal) entry with its own. + """ + module = make_module("fast_manual") + with warnings.catch_warnings(record=True) as caught: + warnings.resetwarnings() + warnings.simplefilter("default") + before = _live_filters()[:] + with catch_warnings(action="ignore", category=DeprecationWarning): + warnings.simplefilter("ignore", DeprecationWarning) + warnings.simplefilter("always", UserWarning) + module.warn() + module.warn() + assert _live_filters() == before + module.warn("dep", DeprecationWarning) + module.warn() + module.warn() + assert [str(w.message) for w in caught] == [ + "warned in fast_manual", # "always" inside, twice + "warned in fast_manual", + "dep", # ignore restored away + "warned in fast_manual", # "default" restored, so only once + ] + + +def test_ignore_fast_path_survives_nested_stdlib_block() -> None: + """Test that a nested standard block copying the filters doesn't confuse it.""" + with warnings.catch_warnings(): + warnings.resetwarnings() + with catch_warnings(action="ignore", category=DeprecationWarning, append=True): + with warnings.catch_warnings(): + warnings.simplefilter("error") + assert len(_live_filters()) == 1 + assert not _live_filters() + + +def test_ignore_fast_path_restores_showwarning_and_rejects_reentry() -> None: + """Test the remaining `catch_warnings` semantics of the fast path.""" + original = warnings.showwarning + block = catch_warnings(action="ignore") + with block: + warnings.showwarning = lambda *args, **kwargs: None + assert warnings.showwarning is original + with pytest.raises(RuntimeError, match="twice"): + with block: + pass + + +def test_ignore_fast_path_still_ignores_and_honours_outer_error( + make_module: Callable[[str], ModuleType], +) -> None: + """Test filtering inside and outside the fast block.""" + module = make_module("fast_error") + with warnings.catch_warnings(): + warnings.simplefilter("error") + with catch_warnings(action="ignore", category=UserWarning): + module.warn() + with pytest.raises(DeprecationWarning): + module.warn("still an error", DeprecationWarning) + with pytest.raises(UserWarning): + module.warn()