RuntimeWarning: coroutine method 'aclose' of '...' was never awaited when breaking out of async for

[hroncok]

Bug report

Edit: See #117536 (comment) and #117536 (comment) for simpler reproducers --@encukou

Bug description:

The following code is extracted from jinja2 tests, as I was debugging pallets/jinja#1900 -- it uses no jinja:

async def auto_aiter(iterable):
    for item in iterable:
        yield item

async def customfilter(iterable):
    items = []
    async for item in auto_aiter(iterable):
        items.append(item)
        if len(items) == 3:
            break
    return "".join(items)

import asyncio
print(asyncio.run(customfilter("yyyNNN")))

$ python3.12 asyncrepro.py 
yyy

$ python3.13 asyncrepro.py 
yyy
sys:1: RuntimeWarning: coroutine method 'aclose' of 'auto_aiter' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

$ python3.13 -X tracemalloc asyncrepro.py 
yyy
sys:1: RuntimeWarning: coroutine method 'aclose' of 'auto_aiter' was never awaited
Object allocated at (most recent call last):
  File ".../asyncrepro.py", lineno 10
    break

This seems to be related to #89091

I am not sure if this is a regression in Python or a bug in Jinja tests, but considering Jinja folks closed my issue, I decided to report it here instead.

CPython versions tested on:

3.13

Operating systems tested on:

Linux

Linked PRs

• GH-117536: ensure asyncgens GCd before runner teardown get aclosed #117577
• gh-117536: Fix asyncio _asyncgen_finalizer_hook() #117751
• GH-117536: GH-117894: fix athrow().throw(...) unawaited warning #117851
• GH-117714: implement athrow().close() and asend().close() using throw #117906

[TeamSpen210]

With async generators, you have to take steps to clean them up properly if you don't iterate through them completely. In this case it's not entirely necessary, but it's good practice in general:

async def customfilter(iterable):
    items = []
    agen = auto_aiter(iterable)
    try:
        async for item in agen:
            items.append(item)
            if len(items) == 3:
                break
        return "".join(items)
    finally:
        await agen.aclose()

This ensures any pending with statements or finally blocks get a chance to run. Instead of directly calling aclose you can also use contextlib.aclosing(). There is a mechanism built in to allow this to eventually be handled - once the generator is ready to be garbage collected, asyncio is able to step in and finalise it. But like file objects getting closed that happens at some arbitrary point, making it difficult to reason about.

With auto_aiter(), there isn't anything to cleanup, so in this case it might not be necessary. But Python can't know that when producing its warnings. Though, maybe auto_aiter() should also attempt to aclose() the iterable it wraps, if that has such a method. Not sure.

[hroncok]

Thank you. This solves the particular case I provided as a reproducer, but I keep getting more and more such problems in Jinja. I've tried to add this pattern at multiple places, but I failed miserably.

I still don't know whether what jinja does has always been "wrong" or whether the check is too strict :(

[graingert]

asyncio.run is supposed to correctly shutdown async generator objects, I'm not sure what's happening here

[graingert]

a smaller repro:

async def agen():
    yield
    yield

async def main():
    a = agen()
    await anext(a)
    del a

import asyncio
asyncio.run(main())

[graingert]

the problem is in _asyncgen_finalizer_hook, and exists in earlier versions of python - except generator.aclose() doesn't warn.

cpython/Lib/asyncio/base_events.py

Lines 557 to 561 in 9ceaee7

	def _asyncgen_finalizer_hook(self, agen):
	self._asyncgens.discard(agen)
	if not self.is_closed():
	self.call_soon_threadsafe(self.create_task, agen.aclose())

what's happening is self.create_task, agen.aclose() is being added to the call_soon_threadsafe callback deque, and the async generator is removed from the self._asyncgens set, then the loop is being closed before the call_soon_threadsafe callback deque is consumed resulting in the async generator not being shutdown

[graingert]

I think it should be something like this? I'll try and get a PR and tests up

    def _asyncgen_finalizer_hook(self, agen):
        def do_close():
            if self._asyncgens_shutdown_called:
                return
            self._asyncgens.discard(agen)
            self.create_task(agen.close())

         if not self.is_closed():
             self.call_soon_threadsafe(do_close)

[graingert]

my original diagnosis was incorrect - I've eliminated asyncio here:

import contextlib

async def agen():
    while True:
        yield


class CancelledError(BaseException):
    pass


async def amain():
    async with contextlib.aclosing(agen()) as a:
        await anext(a)
        try:
            a.aclose().throw(CancelledError)
        except CancelledError:
            pass


try:
    amain().send(None)
except StopIteration as e:
    print(e.value)

output:

/home/graingert/projects/cpython/demo2.py:16: RuntimeWarning: coroutine method 'aclose' of 'agen' was never awaited
  a.aclose().throw(CancelledError)
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
None

it looks like throwing into aclose() doesn't mark the coroutine as awaited

this is caused in asyncio because async generators garbage collected before shutdown_asyncgens agen.aclose() gets cancelled by _cancel_all_tasks before they can run

[graingert]

@kumaraditya303 this looks like an issue in #104611

[encukou]

Simple reproducer that keeps the asyncio and async for/break (this way it might be easier to see that it should work):

async def agen():
    yield 0
    yield 1

async def reproducer():
    async for item in agen():
        break

import asyncio
asyncio.run(reproducer())

This gives:

sys:1: RuntimeWarning: coroutine method 'aclose' of 'agen' was never awaited
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

[encukou]

I'm marking this for consideration as a release blocker.

[encukou]

this is caused in asyncio because async generators garbage collected before shutdown_asyncgens agen.aclose() gets cancelled by _cancel_all_tasks before they can run

This points to a workaround: adding await asyncio.sleep(0) after the for makes the warning go away.
(thanks to @kumaraditya303 for suggesting that)

[gvanrossum]

Who's taking point here? It can't be me, I don't know the first thing about async generators (no need to educate me -- I'm trying to delegate here :-).

[graingert]

Adding the await asyncio.sleep(0) changes the call asyncio makes from agen.aclose().throw(CancelledError) to agen.aclose().send(None) in neither case is asyncio failing to await the aclose()

[encukou]

@kumaraditya303 & @graingert Do you want to work on a fix, or should I look into it?

[vstinner]

I proposed PR gh-117751 to fix this issue: make the finalization of asynchronous generators more reliable.

I don't think that the warning is a regression. The problem is more that the finalization of asynchronous generators is not reliable, but Python now emits a warning if the generator is not finalized properly. It's just that the bug becomes more visible. Correct me if I'm wrong.

[graingert]

@vstinner this is the wrong diagnosis I originally made, throwing into the aclose() object should mark it as awaited

[graingert]

here's a realistic usage example of the problem of the incorrect warning:

import asyncio

async def agenfn():
    return
    yield

class MyExc(Exception):
    pass

async def amain():
    try:
        async with asyncio.TaskGroup() as tg:
            agen = agenfn()
            async for v in agen:
                break
            tg.create_task(agen.aclose())
            raise MyExc()
    except* MyExc:
        pass

asyncio.run(amain())

outputs:

/home/graingert/projects/cpython/Lib/asyncio/base_events.py:2046: RuntimeWarning: coroutine method 'aclose' of 'agenfn' was never awaited
  handle = None  # Needed to break cycles when an exception occurs.
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

when clearly the agen.aclose() was awaited by tg.create_task(age.aclose())

[arthur-tacca]

when clearly the agen.aclose() was awaited by tg.create_task(age.aclose())

Is it clear? You're running into unrelated issue #116048: the task was cancelled before it started running so asyncio doesn't run it at all (unlike Trio which would run it up to the first unshielded await and throw a cancellation exception as usual).

[Edit: an earlier version of this comment was more equivocal but I just tried the code with a couple of extra print statements and I think it really is hitting that issue.]

[arthur-tacca]

By the way, and sorry if this is a total red herring, but perhaps #116048 is not totally unrelated after all. The problem there is that asyncio.Task.cancel() on a task that hasn't been scheduled at all just results on the task being aborted immediately. I haven't followed everything in this issue, but some of the examples seem to hint that generator.aclose() is being scheduled in a task but then hitting the same issue: it's being cancelled by a loop shutdown, so that means the task isn't run at all.

[graingert]

@arthur-tacca yes this is true, but the aclose Coroutine should be considered awaited just like a regular async def

[encukou]

Fixed in main; I'll close this.
The underlying issue is present in 3.12 too, but backporting is tracked in #117894.