Debug

Author: ianthomas23

ipykernel/debugger.py

@@ -17,6 +17,8 @@
 except ImportError:
     from jupyter_client.jsonutil import date_default as json_default
 
+import threading
+
 from .compiler import get_file_name, get_tmp_directory, get_tmp_hash_seed
 
 try:
@@ -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

@@ -6,6 +6,7 @@
 
 import builtins
 import sys
+import threading
 import typing as t
 
 from IPython.core.displayhook import DisplayHook
@@ -44,6 +45,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 +102,8 @@ 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("    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())
 
@@ -280,6 +286,9 @@ def _really_send(self, msg, *args, **kwargs):
         if self.closed:
             return
 
+        #with open("debug.txt", "a") as f:
+        #    f.write(f"{threading.current_thread().ident} _really_send iopub\n")
+
         mp_mode = self._check_mp_mode()
 
         if mp_mode != CHILD:
@@ -646,6 +655,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

@@ -277,6 +277,8 @@ 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 +573,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 +615,11 @@ 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 +633,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,6 +682,10 @@ def _publish_execute_input(self, code, parent, execution_count):
         """Publish the code request on the iopub stream."""
         if not self.session:
             return
+        #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",
@@ -679,6 +698,11 @@ def _publish_status(self, status, channel, parent=None):
         """send status (busy/idle) on IOPub"""
         if not self.session:
             return
+        #with self._iant_lock:
+        #with open("debug.txt", "a") as f:
+        #    f.write(f"{threading.current_thread().ident} iopub_socket status {status}\n")
+        #    f.write(f"SESSION {self.session}\n")
+
         self.session.send(
             self.iopub_socket,
             "status",
@@ -696,6 +720,10 @@ def _publish_status_and_flush(self, status, channel, stream, parent=None):
     def _publish_debug_event(self, event):
         if not self.session:
             return
+        #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",
@@ -763,6 +791,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 +867,14 @@ 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 +894,11 @@ 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} {reply_content}\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)
 
@@ -1170,13 +1235,20 @@ async def create_subshell_request(self, socket, ident, parent) -> None:
             self.log.error("Subshells are not supported by this kernel")
             return
 
+        #with open("debug.txt", "a") as f:
+        #    f.write(f"{threading.current_thread().ident} ? create_subshell_request\n")
+
         assert threading.current_thread().name == CONTROL_THREAD_NAME
 
         # This should only be called in the control thread if it exists.
         # Request is passed to shell channel thread to process.
         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 +1272,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 +1292,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 +1360,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 +1462,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,6 +1645,10 @@ async def _at_shutdown(self):
 
         finally:
             if self._shutdown_message is not None and self.session:
+                #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,

ipykernel/subshell_manager.py

@@ -223,9 +223,18 @@ 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 - start {msg}\n")
+
         with self._lock_shell_socket:
+            #with open("debug.txt", "a") as f:
+            #    f.write(f"   {self._shell_socket}\n")
             self._shell_socket.send_multipart(msg)
 
+        #with open("debug.txt", "a") as f:
+        #    f.write(f"{current_thread().ident} _send_on_shell_channel - end\n")
+
     def _stop_subshell(self, subshell_thread: SubshellThread) -> None:
         """Stop a subshell thread and close all of its resources."""
         assert current_thread().name == SHELL_CHANNEL_THREAD_NAME

ipykernel/thread.py

@@ -7,6 +7,9 @@
 SHELL_CHANNEL_THREAD_NAME = "Shell channel"
 
 
+import threading
+
+
 class BaseThread(Thread):
     """Base class for threads."""
 
@@ -20,6 +23,9 @@ def __init__(self, **kwargs):
     def run(self) -> None:
         """Run the thread."""
         try:
+            #with open("debug.txt", "a") as f:
+            #    f.write(f"{threading.current_thread().ident} io_loop - about to start\n")
+
             self.io_loop.start()
         finally:
             self.io_loop.close()

pyproject.toml

@@ -158,7 +158,8 @@ testpaths = [
     "tests",
 ]
 asyncio_mode = "auto"
-timeout = 300
+#timeout = 300
+timeout = 10
 # Restore this setting to debug failures
 #timeout_method = "thread"
 filterwarnings= [