Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 10 additions & 14 deletions ipykernel/kernelbase.py
Original file line number Diff line number Diff line change
Expand Up @@ -331,10 +331,7 @@ async def process_control(self, msg):

sys.stdout.flush()
sys.stderr.flush()
self._publish_status("idle", "control")
# flush to ensure reply is sent
if self.control_stream:
self.control_stream.flush(zmq.POLLOUT)
self._publish_status_and_flush("idle", "control", self.control_stream)

def should_handle(self, stream, msg, idents):
"""Check whether a shell-channel message should be handled
Expand Down Expand Up @@ -370,11 +367,7 @@ async def dispatch_shell(self, msg):
# Only abort execute requests
if self._aborting and msg_type == "execute_request":
self._send_abort_reply(self.shell_stream, msg, idents)
self._publish_status("idle", "shell")
# flush to ensure reply is sent before
# handling the next request
if self.shell_stream:
self.shell_stream.flush(zmq.POLLOUT)
self._publish_status_and_flush("idle", "shell", self.shell_stream)
return

# Print some info about this message and leave a '--->' marker, so it's
Expand All @@ -384,6 +377,7 @@ async def dispatch_shell(self, msg):
self.log.debug(" Content: %s\n --->\n ", msg["content"])

if not self.should_handle(self.shell_stream, msg, idents):
self._publish_status_and_flush("idle", "shell", self.shell_stream)
return

handler = self.shell_handlers.get(msg_type, None)
Expand Down Expand Up @@ -412,11 +406,7 @@ async def dispatch_shell(self, msg):

sys.stdout.flush()
sys.stderr.flush()
self._publish_status("idle", "shell")
# flush to ensure reply is sent before
# handling the next request
if self.shell_stream:
self.shell_stream.flush(zmq.POLLOUT)
self._publish_status_and_flush("idle", "shell", self.shell_stream)

def pre_handler_hook(self):
"""Hook to execute before calling message handler"""
Expand Down Expand Up @@ -603,6 +593,12 @@ def _publish_status(self, status, channel, parent=None):
ident=self._topic("status"),
)

def _publish_status_and_flush(self, status, channel, stream, parent=None):
"""send status on IOPub and flush specified stream to ensure reply is sent before handling the next reply"""
self._publish_status(status, channel, parent)
if stream:
stream.flush(zmq.POLLOUT)

def _publish_debug_event(self, event):
if not self.session:
return
Expand Down