From c7dbb5c3b70f214be00be349011cfabca3dd1247 Mon Sep 17 00:00:00 2001 From: Piotr Czapla Date: Wed, 10 Jun 2026 15:50:09 +0200 Subject: [PATCH] fix: use socket .get() instead of .getsockopt() to prevent deaf loop This fixes a bug where getsockopt (which pyzmq aliases to SocketBase.get at class level) bypassed the async override in zmq.asyncio.Socket. That bypass drained the FD edge-trigger without rescheduling the asyncio loop, causing the shell channel reader to stall on subsequent replies. --- conkernelclient/core.py | 12 ++++++++---- nbs/00_core.ipynb | 12 ++++++++---- 2 files changed, 16 insertions(+), 8 deletions(-) diff --git a/conkernelclient/core.py b/conkernelclient/core.py index d10f48f..dfb607d 100644 --- a/conkernelclient/core.py +++ b/conkernelclient/core.py @@ -32,12 +32,16 @@ def _send(self, stream, msg_or_type, content=None, parent=None, ident=None, buffers=None, track=False, header=None, metadata=None): msg = self._orig_send(stream, msg_or_type, content=content, parent=parent, ident=ident, buffers=buffers, track=track, header=header, metadata=metadata) - # Force a sync, ensuring the send is fully registered internally - # Avoids a race where the lock releases, another thread immediately calls send(), - # and now 2 threads are interacting with the internal state before I/O thread has caught up + # Session.send sends via a SYNC shadow of the asyncio socket (zmq.Socket.shadow), which runs + # libzmq process_commands() and consumes the FD edge a buffered reply may have signalled — + # a recv parked in poll(None) on that edge would then sleep forever. Reading EVENTS through + # the async override (_AsyncSocket.get) makes pyzmq re-schedule _handle_events for pending + # recv/poll futures, compensating the consumed edge. It MUST be .get(): pyzmq aliases + # `getsockopt = SocketBase.get` at class level, so .getsockopt() bypasses the override + # and consumes the edge without rescheduling. if stream: if hasattr(stream, 'io_thread'): stream.io_thread.socket.get(zmq.EVENTS) - elif hasattr(stream, 'getsockopt'): stream.getsockopt(zmq.EVENTS) + elif hasattr(stream, 'get'): stream.get(zmq.EVENTS) return msg Session.send = _send diff --git a/nbs/00_core.ipynb b/nbs/00_core.ipynb index 2d96afb..419fec4 100644 --- a/nbs/00_core.ipynb +++ b/nbs/00_core.ipynb @@ -103,12 +103,16 @@ " buffers=None, track=False, header=None, metadata=None):\n", " msg = self._orig_send(stream, msg_or_type, content=content, parent=parent,\n", " ident=ident, buffers=buffers, track=track, header=header, metadata=metadata)\n", - " # Force a sync, ensuring the send is fully registered internally\n", - " # Avoids a race where the lock releases, another thread immediately calls send(),\n", - " # and now 2 threads are interacting with the internal state before I/O thread has caught up\n", + " # Session.send sends via a SYNC shadow of the asyncio socket (zmq.Socket.shadow), which runs\n", + " # libzmq process_commands() and consumes the FD edge a buffered reply may have signalled —\n", + " # a recv parked in poll(None) on that edge would then sleep forever. Reading EVENTS through\n", + " # the async override (_AsyncSocket.get) makes pyzmq re-schedule _handle_events for pending\n", + " # recv/poll futures, compensating the consumed edge. It MUST be .get(): pyzmq aliases\n", + " # `getsockopt = SocketBase.get` at class level, so .getsockopt() bypasses the override\n", + " # and consumes the edge without rescheduling.\n", " if stream:\n", " if hasattr(stream, 'io_thread'): stream.io_thread.socket.get(zmq.EVENTS)\n", - " elif hasattr(stream, 'getsockopt'): stream.getsockopt(zmq.EVENTS)\n", + " elif hasattr(stream, 'get'): stream.get(zmq.EVENTS)\n", " return msg\n", "\n", "Session.send = _send"