asyncio.Queue.get() won't resolve when the queue is empty while combined with ThreadPoolExecutor

[bernardolansing]

Bug report

Bug description:

I am experiencing something that I think it's a bug in Python. I'm running a synchronous function in a ThreadPoolExecutor and this function will sometimes put items to a queue. I have also an asynchronous function that runs in a loop and will consume items from that queue. The bug happens when I execute await queue.get() when the queue is empty. If the queue has items, they will be popped nicely.

Below is a code snippet that, at least on my machine, have shown the bug.

import asyncio
from time import sleep
from concurrent.futures import ThreadPoolExecutor

loop = asyncio.get_event_loop()
queue = asyncio.Queue()


def putter_sync():
    while True:
        sleep(1)
        print('will PUT to queue')
        queue.put_nowait(1)
        print('PUT to queue')


async def getter():
    while True:
        # await asyncio.sleep(1.1)  # <-- if you uncomment this, it'll work fine! Items are put every 1 second,
        # so by waiting 1.1 seconds you'll never try to get from an empty queue.
        print('will GET from queue')
        await queue.get()
        print('GOT from queue')


executor = ThreadPoolExecutor(max_workers=1)
loop.run_in_executor(executor=executor, func=putter_sync)
loop.create_task(getter())
loop.run_forever()

This code will print this:

will GET from queue
will PUT to queue
PUT to queue
will PUT to queue
PUT to queue
will PUT to queue
PUT to queue
will PUT to queue
PUT to queue

Note that the get call never resolved. If I uncomment the sleep call, it runs as expected:

will PUT to queue
PUT to queue
will GET from queue
GOT from queue
will PUT to queue
PUT to queue
will GET from queue
GOT from queue
will PUT to queue
PUT to queue
will GET from queue
GOT from queue
will PUT to queue
PUT to queue
will GET from queue
GOT from queue

I also tried to replace the ThreadPoolExecutor by a plain threading.Thread, and the behaviour remained the same. To be honest, I don't actually know the difference between both, if anyone wants to explain it to me I'd appreciate. By the way, is this code even supposed to work? I don't think it is that bad...

CPython versions tested on:

3.11

Operating systems tested on:

Linux

[ronaldoussoren]

asyncio.Queue is documented as being not thread safe and your code tries to use the queue from multiple threads. One way to avoid threading issues is to use loop.call_soon_threadsafe to add items to the queue.

[bernardolansing]

Ohh my bad. I totally forgot about thread unsafety in asyncio. Sorry!

P.S.: It'd be nice if a warning was triggered when someone tried to do something like that. Don't know about how difficult it is to implement, just suggesting.

Thank you!