Python asyncio run_coroutine_threadsafe never running coroutine?

Question

I'm not sure what I'm doing wrong here, I'm trying to have a class which contains a queue and uses a coroutine to consume items on that queue. The wrinkle is that the event loop is being run in a separate thread (in that thread I do loop.run_forever() to get it running).

What I'm seeing though is that the coroutine for consuming items is never fired:

import asyncio from threading import Thread import functools # so print always flushes to stdout print = functools.partial(print, flush=True) def start_loop(loop): def run_forever(loop): print("Setting loop to run forever") asyncio.set_event_loop(loop) loop.run_forever() print("Leaving run forever") asyncio.set_event_loop(loop) print("Spawaning thread") thread = Thread(target=run_forever, args=(loop,)) thread.start() class Foo: def __init__(self, loop): print("in foo init") self.queue = asyncio.Queue() asyncio.run_coroutine_threadsafe(self.consumer(self.queue), loop) async def consumer(self, queue): print("In consumer") while True: message = await queue.get() print(f"Got message {message}") if message == "END OF QUEUE": print(f"exiting consumer") break print(f"Processing {message}...") def main(): loop = asyncio.new_event_loop() start_loop(loop) f = Foo(loop) f.queue.put("this is a message") f.queue.put("END OF QUEUE") loop.call_soon_threadsafe(loop.stop) # wait for the stop to propagate and complete while loop.is_running(): pass if __name__ == "__main__": main()

Output:

 Spawaning thread Setting loop to run forever in foo init Leaving run forever

Community · Accepted Answer · 2020-06-20 09:12:55Z

There are several issues with this code.

First, check the warnings:

test.py:44: RuntimeWarning: coroutine 'Queue.put' was never awaited f.queue.put("this is a message") test.py:45: RuntimeWarning: coroutine 'Queue.put' was never awaited f.queue.put("END OF QUEUE")

That means queue.put is a coroutine, so it has to be run using run_coroutine_threadsafe:

asyncio.run_coroutine_threadsafe(f.queue.put("this is a message"), loop) asyncio.run_coroutine_threadsafe(f.queue.put("END OF QUEUE"), loop)

You could also use queue.put_nowait which is a synchronous method. However, asyncio objects are generally not threadsafe so every synchronous call has to go through call_soon_threadsafe:

loop.call_soon_threadsafe(f.queue.put_nowait, "this is a message") loop.call_soon_threadsafe(f.queue.put_nowait, "END OF QUEUE")

Another issue is that the loop gets stopped before the consumer task can start processing items. You could add a join method to the Foo class to wait for the consumer to finish:

class Foo: def __init__(self, loop): [...] self.future = asyncio.run_coroutine_threadsafe(self.consumer(self.queue), loop) def join(self): self.future.result()

Then make sure to call this method before stopping the loop:

f.join() loop.call_soon_threadsafe(loop.stop)

This should be enough to get the program to work as you expect. However, this code is still problematic on several aspects.

First, the loop should not be set both in the main thread and the extra thread. Asyncio loops are not meant to be shared between threads, so you need to make sure that everything asyncio related happens in the dedicated thread.

Since Foo is responsible for the communication between those two threads, you'll have to be extra careful to make sure every line of code runs in the right thread. For instance, the instantiation of asyncio.Queue has to happen in the asyncio thread.

See this gist for a corrected version of your program.

Also, I'd like to point out that this is not the typical use case for asyncio. You generally want to have an asyncio loop running in the main thread, especially if you need subprocess support:

asyncio supports running subprocesses from different threads, but there are limits:

An event loop must run in the main thread

The child watcher must be instantiated in the main thread, before executing subprocesses from other threads. Call the get_child_watcher() function in the main thread to instantiate the child watcher.

I would suggest designing your application the other way, i.e. running asyncio in the main thread and use run_in_executor for the synchronous blocking code.

Collectives™ on Stack Overflow

Python asyncio run_coroutine_threadsafe never running coroutine?

1 Answer 1

Comments

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Related