End of channel fixes for streams and broadcasting #278
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Without this wakeup you can have tasks which re-enter `.receive()` and get stuck waiting on the wakeup event indefinitely. Whenever a ``trio.EndOfChannel`` arrives we want to make sure all consumers at least know about it and don't block. This previous behaviour was basically a bug. Add some state flags for tracking if the broadcaster was either cancelled or terminated via EOC mostly for testing and debugging purposes though this info might be useful if we decide to offer a `.statistics()` like API in the future.
This actually catches a lot of bugs to do with stream termination and ``MsgStream.subscribe()`` usage where the underlying stream closes from the producer side. When this passes the broadcaster logic will have to ensure non-lossy fan out semantics and closure tracking.
Avoids the issue noted in python-trio/trio-typing#50 to keep CI green.
goodboy
force-pushed
the
end_of_channel_fixes
branch
from
f83d6c4 to
325e550
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This patch set is factored out of #257 and corrects a bunch of previously unknown bugs to do with stream closure and broadcast receiver usage around streams as provided by
MsgStream.subscribe().I decided to pull the work out of the other PR because this is notably different work despite discovering it via the original cached-inter-actor stream test that found it.
Summary:
MsgStream._closed: boolfor internal closure tracking and raisetrio.ClosedResourceErrorin.send()if setMsgStream._eoc: boolinside.aclose()._eocwhen a'stop'message is received.eoc: booland.cancelled: booltracking insideBroadcastState, set these flags on handling of the appropriatetrioexceptions as well as wake all broadcast consumers on each signal to avoid hangs when a wrapped recv channel has closed viatrio.EndOfChannel.subscribe()