[FIXED] How do I collect task garbages in Python asyncio?

Issue

Sorry for the bad English, but I'll try my best.

Consider the following code:

import asyncio
async def main():
    async def printA():
        await asyncio.sleep(1)
        print('a')
    # create a stream
    stream=...
    async for message in stream:
        pass
asyncio.run(main())

Yes, printA is not yet used.

Now I want to invoke printA when I see some types of messages from the stream.

If I can accept that the stream waits printA is done to continue, I can write something like this:

async for message in stream:
    if message=='printA':
        await printA()

But I can't, so I must write at least:

async def main():
    async def printA():
        await asyncio.sleep(1)
        print('a')
    # create a stream
    stream=...
    taskSet=set()
    async for message in stream:
        if message=='printA':
            taskSet.add(asyncio.create_task(printA()))
    await asyncio.gather(*taskSet)

But if the stream is long enough, taskSet would become really big, even if many printA(s) are in fact already done.

So I would want them to be removed as soon as they are done.

I don't know how to write this from now on.

Can I remove that task within printA? The execution of printA() won't be earlier than create_task is invoked, but would it be later than create_task returns? Documentation does not seem to guarantee that. Although I found some says that it is guaranteed by the current implementation.

I can't simply discard the task reference, right? As the doc of create_task says:

Important: Save a reference to the result of this function, to avoid a task disappearing mid execution.

Solution

You can find the answer directly in the bug report concerning the same problem of "fire and forget" tasks that led to the documentation update "Important: Save a reference ..."

https://bugs.python.org/issue44665

I'll copy the recipe for an automatic task removal:

running_tasks = set()
# [...]
task = asyncio.create_task(some_background_function())
running_tasks.add(task)
task.add_done_callback(lambda t: running_tasks.remove(t))

Answered By - VPfB