Should closing receive stream remove waiting senders?

[mbrancato]

Currently, when MemoryObjectReceiveStream.close() is called, it completes the asyncio.Event for the senders, this should allow any senders awaiting the Event to continue.

anyio/src/anyio/streams/memory.py

Lines 142 to 145 in 48efdec

	if self._state.open_receive_channels == 0:
	send_events = list(self._state.waiting_senders.keys())
	for event in send_events:
	event.set()

But it does not remove the sender Event objects from the MemoryObjectStreamState. This appears to lead to BrokenResourceErrors in MemoryObjectSendStream.send() as they have not been removed from the dictionary, but they were completed by calling event.set().

anyio/src/anyio/streams/memory.py

Lines 220 to 221 in 48efdec

	if self._state.waiting_senders.pop(send_event, None): # type: ignore[arg-type]
	raise BrokenResourceError

This appears to at least happen when using async with on an MemoryObjectReceiveStream object.

Is this expected behavior when MemoryObjectReceiveStream.close() is called? To prevent raising an exception, what better way would there be beside using an async context manager on the MemoryObjectReceiveStream object?

Just to note, I'm tracking this down from its use in Starlette where we periodically see this exception raised.

[agronholm]

Sorry for not noticing this before!

Is this expected behavior when MemoryObjectReceiveStream.close() is called?

Yes, as is documented here. The docstring for the send() method was inherited from the superclass and should probably be overridden here.

To prevent raising an exception, what better way would there be beside using an async context manager on the MemoryObjectReceiveStream object?

The producer end should be closed first. It's an error to send an object when nobody is ever going to receive it.