Race condition in asyncio.wait_for()

[jihoonyoon]

In the following program, the prod coroutine produces 0 to 5 and put them into the queue, and the cons coroutine consumes them with timeout. If this program is run, the cons gets item 0, 1, 2, 3, and 5, but not 4. In other words, the item 4 was put into the queue, but it got lost.

It seems that the observed behavior happens because the _release_waiter function does not cancel fut, opening up a potential race condition in which fut progresses even after the waiter is released.

• Program

import asyncio, time

@asyncio.coroutine
def prod(q, loop):
    for i in range(6):
        print("putting an item=%s" % i)
        yield from q.put(i)
        yield from asyncio.sleep(i)
    yield from q.put(None)

@asyncio.coroutine
def cons(q, loop):
    while True:
        try:
            item = yield from asyncio.wait_for(q.get(), 3.0)
            print("got an item=%s" % item)
        except asyncio.TimeoutError as e:
            print("timeout!")
            continue
        if item == None: break
    loop.stop()

loop = asyncio.get_event_loop()
q = asyncio.Queue()
loop.create_task(prod(q, loop))
loop.create_task(cons(q, loop))
loop.run_forever()

• Output

putting an item=0
putting an item=1
got an item=0 <------ 1 received
got an item=1
putting an item=2
got an item=2 <------ 2 received
putting an item=3
got an item=3 <------ 3 received
putting an item=4
timeout!
timeout!               
putting an item=5
got an item=5 <------ 5 received, but where 4 went?
timeout!
got an item=None

[Martiusweb]

Hi,

While I think understand the case you exposed, I can not reproduce it with python 3.5.1. What version of python/asyncio are you using? How frequently do you reproduce the output you posted?

According the the implementation of wait_for() in python 3.5.1, the future waiter is released by setting a value to it, then wait_for() is awoken, sees that the waiter is done (meaning that the timeout expired), and then cancels the future (the task wrapping q.get()) if it is not done.

It means that the cancellation of the future happens one loop iteration after the waiter is released. In the meantime, the task wrapping q.get() could be awoken if q.put() schedules the waiters of q.get() before _release_waiters is called. However, if q.get() is done, the result is returned and no TimeoutError is raised, so we shouldn't get the output you see.

Can the issue be related to the implementation of the queue, and fixed in a more recent version of asyncio (see the pull request #269)?

[jihoonyoon]

Python version and system information are shown below. As you have suggested, I will try it with python 3.5.1 to see if the issue is fixed in the version.

$ uname -srvmpo
Linux 3.13.0-63-generic #103-Ubuntu SMP Fri Aug 14 21:42:59 UTC 2015 x86_64 x86_64 GNU/Linux
$ python3
Python 3.4.3 (default, Oct 14 2015, 20:28:29)
[GCC 4.8.4] on linux

The posted output can be reproduced always. I just ran it again and got the same.

$ python3 aqueue_bug.py
putting an item=0
putting an item=1
got an item=0
got an item=1
putting an item=2
got an item=2
putting an item=3
got an item=3
putting an item=4
timeout!
timeout!
putting an item=5
got an item=5
timeout!
got an item=None

[jihoonyoon]

Confirming that the issue is fixed in python 3.5.1.

root@19900eeb46a4:~# python --version
Python 3.5.1
root@19900eeb46a4:~# python aqueue_bug.py
putting an item=0
putting an item=1
got an item=0
got an item=1
putting an item=2
got an item=2
putting an item=3
got an item=3
putting an item=4    
timeout!
got an item=4  <--- Got it!
timeout!
putting an item=5
got an item=5
timeout!
got an item=None