(Python 3.5) Asyncio and an attempt to run loop.run_until_complete() from within a running loop

[Ian Kelly]

On Fri, Apr 8, 2016 at 5:54 PM, Alexander Myodov <amyodov at gmail.com> wrote:
> Hello.
>
> TLDR: how can I use something like loop.run_until_complete(coro), to execute a coroutine synchronously, while the loop is already running?
>
> More on this:
>
> I was trying to create an aio_map(coro, iterable) function (which would asynchronously launch a coroutine for each iteration over iterable, and collect the data; similary to gevent.pool.Group.imap() from another async world), but stuck while attempting to make it well both from outside the async event loop and from inside one - any help?
>
> My code is at http://paste.pound-python.org/show/EQvN2cSDp0xqXK56dUPy/ - and I stuck around lines 22-28, with the problem that loop.run_until_complete() cannot be executed when the loop is running already (raising "RuntimeError: Event loop is running."). Is it normal? and why is it so restrictive? And what can I do to wait for `coros` Future to be finished?
>
> I tried various mixes of loop.run_forever() and even loop._run_once() there, but was not able to create a stable working code. Am I doing something completely wrong here? Am I expected to create a totally new event loop for synchronously waiting for the Future, if the current event loop is running - and if so, won't the previous event loop miss any its events?

The code is short enough that it would be better to include it inline
in your email for posterity; this thread will be archived, but paste
URLs have a bad tendency of eventually disappearing.


> def aio_map(coro, iterable, loop=None):
>     if loop is None:
>         loop = asyncio.get_event_loop()
>
>     async def wrapped_coro(coro, value):
>         return await coro(value)
>
>     coroutines = (wrapped_coro(coro, v) for v in iterable)

I'm not sure what value wrapped_coro adds here. Couldn't you just do:

    coroutines = (coro(v) for v in interable)

Or even:

    coroutines = map(coro, iterables)

>     coros = asyncio.gather(*coroutines, return_exceptions=True, loop=loop)
>
>     if not loop.is_running():
>         loop.run_until_complete(coros)
>     else:  # problem starts here
>         # If we run loop.run_until_complete(coros) as well,
>         # we get 'RuntimeError: Event loop is running.'

Right, there's no need to run multiple event loops.

>         asyncio.wait(coros)

This doesn't do anything. asyncio.wait is itself a coroutine, so all
this does is to instantiate the asyncio.wait coroutine and then
discard it without ever starting it.

Besides, you don't really want to do this. aio_map isn't a coroutine,
which means it's a synchronous call. In order for it to wait, it would
have to block the event loop thread, which means that the coroutines
it's waiting for would never finish!

This should get you the result that you're looking for:

def aio_map(coro, iterable, loop=None):
    if loop is None:
        loop = asyncio.get_event_loop()

    coroutines = map(coro, iterable)
    coros = asyncio.gather(*coroutines, return_exceptions=True, loop=loop)

    if loop.is_running():
        return coros
    else:
        return loop.run_until_complete(coros)

Note that this does something slightly different. Instead of returning
the *result* if we're running in the event loop, we're going to return
the *future* instead. This allows the *caller* to await the result, so
that you don't end up blocking the event thread.

This does mean that the pattern for calling aio_map from outside the
event loop is different from calling it inside the event loop. Your
main_loop coroutine becomes (note the addition of the "await"):

async def main_loop(loop):
    results = list(await aio_map(fetch_aio, range(5)))
    assert(results == [{'aio_arg': '0'}, {'aio_arg': '1'}, {'aio_arg':
'2'}, {'aio_arg': '3'}, {'aio_arg': '4'}]), results
    print('Assert ok!')