In trying to debug asyncio code, I’ve run into some various issues, and ended up settling on this idiom for task spawning. I’m curious how other people do things, and if there should be a recommended way in the docs? Conceptually, I want to spawn a task the same way you’d spin off a thread or start an asynchronous subprocess, where I don’t ever want to await that, so any exceptions should be logged to the console.
import asyncio
import traceback
all_tasks = [] # kinda like threading.all_threads()
def task_done(task):
all_tasks.remove(task)
exc = task.exception() # Also marks that the exception has been handled
if exc: traceback.print_exception(exc)
def spawn(awaitable):
"""Spawn an awaitable as a stand-alone task"""
task = asyncio.create_task(awaitable)
all_tasks.append(task)
task.add_done_callback(task_done)
return task
Usage would be something like this:
async def task1():
print("Starting task 1")
await asyncio.sleep(1)
print("Ending task 1")
async def task2():
print("Starting task 2")
await asyncio.slep(2) # oopsie typo
async def main():
...
print("blah blah stuff goes here")
...
spawn(task1())
...
spawn(task2())
...
await asyncio.sleep(3)
print("Shutting down") # No hard feelings. I don't blame you. Critical error.
if __name__ == "__main__":
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(main())
print("Unfinished tasks:", all_tasks)
Critically, every spawned task is retained (see docs note which wasn’t there in Python 3.7 or 3.8), and when the task is done, any exception is immediately reported.
I’d very much like for this to have been more discoverable. Can something be added to the docs, or maybe even have an asyncio.spawn() function that behaves like this?
And I’d much rather use a standard or well-known idiom than try to invent my own.