Hang in AsyncQueue.join()

[x42005e1f]

Due to the way call_soon_threadsafe() works, calling SyncQueue.task_done() has one side effect: a race is possible between queue._finished.set() and queue._finished.clear(). As a result, queue._finished (asyncio.Event instance) can be set even when there were more put() calls than task_done() calls, causing an infinite loop with a complete blocking of the event loop and high CPU load:

queue = janus.Queue()

await queue.async_q.put(1)  # queue._finished.clear()
assert queue.async_q.unfinished_tasks == 1

queue.sync_q.task_done()  # call_soon_threadsafe(queue._finished.set)
assert queue.async_q.unfinished_tasks == 0

await queue.async_q.put(2)  # queue._finished.clear()
assert queue.async_q.unfinished_tasks == 1

async def func():
    print(1)  # 1
    await asyncio.sleep(0)
    print(2)  # will never be printed!

asyncio.create_task(func())

await queue.async_q.join()  # infinite loop!

A possible solution would be to replace asyncio.Event with asyncio.Condition.

[asvetlov]

Yes, recently during the code review I've been asking myself: "Why ._finished is not a condition variable?"
Perhaps because .join() was implemented by third-party contributor; and people are hard in understanding conditions-vars.
I don't know why, but mutex+event fits their brain better, and many even experienced Python devs are not familiar with condition-vars at all.

I'll do the change if a champion don't overtake me.

[x42005e1f]

._finished came from the asyncio.Queue implementation, since Janus is essentially a symbiosis of that queue and queue.Queue. Because asyncio.Queue is not thread-safe, its methods do not need a mutex. So asyncio.Event has the distinct advantage: it is simpler and faster, which matters in the standard library.

[x42005e1f]

Moreover, I will even say that it was added in 167cc80. And it looks like it first appeared in python-tulip as part of JoinableQueue.