flush control queue prior to handling shell messages #658
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
preserves control channel priority over shell channel - when control thread is running, resolves message-processing races - when control thread is not running, ensures control messages are processed before shell requests
| awaitable_future = asyncio.wrap_future(tracer_future) | ||
| else: | ||
| control_loop = self.io_loop | ||
| tracer_future = awaitable_future = asyncio.Future() |
There was a problem hiding this comment.
Nice way to make the flush awaitable across threads. I learned something.
|
This looks good to me! |
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.
preserves control channel priority over shell channel
Always flush the control queue before processing shell messages. When control thread wasn't running, shell messages would often get handled before control messages that had arrived before they started processing, due to how the queues are flushed from zmq to processing queues and the removal of synchronization in #585.
This adds a flush immediately before any shell message is handled, which does:
control_stream.flush()pops messages from zmq tocontrol_queuecontrol_queueto allow waiting for any pending messages to process (I first triedcontrol_queue.qsize(), but that doesn't work because it goes to 0 when control starts processing the last request, not when it finishes)This is much less of an issue when the control thread is running because control messages are usually quick, so the control queue is usually empty. So we could do this only when control thread is not running. However, that's not necessarily a safe assumption, and the behavior is now more correct in both cases, though it was usually fine with control_thread and always wrong without it.
with this, all ipyparallel tests are passing for me (#635)