Wsrtjtyk

I've been playing around with Python's asyncio. I think I have a reasonable understanding by now. But the following behavior puzzles me.

test.py:

from threading import Thread

import asyncio



async def wait(t):

    await asyncio.sleep(t)

    print(f'waited {t} sec')



def run(loop):

    loop.run_until_complete(wait(2))



loop = asyncio.get_event_loop()

t = Thread(target=run, args=(loop,))

t.start()

loop.run_until_complete(wait(1))

t.join()

This code is wrong. I know that. The event loop can't be run while it's running, and it's generally not thread safe.

My question: why can wait(1) sometimes still finish its job?

Here's the output from two consecutive runs:

>>> py test.py

... Traceback (most recent call last):

...   File "test.py", line 14, in <module>

...     loop.run_until_complete(wait(1))

...   File "C:PythonPython37libasynciobase_events.py", line 555, in run_until_complete

...     self.run_forever()

...   File "C:PythonPython37libasynciobase_events.py", line 510, in run_forever

... 

...     raise RuntimeError('This event loop is already running')

... RuntimeError: This event loop is already running

... waited 2 sec



>>> py test.py

... Traceback (most recent call last):

...   File "test.py", line 14, in <module>

...     loop.run_until_complete(wait(1))

...   File "C:PythonPython37libasynciobase_events.py", line 555, in run_until_c

... omplete

...     self.run_forever()

...   File "C:PythonPython37libasynciobase_events.py", line 510, in run_forever

... 

...     raise RuntimeError('This event loop is already running')

... RuntimeError: This event loop is already running

... waited 1 sec

... waited 2 sec

The first run's behavior is what I expected - the main thread fails, but the event loop still runs wait(2) to finish in the thread t.

The second run is puzzling, how can wait(1) do its job when the RuntimeError is already thrown? I guess it has to do with thread synchronization and the non-thread-safe nature of the event loop. But I don't know exactly how this works.

Ohhh... never mind. I read the code of asyncio and figured it out. It's actually quite simple.

run_until_complete calls ensure_future(future, loop=self) before it checks self.is_running() (which is done in run_forever). Since the loop is already running, it can pick up the task before the RuntimeError is thrown. Of course it doesn't always happen because of the race condition.

Exceptions are thrown per thread. The runtime error is raised in a different thread from the event loop. The event loop continues to execute, regardless.

And wait(1) can sometimes finish it's job because you can get lucky. The asyncio loop internal data structures are not guarded against race conditions caused by using threads (which is why there are specific thread-support methods you should use instead). But the nature of race conditions is such that it depends on the exact order of events and that order can change each time you run your program, depending on what else your OS is doing at the time.

The run_until_complete() method first calls asyncio.ensure_task() to add the coroutine to the task queue with a 'done' callback attached that will stop the event loop again, then calls loop.run_forever(). When the coroutine returns, the callback stops the loop. The loop.run_forever() call throws the RuntimeError here.

When you do this from a thread, the task gets added to a deque object attached to the loop, and if that happens at the right moment (e.g. when the running loop is not busy emptying the queue), the running loop in the main thread will find it, and execute it, even if the loop.run_forever() call raised an exception.

All this relies on implementation details. Different versions of Python will probably exhibit different behaviour here, and if you install an alternative loop (e.g. uvloop), there will almost certainly be different behaviour again.

If you want to schedule coroutines from a different thread, use asyncio.run_coroutine_threadsafe(); it would :

from threading import Thread

import asyncio



async def wait(t):

    print(f'going to wait {t} seconds')

    await asyncio.sleep(t)

    print(f'waited {t} sec')



def run(loop):

    asyncio.run_coroutine_threadsafe(wait(2), loop)



loop = asyncio.get_event_loop()

t = Thread(target=run, args=(loop,))

t.start()

loop.run_until_complete(wait(1))

t.join()

The above doesn't actually complete the wait(2) coroutine because the wait(1) coroutine is being run with loop.run_until_complete() so its callback stops the loop again before the 2 second wait is over. But the coroutine is actually started:

going to wait 1 seconds

going to wait 2 seconds

waited 1 sec

but if you made the main-thread coroutine take longer (with, say, wait(3)) then the one scheduled from the thread would also complete. You'd have to do additional work to ensure that there are no more pending tasks scheduled to run with the loop before you shut it down.