Fix in-process kernels

ipykernel/iostream.py

@@ -18,7 +18,7 @@
 
 import zmq
 from jupyter_client.session import extract_header
-from zmq.eventloop.zmqstream import ZMQStream
+from .zmqstream import ZMQStream
 
 # -----------------------------------------------------------------------------
 # Globals
@@ -184,7 +184,7 @@ def stop(self):
         if not self.thread.is_alive():
             return
         self.io_loop.call_soon_threadsafe(self.io_loop.stop)
-        self.thread.join()
+        # self.thread.join()
         # close *all* event pipes, created in any thread
         # event pipes can only be used from other threads while self.thread.is_alive()
         # so after thread.join, this should be safe

ipykernel/ipkernel.py

@@ -11,7 +11,6 @@
 from IPython.core import release
 from IPython.utils.tokenutil import line_at_cursor, token_at_cursor
 from traitlets import Any, Bool, Instance, List, Type, observe, observe_compat
-from zmq.eventloop.zmqstream import ZMQStream
 
 from .comm import CommManager
 from .compiler import XCachingCompiler
@@ -20,6 +19,7 @@
 from .kernelbase import Kernel as KernelBase
 from .kernelbase import _accepts_cell_id
 from .zmqshell import ZMQInteractiveShell
+from .zmqstream import ZMQStream
 
 try:
     from IPython.core.interactiveshell import _asyncio_runner

ipykernel/kernelapp.py

@@ -44,7 +44,6 @@
 )
 from traitlets.utils import filefind
 from traitlets.utils.importstring import import_item
-from zmq.eventloop.zmqstream import ZMQStream
 
 from .control import ControlThread
 from .heartbeat import Heartbeat

ipykernel/kernelbase.py

@@ -78,7 +78,7 @@ def _update_eventloop(self, change):
         """schedule call to eventloop from IOLoop"""
         loop = asyncio.get_event_loop()
         if change.new is not None:
-            loop.call_soon(self.enter_eventloop())
+            loop.call_soon(self.enter_eventloop)
 
     session = Instance(Session, allow_none=True)
     profile_dir = Instance("IPython.core.profiledir.ProfileDir", allow_none=True)

ipykernel/zmqstream.py

@@ -1,4 +1,5 @@
 import asyncio
+from inspect import isawaitable
 
 
 class ZMQStream:
@@ -21,7 +22,10 @@ def on_recv(self, callback, copy=True):
 
     async def recv_task(self, callback, copy):
         while True:
-            msg_list = await self.socket.recv_multipart(copy=copy)
+            msg_list = self.socket.recv_multipart(copy=copy)
+            if isawaitable(msg_list):
+                # in-process kernels are not async
+                msg_list = await msg_list
             callback(msg_list)
 
     def send_multipart(self, msg_list, flags=0, copy=True, track=False, **kwargs):