[FIXED] await does not return the value of a future with the value set in a thread

Issue

This code should print "hi" 3 times, but it doesn't always print.

I made a gif that shows the code being executed:

from asyncio import get_event_loop, wait_for, new_event_loop
from threading import Thread


class A:
    def __init__(self):
        self.fut = None

    def start(self):
        """
        Expects a future to be created and puts "hi" as a result
        """
        async def foo():
            while True:
                if self.fut:
                    self.fut.set_result('hi')
                    self.fut = None
        
        new_event_loop().run_until_complete(foo())

    async def make(self):
        """
        Create a future and print your result when it runs out
        """
        future = get_event_loop().create_future()
        self.fut = future
        print(await future)

a = A()

Thread(target=a.start).start()

for _ in range(3):
    get_event_loop().run_until_complete(a.make())

This is caused by await future, because when I change

print(await future)

by

while not future.done():
    pass
print(future.result())

the code always prints "hi" 3 times.

• Is there anything in my code that causes this problem in await future?

Solution

Asyncio functions are not thread-safe, except where explicitly noted. For set_result to work from another thread, you'd need to call it through call_soon_threadsafe.

But in your case this wouldn't work because A.start creates a different event loop than the one the main thread executes. This creates issues because the futures created in one loop cannot be awaited in another one. Because of this, and also because there is just no need to create multiple event loops, you should pass the event loop instance to A.start and use it for your async needs.

But - when using the event loop from the main thread, A.start cannot call run_until_complete() because that would try to run an already running event loop. Instead, it must call asyncio.run_coroutine_threadsafe to submit the coroutine to the event loop running in the main thread. This will return a concurrent.futures.Future (not to be confused with an asyncio Future) whose result() method can be used to wait for it to execute and propagate the result or exception, just like run_until_complete() would have done. Since foo will now run in the same thread as the event loop, it can just call set_result without call_soon_threadsafe.

One final problem is that foo contains an infinite loop that doesn't await anything, which blocks the event loop. (Remember that asyncio is based on cooperative multitasking, and a coroutine that spins without awaiting doesn't cooperate.) To fix that, you can have foo monitor an event that gets triggered when a new future is available.

Applying the above to your code can look like this, which prints "hi" three times as desired:

import asyncio
from asyncio import get_event_loop
from threading import Thread

class A:
    def __init__(self):
        self.fut = None
        self.have_fut = asyncio.Event()

    def start(self, loop):
        async def foo():
            while True:
                await self.have_fut.wait()
                self.have_fut.clear()
                if self.fut:
                    self.fut.set_result('hi')
                    self.fut = None

        asyncio.run_coroutine_threadsafe(foo(), loop).result()

    async def make(self):
        future = get_event_loop().create_future()
        self.fut = future
        self.have_fut.set()
        print(await future)

a = A()

Thread(target=a.start, args=(get_event_loop(),), daemon=True).start()

for _ in range(3):
    get_event_loop().run_until_complete(a.make())

Answered By - user4815162342