Avoid dropping of messages after breaking from Select blocks

[shsms]

The current implementation of Select.ready(), instead of just checking that some of the receivers have messages, also fetches them.

And in cases where there are multiple receivers with messages waiting to be read, Select.ready() consumes the latest message from each of them, and if subsequent user code decides to drop the Select object, and recreates it for some reason, the unprocessed messages that Select.ready() had fetched will get lost.

This is incorrect usage of Select, but the implementation should still try to ensure messages don't get dropped in such cases. This PR addresses this issue as follows:

1. the Receiver interface now requires implementations to provide two private methods _ready and _get, which are used to implement __anext__ on the interface. So Receiver objects would retain the original interface with __anext__, and receive, but now provide methods to only check if a message is ready or wait until one is, and a separate method to get a message after it becomes ready.
2. Update Select.ready() to use the Receiver._ready() method to know when a message is ready and fetch the message with _get() only when the user code is trying to fetch the message.
3. This would mean the Select implementation no longer be able to operate on async iterators like it used to, but that's probably a good thing, taking away some of its power.

[leandro-lucarella-frequenz]

OK, I think I can conclude that as_completed() does exactly what we want, except it doesn't return the finished task but its .result(), so we don't have a way to know which of all the tasks finished, so we can't re-schedule it (unless we start wrapping the tasks to return also some way to identify which one finished as part of the .result()).

Another option would be to implement our own as_completed(), the code is not that complicated, which returns the task instead of the .result(): https://github.com/python/cpython/blob/db115682bd639a2642c617f0b7d5b30cd7d7f472/Lib/asyncio/tasks.py#L561-L615

This approach is not making the code simpler, but an advantage over this approach, IMHO, is that we keep the complexity in select() only, and receivers can still be simple async iterators, which for me it makes a lot of sense, and avoid people wanting to provide new channels to worry about implementing readiness primitives.

I gave it a shot by doing the mapping of the done coroutine returned by as_completed() and our pending items, and the code is greatly simplified but for some reason it is not working completely as expected.

#46

[leandro-lucarella-frequenz]

yes, but you are operating on the same tasks a1 and a2 after you break out of the loop. But in the select implementation, a1 and a2 will be owned by select, and when we break out of the loop, the task objects get lost and hence the values they hold.

Not if we break the loop but keep the Select object alive. But yeah, I see your points, that would lead to a select() function implementation that it is not correct. I see now we have another layer of complexity, which is we are consuming from the async iterators in a function running inside a task, so if the task already consumed the value and we cancel the task for some reason, then the value will be lost too. The alternative I presented above suffers from this issue too.

So I think you finally convinced me that we can't have both a nice simple and safe async iterator interface and not introduce the ready test to receivers.

I really hate moving the complexity to channels implementators, but I guess there is no way around that is not moving complexity to users of select(), which is even worse.

[leandro-lucarella-frequenz]

This looks good. I like it. It's also not that bad in terms of complexity as it only splits the work that needed to be done by receive(), so I'm less worried about complexity for channels implementors too.

Before this PR we also had the issue about potential messages being lost when breaking in Merge/MergeNamed, right?

This also needs a RELEASE_NOTES entry!

In a followup PR I think we should definitely go for the async iterable interface for Select.ready(), it will simplify the implementation too (as we won't need the extra class for lazy evaluation for example).

[leandro-lucarella-frequenz]

Needs a couple of documentation fixes and it is ready to go! Feel free to merge after those fixes.