Send the shell reply through the ZMQStream instead of raw on its socket

On Windows, ipykernel 7 intermittently drops an execute_request on the
shell channel: the kernel goes idle and never replies, and the client
times out waiting for execute_reply (historically ~30% of headless
notebook runs in our measurements; which cell hangs wanders run to run).

Root cause: the shell ROUTER socket is dual-use on the shell-channel
thread. A ZMQStream reads execute_requests off it, while replies are sent
back over the SAME socket out-of-band via a raw send_multipart in
SubshellManager._send_on_shell_channel. That out-of-band send drains the
socket's edge-triggered ZMQ_FD read edge (a documented libzmq corollary:
after zmq_send the socket may become readable without a new edge). Because
the send is not ZMQStream-mediated, the stream is never re-armed and a
request that arrived concurrently strands unread on a registered-but-
non-readable fd. The strand is terminal: no later arrival re-edges it.

Fix: send the reply through shell_stream.send_multipart rather than raw on
shell_socket. This keeps the ZMQStream the sole user of the socket: the
send is serviced by the stream's own _handle_events, which recvs any
concurrently-queued request first and then re-arms POLLIN via
_rebuild_io_state, so the request cannot strand. It removes the root cause
rather than re-arming after it, and needs no reach into the private
_handle_events. The shell_stream (built in kernelapp.init_kernel) is
threaded through ShellChannelThread into SubshellManager so the reply path
can reach it, and falls back to the raw socket if no stream was threaded in.

Verified by the regression test added next, which reproduces the strand
precondition deterministically and asserts the queued request is still
delivered once the reply is routed through the stream.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: BoykoNeov

ipykernel/kernelapp.py

@@ -608,6 +608,9 @@ def init_kernel(self):
         """Create the Kernel object itself"""
         if self.shell_channel_thread:
             shell_stream = ZMQStream(self.shell_socket, self.shell_channel_thread.io_loop)
+            # Hand the stream to the shell-channel thread so SubshellManager can send the
+            # out-of-band reply through the stream rather than raw on the socket (the wedge fix).
+            self.shell_channel_thread.shell_stream = shell_stream
         else:
             shell_stream = ZMQStream(self.shell_socket)
         control_stream = ZMQStream(self.control_socket, self.control_thread.io_loop)

ipykernel/shellchannel.py

@@ -29,6 +29,10 @@ def __init__(
         self._manager: SubshellManager | None = None
         self._zmq_context = context  # Avoid use of self._context
         self._shell_socket = shell_socket
+        # Set by kernelapp.init_kernel after it builds the shell ZMQStream (this thread
+        # is created before the stream). Threaded into SubshellManager so the out-of-band
+        # reply is sent through the stream rather than raw on the socket (the wedge fix).
+        self.shell_stream = None
         # Record the parent thread - the thread that started the app (usually the main thread)
         self.parent_thread = current_thread()
 
@@ -43,6 +47,7 @@ def manager(self) -> SubshellManager:
                 self._zmq_context,
                 self.io_loop,
                 self._shell_socket,
+                self.shell_stream,
             )
         return self._manager
 

ipykernel/subshell_manager.py

@@ -40,13 +40,19 @@ def __init__(
         context: zmq.Context[t.Any],
         shell_channel_io_loop: IOLoop,
         shell_socket: zmq.Socket[t.Any],
+        shell_stream=None,
     ):
         """Initialize the subshell manager."""
         self._parent_thread = current_thread()
 
         self._context: zmq.Context[t.Any] = context
         self._shell_channel_io_loop = shell_channel_io_loop
         self._shell_socket = shell_socket
+        # ZMQStream that reads `shell_socket`. The out-of-band reply below is sent through
+        # this stream rather than raw on the socket: a raw send would drain the socket's
+        # edge-triggered ZMQ_FD read edge without the stream re-arming, stranding a
+        # concurrently-arrived request unread (the wedge).
+        self._shell_stream = shell_stream
         self._cache: dict[str, SubshellThread] = {}
         self._lock_cache = Lock()  # Sync lock across threads when accessing cache.
 
@@ -225,7 +231,19 @@ def _process_control_request(
 
     def _send_on_shell_channel(self, msg) -> None:
         assert current_thread().name == SHELL_CHANNEL_THREAD_NAME
-        self._shell_socket.send_multipart(msg)
+        # Send the reply through the shell ZMQStream rather than raw on its socket. A raw
+        # send_multipart on the dual-use shell ROUTER drains its edge-triggered ZMQ_FD read
+        # edge; because the stream never sees that send, it is never re-armed, so a request
+        # that arrived concurrently can strand unread on a registered-but-non-readable fd
+        # (the wedge). Routing the send through the stream keeps the stream the sole user of
+        # the socket: the send is serviced by the stream's own _handle_events, which recvs
+        # any pending request first and then re-arms POLLIN via _rebuild_io_state, so the
+        # request cannot strand. Fall back to the raw socket if no stream was threaded in.
+        stream = self._shell_stream
+        if stream is not None and stream.socket is self._shell_socket:
+            stream.send_multipart(msg)
+        else:
+            self._shell_socket.send_multipart(msg)
 
     def _stop_subshell(self, subshell_thread: SubshellThread) -> None:
         """Stop a subshell thread and close all of its resources."""