When using asyncio, how do you allow all running tasks to finish before shutting down the event loop
You can retrieve unfinished tasks and run the loop again until they finished, then close the loop or exit your program.
pending = asyncio.all_tasks()
loop.run_until_complete(asyncio.gather(*pending))
pending
is a list of pending tasks.asyncio.gather()
allows to wait on several tasks at once.
If you want to ensure all the tasks are completed inside a coroutine (maybe you have a "main" coroutine), you can do it this way, for instance:
async def do_something_periodically():
while True:
asyncio.create_task(my_expensive_operation())
await asyncio.sleep(my_interval)
if shutdown_flag_is_set:
print("Shutting down")
break
await asyncio.gather(*asyncio.all_tasks())
Also, in this case, since all the tasks are created in the same coroutine, you already have access to the tasks:
async def do_something_periodically():
tasks = []
while True:
tasks.append(asyncio.create_task(my_expensive_operation()))
await asyncio.sleep(my_interval)
if shutdown_flag_is_set:
print("Shutting down")
break
await asyncio.gather(*tasks)
As of Python 3.7 the above answer uses multiple deprecated APIs (asyncio.async and Task.all_tasks,@asyncio.coroutine, yield from, etc.) and you should rather use this:
import asyncio
async def my_expensive_operation(expense):
print(await asyncio.sleep(expense, result="Expensive operation finished."))
async def do_something_periodically(expense, interval):
while True:
asyncio.create_task(my_expensive_operation(expense))
await asyncio.sleep(interval)
loop = asyncio.get_event_loop()
coro = do_something_periodically(1, 1)
try:
loop.run_until_complete(coro)
except KeyboardInterrupt:
coro.close()
tasks = asyncio.all_tasks(loop)
expensive_tasks = {task for task in tasks if task._coro.__name__ != coro.__name__}
loop.run_until_complete(asyncio.gather(*expensive_tasks))
I'm not sure if this is what you've asked for but I had a similar problem and here is the ultimate solution that I came up with.
The code is python 3 compatible and uses only public asyncio APIs (meaning no hacky _coro
and no deprecated APIs).
import asyncio
async def fn():
await asyncio.sleep(1.5)
print('fn')
async def main():
print('main start')
asyncio.create_task(fn()) # run in parallel
await asyncio.sleep(0.2)
print('main end')
def async_run_and_await_all_tasks(main):
def get_pending_tasks():
tasks = asyncio.Task.all_tasks()
pending = [task for task in tasks if task != run_main_task and not task.done()]
return pending
async def run_main():
await main()
while True:
pending_tasks = get_pending_tasks()
if len(pending_tasks) == 0: return
await asyncio.gather(*pending_tasks)
loop = asyncio.new_event_loop()
run_main_coro = run_main()
run_main_task = loop.create_task(run_main_coro)
loop.run_until_complete(run_main_task)
# asyncio.run(main()) # doesn't print from fn task, because main finishes earlier
async_run_and_await_all_tasks(main)
output (as expected):
main start
main end
fn
That async_run_and_await_all_tasks function will make python to behave in a nodejs manner: exit only when there are no unfinished tasks.