Object stream randomly drops items

[mjwestcott]

This issue concerns master@23803be

Memory object streams appear to randomly drop items in the following scenario (across all backends):

import anyio


async def receiver(r):
    while True:
        async with anyio.move_on_after(0.1):
            print(await r.receive())


async def main():
    s, r = anyio.create_memory_object_stream()

    async with anyio.create_task_group() as tg:
        await tg.spawn(receiver, r)

        for i in range(10):
            await anyio.sleep(0.2)
            await s.send(i)
        await tg.cancel_scope.cancel()


anyio.run(main)

Whereas the equivalent Trio code works as expected (every item is received and printed):

import trio


async def receiver(r):
    while True:
        with trio.move_on_after(0.1):
            print(await r.receive())


async def main():
    s, r = trio.open_memory_channel(max_buffer_size=0)

    async with trio.open_nursery() as n:
        n.start_soon(receiver, r)

        for i in range(10):
            await trio.sleep(0.2)
            await s.send(i)
        n.cancel_scope.cancel()


trio.run(main)

[mjwestcott]

diff --git a/src/anyio/streams/memory.py b/src/anyio/streams/memory.py
index 9b8e8ee..693caba 100644
--- a/src/anyio/streams/memory.py
+++ b/src/anyio/streams/memory.py
@@ -70,6 +70,8 @@ class MemoryObjectReceiveStream(Generic[T_Item], ObjectReceiveStream[T_Item]):
             try:
                 await receive_event.wait()
             except BaseException:
+                if container:
+                    self._state.buffer.append(container[0])
                 self._state.waiting_receivers.pop(receive_event, None)
                 raise

The basic problem: when receive is cancelled, it might have already been given the item. This patch just illustrates one avenue to explore.

[agronholm]

Thanks for the report, the accurate analysis and solution! The only change I would make is to use .appendleft() instead.

[njsmith]

I don't think returning the object to the internal buffer like that quite fixes the bug, because the stream is supposed to deliver objects in FIFO order, right? Consider if task A grabs an item from the buffer, then task B grabs an item from the buffer, then task A "puts back" the item it grabbed. Now the stream has re-ordered the objects.

…

On Wed, Aug 12, 2020 at 11:20 PM Alex Grönholm ***@***.***> wrote: Thanks for the report, the accurate analysis and solution! The only change I would make is to use .appendleft() instead. — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#146 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAEU42EBXTGWH5H2NHBAETTSAOA2HANCNFSM4P5P2OAA> .

-- Nathaniel J. Smith -- https://vorpus.org <http://vorpus.org>

[agronholm]

I'm not sure how to fix it then, but I will look at trio's implementation for inspiration.

[agronholm]

I don't understand how trio circumvents the same problem. If a task is waiting to receive from a memory channel and gets cancelled, and then a sender task gets scheduled, won't it try to reschedule that cancelled task with the value to be sent?

[njsmith]

@agronholm when wait_task_rescheduled is cancelled, it runs the abort_fn immediately, without waiting for the task to be scheduled again. And in this case, the abort_fn removes the cancelled task from the data structure that's tracking receivers, so a sender will never try to pass an object to a receiver that was cancelled.

[agronholm]

I see. But since AnyIO does not have the same low level facilities at its disposal, I will have to think of something else.

[njsmith]

Maybe you could detect when you've been simultaneously awoken by a cancellation + receiving an object to return, and "undo" the cancellation so the operation completes successfully?

[mjwestcott]

@njsmith I thought about that, but I wasn't sure that it's correct to ignore the cancellation. I guess if your 'simultaneously awoken by two things' interpretation is correct, then simply picking one of them is legit. Here's an implementation:

diff --git a/src/anyio/streams/memory.py b/src/anyio/streams/memory.py
index 9b8e8ee..6dc08f1 100644
--- a/src/anyio/streams/memory.py
+++ b/src/anyio/streams/memory.py
@@ -69,7 +69,11 @@ class MemoryObjectReceiveStream(Generic[T_Item], ObjectReceiveStream[T_Item]):

             try:
                 await receive_event.wait()
-            except BaseException:
+            except BaseException as e:
+                if type(e) == anyio.get_cancelled_exc_class() and container:
+                    # We were simultaneously awoken by a cancellation and receiving an
+                    # item. We choose the latter to ensure that items are not dropped.
+                    return container[0]
                 self._state.waiting_receivers.pop(receive_event, None)
                 raise

It seems to work on the asyncio and Trio backends, but not Curio.

[mjwestcott]

Curio is raising TaskTimeout instead of TaskCancelled. Hopefully that's the only thing wrong with the patch

[agronholm]

That really rubs me the wrong way. I've opened a PR with an alternative fix.