asyncio.Queue won't wake up when put is executed in separate thread

[sergey-p3]

Bug report

When await queue.put() is called from another thread it never awakens original thread waiting for the value i.e. await queue.get()

Minimal, Reproducible Example

import asyncio
import threading


async def delayed_execution(queue, some_value):
    await asyncio.sleep(0.1)
    print("Delayed execution triggered")
    await queue.put(some_value)
    print("Delayed execution finished")


async def main():
    queue = asyncio.Queue()
    thread = threading.Thread(target=asyncio.run, args=(delayed_execution(queue, 1),))
    thread.start()
    print("Delayed execution 1 result:", await queue.get()) # never printed


if __name__ == "__main__":
    asyncio.run(main())

Your environment

• Python 3.10.7, Ubuntu 20.04.5 LTS
• Docker amd64/python:3.9-slim-buster

When the code is launched with ensure_future instead of a thread, it works as expected. (see main2)

When the code is launched without with a while loop checking for the queue not to be empty, the value is available. (see main3)

The same code as above, with the two extra tests

import asyncio
import threading


async def delayed_execution(queue, some_value):
    await asyncio.sleep(0.1)
    print("Delayed execution triggered")
    await queue.put(some_value)
    print("Delayed execution finished")


async def main():
    queue = asyncio.Queue()

    thread = threading.Thread(target=asyncio.run, args=(delayed_execution(queue, 1),))
    thread.start()

    print("Delayed execution 1 result:", await queue.get()) # never printed


async def main2():
    queue = asyncio.Queue()

    asyncio.ensure_future(delayed_execution(queue, 2))

    print("Delayed execution 2 result:", await queue.get()) # printed


async def main3():
    queue = asyncio.Queue()

    thread = threading.Thread(target=asyncio.run, args=(delayed_execution(queue, 3),))
    thread.start()

    while queue.empty():
        await asyncio.sleep(0.1)
    print("Delayed execution 3 result:", await queue.get()) # printed


if __name__ == "__main__":
    asyncio.run(main2())
    asyncio.run(main3())
    asyncio.run(main())

[sergey-p3]

asyncio.Queue uses future. The code can be simplified to remove a middleware

import asyncio
import threading


async def delayed_execution(queue, some_value):
    await asyncio.sleep(0.1)
    print("Delayed execution triggered")
    await queue.put(some_value)
    print("Delayed execution finished")


async def delayed_execution_future(future, some_value):
    await asyncio.sleep(0.1)
    print("Delayed execution triggered")
    future.set_result(some_value)
    print("Delayed execution finished")


async def main():
    future = asyncio.get_event_loop().create_future()

    thread = threading.Thread(target=asyncio.run, args=(delayed_execution_future(future, 1),))
    thread.start()

    await future
    print("Delayed execution 1 result:", future.result())


async def main2():
    future = asyncio.get_event_loop().create_future()

    asyncio.ensure_future(delayed_execution_future(future, 2))

    await future
    print("Delayed execution 2 result:", future.result())


async def main3():
    future = asyncio.get_event_loop().create_future()

    thread = threading.Thread(target=asyncio.run, args=(delayed_execution_future(future, 3),))
    thread.start()

    while not future.done():
        await asyncio.sleep(0.1)
    print("Delayed execution 3 result:", future.result())


if __name__ == "__main__":
    asyncio.run(main2())
    asyncio.run(main3())
    asyncio.run(main())

[kumaraditya303]

asyncio.Queue is documented to be not thread safe and your code is not using thread safe APIs so it's bug in your code.

[gvanrossum]

If you want to put something into an asyncio.Queue from another thread, you must use loop.call_soon_threadsafe(q.put, item). This is not an asyncio bug.

[sergey-p3]

@kumaraditya303 the problem is that the other thread is launched by a third-party library, azure-iot-device to be precise, and I discovered the issue only after digging into source code of BOTH libraries.
The use case I built is to simplify the reproducible example.

@gvanrossum the solution you proposed doesn't work either. q.put would not work because it is async and "was not awaited" error is raised, ensure_future doesn't wake the main thread, q.put_nowait neither wakes up original thread.
Since 3.10 asyncio.Queue won't accept loop as an argument.

Overall, the main issue is that there is no way of identifying whether the "put" and "get" are called from different threads. And to make the things worse, it is not always possible or not simple at best to know if another thread got spawned to process your code.

[gvanrossum]

You're right, we'd have to use call_soon_threadsafe to create a task running a coroutine that calls q.put(item).

The Queue class could defend itself against this misuse by having a _loop instance variable initialized to None that gets set to the current loop in the first call to put() or get(), and which every following put() or get() call compares to the current loop. (Arguably when the loop= argument was removed we went too far.)

Irregardless, mixing threads and asyncio is generally a bad idea, especially if you don't even know whether your code may be run in another thread.