diff --git a/newsfragments/2378.bugfix.rst b/newsfragments/2378.bugfix.rst new file mode 100644 index 0000000000..798257a305 --- /dev/null +++ b/newsfragments/2378.bugfix.rst @@ -0,0 +1,3 @@ +Issue a :class:`ResourceWarning` for unfinished async generators finalized at +:func:`trio.run` shutdown, including generators that still have strong references. +Warnings promoted to errors are logged without interrupting generator cleanup. diff --git a/src/trio/_core/_asyncgens.py b/src/trio/_core/_asyncgens.py index fea41e0e4d..e30642aa81 100644 --- a/src/trio/_core/_asyncgens.py +++ b/src/trio/_core/_asyncgens.py @@ -176,7 +176,8 @@ async def finalize_remaining(self, runner: _run.Runner) -> None: await _core.wait_task_rescheduled( lambda _: _core.Abort.FAILED, # pragma: no cover ) - self.alive.update(self.trailing_needs_finalize) + already_warned = self.trailing_needs_finalize.copy() + self.alive.update(already_warned) self.trailing_needs_finalize.clear() # None of the still-living tasks use async generators, so @@ -215,7 +216,28 @@ async def finalize_remaining(self, runner: _run.Runner) -> None: batch = self.alive self.alive = _ASYNC_GEN_SET() for agen in batch: - await self._finalize_one(agen, name_asyncgen(agen)) + agen_name = name_asyncgen(agen) + needs_warning = agen.ag_frame is not None and agen not in already_warned + await self._finalize_one(agen, agen_name) + if needs_warning: + try: + warnings.warn( + f"Async generator {agen_name!r} was finalized at trio.run() " + "shutdown before it had been exhausted. Surround its use " + "in 'async with aclosing(...):' to ensure that it gets " + "cleaned up as soon as you're done using it.", + ResourceWarning, + stacklevel=2, + source=agen, + ) + except BaseException: + # Warning filters must not prevent other generators from + # being closed or replace the main task's exception. + ASYNCGEN_LOGGER.exception( + "Exception ignored while warning about finalization " + "of async generator %r", + agen_name, + ) def close(self) -> None: sys.set_asyncgen_hooks(*self.prev_hooks) diff --git a/src/trio/_core/_tests/test_asyncgen.py b/src/trio/_core/_tests/test_asyncgen.py index a5c27ed5fb..55dbacd69d 100644 --- a/src/trio/_core/_tests/test_asyncgen.py +++ b/src/trio/_core/_tests/test_asyncgen.py @@ -2,6 +2,7 @@ import contextlib import sys +import warnings import weakref from math import inf from types import AsyncGeneratorType @@ -14,6 +15,7 @@ if TYPE_CHECKING: from collections.abc import AsyncGenerator + from typing import Literal def test_asyncgen_basics() -> None: @@ -119,6 +121,119 @@ async def agen() -> AsyncGenerator[int, None]: assert "during finalization of async generator" in caplog.records[0].message +@pytest.mark.parametrize("warning_filter", ["always", "error"]) +@pytest.mark.parametrize("main_raises", [False, True]) +@pytest.mark.parametrize("cleanup_raises", [False, True]) +def test_asyncgen_shutdown_warning( + warning_filter: Literal["always", "error"], + main_raises: bool, + cleanup_raises: bool, + caplog: pytest.LogCaptureFixture, +) -> None: + saved = [] + closed = [] + main_error = ValueError("main failed") + cleanup_error = LookupError("cleanup failed") + + async def agen(index: int) -> AsyncGenerator[int, None]: + try: + yield index + finally: + await _core.cancel_shielded_checkpoint() + closed.append(index) + if cleanup_raises and index == 0: + raise cleanup_error + + async def async_main() -> None: + for index in range(3): + saved.append(agen(index)) + assert await saved[-1].asend(None) == index + if main_raises: + raise main_error + + with warnings.catch_warnings(record=True) as emitted: + warnings.simplefilter(warning_filter, ResourceWarning) + if main_raises: + with pytest.raises(ValueError, match="main failed") as caught: + _core.run(async_main) + assert caught.value is main_error + else: + _core.run(async_main) + + assert sorted(closed) == [0, 1, 2] + for saved_agen in saved: + assert isinstance(saved_agen, AsyncGeneratorType) + assert saved_agen.ag_frame is None + warning_errors = [ + record + for record in caplog.records + if record.exc_info and record.exc_info[0] is ResourceWarning + ] + if warning_filter == "always": + assert len(emitted) == 3 + assert all(warning.category is ResourceWarning for warning in emitted) + assert all("trio.run() shutdown" in str(warning.message) for warning in emitted) + assert {id(warning.source) for warning in emitted} == { + id(agen) for agen in saved + } + assert warning_errors == [] + else: + assert emitted == [] + assert len(warning_errors) == 3 + cleanup_errors = [ + record.exc_info[1] + for record in caplog.records + if record.exc_info and record.exc_info[0] is LookupError + ] + assert cleanup_errors == ([cleanup_error] if cleanup_raises else []) + + +@pytest.mark.parametrize("state", ["closed", "exhausted", "unstarted"]) +def test_asyncgen_shutdown_warning_only_for_unfinished(state: str) -> None: + saved = [] + + async def agen() -> AsyncGenerator[int, None]: + yield 1 + + async def async_main() -> None: + saved.append(agen()) + if state == "closed": + assert await saved[-1].asend(None) == 1 + await saved[-1].aclose() + elif state == "exhausted": + async for value in saved[-1]: + assert value == 1 + + with warnings.catch_warnings(record=True) as emitted: + warnings.simplefilter("always", ResourceWarning) + _core.run(async_main) + assert emitted == [] + + +def test_asyncgen_shutdown_warning_for_generator_created_during_cleanup() -> None: + saved = [] + closed = [] + + async def agen(index: int) -> AsyncGenerator[int, None]: + try: + yield index + finally: + closed.append(index) + if index: + saved.append(agen(0)) + assert await saved[-1].asend(None) == 0 + + async def async_main() -> None: + saved.append(agen(1)) + assert await saved[-1].asend(None) == 1 + + with pytest.warns(ResourceWarning, match="trio.run.*shutdown") as emitted: + _core.run(async_main) + assert closed == [1, 0] + assert len(emitted) == 2 + assert {id(warning.source) for warning in emitted} == {id(agen) for agen in saved} + + def test_firstiter_after_closing() -> None: saved = [] record = [] @@ -234,10 +349,13 @@ async def async_main() -> None: needs_retry = False record.clear() saved.clear() - _core.run(async_main) + with warnings.catch_warnings(record=True) as emitted: + warnings.simplefilter("always", ResourceWarning) + _core.run(async_main) if needs_retry: # pragma: no cover assert record == ["cleaned up"] else: + assert len(emitted) == 1 assert record == ["final collection", "done", "cleaned up"] break else: # pragma: no cover