What is the pythonic way of running an asyncio event loop forever?

Viewed 3893

From the docs it seems the recommended way to kickstart a asyncio application is to use asyncio.run(), so my application looks like this:

async def async_main():
    # Everything here can use asyncio.create_task():
    o = ObjectThatMustBeKeptReferenced()
    create_tasks_and_register_callbacks(o)

    # Wait forever, ugly:
    while True:
        await asyncio.sleep(10000)

asyncio.run(async_main())

This infinite loop at the end of async_main() feels very wrong. In other languages, there is where I would call the event loop forever. So I tried this:

def main():
    loop = asyncio.get_event_loop()

    # Everything here can use asyncio.create_task():
    o = ObjectThatMustBeKeptReferenced()
    create_tasks_and_register_callbacks(o)

    # Wait forever, pretty:
    loop.run_forever()

main()

The problem here is this will fail with an error of the sorts RuntimeError: no running event loop when I call asyncio.create_task() inside my functions, even though the event loop is created and registered on the thread.

What is the pythonic, one way of sleeping forever on the asyncio event loop?

4 Answers

You can simply change the sleep loop to an ad-hoc event which is never set:

# wait forever
await asyncio.Event().wait()

If needed, you can easily modify this to store the event into a variable and propagate it as a shutdown signal.

Another option is for your function that creates the tasks to return the tasks it has created, in which case you can await them even though (or precisely because) they'll never complete:

async def async_main():
    o = ObjectThatMustBeKeptReferenced()
    tasks = create_tasks_and_register_callbacks(o)
    # wait forever, or until a task raises
    await asyncio.gather(*tasks)

While this doesn't communicate the intention of looping forever as clearly, it has the advantage that it will halt the program (and propagate the exception) if any of the tasks raises an unhandled exception.

Infinite while True is quite pythonic for asyncio-based scripts. It represents main loop of your program. This decouples script logic from low-level event loop and allows you to control the lifecycle of the program from the point of view of the application logic. Do some periodic actions and checkings.

Such loops wrapped in a try-except or finally can be placed in other modules and classes, which imply long-term periodic work. This is all consistent with asyncio as it uses cooperative multitasking and such loops give control to the event loop (e.g. by means of await asyncio.sleep).

asyncio.run is more preferable, because it still boils down to the calling of run_forever, but when exiting the main coroutine of the script, it cancels and clears running tasks, asynchronous generators, etc.

Real-world examples of infinite loops:

Uvicorn

    #...
    async def main_loop(self):
        counter = 0
        should_exit = await self.on_tick(counter)
        while not should_exit:
            counter += 1
            counter = counter % 864000
            await asyncio.sleep(0.1)
            should_exit = await self.on_tick(counter)

aiohttp:

        #...
        # sleep forever by 1 hour intervals,
        # on Windows before Python 3.8 wake up every 1 second to handle
        # Ctrl+C smoothly
        if sys.platform == "win32" and sys.version_info < (3, 8):
            delay = 1
        else:
            delay = 3600

        while True:
            await asyncio.sleep(delay)
    finally:
        await runner.cleanup()

The way I had done this in the past was to use asyncio.async which is a deprecated alias to asyncip.ensure_future. The biggest issue people faced was that they would call ensure_future without a running loop which doesn't ensure anything, so the higher-level API asyncio.create_task was added to assert there is a running loop so that the task would actually get run. In your case you want to schedule the main task then run the loop forever so the lower-level API of ensure_future is indeed what you want:

async def async_main():
    # Everything here can use asyncio.create_task():
    o = ObjectThatMustBeKeptReferenced()
    create_tasks_and_register_callbacks(o)
    
asyncio.ensure_future(async_main())
asyncio.get_event_loop().run_forever()

this basically indicates that when the loop is running we ensure async_main will execute but unlike asyncio.run we get the loop to run forever not just until it finishes the main entry point.


But you can do better, whether a given task will run forever or not shouldn't matter, you should ideally keep track of them and have your function await on all of them, if some of them run forever then your function will run forever as well by extension, but if not then as soon as all your async tasks have exited then your program exits as well. All that would be required is to keep track of the return value of each create_task that is run and pass them all to an asyncio.gather as @user4815162342 has done.

The classes don't have to expose their spawned tasks to accomplish this, just provide a method to wait on them, this is an example to illustrate the idea:

class NetworkAsyncHandler():
    def __init__(self, some_data):
        # don't just create floating tasks, keep a record of them
        self._floating_coroutines = []
        self.remembered_data = some_data
        self.connect_to_some_server_idk_this_is_is_an_example("localhost")
    def connect_to_some_server_idk_this_is_is_an_example(self, host):
        # ALWAYS KEEP A REFERENCE TO THE FUTURES YOU CREATE
        a = asyncio.create_task(DO_STUFF(host, self.remembered_data))
        # assuming you were never explicitly awaiting on a yourself, then just add it to the list
        self._floating_coroutines.append(a)

    async def join(self):
        """awaits on all floating coroutines created"""
        await asyncio.gather(*self._floating_coroutines)


async def async_main():
    a = NetworkAsyncHandler("A")
    b = NetworkAsyncHandler("B")
    # this function will wait on all created subroutines and return only when all
    # sub routines end.  If some run forever then so do we.
    await asyncio.gather(a.join(), b.join())
    
# now this works exactly as you would want it to.
asyncio.run(async_main())

you could imagine that typically this NetworkAsyncHandler might run forever but maybe there is a signal it can receive from the server to shut down which causes it to return, if both servers received this signal you would want the python program to finish. The first case where you call run_forever this doesn't happen, in fact you don't necessarily even know if the routines are still running but in this architecture each function awaits on the routines it actually depends on and asyncio.run does exactly what it is intended to do.

What is the pythonic, one way of sleeping forever on the asyncio event loop?

asyncio.run is a higher level API and is typically the preferred way of running an event loop. There's nothing wrong with using the lower level run_forever, though.

The problem here is this will fail with an error of the sorts RuntimeError: no running event loop when I call asyncio.create_task()

This doesn't work since create_task can't get the running event loop. Fortunately there is also a create_task method on loops. You'll need to update create_tasks_and_register_callbacks to accept the loop as an argument

def create_tasks_and_register_callbacks(obj, loop):

and then change any references to asyncio.create_task to loop.create_task inside its definition.

Related