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

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related