PR: Run GUI eventloop while waiting for input

ipykernel/eventloops.py

@@ -116,7 +116,7 @@ def loop_qt4(kernel):
     kernel.app = get_app_qt4([" "])
     kernel.app.setQuitOnLastWindowClosed(False)
 
-    for s in kernel.shell_streams:
+    for s in kernel.shell_streams + [kernel.stdin_stream]:
         _notify_stream_qt(kernel, s)
 
     _loop_qt(kernel.app)
@@ -159,7 +159,7 @@ def loop_wx(kernel):
 
     def wake():
         """wake from wx"""
-        for stream in kernel.shell_streams:
+        for stream in kernel.shell_streams + [kernel.stdin_stream]:
             if stream.flush(limit=1):
                 kernel.app.ExitMainLoop()
                 return
@@ -225,7 +225,7 @@ def process_stream_events(stream, *a, **kw):
     # For Tkinter, we create a Tk object and call its withdraw method.
     kernel.app = app = Tk()
     kernel.app.withdraw()
-    for stream in kernel.shell_streams:
+    for stream in kernel.shell_streams + [kernel.stdin_stream]:
         notifier = partial(process_stream_events, stream)
         # seems to be needed for tk
         notifier.__name__ = 'notifier'
@@ -296,7 +296,7 @@ def handle_int(etype, value, tb):
                 # don't let interrupts during mainloop invoke crash_handler:
                 sys.excepthook = handle_int
                 mainloop(kernel._poll_interval)
-                for stream in kernel.shell_streams:
+                for stream in kernel.shell_streams + [kernel.stdin_stream]:
                     if stream.flush(limit=1):
                         # events to process, return control to kernel
                         return
@@ -337,7 +337,7 @@ def process_stream_events(stream):
         if stream.flush(limit=1):
             loop.stop()
 
-    for stream in kernel.shell_streams:
+    for stream in kernel.shell_streams + [kernel.stdin_stream]:
         fd = stream.getsockopt(zmq.FD)
         notifier = partial(process_stream_events, stream)
         loop.add_reader(fd, notifier)

ipykernel/kernelbase.py

@@ -13,6 +13,7 @@
 import sys
 import time
 import uuid
+import threading
 
 try:
     # jupyter_client >= 5, use tz-aware now
@@ -40,6 +41,7 @@
 from jupyter_client.session import Session
 
 from ._version import kernel_protocol_version
+from .inprocess.socket import DummySocket
 
 CONTROL_PRIORITY = 1
 SHELL_PRIORITY = 10
@@ -170,6 +172,26 @@ def __init__(self, **kwargs):
         for msg_type in self.control_msg_types:
             self.control_handlers[msg_type] = getattr(self, msg_type)
 
+        self._stdin_msg = None
+        self._stdin_lock = threading.Lock()
+
+        if isinstance(self.stdin_socket, DummySocket):
+            # This is a test
+            self.stdin_stream = None
+            return
+        self.stdin_stream = ZMQStream(self.stdin_socket)
+
+        def handle_msg(msg):
+            idents, msg = self.session.feed_identities(msg, copy=False)
+            try:
+                msg = self.session.deserialize(msg, content=True, copy=False)
+            except Exception:
+                self.log.error("Invalid Message", exc_info=True)
+                return
+            self._stdin_msg = msg
+
+        self.stdin_stream.on_recv(handle_msg, copy=False)
+
     @gen.coroutine
     def dispatch_control(self, msg):
         """dispatch control requests"""
@@ -860,6 +882,7 @@ def raw_input(self, prompt=''):
         )
 
     def _input_request(self, prompt, ident, parent, password=False):
+        """Send an input request to the frontend and wait for the reply."""
         # Flush output before making the request.
         sys.stderr.flush()
         sys.stdout.flush()
@@ -873,22 +896,15 @@ def _input_request(self, prompt, ident, parent, password=False):
                 else:
                     raise
 
-        # Send the input request.
-        content = json_clean(dict(prompt=prompt, password=password))
-        self.session.send(self.stdin_socket, u'input_request', content, parent,
-                          ident=ident)
+        with self._stdin_lock:
+            self._stdin_msg = None
+            # Send the input request.
+            content = json_clean(dict(prompt=prompt, password=password))
+            self.session.send(self.stdin_socket, u'input_request', content, parent,
+                              ident=ident)
+            # Await a response.
+            reply = self._wait_input_request_reply()
 
-        # Await a response.
-        while True:
-            try:
-                ident, reply = self.session.recv(self.stdin_socket, 0)
-            except Exception:
-                self.log.warning("Invalid Message:", exc_info=True)
-            except KeyboardInterrupt:
-                # re-raise KeyboardInterrupt, to truncate traceback
-                raise KeyboardInterrupt
-            else:
-                break
         try:
             value = py3compat.unicode_to_str(reply['content']['value'])
         except:
@@ -899,6 +915,41 @@ def _input_request(self, prompt, ident, parent, password=False):
             raise EOFError
         return value
 
+    def _wait_input_request_reply(self):
+        """Wait for an input request reply.
+
+        Raises
+        ------
+        KeyboardInterrupt if a keyboard interrupt is recieved.
+        """
+        # Await a response.
+        reply = None
+        while reply is None:
+            try:
+                reply = self._input_request_loop_step()
+            except Exception:
+                self.log.warning("Invalid Message:", exc_info=True)
+            except KeyboardInterrupt:
+                # re-raise KeyboardInterrupt, to truncate traceback
+                raise KeyboardInterrupt
+        return reply
+
+    def _input_request_loop_step(self):
+        """Do one step of the input request loop."""
+        # Allow GUI event loop to update
+        if sys.version_info >= (3, 4):
+            is_main_thread = (threading.current_thread() is
+                              threading.main_thread())
+        else:
+            is_main_thread = isinstance(threading.current_thread(),
+                                        threading._MainThread)
+        if is_main_thread and self.eventloop:
+            self.eventloop(self)
+            return self._stdin_msg
+        else:
+            ident, reply = self.session.recv(self.stdin_socket, 0)
+            return reply
+
     def _at_shutdown(self):
         """Actions taken at shutdown by the kernel, called by python's atexit.
         """