debug

Author: ianthomas23

ipykernel/kernelbase.py

@@ -277,7 +277,7 @@ def __init__(self, **kwargs):
         """Initialize the kernel."""
         super().__init__(**kwargs)
 
-        self._iant_lock = threading.Lock()
+        #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`
@@ -682,33 +682,33 @@ 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",
-                {"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
-        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"),
-            )
+        #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"""
@@ -719,17 +719,17 @@ 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",
-                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
@@ -895,7 +895,7 @@ async def execute_request(self, stream, ident, parent):
 
         with open("debug.txt", "a") as f:
             f.write(
-                f"{threading.current_thread().ident} execute_reply            {msg_id} {subshell_id}\n"
+                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]
@@ -1234,6 +1234,11 @@ 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.
@@ -1641,15 +1646,15 @@ 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,
-                        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

@@ -225,11 +225,16 @@ 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")
+            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

tests/test_subshells.py

@@ -20,7 +20,7 @@ def create_subshell_helper(kc: BlockingKernelClient):
     kc.control_channel.send(msg)
     msg_id = msg["header"]["msg_id"]
     reply = get_reply(kc, msg_id, TIMEOUT, channel="control")
-    wait_for_idle(kc)
+    #wait_for_idle(kc, msg_id)
     return reply["content"]
 
 
@@ -29,7 +29,7 @@ def delete_subshell_helper(kc: BlockingKernelClient, subshell_id: str):
     kc.control_channel.send(msg)
     msg_id = msg["header"]["msg_id"]
     reply = get_reply(kc, msg_id, TIMEOUT, channel="control")
-    wait_for_idle(kc)
+    #wait_for_idle(kc, msg_id)
     return reply["content"]
 
 
@@ -38,7 +38,7 @@ def list_subshell_helper(kc: BlockingKernelClient):
     kc.control_channel.send(msg)
     msg_id = msg["header"]["msg_id"]
     reply = get_reply(kc, msg_id, TIMEOUT, channel="control")
-    wait_for_idle(kc)
+    #wait_for_idle(kc, msg_id)
     return reply["content"]
 
 
@@ -52,9 +52,25 @@ def execute_request(kc: BlockingKernelClient, code: str, subshell_id: str | None
 def execute_request_subshell_id(
     kc: BlockingKernelClient, code: str, subshell_id: str | None, terminator: str = "\n"
 ):
+
+    with open("debug.txt", "a") as f:
+        f.write("CHECK 1\n")
+
     msg = execute_request(kc, code, subshell_id)
+
+    with open("debug.txt", "a") as f:
+        f.write("CHECK 2\n")
+
     msg_id = msg["header"]["msg_id"]
+
+    with open("debug.txt", "a") as f:
+        f.write("CHECK 3\n")
+
     stdout, _ = assemble_output(kc.get_iopub_msg, None, msg_id)
+
+    with open("debug.txt", "a") as f:
+        f.write("CHECK 4\n")
+
     return stdout.strip()
 
 
@@ -110,7 +126,7 @@ def test_thread_ids():
         thread_id, main_thread_id = execute_thread_ids(kc)
         assert thread_id == main_thread_id
 
-        thread_id, main_thread_id = execute_thread_ids(kc, subshell_id)
+        thread_id, main_thread_id = execute_thread_ids(kc, subshell_id)   # This is the problem
         assert thread_id != main_thread_id
 
         delete_subshell_helper(kc, subshell_id)

tests/utils.py

@@ -178,6 +178,8 @@ def assemble_output(get_msg, timeout=1, parent_msg_id: str | None = None):
     stderr = ""
     while True:
         msg = get_msg(timeout=timeout)
+        #with open("debug.txt", "a") as f:
+        #    f.write(f"assembly output {msg}\n")
         msg_type = msg["msg_type"]
         content = msg["content"]
         if (