Subshells implemented using tornado event loops

Author: ianthomas23

ipykernel/control.py

@@ -1,32 +1,11 @@
 """A thread for a control channel."""
-from threading import Thread
 
-from tornado.ioloop import IOLoop
+from .thread import CONTROL_THREAD_NAME, BaseThread
 
-CONTROL_THREAD_NAME = "Control"
 
-
-class ControlThread(Thread):
+class ControlThread(BaseThread):
     """A thread for a control channel."""
 
     def __init__(self, **kwargs):
         """Initialize the thread."""
-        Thread.__init__(self, name=CONTROL_THREAD_NAME, **kwargs)
-        self.io_loop = IOLoop(make_current=False)
-        self.pydev_do_not_trace = True
-        self.is_pydev_daemon_thread = True
-
-    def run(self):
-        """Run the thread."""
-        self.name = CONTROL_THREAD_NAME
-        try:
-            self.io_loop.start()
-        finally:
-            self.io_loop.close()
-
-    def stop(self):
-        """Stop the thread.
-
-        This method is threadsafe.
-        """
-        self.io_loop.add_callback(self.io_loop.stop)
+        super().__init__(name=CONTROL_THREAD_NAME, **kwargs)

ipykernel/ipkernel.py

@@ -361,6 +361,11 @@ def set_sigint_result():
             # restore the previous sigint handler
             signal.signal(signal.SIGINT, save_sigint)
 
+    @contextmanager
+    def _dummy_context_manager(self, *args):
+        # Signals only work in main thread, so cannot use _cancel_on_sigint in subshells.
+        yield
+
     async def execute_request(self, stream, ident, parent):
         """Override for cell output - cell reconciliation."""
         parent_header = extract_header(parent)
@@ -439,7 +444,12 @@ async def run_cell(*args, **kwargs):
 
                 coro_future = asyncio.ensure_future(coro)
 
-                with self._cancel_on_sigint(coro_future):
+                cm = (
+                    self._cancel_on_sigint
+                    if threading.current_thread() == threading.main_thread()
+                    else self._dummy_context_manager
+                )
+                with cm(coro_future):  # type:ignore[operator]
                     res = None
                     try:
                         res = await coro_future

ipykernel/kernelapp.py

@@ -53,6 +53,7 @@
 from .iostream import IOPubThread
 from .ipkernel import IPythonKernel
 from .parentpoller import ParentPollerUnix, ParentPollerWindows
+from .shellchannel import ShellChannelThread
 from .zmqshell import ZMQInteractiveShell
 
 # -----------------------------------------------------------------------------
@@ -143,6 +144,7 @@ class IPKernelApp(BaseIPythonApplication, InteractiveShellApp, ConnectionFileMix
     iopub_socket = Any()
     iopub_thread = Any()
     control_thread = Any()
+    shell_channel_thread = Any()
 
     _ports = Dict()
 
@@ -367,6 +369,11 @@ def init_control(self, context):
             self.control_socket.router_handover = 1
 
         self.control_thread = ControlThread(daemon=True)
+        self.shell_channel_thread = ShellChannelThread(
+            context,
+            self.shell_socket,
+            daemon=True,
+        )
 
     def init_iopub(self, context):
         """Initialize the iopub channel."""
@@ -406,6 +413,10 @@ def close(self):
             self.log.debug("Closing control thread")
             self.control_thread.stop()
             self.control_thread.join()
+        if self.shell_channel_thread and self.shell_channel_thread.is_alive():
+            self.log.debug("Closing shell channel thread")
+            self.shell_channel_thread.stop()
+            self.shell_channel_thread.join()
 
         if self.debugpy_socket and not self.debugpy_socket.closed:
             self.debugpy_socket.close()
@@ -546,10 +557,17 @@ def init_signal(self):
 
     def init_kernel(self):
         """Create the Kernel object itself"""
-        shell_stream = ZMQStream(self.shell_socket)
+        if self.shell_channel_thread:
+            shell_stream = ZMQStream(self.shell_socket, self.shell_channel_thread.io_loop)
+        else:
+            shell_stream = ZMQStream(self.shell_socket)
         control_stream = ZMQStream(self.control_socket, self.control_thread.io_loop)
         debugpy_stream = ZMQStream(self.debugpy_socket, self.control_thread.io_loop)
+
         self.control_thread.start()
+        if self.shell_channel_thread:
+            self.shell_channel_thread.start()
+
         kernel_factory = self.kernel_class.instance  # type:ignore[attr-defined]
 
         kernel = kernel_factory(
@@ -560,6 +578,7 @@ def init_kernel(self):
             debug_shell_socket=self.debug_shell_socket,
             shell_stream=shell_stream,
             control_thread=self.control_thread,
+            shell_channel_thread=self.shell_channel_thread,
             iopub_thread=self.iopub_thread,
             iopub_socket=self.iopub_socket,
             stdin_socket=self.stdin_socket,