asyncio.Queue.put_nowait(), followed get() task cancellation leads to item being lost

[gustavo]

BPO	23812
Nosy	@gvanrossum, @vstinner, @1st1
Files	bug-test.diff: test case patch Issue23812.diff: A patch that fixes the problem queue_bug.py p.diff: patch to refactor Queue.get()

^{Note: these values reflect the state of the issue at the time it was migrated and might not reflect the current state.}

Show more details

GitHub fields:

assignee = None
closed_at = <Date 2015-08-05.17:59:43.619>
created_at = <Date 2015-03-30.13:59:52.739>
labels = ['deferred-blocker', 'type-bug', 'expert-asyncio']
title = 'asyncio.Queue.put_nowait(), followed get() task cancellation leads to item being lost'
updated_at = <Date 2015-08-07.16:37:54.660>
user = 'https://bugs.python.org/gustavo'

bugs.python.org fields:

activity = <Date 2015-08-07.16:37:54.660>
actor = 'yselivanov'
assignee = 'none'
closed = True
closed_date = <Date 2015-08-05.17:59:43.619>
closer = 'yselivanov'
components = ['asyncio']
creation = <Date 2015-03-30.13:59:52.739>
creator = 'gustavo'
dependencies = []
files = ['38740', '38741', '38820', '40142']
hgrepos = []
issue_num = 23812
keywords = ['patch']
message_count = 24.0
messages = ['239611', '239624', '239625', '239627', '239628', '239679', '239738', '240006', '246762', '248055', '248056', '248137', '248138', '248144', '248145', '248146', '248148', '248150', '248157', '248181', '248187', '248203', '248206', '248207']
nosy_count = 5.0
nosy_names = ['gvanrossum', 'gustavo', 'vstinner', 'python-dev', 'yselivanov']
pr_nums = []
priority = 'deferred blocker'
resolution = 'fixed'
stage = 'resolved'
status = 'closed'
superseder = None
type = 'behavior'
url = 'https://bugs.python.org/issue23812'
versions = ['Python 3.4', 'Python 3.5']

[gustavo]

I have a pattern where I read from a queue with a timeout, generally like this:

while True:
  reader = asyncio.async(wait_for(queue.get(), 0.1))
  try:
    item = (yield from reader)
  except asyncio.TimeoutError:
    reader.cancel()
    continue

This is to have a loop where we try to get items from the queue with a timeout. Prior to Python 3.5, wait_for() doesn't automatically cancel the reading task, so I have to do it explicitly above.

The code has a strange race condition where, if a taks calls put_nowait on a reader task that is just about to be cancelled due to timeout, then the item that wast put onto the queue gets lost.

In the tests framework, the minimal test case I can come up with is mainly this code:

  q = asyncio.Queue(loop=loop)
  reader = loop.create_task(q.get())
  loop.run_until_complete(asyncio.sleep(0.01, loop=loop))

  q.put_nowait(1)
  q.put_nowait(2)
  reader.cancel()

When the reader gets cancelled, the item 1 is lost into the ether, and when you create another reader for the same queue, it only gets the second item 2. I would expect that, if a reader task is cancelled, the item at the head of the queue doesn't get lost. Either the reader succeeds and the item is removed, or it doesn't succeed and the item is left alone.

I attach a patch to the tulip tests that reproduces the problem in both Python 3.4.2 and tulip hg.

[gvanrossum]

Looks like a valid bug report, I like the test you provided, and the fix seems on the right track. Comments on the fix:

• I'd really like to see a rietveld diff for both patches.

• Are there other places where a cancellation can have a similar effect? Maybe the same logic in put()?

• The fix uses waiter.result() but if waiter itself is cancelled this will raise a CancelledError. This is probably fine because in that case no value has been put into it. But now this got me thinking -- how can the yield-from in get() receive a CancelledError without the waiter being cancelled? IOW how does the fix actually work??? (Maybe because the generator has already returned when the Task wrapping it is cancelled?)

[gustavo]

I created a codereview issue: https://codereview.appspot.com/222930043

[gustavo]

• Are there other places where a cancellation can have a similar effect? Maybe the same logic in put()?

Hm.. I didn't look, but yes, it does look like it might be affected by the same issue. I'll try to create a test for that to confirm.

how can the yield-from in get() receive a CancelledError without the waiter being cancelled?

Well, it definitely gets a CancelledError in the return (yield from waiter). I added extensive logging and could confirm this is the case.

What happens is that

1. Test creates a task q.get()
2. The task coroutine, inside Queue.get, blocks on the yield from waiter
3. a couple of items are put in the queue
4. Test code calls task.cancel() (where task is running the coroutine q.get)
5. The Queue.get task receives a CancelledError because it was cancelled. The exception appears to come from the yield from waiter only because this is the place where the Queue.get task was suspended, but waiter itself is not cancelled.

[gvanrossum]

Make sense. I'll be waiting for your updated patch. Thanks for both the bug report and the fix!

[gustavo]

So I uploaded a new patch version fixing a similar problem in put().

[gvanrossum]

I'm sorry, I don't have time to review this (and it's subtle enough that I don't want to approve it without understanding).

Maybe Victor understands?

[vstinner]

queue_bug.py: script to reproduce the bug.

I confirm that Queue.get() sometimes looses items when it is cancelled. The waiter contains the result, but the waiter is lost when get() is cancelled.

Queue.get() waiter got a result, but Queue.get() wakeup is only *scheduled* yet. It may even be executed in the same iteration, or maybe in the next loop iteration.

I didn't review patches yet.

[gustavo]

Don't know if it helps, but I made a github pull request for this:

python/asyncio#256

[python-dev]

New changeset 7aa2d3e1c885 by Yury Selivanov in branch '3.4':
Issue bpo-23812: Fix asyncio.Queue.get() to avoid loosing items on cancellation.
https://hg.python.org/cpython/rev/7aa2d3e1c885

New changeset d5644d7e222d by Yury Selivanov in branch '3.5':
Issue bpo-23812: Fix asyncio.Queue.get() to avoid loosing items on cancellation.
https://hg.python.org/cpython/rev/d5644d7e222d

New changeset 8f581da70ccd by Yury Selivanov in branch 'default':
Issue bpo-23812: Fix asyncio.Queue.get() to avoid loosing items on cancellation.
https://hg.python.org/cpython/rev/8f581da70ccd

[1st1]

The fix is committed. Closing the issue. Thanks a lot, Gustavo!