debug

Author: ianthomas23

ipykernel/debugger.py

@@ -19,6 +19,8 @@
 
 from .compiler import get_file_name, get_tmp_directory, get_tmp_hash_seed
 
+import threading
+
 try:
     # This import is required to have the next ones working...
     from debugpy.server import api  # noqa: F401
@@ -420,6 +422,10 @@ def start(self):
             code = "import debugpy;"
             code += 'debugpy.listen(("' + host + '",' + port + "))"
             content = {"code": code, "silent": True}
+
+            with open("debug.txt", "a") as f:
+                f.write(f"{threading.current_thread().ident} shell_socket execute_request\n")
+
             self.session.send(
                 self.shell_socket,
                 "execute_request",

ipykernel/displayhook.py

@@ -15,6 +15,9 @@
 from ipykernel.jsonutil import encode_images, json_clean
 
 
+import threading
+
+
 class ZMQDisplayHook:
     """A simple displayhook that publishes the object's repr over a ZeroMQ
     socket."""
@@ -44,6 +47,10 @@ def __call__(self, obj):
             "data": {"text/plain": repr(obj)},
             "metadata": {},
         }
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} pub_socket execute_result\n")
+
         self.session.send(
             self.pub_socket, "execute_result", contents, parent=self.parent_header, ident=self.topic
         )
@@ -97,5 +104,9 @@ def finish_displayhook(self):
         sys.stdout.flush()
         sys.stderr.flush()
         if self.msg and self.msg["content"]["data"] and self.session:
+
+            with open("debug.txt", "a") as f:
+                f.write(f"{threading.current_thread().ident} pub_socket ?finish_displayhook\n")
+
             self.session.send(self.pub_socket, self.msg, ident=self.topic)
         self.msg = None

ipykernel/iostream.py

@@ -78,9 +78,15 @@ def __init__(self, socket, pipe=False):
         self.thread.is_pydev_daemon_thread = True  # type:ignore[attr-defined]
         self.thread.name = "IOPub"
 
+        with open("debug.txt", "a") as f:
+            f.write(f"    IOPubThread.__init__\n")
+
     def _thread_main(self):
         """The inner loop that's actually run in a thread"""
 
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} IOPubThread._thread_main\n")
+
         def _start_event_gc():
             self._event_pipe_gc_task = asyncio.ensure_future(self._run_event_pipe_gc())
 
@@ -646,6 +652,9 @@ def _flush(self):
                     if msg is None:
                         return
 
+                with open("debug.txt", "a") as f:
+                    f.write(f"{threading.current_thread().ident} pub_thread ?OutStream.flush\n")
+
                 self.session.send(
                     self.pub_thread,
                     msg,

ipykernel/kernelbase.py

@@ -79,7 +79,6 @@ def _accepts_parameters(meth, param_names):
 
 class Kernel(SingletonConfigurable):
     """The base kernel class."""
-
     # ---------------------------------------------------------------------------
     # Kernel interface
     # ---------------------------------------------------------------------------
@@ -277,6 +276,12 @@ def __init__(self, **kwargs):
         """Initialize the kernel."""
         super().__init__(**kwargs)
 
+
+
+        self._iant_lock = threading.Lock()
+
+
+
         # Kernel application may swap stdout and stderr to OutStream,
         # which is the case in `IPKernelApp.init_io`, hence `sys.stdout`
         # can already by different from TextIO at initialization time.
@@ -571,6 +576,10 @@ def schedule_dispatch(self, dispatch, *args):
 
     def start(self):
         """register dispatchers for streams"""
+
+        with open("debug.txt", "a") as f:
+            f.write(f"--- kernelbase start --- {threading.main_thread().ident}\n")
+
         self.io_loop = ioloop.IOLoop.current()
         self.msg_queue: Queue[t.Any] = Queue()
         if not self.shell_channel_thread:
@@ -609,6 +618,9 @@ async def shell_channel_thread_main(self, msg):
             msg3 = self.session.deserialize(msg2, content=False, copy=False)
             subshell_id = msg3["header"].get("subshell_id")
 
+            with open("debug.txt", "a") as f:
+                f.write(f"{threading.current_thread().ident} shell_channel_thread_main msg received for {subshell_id}\n")
+
             # Find inproc pair socket to use to send message to correct subshell.
             subshell_manager = self.shell_channel_thread.manager
             socket = subshell_manager.get_shell_channel_stream(subshell_id)
@@ -622,6 +634,10 @@ async def shell_channel_thread_main(self, msg):
 
     async def shell_main(self, subshell_id: str | None, msg):
         """Handler of shell messages for a single subshell"""
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} shell_main msg recvd on {subshell_id}\n")
+
         if self._supports_kernel_subshells:
             if subshell_id is None:
                 assert threading.current_thread() == threading.main_thread()
@@ -667,25 +683,35 @@ def _publish_execute_input(self, code, parent, execution_count):
         """Publish the code request on the iopub stream."""
         if not self.session:
             return
-        self.session.send(
-            self.iopub_socket,
-            "execute_input",
-            {"code": code, "execution_count": execution_count},
-            parent=parent,
-            ident=self._topic("execute_input"),
-        )
+        with self._iant_lock:
+
+            with open("debug.txt", "a") as f:
+                f.write(f"{threading.current_thread().ident} iopub_socket execute_input\n")
+
+            self.session.send(
+                self.iopub_socket,
+                "execute_input",
+                {"code": code, "execution_count": execution_count},
+                parent=parent,
+                ident=self._topic("execute_input"),
+            )
 
     def _publish_status(self, status, channel, parent=None):
         """send status (busy/idle) on IOPub"""
         if not self.session:
             return
-        self.session.send(
-            self.iopub_socket,
-            "status",
-            {"execution_state": status},
-            parent=parent or self.get_parent(channel),
-            ident=self._topic("status"),
-        )
+        with self._iant_lock:
+
+            with open("debug.txt", "a") as f:
+                f.write(f"{threading.current_thread().ident} iopub_socket status {status}\n")
+
+            self.session.send(
+                self.iopub_socket,
+                "status",
+                {"execution_state": status},
+                parent=parent or self.get_parent(channel),
+                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"""
@@ -696,13 +722,18 @@ def _publish_status_and_flush(self, status, channel, stream, parent=None):
     def _publish_debug_event(self, event):
         if not self.session:
             return
-        self.session.send(
-            self.iopub_socket,
-            "debug_event",
-            event,
-            parent=self.get_parent(),
-            ident=self._topic("debug_event"),
-        )
+        with self._iant_lock:
+
+            with open("debug.txt", "a") as f:
+                f.write(f"{threading.current_thread().ident} iopub_socket debug_event\n")
+
+            self.session.send(
+                self.iopub_socket,
+                "debug_event",
+                event,
+                parent=self.get_parent(),
+                ident=self._topic("debug_event"),
+            )
 
     def set_parent(self, ident, parent, channel="shell"):
         """Set the current parent request
@@ -763,6 +794,10 @@ def send_response(
         """
         if not self.session:
             return None
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} ? ?send_response\n")
+
         return self.session.send(
             stream,
             msg_or_type,
@@ -835,6 +870,13 @@ async def execute_request(self, stream, ident, parent):
         if self._do_exec_accepted_params["cell_id"]:
             do_execute_args["cell_id"] = cell_id
 
+
+        subshell_id = parent["header"].get("subshell_id")
+        msg_id = parent["header"].get("msg_id")
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} about to execute_request {msg_id} {subshell_id} {code}\n")
+
         # Call do_execute with the appropriate arguments
         reply_content = self.do_execute(**do_execute_args)
 
@@ -854,6 +896,9 @@ async def execute_request(self, stream, ident, parent):
         reply_content = json_clean(reply_content)
         metadata = self.finish_metadata(parent, metadata, reply_content)
 
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} execute_reply            {msg_id} {subshell_id}\n")
+
         reply_msg: dict[str, t.Any] = self.session.send(  # type:ignore[assignment]
             stream,
             "execute_reply",
@@ -943,6 +988,10 @@ async def history_request(self, stream, ident, parent):
             reply_content = await reply_content
 
         reply_content = json_clean(reply_content)
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} ? history_reply\n")
+
         msg = self.session.send(stream, "history_reply", reply_content, parent, ident)
         self.log.debug("%s", msg)
 
@@ -967,6 +1016,10 @@ async def connect_request(self, stream, ident, parent):
             return
         content = self._recorded_ports.copy() if self._recorded_ports else {}
         content["status"] = "ok"
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} ? connect_reply\n")
+
         msg = self.session.send(stream, "connect_reply", content, parent, ident)
         self.log.debug("%s", msg)
 
@@ -991,6 +1044,10 @@ async def kernel_info_request(self, stream, ident, parent):
             return
         content = {"status": "ok"}
         content.update(self.kernel_info)
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} ? kernel_info_reply\n")
+
         msg = self.session.send(stream, "kernel_info_reply", content, parent, ident)
         self.log.debug("%s", msg)
 
@@ -1058,6 +1115,10 @@ async def shutdown_request(self, stream, ident, parent):
         content = self.do_shutdown(parent["content"]["restart"])
         if inspect.isawaitable(content):
             content = await content
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} ? shutdown_reply\n")
+
         self.session.send(stream, "shutdown_reply", content, parent, ident=ident)
         # same content, but different msg_id for broadcasting on IOPub
         self._shutdown_message = self.session.msg("shutdown_reply", content, parent)
@@ -1108,6 +1169,10 @@ async def debug_request(self, stream, ident, parent):
         if inspect.isawaitable(reply_content):
             reply_content = await reply_content
         reply_content = json_clean(reply_content)
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} ? debug_reply\n")
+
         reply_msg = self.session.send(stream, "debug_reply", reply_content, parent, ident)
         self.log.debug("%s", reply_msg)
 
@@ -1177,6 +1242,10 @@ async def create_subshell_request(self, socket, ident, parent) -> None:
         other_socket = self.shell_channel_thread.manager.get_control_other_socket()
         other_socket.send_json({"type": "create"})
         reply = other_socket.recv_json()
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} ? create_subshell_reply\n")
+
         self.session.send(socket, "create_subshell_reply", reply, parent, ident)
 
     async def delete_subshell_request(self, socket, ident, parent) -> None:
@@ -1200,6 +1269,10 @@ async def delete_subshell_request(self, socket, ident, parent) -> None:
         other_socket = self.shell_channel_thread.manager.get_control_other_socket()
         other_socket.send_json({"type": "delete", "subshell_id": subshell_id})
         reply = other_socket.recv_json()
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} ? delete_subshell_reply\n")
+
         self.session.send(socket, "delete_subshell_reply", reply, parent, ident)
 
     async def list_subshell_request(self, socket, ident, parent) -> None:
@@ -1216,6 +1289,10 @@ async def list_subshell_request(self, socket, ident, parent) -> None:
         other_socket = self.shell_channel_thread.manager.get_control_other_socket()
         other_socket.send_json({"type": "list"})
         reply = other_socket.recv_json()
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} ? list_subshell_reply\n")
+
         self.session.send(socket, "list_subshell_reply", reply, parent, ident)
 
     # ---------------------------------------------------------------------------
@@ -1280,6 +1357,10 @@ async def abort_request(self, stream, ident, parent):  # pragma: no cover
         content = dict(status="ok")
         if not self.session:
             return
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} ? abort_reply\n")
+
         reply_msg = self.session.send(
             stream, "abort_reply", content=content, parent=parent, ident=ident
         )
@@ -1378,6 +1459,9 @@ def _send_abort_reply(self, stream, msg, idents):
         md = self.finish_metadata(msg, md, status)
         md.update(status)
 
+        with open("debug.txt", "a") as f:
+            f.write(f"{threading.current_thread().ident} ? {reply_type}\n")
+
         self.session.send(
             stream,
             reply_type,
@@ -1558,11 +1642,16 @@ async def _at_shutdown(self):
 
         finally:
             if self._shutdown_message is not None and self.session:
-                self.session.send(
-                    self.iopub_socket,
-                    self._shutdown_message,
-                    ident=self._topic("shutdown"),
-                )
+                with self._iant_lock:
+
+                    with open("debug.txt", "a") as f:
+                        f.write(f"{threading.current_thread().ident} ? _shutdown\n")
+
+                    self.session.send(
+                        self.iopub_socket,
+                        self._shutdown_message,
+                        ident=self._topic("shutdown"),
+                    )
                 self.log.debug("%s", self._shutdown_message)
             if self.control_stream:
                 self.control_stream.flush(zmq.POLLOUT)

ipykernel/subshell_manager.py

@@ -223,6 +223,10 @@ def _process_control_request(
 
     def _send_on_shell_channel(self, msg) -> None:
         assert current_thread().name == SHELL_CHANNEL_THREAD_NAME
+
+        with open("debug.txt", "a") as f:
+            f.write(f"{current_thread().ident} _send_on_shell_channel {msg}\n")
+
         with self._lock_shell_socket:
             self._shell_socket.send_multipart(msg)