Queue.get_nowait() and Queue.get() raise AssertionError when there are multiple tasks waiting for free slots.

[manipopopo]

The following code produce an AssertionError:

import asyncio

queue = asyncio.Queue(2)

async def putter(item):
    await queue.put(item)

async def getter():
    await asyncio.sleep(1)
    num = queue.qsize()
    try:
        for _ in range(num):
            item = queue.get_nowait()
    except AssertionError as e:
        print(e)

asyncio.ensure_future(putter(0))
asyncio.ensure_future(putter(1))
asyncio.ensure_future(putter(2))
asyncio.ensure_future(putter(3))
asyncio.get_event_loop().run_until_complete(getter())

When getter called get_nowait() (or await queue.get()), one pending putter would be scheduled to put an item in the next step of the event loop. When get_nowait() (or await queue.get()) was called for the second time, the queue was not full and the assertion in Queue.get_nowait (and Queue.get)

assert self.full(), 'queue not full, why are putters waiting?'

failed.
It seems 1dd213e causes the problem.

[gjcarneiro]

On 23 September 2015 at 07:14, manipopopo notifications@github.com wrote:

The following code produce an AssertionError:

import asyncio

queue = asyncio.Queue(2)

async def putter(item):
await queue.put(item)

async def getter():
await asyncio.sleep(1)
num = queue.qsize()
try:
for _ in range(num):
item = queue.get_nowait()
except AssertionError as e:
print(e)

asyncio.ensure_future(putter(0))
asyncio.ensure_future(putter(1))
asyncio.ensure_future(putter(2))
asyncio.ensure_future(putter(3))
asyncio.get_event_loop().run_until_complete(getter())

When getter called get_nowait() (or await queue.get()), one pending
putter would be scheduled to put an item in the next step of the event
loop. When get_nowait() (or await queue.get()) was called for the second
time, the queue was not full and the assertion in Queue.get_nowait (and
Queue.get)

assert self.full(), 'queue not full, why are putters waiting?'

failed.
It seems 1dd213e
1dd213e
cause the problem.

I don't know if 1dd213e caused the problem, but I debugged it and found
out the reason why this happens is that you do two consecutive get_nowait
calls, without letting the main loop run in between. Adding a "yield from
asyncio.sleep(0)" after the get_nowait() call solves (or works around) this
problem.

What happened is that in the second get_nowait() call() the putter(3)
waiter (a Future) is marked as done, but the coroutine code that actually
was suspended in Queue.put() (from putter(3)) did not actually finish
running, it is waiting for the main loop to resume it.

Thinking out loud, we might add a "hack" to sort of fix the problem. If we
keep a counter with "number of items are pending to put in the queue", the
assertion would be modified to check for qsize() plus this new counter.

But clearly, it would be desirable that get_nowait() can be called any
number of times without giving back control to the main loop without
risking the assertion getting triggered.

I think this is an interesting problem and deserves a bug tracker issue
hint

Gustavo J. A. M. Carneiro
Gambit Research
"The universe is always one step beyond logic." -- Frank Herbert

[manipopopo]

I noticed that Queue.get_nowait (and Queue.get) will check if there is any putter waiting for free slots. And if there is at least one pending putter, Queue.get_nowait prior to 1dd213e will invoke a putter immediately (instead of waiting for the main loop to resume the putter).

I use asyncio.Queue(maxsize=nonzero_value) as a container of numbers. There are many producers putting numbers into the queue, and one consumer who will get all numbers in the queue (at that moment) and calculate statistics for the numbers. For examples,

# in the consumer
if not queue.empty():
    numbers = [queue.get_nowait() for _ in range(queue.qsize())]
    print(max(numbers))

The program works (probably) on Python 3.4, but fails on Python 3.5. It seems that the document (Python 3.5) of asyncio.Queue doesn't clarify this issue.

Thanks for your help!