Repository: jupyter/jupyter_client
Pull request #835 (2022-09-14 00:46:09)

Remove nest-asyncio dependency
- Remove the need for `nest_asyncio` by only calling asynchronous methods on self from within an asynchronous method and using the task runner method described in https://github.com/jupyter/jupyter_client/issues/755

- The blocking channel/client/manager methods call `run_sync` to run ad-hoc and background loop(s) as needed, and use `zmq.Context`.  The async channel/client/manager use `async/await` and `zmq.asyncio.Context`.   

- For session objects, we make all of the methods synchronous and check for futures returned by socket methods, converting them to values as needed.  We had to subclass `ZMQStream` to do the same, but perhaps this logic should go in `pyzmq` itself.

- Add tests for async and threaded clients


Patch:
diff --git a/.github/workflows/downstream.yml b/.github/workflows/downstream.yml
index 6b19e1c..1380478 100644
--- a/.github/workflows/downstream.yml
+++ b/.github/workflows/downstream.yml
@@ -12,7 +12,7 @@ concurrency:
 jobs:
   ipykernel:
     runs-on: ubuntu-latest
-    timeout-minutes: 10
+    timeout-minutes: 15
     steps:
       - uses: actions/checkout@v2
       - uses: jupyterlab/maintainer-tools/.github/actions/base-setup@v1
@@ -23,7 +23,7 @@ jobs:
 
   nbclient:
     runs-on: ubuntu-latest
-    timeout-minutes: 10
+    timeout-minutes: 15
     steps:
       - uses: actions/checkout@v2
       - uses: jupyterlab/maintainer-tools/.github/actions/base-setup@v1
@@ -34,7 +34,7 @@ jobs:
 
   nbconvert:
     runs-on: ubuntu-latest
-    timeout-minutes: 10
+    timeout-minutes: 15
     steps:
       - uses: actions/checkout@v2
       - uses: jupyterlab/maintainer-tools/.github/actions/base-setup@v1
@@ -45,7 +45,7 @@ jobs:
 
   jupyter_server:
     runs-on: ubuntu-latest
-    timeout-minutes: 10
+    timeout-minutes: 15
     steps:
       - uses: actions/checkout@v2
       - uses: jupyterlab/maintainer-tools/.github/actions/base-setup@v1
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index 190b887..c2c47cb 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -76,7 +76,8 @@ jobs:
       - name: Run the tests on pypy and windows
         if: ${{ startsWith(matrix.python-version, 'pypy') || startsWith(matrix.os, 'windows') }}
         run: |
-          python -m pytest -vv || python -m pytest -vv --lf
+          # Ignore warnings on Windows and PyPI
+          python -m pytest -vv -W ignore || python -m pytest -vv -W ignore --lf
 
       - name: Code coverage
         run: codecov
diff --git a/jupyter_client/asynchronous/client.py b/jupyter_client/asynchronous/client.py
index aa5aa41..fa61390 100644
--- a/jupyter_client/asynchronous/client.py
+++ b/jupyter_client/asynchronous/client.py
@@ -1,10 +1,14 @@
 """Implements an async kernel client"""
 # Copyright (c) Jupyter Development Team.
 # Distributed under the terms of the Modified BSD License.
+import asyncio
+
+import zmq.asyncio
+from traitlets import Instance
 from traitlets import Type
 
+from jupyter_client.channels import AsyncZMQSocketChannel
 from jupyter_client.channels import HBChannel
-from jupyter_client.channels import ZMQSocketChannel
 from jupyter_client.client import KernelClient
 from jupyter_client.client import reqrep
 
@@ -14,9 +18,11 @@ def wrapped(meth, channel):
         reply = kwargs.pop("reply", False)
         timeout = kwargs.pop("timeout", None)
         msg_id = meth(self, *args, **kwargs)
+        fut: asyncio.Future = asyncio.Future()
+        fut.set_result(msg_id)
         if not reply:
-            return msg_id
-        return self._async_recv_reply(msg_id, timeout=timeout, channel=channel)
+            return fut
+        return self._recv_reply(msg_id, timeout=timeout, channel=channel)
 
     return _
 
@@ -28,6 +34,12 @@ class AsyncKernelClient(KernelClient):
     raising :exc:`queue.Empty` if no message arrives within ``timeout`` seconds.
     """
 
+    context = Instance(zmq.asyncio.Context)
+
+    def _context_default(self) -> zmq.asyncio.Context:
+        self._created_context = True
+        return zmq.asyncio.Context()
+
     # --------------------------------------------------------------------------
     # Channel proxy methods
     # --------------------------------------------------------------------------
@@ -40,11 +52,11 @@ class AsyncKernelClient(KernelClient):
     wait_for_ready = KernelClient._async_wait_for_ready
 
     # The classes to use for the various channels
-    shell_channel_class = Type(ZMQSocketChannel)
-    iopub_channel_class = Type(ZMQSocketChannel)
-    stdin_channel_class = Type(ZMQSocketChannel)
+    shell_channel_class = Type(AsyncZMQSocketChannel)
+    iopub_channel_class = Type(AsyncZMQSocketChannel)
+    stdin_channel_class = Type(AsyncZMQSocketChannel)
     hb_channel_class = Type(HBChannel)
-    control_channel_class = Type(ZMQSocketChannel)
+    control_channel_class = Type(AsyncZMQSocketChannel)
 
     _recv_reply = KernelClient._async_recv_reply
 
@@ -52,6 +64,7 @@ class AsyncKernelClient(KernelClient):
     execute = reqrep(wrapped, KernelClient.execute)
     history = reqrep(wrapped, KernelClient.history)
     complete = reqrep(wrapped, KernelClient.complete)
+    is_complete = reqrep(wrapped, KernelClient.is_complete)
     inspect = reqrep(wrapped, KernelClient.inspect)
     kernel_info = reqrep(wrapped, KernelClient.kernel_info)
     comm_info = reqrep(wrapped, KernelClient.comm_info)
diff --git a/jupyter_client/blocking/client.py b/jupyter_client/blocking/client.py
index 27604fc..2303519 100644
--- a/jupyter_client/blocking/client.py
+++ b/jupyter_client/blocking/client.py
@@ -20,7 +20,7 @@ def wrapped(meth, channel):
         msg_id = meth(self, *args, **kwargs)
         if not reply:
             return msg_id
-        return run_sync(self._async_recv_reply)(msg_id, timeout=timeout, channel=channel)
+        return self._recv_reply(msg_id, timeout=timeout, channel=channel)
 
     return _
 
diff --git a/jupyter_client/channels.py b/jupyter_client/channels.py
index c340c08..978cf68 100644
--- a/jupyter_client/channels.py
+++ b/jupyter_client/channels.py
@@ -14,6 +14,7 @@ import zmq.asyncio
 from .channelsabc import HBChannelABC
 from .session import Session
 from jupyter_client import protocol_version_info
+from jupyter_client.utils import ensure_async
 
 # import ZMQError in top-level namespace, to avoid ugly attribute-error messages
 # during garbage collection of threads at exit
@@ -49,7 +50,7 @@ class HBChannel(Thread):
 
     def __init__(
         self,
-        context: t.Optional[zmq.asyncio.Context] = None,
+        context: t.Optional[zmq.Context] = None,
         session: t.Optional[Session] = None,
         address: t.Union[t.Tuple[str, int], str] = "",
     ):
@@ -57,7 +58,7 @@ class HBChannel(Thread):
 
         Parameters
         ----------
-        context : :class:`zmq.asyncio.Context`
+        context : :class:`zmq.Context`
             The ZMQ context to use.
         session : :class:`session.Session`
             The session to use.
@@ -106,12 +107,6 @@ class HBChannel(Thread):
 
         self.poller.register(self.socket, zmq.POLLIN)
 
-    def run(self) -> None:
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-        loop.run_until_complete(self._async_run())
-        loop.close()
-
     async def _async_run(self) -> None:
         """The thread's main activity.  Call start() instead."""
         self._create_socket()
@@ -127,8 +122,8 @@ class HBChannel(Thread):
 
             since_last_heartbeat = 0.0
             # no need to catch EFSM here, because the previous event was
-            # either a recv or connect, which cannot be followed by EFSM
-            await self.socket.send(b"ping")
+            # either a recv or connect, which cannot be followed by EFSM)
+            await ensure_async(self.socket.send(b"ping"))
             request_time = time.time()
             # Wait until timeout
             self._exit.wait(self.time_to_dead)
@@ -136,7 +131,7 @@ class HBChannel(Thread):
             self._beating = bool(self.poller.poll(0))
             if self._beating:
                 # the poll above guarantees we have something to recv
-                await self.socket.recv()
+                await ensure_async(self.socket.recv())
                 continue
             elif self._running:
                 # nothing was received within the time limit, signal heart failure
@@ -146,6 +141,12 @@ class HBChannel(Thread):
                 self._create_socket()
                 continue
 
+    def run(self) -> None:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        loop.run_until_complete(self._async_run())
+        loop.close()
+
     def pause(self) -> None:
         """Pause the heartbeat."""
         self._pause = True
@@ -191,14 +192,14 @@ HBChannelABC.register(HBChannel)
 
 
 class ZMQSocketChannel(object):
-    """A ZMQ socket in an async API"""
+    """A ZMQ socket wrapper"""
 
-    def __init__(self, socket: zmq.asyncio.Socket, session: Session, loop: t.Any = None) -> None:
+    def __init__(self, socket: zmq.Socket, session: Session, loop: t.Any = None) -> None:
         """Create a channel.
 
         Parameters
         ----------
-        socket : :class:`zmq.asyncio.Socket`
+        socket : :class:`zmq.Socket`
             The ZMQ socket to use.
         session : :class:`session.Session`
             The session to use.
@@ -207,42 +208,41 @@ class ZMQSocketChannel(object):
         """
         super().__init__()
 
-        self.socket: t.Optional[zmq.asyncio.Socket] = socket
+        self.socket: t.Optional[zmq.Socket] = socket
         self.session = session
 
-    async def _recv(self, **kwargs: t.Any) -> t.Dict[str, t.Any]:
+    def _recv(self, **kwargs: t.Any) -> t.Dict[str, t.Any]:
         assert self.socket is not None
-        msg = await self.socket.recv_multipart(**kwargs)
+        msg = self.socket.recv_multipart(**kwargs)
         ident, smsg = self.session.feed_identities(msg)
         return self.session.deserialize(smsg)
 
-    async def get_msg(self, timeout: t.Optional[float] = None) -> t.Dict[str, t.Any]:
+    def get_msg(self, timeout: t.Optional[float] = None) -> t.Dict[str, t.Any]:
         """Gets a message if there is one that is ready."""
         assert self.socket is not None
         if timeout is not None:
             timeout *= 1000  # seconds to ms
-        ready = await self.socket.poll(timeout)
-
+        ready = self.socket.poll(timeout)
         if ready:
-            res = await self._recv()
+            res = self._recv()
             return res
         else:
             raise Empty
 
-    async def get_msgs(self) -> t.List[t.Dict[str, t.Any]]:
+    def get_msgs(self) -> t.List[t.Dict[str, t.Any]]:
         """Get all messages that are currently ready."""
         msgs = []
         while True:
             try:
-                msgs.append(await self.get_msg())
+                msgs.append(self.get_msg())
             except Empty:
                 break
         return msgs
 
-    async def msg_ready(self) -> bool:
+    def msg_ready(self) -> bool:
         """Is there a message that has been received?"""
         assert self.socket is not None
-        return bool(await self.socket.poll(timeout=0))
+        return bool(self.socket.poll(timeout=0))
 
     def close(self) -> None:
         if self.socket is not None:
@@ -264,3 +264,60 @@ class ZMQSocketChannel(object):
 
     def start(self) -> None:
         pass
+
+
+class AsyncZMQSocketChannel(ZMQSocketChannel):
+    """A ZMQ socket in an async API"""
+
+    socket: zmq.asyncio.Socket
+
+    def __init__(self, socket: zmq.asyncio.Socket, session: Session, loop: t.Any = None) -> None:
+        """Create a channel.
+
+        Parameters
+        ----------
+        socket : :class:`zmq.asyncio.Socket`
+            The ZMQ socket to use.
+        session : :class:`session.Session`
+            The session to use.
+        loop
+            Unused here, for other implementations
+        """
+        if not isinstance(socket, zmq.asyncio.Socket):
+            raise ValueError('Socket must be asyncio')
+        super().__init__(socket, session)
+
+    async def _recv(self, **kwargs: t.Any) -> t.Dict[str, t.Any]:  # type:ignore[override]
+        assert self.socket is not None
+        msg = await self.socket.recv_multipart(**kwargs)
+        _, smsg = self.session.feed_identities(msg)
+        return self.session.deserialize(smsg)
+
+    async def get_msg(  # type:ignore[override]
+        self, timeout: t.Optional[float] = None
+    ) -> t.Dict[str, t.Any]:
+        """Gets a message if there is one that is ready."""
+        assert self.socket is not None
+        if timeout is not None:
+            timeout *= 1000  # seconds to ms
+        ready = await self.socket.poll(timeout)
+        if ready:
+            res = await self._recv()
+            return res
+        else:
+            raise Empty
+
+    async def get_msgs(self) -> t.List[t.Dict[str, t.Any]]:  # type:ignore[override]
+        """Get all messages that are currently ready."""
+        msgs = []
+        while True:
+            try:
+                msgs.append(await self.get_msg())
+            except Empty:
+                break
+        return msgs
+
+    async def msg_ready(self) -> bool:  # type:ignore[override]
+        """Is there a message that has been received?"""
+        assert self.socket is not None
+        return bool(await self.socket.poll(timeout=0))
diff --git a/jupyter_client/client.py b/jupyter_client/client.py
index 6be2644..f3d9a10 100644
--- a/jupyter_client/client.py
+++ b/jupyter_client/client.py
@@ -21,8 +21,8 @@ from .channelsabc import HBChannelABC
 from .clientabc import KernelClientABC
 from .connect import ConnectionFileMixin
 from .session import Session
-from .utils import ensure_async
 from jupyter_client.channels import major_protocol_version
+from jupyter_client.utils import ensure_async
 
 # some utilities to validate message structure, these might get moved elsewhere
 # if they prove to have more generic utility
@@ -92,13 +92,13 @@ class KernelClient(ConnectionFileMixin):
     """
 
     # The PyZMQ Context to use for communication with the kernel.
-    context = Instance(zmq.asyncio.Context)
+    context = Instance(zmq.Context)
 
     _created_context = Bool(False)
 
-    def _context_default(self) -> zmq.asyncio.Context:
+    def _context_default(self) -> zmq.Context:
         self._created_context = True
-        return zmq.asyncio.Context()
+        return zmq.Context()
 
     # The classes to use for the various channels
     shell_channel_class = Type(ChannelABC)
@@ -140,19 +140,19 @@ class KernelClient(ConnectionFileMixin):
 
     async def _async_get_shell_msg(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
         """Get a message from the shell channel"""
-        return await self.shell_channel.get_msg(*args, **kwargs)
+        return await ensure_async(self.shell_channel.get_msg(*args, **kwargs))
 
     async def _async_get_iopub_msg(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
         """Get a message from the iopub channel"""
-        return await self.iopub_channel.get_msg(*args, **kwargs)
+        return await ensure_async(self.iopub_channel.get_msg(*args, **kwargs))
 
     async def _async_get_stdin_msg(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
         """Get a message from the stdin channel"""
-        return await self.stdin_channel.get_msg(*args, **kwargs)
+        return await ensure_async(self.stdin_channel.get_msg(*args, **kwargs))
 
     async def _async_get_control_msg(self, *args: t.Any, **kwargs: t.Any) -> t.Dict[str, t.Any]:
         """Get a message from the control channel"""
-        return await self.control_channel.get_msg(*args, **kwargs)
+        return await ensure_async(self.control_channel.get_msg(*args, **kwargs))
 
     async def _async_wait_for_ready(self, timeout: t.Optional[float] = None) -> None:
         """Waits for a response when a client is blocked
@@ -173,7 +173,7 @@ class KernelClient(ConnectionFileMixin):
             # This Client was not created by a KernelManager,
             # so wait for kernel to become responsive to heartbeats
             # before checking for kernel_info reply
-            while not await ensure_async(self.is_alive()):
+            while not await self._async_is_alive():
                 if time.time() > abs_timeout:
                     raise RuntimeError(
                         "Kernel didn't respond to heartbeats in %d seconds and timed out" % timeout
@@ -184,21 +184,21 @@ class KernelClient(ConnectionFileMixin):
         while True:
             self.kernel_info()
             try:
-                msg = await self.shell_channel.get_msg(timeout=1)
+                msg = await ensure_async(self.shell_channel.get_msg(timeout=1))
             except Empty:
                 pass
             else:
                 if msg["msg_type"] == "kernel_info_reply":
                     # Checking that IOPub is connected. If it is not connected, start over.
                     try:
-                        await self.iopub_channel.get_msg(timeout=0.2)
+                        await ensure_async(self.iopub_channel.get_msg(timeout=0.2))
                     except Empty:
                         pass
                     else:
                         self._handle_kernel_info_reply(msg)
                         break
 
-            if not await ensure_async(self.is_alive()):
+            if not await self._async_is_alive():
                 raise RuntimeError("Kernel died before replying to kernel_info")
 
             # Check if current time is ready check time plus timeout
@@ -208,7 +208,7 @@ class KernelClient(ConnectionFileMixin):
         # Flush IOPub channel
         while True:
             try:
-                msg = await self.iopub_channel.get_msg(timeout=0.2)
+                msg = await ensure_async(self.iopub_channel.get_msg(timeout=0.2))
             except Empty:
                 break
 
@@ -403,7 +403,7 @@ class KernelClient(ConnectionFileMixin):
         if isinstance(self.parent, KernelManager):
             # This KernelClient was created by a KernelManager,
             # we can ask the parent KernelManager:
-            return await ensure_async(self.parent.is_alive())
+            return await self.parent._async_is_alive()
         if self._hb_channel is not None:
             # We don't have access to the KernelManager,
             # so we use the heartbeat.
@@ -484,13 +484,15 @@ class KernelClient(ConnectionFileMixin):
             allow_stdin = self.allow_stdin
         if allow_stdin and not self.stdin_channel.is_alive():
             raise RuntimeError("stdin channel must be running to allow input")
-        msg_id = self.execute(
-            code,
-            silent=silent,
-            store_history=store_history,
-            user_expressions=user_expressions,
-            allow_stdin=allow_stdin,
-            stop_on_error=stop_on_error,
+        msg_id = await ensure_async(
+            self.execute(
+                code,
+                silent=silent,
+                store_history=store_history,
+                user_expressions=user_expressions,
+                allow_stdin=allow_stdin,
+                stop_on_error=stop_on_error,
+            )
         )
         if stdin_hook is None:
             stdin_hook = self._stdin_hook_default
@@ -536,7 +538,7 @@ class KernelClient(ConnectionFileMixin):
             if not events:
                 raise TimeoutError("Timeout waiting for output")
             if stdin_socket in events:
-                req = await self.stdin_channel.get_msg(timeout=0)
+                req = await ensure_async(self.stdin_channel.get_msg(timeout=0))
                 res = stdin_hook(req)
                 if inspect.isawaitable(res):
                     await res
@@ -544,7 +546,7 @@ class KernelClient(ConnectionFileMixin):
             if iopub_socket not in events:
                 continue
 
-            msg = await self.iopub_channel.get_msg(timeout=0)
+            msg = await ensure_async(self.iopub_channel.get_msg(timeout=0))
 
             if msg["parent_header"].get("msg_id") != msg_id:
                 # not from my request
@@ -768,7 +770,12 @@ class KernelClient(ConnectionFileMixin):
             self.session.adapt_version = adapt_version
 
     def is_complete(self, code: str) -> str:
-        """Ask the kernel whether some code is complete and ready to execute."""
+        """Ask the kernel whether some code is complete and ready to execute.
+
+        Returns
+        -------
+        The ID of the message sent.
+        """
         msg = self.session.msg("is_complete_request", {"code": code})
         self.shell_channel.send(msg)
         return msg["header"]["msg_id"]
@@ -778,6 +785,10 @@ class KernelClient(ConnectionFileMixin):
 
         This should only be called in response to the kernel sending an
         ``input_request`` message on the stdin channel.
+
+        Returns
+        -------
+        The ID of the message sent.
         """
         content = dict(value=string)
         msg = self.session.msg("input_reply", content)
diff --git a/jupyter_client/ioloop/manager.py b/jupyter_client/ioloop/manager.py
index 23713ac..8a98025 100644
--- a/jupyter_client/ioloop/manager.py
+++ b/jupyter_client/ioloop/manager.py
@@ -1,6 +1,7 @@
 """A kernel manager with a tornado IOLoop"""
 # Copyright (c) Jupyter Development Team.
 # Distributed under the terms of the Modified BSD License.
+import zmq
 from tornado import ioloop
 from traitlets import Instance
 from traitlets import Type
@@ -14,7 +15,17 @@ from jupyter_client.manager import KernelManager
 
 def as_zmqstream(f):
     def wrapped(self, *args, **kwargs):
-        socket = f(self, *args, **kwargs)
+        save_socket_class = None
+        # zmqstreams only support sync sockets
+        if self.context._socket_class is not zmq.Socket:
+            save_socket_class = self.context._socket_class
+            self.context._socket_class = zmq.Socket
+        try:
+            socket = f(self, *args, **kwargs)
+        finally:
+            if save_socket_class:
+                # restore default socket class
+                self.context._socket_class = save_socket_class
         return ZMQStream(socket, self.loop)
 
     return wrapped
diff --git a/jupyter_client/manager.py b/jupyter_client/manager.py
index 65a0c22..ea8c97b 100644
--- a/jupyter_client/manager.py
+++ b/jupyter_client/manager.py
@@ -32,7 +32,6 @@ from .connect import ConnectionFileMixin
 from .managerabc import KernelManagerABC
 from .provisioning import KernelProvisionerBase
 from .provisioning import KernelProvisionerFactory as KPF
-from .utils import ensure_async
 from .utils import run_sync
 from jupyter_client import DEFAULT_EVENTS_SCHEMA_PATH
 from jupyter_client import JUPYTER_CLIENT_EVENTS_URI
@@ -409,12 +408,12 @@ class KernelManager(ConnectionFileMixin):
              keyword arguments that are passed down to build the kernel_cmd
              and launching the kernel (e.g. Popen kwargs).
         """
-        kernel_cmd, kw = await ensure_async(self.pre_start_kernel(**kw))
+        kernel_cmd, kw = await self._async_pre_start_kernel(**kw)
 
         # launch the kernel subprocess
         self.log.debug("Starting kernel: %s", kernel_cmd)
-        await ensure_async(self._launch_kernel(kernel_cmd, **kw))
-        await ensure_async(self.post_start_kernel(**kw))
+        await self._async_launch_kernel(kernel_cmd, **kw)
+        await self._async_post_start_kernel(**kw)
 
     start_kernel = run_sync(_async_start_kernel)
 
@@ -436,7 +435,7 @@ class KernelManager(ConnectionFileMixin):
         self,
         waittime: t.Optional[float] = None,
         pollinterval: float = 0.1,
-        restart: t.Optional[bool] = False,
+        restart: bool = False,
     ) -> None:
         """Wait for kernel shutdown, then kill process if it doesn't shutdown.
 
@@ -455,7 +454,7 @@ class KernelManager(ConnectionFileMixin):
         except asyncio.TimeoutError:
             self.log.debug("Kernel is taking too long to finish, terminating")
             self._shutdown_status = _ShutdownStatus.SigtermRequest
-            await ensure_async(self._send_kernel_sigterm())
+            await self._async_send_kernel_sigterm()
 
         try:
             await asyncio.wait_for(
@@ -464,7 +463,7 @@ class KernelManager(ConnectionFileMixin):
         except asyncio.TimeoutError:
             self.log.debug("Kernel is taking too long to finish, killing")
             self._shutdown_status = _ShutdownStatus.SigkillRequest
-            await ensure_async(self._kill_kernel(restart=restart))
+            await self._async_kill_kernel(restart=restart)
         else:
             # Process is no longer alive, wait and clear
             if self.has_kernel:
@@ -517,18 +516,18 @@ class KernelManager(ConnectionFileMixin):
         self.stop_restarter()
 
         if self.has_kernel:
-            await ensure_async(self.interrupt_kernel())
+            await self._async_interrupt_kernel()
 
         if now:
-            await ensure_async(self._kill_kernel())
+            await self._async_kill_kernel()
         else:
-            await ensure_async(self.request_shutdown(restart=restart))
+            await self._async_request_shutdown(restart=restart)
             # Don't send any additional kernel kill messages immediately, to give
             # the kernel a chance to properly execute shutdown actions. Wait for at
             # most 1s, checking every 0.1s.
-            await ensure_async(self.finish_shutdown(restart=restart))
+            await self._async_finish_shutdown(restart=restart)
 
-        await ensure_async(self.cleanup_resources(restart=restart))
+        await self._async_cleanup_resources(restart=restart)
         self._emit(action="shutdown_finished")
 
     shutdown_kernel = run_sync(_async_shutdown_kernel)
@@ -565,14 +564,14 @@ class KernelManager(ConnectionFileMixin):
             raise RuntimeError("Cannot restart the kernel. No previous call to 'start_kernel'.")
 
         # Stop currently running kernel.
-        await ensure_async(self.shutdown_kernel(now=now, restart=True))
+        await self._async_shutdown_kernel(now=now, restart=True)
 
         if newports:
             self.cleanup_random_ports()
 
         # Start new kernel.
         self._launch_args.update(kw)
-        await ensure_async(self.start_kernel(**self._launch_args))
+        await self._async_start_kernel(**self._launch_args)
         self._emit(action="restart_finished")
 
     restart_kernel = run_sync(_async_restart_kernel)
@@ -624,7 +623,7 @@ class KernelManager(ConnectionFileMixin):
             assert self.kernel_spec is not None
             interrupt_mode = self.kernel_spec.interrupt_mode
             if interrupt_mode == "signal":
-                await ensure_async(self.signal_kernel(signal.SIGINT))
+                await self._async_signal_kernel(signal.SIGINT)
 
             elif interrupt_mode == "message":
                 msg = self.session.msg("interrupt_request", content={})
@@ -668,17 +667,26 @@ class KernelManager(ConnectionFileMixin):
         # not alive.  If we find the process is no longer alive, complete
         # its cleanup via the blocking wait().  Callers are responsible for
         # issuing calls to wait() using a timeout (see _kill_kernel()).
-        while await ensure_async(self.is_alive()):
+        while await self._async_is_alive():
             await asyncio.sleep(pollinterval)
 
 
 class AsyncKernelManager(KernelManager):
+
     # the class to create with our `client` method
     client_class: DottedObjectName = DottedObjectName(
         "jupyter_client.asynchronous.AsyncKernelClient"
     )
     client_factory: Type = Type(klass="jupyter_client.asynchronous.AsyncKernelClient")
 
+    # The PyZMQ Context to use for communication with the kernel.
+    context: Instance = Instance(zmq.asyncio.Context)
+
+    @default("context")  # type:ignore[misc]
+    def _context_default(self) -> zmq.asyncio.Context:
+        self._created_context = True
+        return zmq.asyncio.Context()
+
     _launch_kernel = KernelManager._async_launch_kernel
     start_kernel = KernelManager._async_start_kernel
     pre_start_kernel = KernelManager._async_pre_start_kernel
diff --git a/jupyter_client/multikernelmanager.py b/jupyter_client/multikernelmanager.py
index 7dceb94..5a2ffb5 100644
--- a/jupyter_client/multikernelmanager.py
+++ b/jupyter_client/multikernelmanager.py
@@ -309,7 +309,7 @@ class MultiKernelManager(LoggingConfigurable):
         kids = self.list_kernel_ids()
         kids += list(self._pending_kernels)
         kms = list(self._kernels.values())
-        futs = [ensure_async(self.shutdown_kernel(kid, now=now)) for kid in set(kids)]
+        futs = [self._async_shutdown_kernel(kid, now=now) for kid in set(kids)]
         await asyncio.gather(*futs)
         # If using pending kernels, the kernels will not have been fully shut down.
         if self._using_pending_kernels():
@@ -545,6 +545,13 @@ class AsyncMultiKernelManager(MultiKernelManager):
         kernel has a `.ready` future which can be awaited before connecting""",
     ).tag(config=True)
 
+    context = Instance("zmq.asyncio.Context")
+
+    @default("context")  # type:ignore[misc]
+    def _context_default(self) -> zmq.asyncio.Context:
+        self._created_context = True
+        return zmq.asyncio.Context()
+
     start_kernel = MultiKernelManager._async_start_kernel
     restart_kernel = MultiKernelManager._async_restart_kernel
     shutdown_kernel = MultiKernelManager._async_shutdown_kernel
diff --git a/jupyter_client/provisioning/local_provisioner.py b/jupyter_client/provisioning/local_provisioner.py
index c94a827..875995f 100644
--- a/jupyter_client/provisioning/local_provisioner.py
+++ b/jupyter_client/provisioning/local_provisioner.py
@@ -42,7 +42,6 @@ class LocalProvisioner(KernelProvisionerBase):
         return self.process is not None
 
     async def poll(self) -> Optional[int]:
-
         ret = 0
         if self.process:
             ret = self.process.poll()
diff --git a/jupyter_client/session.py b/jupyter_client/session.py
index 63753f8..4a2ebda 100644
--- a/jupyter_client/session.py
+++ b/jupyter_client/session.py
@@ -29,7 +29,7 @@ from hmac import (
 from typing import Optional
 from typing import Union
 
-import zmq
+import zmq.asyncio
 from traitlets import Any
 from traitlets import Bool
 from traitlets import CBytes
@@ -807,6 +807,10 @@ class Session(Configurable):
             # ZMQStreams and dummy sockets do not support tracking.
             track = False
 
+        if isinstance(stream, zmq.asyncio.Socket):
+            assert stream is not None
+            stream = zmq.Socket.shadow(stream.underlying)
+
         if isinstance(msg_or_type, (Message, dict)):
             # We got a Message or message dict, not a msg_type so don't
             # build a new Message.
@@ -896,6 +900,8 @@ class Session(Configurable):
         # Don't include buffers in signature (per spec).
         to_send.append(self.sign(msg_list[0:4]))
         to_send.extend(msg_list)
+        if isinstance(stream, zmq.asyncio.Socket):
+            stream = zmq.Socket.shadow(stream.underlying)
         stream.send_multipart(to_send, flags, copy=copy)
 
     def recv(
@@ -920,6 +926,9 @@ class Session(Configurable):
         """
         if isinstance(socket, ZMQStream):
             socket = socket.socket
+        if isinstance(socket, zmq.asyncio.Socket):
+            socket = zmq.Socket.shadow(socket.underlying)
+
         try:
             msg_list = socket.recv_multipart(mode, copy=copy)
         except zmq.ZMQError as e:
diff --git a/jupyter_client/threaded.py b/jupyter_client/threaded.py
index 6c46219..ca61ff7 100644
--- a/jupyter_client/threaded.py
+++ b/jupyter_client/threaded.py
@@ -3,22 +3,18 @@ replies.
 """
 import asyncio
 import atexit
-import errno
 import time
 from threading import Event
 from threading import Thread
 from typing import Any
-from typing import Awaitable
 from typing import Dict
 from typing import List
 from typing import Optional
-from typing import Union
 
 import zmq
+from tornado.ioloop import IOLoop
 from traitlets import Instance
 from traitlets import Type
-from zmq import ZMQError
-from zmq.eventloop import ioloop
 from zmq.eventloop import zmqstream
 
 from .session import Session
@@ -30,10 +26,6 @@ from jupyter_client.channels import HBChannel
 # during garbage collection of threads at exit
 
 
-async def get_msg(msg: Awaitable) -> Union[List[bytes], List[zmq.Message]]:
-    return await msg
-
-
 class ThreadedZMQSocketChannel(object):
     """A ZMQ socket invoking a callback in the ioloop"""
 
@@ -47,7 +39,7 @@ class ThreadedZMQSocketChannel(object):
         self,
         socket: Optional[zmq.Socket],
         session: Optional[Session],
-        loop: Optional[zmq.eventloop.ioloop.ZMQIOLoop],
+        loop: Optional[IOLoop],
     ) -> None:
         """Create a channel.
 
@@ -70,7 +62,7 @@ class ThreadedZMQSocketChannel(object):
         def setup_stream():
             assert self.socket is not None
             self.stream = zmqstream.ZMQStream(self.socket, self.ioloop)
-            self.stream.on_recv(self._handle_recv)  # type:ignore[arg-type]
+            self.stream.on_recv(self._handle_recv)
             evt.set()
 
         assert self.ioloop is not None
@@ -114,13 +106,12 @@ class ThreadedZMQSocketChannel(object):
         assert self.ioloop is not None
         self.ioloop.add_callback(thread_send)
 
-    def _handle_recv(self, future_msg: Awaitable) -> None:
+    def _handle_recv(self, msg_list: List) -> None:
         """Callback for stream.on_recv.
 
         Unpacks message, and calls handlers with it.
         """
         assert self.ioloop is not None
-        msg_list = self.ioloop._asyncio_event_loop.run_until_complete(get_msg(future_msg))
         assert self.session is not None
         ident, smsg = self.session.feed_identities(msg_list)
         msg = self.session.deserialize(smsg)
@@ -211,24 +202,15 @@ class IOLoopThread(Thread):
         """Run my loop, ignoring EINTR events in the poller"""
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)
-        self.ioloop = ioloop.IOLoop()
-        self.ioloop._asyncio_event_loop = loop
+        loop.run_until_complete(self._async_run())
+
+    async def _async_run(self):
+        self.ioloop = IOLoop.current()
         # signal that self.ioloop is defined
         self._start_event.set()
         while True:
-            try:
-                self.ioloop.start()
-            except ZMQError as e:
-                if e.errno == errno.EINTR:
-                    continue
-                else:
-                    raise
-            except Exception:
-                if self._exiting:
-                    break
-                else:
-                    raise
-            else:
+            await asyncio.sleep(1)
+            if self._exiting:
                 break
 
     def stop(self) -> None:
@@ -238,8 +220,7 @@ class IOLoopThread(Thread):
         terminates. :class:`RuntimeError` will be raised if
         :meth:`~threading.Thread.start` is called again.
         """
-        if self.ioloop is not None:
-            self.ioloop.add_callback(self.ioloop.stop)
+        self._exiting = True
         self.join()
         self.close()
         self.ioloop = None
diff --git a/jupyter_client/utils.py b/jupyter_client/utils.py
index 585bf1b..d94654d 100644
--- a/jupyter_client/utils.py
+++ b/jupyter_client/utils.py
@@ -4,36 +4,80 @@ utils:
 - vendor functions from ipython_genutils that should be retired at some point.
 """
 import asyncio
+import atexit
 import inspect
 import os
+import threading
+from typing import Optional
+
+
+class _TaskRunner:
+    """A task runner that runs an asyncio event loop on a background thread."""
+
+    def __init__(self):
+        self.__io_loop: Optional[asyncio.AbstractEventLoop] = None
+        self.__runner_thread: Optional[threading.Thread] = None
+        self.__lock = threading.Lock()
+        atexit.register(self._close)
+
+    def _close(self):
+        if self.__io_loop:
+            self.__io_loop.stop()
+
+    def _runner(self):
+        loop = self.__io_loop
+        assert loop is not None
+        try:
+            loop.run_forever()
+        finally:
+            loop.close()
+
+    def run(self, coro):
+        """Synchronously run a coroutine on a background thread."""
+        with self.__lock:
+            name = f"{threading.current_thread().name} - runner"
+            if self.__io_loop is None:
+                self.__io_loop = asyncio.new_event_loop()
+                self.__runner_thread = threading.Thread(target=self._runner, daemon=True, name=name)
+                self.__runner_thread.start()
+        fut = asyncio.run_coroutine_threadsafe(coro, self.__io_loop)
+        return fut.result(None)
+
+
+_runner_map = {}
+_loop_map = {}
 
 
 def run_sync(coro):
     def wrapped(*args, **kwargs):
+        name = threading.current_thread().name
+        inner = coro(*args, **kwargs)
         try:
-            loop = asyncio.get_running_loop()
+            # If a loop is currently running in this thread,
+            # use a task runner.
+            asyncio.get_running_loop()
+            if name not in _runner_map:
+                _runner_map[name] = _TaskRunner()
+            return _runner_map[name].run(inner)
         except RuntimeError:
-            # Workaround for bugs.python.org/issue39529.
-            try:
-                loop = asyncio.get_event_loop_policy().get_event_loop()
-            except RuntimeError:
-                loop = asyncio.new_event_loop()
-                asyncio.set_event_loop(loop)
-        import nest_asyncio  # type: ignore
-
-        nest_asyncio.apply(loop)
-        future = asyncio.ensure_future(coro(*args, **kwargs), loop=loop)
-        try:
-            return loop.run_until_complete(future)
-        except BaseException as e:
-            future.cancel()
-            raise e
+            pass
+
+        # Run the loop for this thread.
+        if name not in _loop_map:
+            _loop_map[name] = asyncio.new_event_loop()
+        loop = _loop_map[name]
+        return loop.run_until_complete(inner)
 
     wrapped.__doc__ = coro.__doc__
     return wrapped
 
 
 async def ensure_async(obj):
+    """Ensure a returned object is asynchronous.L
+
+    NOTE: This should only be used on methods of external classes,
+    not on a `self` method.
+    """
     if inspect.isawaitable(obj):
         return await obj
     return obj
diff --git a/pyproject.toml b/pyproject.toml
index 6850768..d2cfe69 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -26,7 +26,6 @@ requires-python = ">=3.7"
 dependencies = [
     "entrypoints",
     "jupyter_core>=4.9.2",
-    "nest-asyncio>=1.5.4",
     "python-dateutil>=2.8.2",
     "pyzmq>=23.0",
     "tornado>=6.2",
@@ -117,27 +116,13 @@ testpaths = [
     "jupyter_client",
     "tests/"
 ]
-timeout = 300
+timeout = 100
 # Restore this setting to debug failures
-# timeout_method = "thread"
+timeout_method = "thread"
 asyncio_mode = "auto"
 filterwarnings= [
   # Fail on warnings
   "error",
-
-  # We need to handle properly closing loops as part of https://github.com/jupyter/jupyter_client/issues/755.
-  "ignore:unclosed <socket.socket:ResourceWarning",
-  "ignore:unclosed event loop:ResourceWarning",
-
-  # Workaround for https://github.com/tornadoweb/tornado/issues/3106
-  # (To be fixed in Tornado 6.2)
-  "ignore:There is no current event loop:DeprecationWarning:tornado",
-
-  # ZMQ uses Future internally, which raises a DeprecationWarning
-  # When there is no loop running.
-  # We could eventually find a way to make sure these are only created
-  # when there is a running event loop.
-  "ignore:There is no current event loop:DeprecationWarning:zmq",
 ]
 
 [tool.mypy]


Test patch:
diff --git a/tests/test_client.py b/tests/test_client.py
index b77859c..6792395 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -2,16 +2,23 @@
 # Copyright (c) Jupyter Development Team.
 # Distributed under the terms of the Modified BSD License.
 import os
+from threading import Event
 from unittest import TestCase
 
 import pytest
 from IPython.utils.capture import capture_output
+from traitlets import DottedObjectName
+from traitlets import Type
 
 from .utils import test_env
 from jupyter_client.kernelspec import KernelSpecManager
 from jupyter_client.kernelspec import NATIVE_KERNEL_NAME
 from jupyter_client.kernelspec import NoSuchKernel
+from jupyter_client.manager import KernelManager
+from jupyter_client.manager import start_new_async_kernel
 from jupyter_client.manager import start_new_kernel
+from jupyter_client.threaded import ThreadedKernelClient
+from jupyter_client.threaded import ThreadedZMQSocketChannel
 
 TIMEOUT = 30
 
@@ -92,3 +99,176 @@ class TestKernelClient(TestCase):
         kc = self.kc
         msg_id = kc.shutdown()
         self.assertIsInstance(msg_id, str)
+
+
+@pytest.fixture
+async def kc():
+    env_patch = test_env()
+    env_patch.start()
+    try:
+        KernelSpecManager().get_kernel_spec(NATIVE_KERNEL_NAME)
+    except NoSuchKernel:
+        pytest.skip()
+    km, kc = await start_new_async_kernel(kernel_name=NATIVE_KERNEL_NAME)
+    yield kc
+    env_patch.stop()
+    await km.shutdown_kernel()
+    kc.stop_channels()
+
+
+class TestAsyncKernelClient:
+    async def test_execute_interactive(self, kc):
+        with capture_output() as io:
+            reply = await kc.execute_interactive("print('hello')", timeout=TIMEOUT)
+        assert "hello" in io.stdout
+        assert reply["content"]["status"] == "ok"
+
+    def _check_reply(self, reply_type, reply):
+        assert isinstance(reply, dict)
+        assert reply["header"]["msg_type"] == reply_type + "_reply"
+        assert reply["parent_header"]["msg_type"] == reply_type + "_request"
+
+    async def test_history(self, kc):
+        msg_id = await kc.history(session=0)
+        assert isinstance(msg_id, str)
+        reply = await kc.history(session=0, reply=True, timeout=TIMEOUT)
+        self._check_reply("history", reply)
+
+    async def test_inspect(self, kc):
+        msg_id = await kc.inspect("who cares")
+        assert isinstance(msg_id, str)
+        reply = await kc.inspect("code", reply=True, timeout=TIMEOUT)
+        self._check_reply("inspect", reply)
+
+    async def test_complete(self, kc):
+        msg_id = await kc.complete("who cares")
+        assert isinstance(msg_id, str)
+        reply = await kc.complete("code", reply=True, timeout=TIMEOUT)
+        self._check_reply("complete", reply)
+
+    async def test_kernel_info(self, kc):
+        msg_id = await kc.kernel_info()
+        assert isinstance(msg_id, str)
+        reply = await kc.kernel_info(reply=True, timeout=TIMEOUT)
+        self._check_reply("kernel_info", reply)
+
+    async def test_comm_info(self, kc):
+        msg_id = await kc.comm_info()
+        assert isinstance(msg_id, str)
+        reply = await kc.comm_info(reply=True, timeout=TIMEOUT)
+        self._check_reply("comm_info", reply)
+
+    async def test_shutdown(self, kc):
+        reply = await kc.shutdown(reply=True, timeout=TIMEOUT)
+        self._check_reply("shutdown", reply)
+
+    async def test_shutdown_id(self, kc):
+        msg_id = await kc.shutdown()
+        assert isinstance(msg_id, str)
+
+
+class ThreadedKernelManager(KernelManager):
+    client_class = DottedObjectName('tests.test_client.CustomThreadedKernelClient')
+
+
+class CustomThreadedZMQSocketChannel(ThreadedZMQSocketChannel):
+    last_msg = None
+
+    def __init__(self, *args, **kwargs):
+        self.msg_recv = Event()
+        super().__init__(*args, **kwargs)
+
+    def call_handlers(self, msg):
+        self.last_msg = msg
+        self.msg_recv.set()
+
+
+class CustomThreadedKernelClient(ThreadedKernelClient):
+    iopub_channel_class = Type(CustomThreadedZMQSocketChannel)
+    shell_channel_class = Type(CustomThreadedZMQSocketChannel)
+    stdin_channel_class = Type(CustomThreadedZMQSocketChannel)
+    control_channel_class = Type(CustomThreadedZMQSocketChannel)
+
+
+class TestThreadedKernelClient(TestKernelClient):
+    def setUp(self):
+        self.env_patch = test_env()
+        self.env_patch.start()
+        self.addCleanup(self.env_patch.stop)
+        try:
+            KernelSpecManager().get_kernel_spec(NATIVE_KERNEL_NAME)
+        except NoSuchKernel:
+            pytest.skip()
+        self.km = km = ThreadedKernelManager(kernel_name=NATIVE_KERNEL_NAME)
+        km.start_kernel()
+        self.kc = kc = km.client()
+        kc.start_channels()
+
+    def tearDown(self):
+        self.env_patch.stop()
+        self.km.shutdown_kernel()
+        self.kc.stop_channels()
+
+    def _check_reply(self, reply_type, reply):
+        self.assertIsInstance(reply, dict)
+        self.assertEqual(reply["header"]["msg_type"], reply_type + "_reply")
+        self.assertEqual(reply["parent_header"]["msg_type"], reply_type + "_request")
+
+    def test_execute_interactive(self):
+        pytest.skip('Not supported')
+
+    def test_history(self):
+        kc = self.kc
+        msg_id = kc.history(session=0)
+        self.assertIsInstance(msg_id, str)
+        kc.history(session=0)
+        kc.shell_channel.msg_recv.wait()
+        reply = kc.shell_channel.last_msg
+        self._check_reply("history", reply)
+
+    def test_inspect(self):
+        kc = self.kc
+        msg_id = kc.inspect("who cares")
+        self.assertIsInstance(msg_id, str)
+        kc.inspect("code")
+        kc.shell_channel.msg_recv.wait()
+        reply = kc.shell_channel.last_msg
+        self._check_reply("inspect", reply)
+
+    def test_complete(self):
+        kc = self.kc
+        msg_id = kc.complete("who cares")
+        self.assertIsInstance(msg_id, str)
+        kc.complete("code")
+        kc.shell_channel.msg_recv.wait()
+        reply = kc.shell_channel.last_msg
+        self._check_reply("complete", reply)
+
+    def test_kernel_info(self):
+        kc = self.kc
+        msg_id = kc.kernel_info()
+        self.assertIsInstance(msg_id, str)
+        kc.kernel_info()
+        kc.shell_channel.msg_recv.wait()
+        reply = kc.shell_channel.last_msg
+        self._check_reply("kernel_info", reply)
+
+    def test_comm_info(self):
+        kc = self.kc
+        msg_id = kc.comm_info()
+        self.assertIsInstance(msg_id, str)
+        kc.shell_channel.msg_recv.wait()
+        reply = kc.shell_channel.last_msg
+        self._check_reply("comm_info", reply)
+
+    def test_shutdown(self):
+        kc = self.kc
+        kc.shutdown()
+        kc.control_channel.msg_recv.wait()
+        reply = kc.control_channel.last_msg
+        self._check_reply("shutdown", reply)
+
+    def test_shutdown_id(self):
+        kc = self.kc
+        msg_id = kc.shutdown()
+        self.assertIsInstance(msg_id, str)
diff --git a/tests/test_kernelmanager.py b/tests/test_kernelmanager.py
index 4c560ad..67171b7 100644
--- a/tests/test_kernelmanager.py
+++ b/tests/test_kernelmanager.py
@@ -26,7 +26,7 @@ from jupyter_client.manager import start_new_kernel
 
 pjoin = os.path.join
 
-TIMEOUT = 30
+TIMEOUT = 60
 
 
 @pytest.fixture(params=["tcp", "ipc"])
@@ -108,7 +108,7 @@ def km_subclass(config, jp_event_logger):
 def zmq_context():
     import zmq
 
-    ctx = zmq.Context()
+    ctx = zmq.asyncio.Context()
     yield ctx
     ctx.term()
 
@@ -194,7 +194,6 @@ class TestKernelManagerShutDownGracefully:
 
         assert km._shutdown_status in expected
 
-    @pytest.mark.asyncio
     @pytest.mark.skipif(sys.platform == "win32", reason="Windows doesn't support signals")
     @pytest.mark.parametrize(*parameters)
     async def test_async_signal_kernel_subprocesses(self, name, install, expected):
@@ -352,7 +351,7 @@ class TestKernelManager:
 
         km_subclass.start_kernel(stdout=PIPE, stderr=PIPE)
         assert km_subclass.call_count("start_kernel") == 1
-        assert km_subclass.call_count("_launch_kernel") == 1
+        assert km_subclass.call_count("_async_launch_kernel") == 1
 
         is_alive = km_subclass.is_alive()
         assert is_alive
@@ -360,13 +359,12 @@ class TestKernelManager:
 
         km_subclass.restart_kernel(now=True)
         assert km_subclass.call_count("restart_kernel") == 1
-        assert km_subclass.call_count("shutdown_kernel") == 1
-        assert km_subclass.call_count("interrupt_kernel") == 1
-        assert km_subclass.call_count("_kill_kernel") == 1
-        assert km_subclass.call_count("cleanup_resources") == 1
-        assert km_subclass.call_count("start_kernel") == 1
-        assert km_subclass.call_count("_launch_kernel") == 1
-        assert km_subclass.call_count("signal_kernel") == 1
+        assert km_subclass.call_count("_async_shutdown_kernel") == 1
+        assert km_subclass.call_count("_async_interrupt_kernel") == 1
+        assert km_subclass.call_count("_async_kill_kernel") == 1
+        assert km_subclass.call_count("_async_cleanup_resources") == 1
+        assert km_subclass.call_count("_async_launch_kernel") == 1
+        assert km_subclass.call_count("_async_signal_kernel") == 1
 
         is_alive = km_subclass.is_alive()
         assert is_alive
@@ -374,24 +372,21 @@ class TestKernelManager:
         km_subclass.reset_counts()
 
         km_subclass.interrupt_kernel()
-        assert km_subclass.call_count("interrupt_kernel") == 1
-        assert km_subclass.call_count("signal_kernel") == 1
+        assert km_subclass.call_count("_async_signal_kernel") == 1
 
         assert isinstance(km_subclass, KernelManager)
         km_subclass.reset_counts()
 
         km_subclass.shutdown_kernel(now=False)
         assert km_subclass.call_count("shutdown_kernel") == 1
-        assert km_subclass.call_count("interrupt_kernel") == 1
-        assert km_subclass.call_count("request_shutdown") == 1
-        assert km_subclass.call_count("finish_shutdown") == 1
-        assert km_subclass.call_count("cleanup_resources") == 1
-        assert km_subclass.call_count("signal_kernel") == 1
-        assert km_subclass.call_count("is_alive") >= 1
+        assert km_subclass.call_count("_async_interrupt_kernel") == 1
+        assert km_subclass.call_count("_async_cleanup_resources") == 1
+        assert km_subclass.call_count("_async_signal_kernel") == 1
+        assert km_subclass.call_count("_async_is_alive") >= 1
 
         is_alive = km_subclass.is_alive()
         assert is_alive is False
-        assert km_subclass.call_count("is_alive") >= 1
+        assert km_subclass.call_count("_async_is_alive") >= 1
         assert km_subclass.context.closed
 
 
@@ -531,7 +526,7 @@ class TestAsyncKernelManager:
         km, kc = start_async_kernel
 
         async def execute(cmd):
-            request_id = kc.execute(cmd)
+            request_id = await kc.execute(cmd)
             while True:
                 reply = await kc.get_shell_msg(TIMEOUT)
                 if reply["parent_header"]["msg_id"] == request_id:
@@ -551,7 +546,7 @@ class TestAsyncKernelManager:
         assert reply["user_expressions"]["poll"] == [None] * N
 
         # start a job on the kernel to be interrupted
-        request_id = kc.execute("sleep")
+        request_id = await kc.execute("sleep")
         await asyncio.sleep(1)  # ensure sleep message has been handled before we interrupt
         await km.interrupt_kernel()
         while True:
@@ -584,7 +579,7 @@ class TestAsyncKernelManager:
 
         await async_km_subclass.start_kernel(stdout=PIPE, stderr=PIPE)
         assert async_km_subclass.call_count("start_kernel") == 1
-        assert async_km_subclass.call_count("_launch_kernel") == 1
+        assert async_km_subclass.call_count("_async_launch_kernel") == 1
 
         is_alive = await async_km_subclass.is_alive()
         assert is_alive
@@ -593,13 +588,12 @@ class TestAsyncKernelManager:
 
         await async_km_subclass.restart_kernel(now=True)
         assert async_km_subclass.call_count("restart_kernel") == 1
-        assert async_km_subclass.call_count("shutdown_kernel") == 1
-        assert async_km_subclass.call_count("interrupt_kernel") == 1
-        assert async_km_subclass.call_count("_kill_kernel") == 1
-        assert async_km_subclass.call_count("cleanup_resources") == 1
-        assert async_km_subclass.call_count("start_kernel") == 1
-        assert async_km_subclass.call_count("_launch_kernel") == 1
-        assert async_km_subclass.call_count("signal_kernel") == 1
+        assert async_km_subclass.call_count("_async_shutdown_kernel") == 1
+        assert async_km_subclass.call_count("_async_interrupt_kernel") == 1
+        assert async_km_subclass.call_count("_async_kill_kernel") == 1
+        assert async_km_subclass.call_count("_async_cleanup_resources") == 1
+        assert async_km_subclass.call_count("_async_launch_kernel") == 1
+        assert async_km_subclass.call_count("_async_signal_kernel") == 1
 
         is_alive = await async_km_subclass.is_alive()
         assert is_alive
@@ -608,21 +602,19 @@ class TestAsyncKernelManager:
 
         await async_km_subclass.interrupt_kernel()
         assert async_km_subclass.call_count("interrupt_kernel") == 1
-        assert async_km_subclass.call_count("signal_kernel") == 1
+        assert async_km_subclass.call_count("_async_signal_kernel") == 1
 
         assert isinstance(async_km_subclass, AsyncKernelManager)
         async_km_subclass.reset_counts()
 
         await async_km_subclass.shutdown_kernel(now=False)
         assert async_km_subclass.call_count("shutdown_kernel") == 1
-        assert async_km_subclass.call_count("interrupt_kernel") == 1
-        assert async_km_subclass.call_count("request_shutdown") == 1
-        assert async_km_subclass.call_count("finish_shutdown") == 1
-        assert async_km_subclass.call_count("cleanup_resources") == 1
-        assert async_km_subclass.call_count("signal_kernel") == 1
-        assert async_km_subclass.call_count("is_alive") >= 1
+        assert async_km_subclass.call_count("_async_interrupt_kernel") == 1
+        assert async_km_subclass.call_count("_async_cleanup_resources") == 1
+        assert async_km_subclass.call_count("_async_signal_kernel") == 1
+        assert async_km_subclass.call_count("_async_is_alive") >= 1
 
         is_alive = await async_km_subclass.is_alive()
         assert is_alive is False
-        assert async_km_subclass.call_count("is_alive") >= 1
+        assert async_km_subclass.call_count("_async_is_alive") >= 1
         assert async_km_subclass.context.closed
diff --git a/tests/test_multikernelmanager.py b/tests/test_multikernelmanager.py
index 20d9527..a22f732 100644
--- a/tests/test_multikernelmanager.py
+++ b/tests/test_multikernelmanager.py
@@ -178,6 +178,10 @@ class TestKernelManager(TestCase):
         (sys.platform == "darwin") and (sys.version_info >= (3, 6)) and (sys.version_info < (3, 8)),
         reason='"Bad file descriptor" error',
     )
+    @pytest.mark.skipif(
+        sys.platform == "linux",
+        reason='Kernel refuses to start in process pool',
+    )
     def test_start_parallel_process_kernels(self):
         self.test_tcp_lifecycle()
 
@@ -197,7 +201,7 @@ class TestKernelManager(TestCase):
         assert km.call_count("start_kernel") == 1
         assert isinstance(km.get_kernel(kid), SyncKMSubclass)
         assert km.get_kernel(kid).call_count("start_kernel") == 1
-        assert km.get_kernel(kid).call_count("_launch_kernel") == 1
+        assert km.get_kernel(kid).call_count("_async_launch_kernel") == 1
 
         assert km.is_alive(kid)
         assert kid in km
@@ -210,12 +214,11 @@ class TestKernelManager(TestCase):
         assert km.call_count("restart_kernel") == 1
         assert km.call_count("get_kernel") == 1
         assert km.get_kernel(kid).call_count("restart_kernel") == 1
-        assert km.get_kernel(kid).call_count("shutdown_kernel") == 1
-        assert km.get_kernel(kid).call_count("interrupt_kernel") == 1
-        assert km.get_kernel(kid).call_count("_kill_kernel") == 1
-        assert km.get_kernel(kid).call_count("cleanup_resources") == 1
-        assert km.get_kernel(kid).call_count("start_kernel") == 1
-        assert km.get_kernel(kid).call_count("_launch_kernel") == 1
+        assert km.get_kernel(kid).call_count("_async_shutdown_kernel") == 1
+        assert km.get_kernel(kid).call_count("_async_interrupt_kernel") == 1
+        assert km.get_kernel(kid).call_count("_async_kill_kernel") == 1
+        assert km.get_kernel(kid).call_count("_async_cleanup_resources") == 1
+        assert km.get_kernel(kid).call_count("_async_launch_kernel") == 1
 
         assert km.is_alive(kid)
         assert kid in km.list_kernel_ids()
@@ -236,7 +239,6 @@ class TestKernelManager(TestCase):
         km.get_kernel(kid).reset_counts()
         km.reset_counts()
         km.shutdown_all(now=True)
-        assert km.call_count("shutdown_kernel") == 1
         assert km.call_count("remove_kernel") == 1
         assert km.call_count("request_shutdown") == 0
         assert km.call_count("finish_shutdown") == 0
@@ -244,6 +246,38 @@ class TestKernelManager(TestCase):
 
         assert kid not in km, f"{kid} not in {km}"
 
+    def test_stream_on_recv(self):
+        mkm = self._get_tcp_km()
+        kid = mkm.start_kernel(stdout=PIPE, stderr=PIPE)
+        stream = mkm.connect_iopub(kid)
+
+        km = mkm.get_kernel(kid)
+        client = km.client()
+        session = km.session
+        called = False
+
+        def record_activity(msg_list):
+            nonlocal called
+            """Record an IOPub message arriving from a kernel"""
+            idents, fed_msg_list = session.feed_identities(msg_list)
+            msg = session.deserialize(fed_msg_list, content=False)
+
+            msg_type = msg["header"]["msg_type"]
+            stream.send(msg)
+            called = True
+
+        stream.on_recv(record_activity)
+        while True:
+            client.kernel_info()
+            import time
+
+            time.sleep(0.1)
+            if called:
+                break
+
+        client.stop_channels()
+        km.shutdown_kernel(now=True)
+
 
 class TestAsyncKernelManager(AsyncTestCase):
     def setUp(self):
@@ -525,7 +559,7 @@ class TestAsyncKernelManager(AsyncTestCase):
         assert mkm.call_count("start_kernel") == 1
         assert isinstance(mkm.get_kernel(kid), AsyncKMSubclass)
         assert mkm.get_kernel(kid).call_count("start_kernel") == 1
-        assert mkm.get_kernel(kid).call_count("_launch_kernel") == 1
+        assert mkm.get_kernel(kid).call_count("_async_launch_kernel") == 1
 
         assert await mkm.is_alive(kid)
         assert kid in mkm
@@ -538,12 +572,10 @@ class TestAsyncKernelManager(AsyncTestCase):
         assert mkm.call_count("restart_kernel") == 1
         assert mkm.call_count("get_kernel") == 1
         assert mkm.get_kernel(kid).call_count("restart_kernel") == 1
-        assert mkm.get_kernel(kid).call_count("shutdown_kernel") == 1
-        assert mkm.get_kernel(kid).call_count("interrupt_kernel") == 1
-        assert mkm.get_kernel(kid).call_count("_kill_kernel") == 1
-        assert mkm.get_kernel(kid).call_count("cleanup_resources") == 1
-        assert mkm.get_kernel(kid).call_count("start_kernel") == 1
-        assert mkm.get_kernel(kid).call_count("_launch_kernel") == 1
+        assert mkm.get_kernel(kid).call_count("_async_interrupt_kernel") == 1
+        assert mkm.get_kernel(kid).call_count("_async_kill_kernel") == 1
+        assert mkm.get_kernel(kid).call_count("_async_cleanup_resources") == 1
+        assert mkm.get_kernel(kid).call_count("_async_launch_kernel") == 1
 
         assert await mkm.is_alive(kid)
         assert kid in mkm.list_kernel_ids()
@@ -564,11 +596,10 @@ class TestAsyncKernelManager(AsyncTestCase):
         mkm.get_kernel(kid).reset_counts()
         mkm.reset_counts()
         await mkm.shutdown_all(now=True)
-        assert mkm.call_count("shutdown_kernel") == 1
         assert mkm.call_count("remove_kernel") == 1
-        assert mkm.call_count("request_shutdown") == 0
-        assert mkm.call_count("finish_shutdown") == 0
-        assert mkm.call_count("cleanup_resources") == 0
+        assert mkm.call_count("_async_request_shutdown") == 0
+        assert mkm.call_count("_async_finish_shutdown") == 0
+        assert mkm.call_count("_async_cleanup_resources") == 0
 
         assert kid not in mkm, f"{kid} not in {mkm}"
 
@@ -599,3 +630,34 @@ class TestAsyncKernelManager(AsyncTestCase):
         assert kernel_id in km.list_kernel_ids()
         await ensure_future(km.shutdown_kernel(kernel_id))
         assert kernel_id not in km.list_kernel_ids()
+
+    @gen_test
+    async def test_stream_on_recv(self):
+        mkm = self._get_tcp_km()
+        kid = await mkm.start_kernel(stdout=PIPE, stderr=PIPE)
+        stream = mkm.connect_iopub(kid)
+
+        km = mkm.get_kernel(kid)
+        client = km.client()
+        session = km.session
+        called = False
+
+        def record_activity(msg_list):
+            nonlocal called
+            """Record an IOPub message arriving from a kernel"""
+            idents, fed_msg_list = session.feed_identities(msg_list)
+            msg = session.deserialize(fed_msg_list, content=False)
+
+            msg_type = msg["header"]["msg_type"]
+            stream.send(msg)
+            called = True
+
+        stream.on_recv(record_activity)
+        while True:
+            await client.kernel_info()
+            if called:
+                break
+            await asyncio.sleep(0.1)
+
+        client.stop_channels()
+        await km.shutdown_kernel(now=True)
diff --git a/tests/test_restarter.py b/tests/test_restarter.py
index 5e93431..4c85686 100644
--- a/tests/test_restarter.py
+++ b/tests/test_restarter.py
@@ -5,6 +5,7 @@ import asyncio
 import json
 import os
 import sys
+from concurrent.futures import Future
 
 import pytest
 from jupyter_core import paths
@@ -89,7 +90,7 @@ async def test_restart_check(config, install_kernel, debug_logging):
     km = IOLoopKernelManager(kernel_name=install_kernel, config=config)
 
     cbs = 0
-    restarts = [asyncio.Future() for i in range(N_restarts)]
+    restarts = [Future() for i in range(N_restarts)]
 
     def cb():
         nonlocal cbs
@@ -115,7 +116,7 @@ async def test_restart_check(config, install_kernel, debug_logging):
             if i < N_restarts:
                 # Kill without cleanup to simulate crash:
                 await km.provisioner.kill()
-                await restarts[i]
+                restarts[i].result()
                 # Wait for kill + restart
                 max_wait = 10.0
                 waited = 0.0
@@ -145,7 +146,7 @@ async def test_restarter_gives_up(config, install_fail_kernel, debug_logging):
     km = IOLoopKernelManager(kernel_name=install_fail_kernel, config=config)
 
     cbs = 0
-    restarts = [asyncio.Future() for i in range(N_restarts)]
+    restarts = [Future() for i in range(N_restarts)]
 
     def cb():
         nonlocal cbs
@@ -154,7 +155,7 @@ async def test_restarter_gives_up(config, install_fail_kernel, debug_logging):
         restarts[cbs].set_result(True)
         cbs += 1
 
-    died = asyncio.Future()
+    died = Future()
 
     def on_death():
         died.set_result(True)
@@ -170,9 +171,9 @@ async def test_restarter_gives_up(config, install_fail_kernel, debug_logging):
 
     try:
         for i in range(N_restarts):
-            await restarts[i]
+            restarts[i].result()
 
-        assert await died
+        assert died.result()
         assert cbs == N_restarts
 
     finally:
diff --git a/tests/test_session.py b/tests/test_session.py
index bd59561..60f879d 100644
--- a/tests/test_session.py
+++ b/tests/test_session.py
@@ -12,7 +12,6 @@ import pytest
 import zmq
 from tornado import ioloop
 from zmq.eventloop.zmqstream import ZMQStream
-from zmq.tests import BaseZMQTestCase
 
 from jupyter_client import jsonutil
 from jupyter_client import session as ss
@@ -26,12 +25,6 @@ def _bad_unpacker(bytes):
     raise TypeError("I don't work either")
 
 
-class SessionTestCase(BaseZMQTestCase):
-    def setUp(self):
-        BaseZMQTestCase.setUp(self)
-        self.session = ss.Session()
-
-
 @pytest.fixture
 def no_copy_threshold():
     """Disable zero-copy optimizations in pyzmq >= 17"""
@@ -39,11 +32,22 @@ def no_copy_threshold():
         yield
 
 
+@pytest.fixture()
+def session():
+    return ss.Session()
+
+
 @pytest.mark.usefixtures("no_copy_threshold")
-class TestSession(SessionTestCase):
-    def test_msg(self):
+class TestSession:
+    def assertEqual(self, a, b):
+        assert a == b, (a, b)
+
+    def assertTrue(self, a):
+        assert a, a
+
+    def test_msg(self, session):
         """message format"""
-        msg = self.session.msg("execute")
+        msg = session.msg("execute")
         thekeys = set("header parent_header metadata content msg_type msg_id".split())
         s = set(msg.keys())
         self.assertEqual(s, thekeys)
@@ -56,11 +60,11 @@ class TestSession(SessionTestCase):
         self.assertEqual(msg["header"]["msg_type"], "execute")
         self.assertEqual(msg["msg_type"], "execute")
 
-    def test_serialize(self):
-        msg = self.session.msg("execute", content=dict(a=10, b=1.1))
-        msg_list = self.session.serialize(msg, ident=b"foo")
-        ident, msg_list = self.session.feed_identities(msg_list)
-        new_msg = self.session.deserialize(msg_list)
+    def test_serialize(self, session):
+        msg = session.msg("execute", content=dict(a=10, b=1.1))
+        msg_list = session.serialize(msg, ident=b"foo")
+        ident, msg_list = session.feed_identities(msg_list)
+        new_msg = session.deserialize(msg_list)
         self.assertEqual(ident[0], b"foo")
         self.assertEqual(new_msg["msg_id"], msg["msg_id"])
         self.assertEqual(new_msg["msg_type"], msg["msg_type"])
@@ -71,22 +75,22 @@ class TestSession(SessionTestCase):
         # ensure floats don't come out as Decimal:
         self.assertEqual(type(new_msg["content"]["b"]), type(new_msg["content"]["b"]))
 
-    def test_default_secure(self):
-        self.assertIsInstance(self.session.key, bytes)
-        self.assertIsInstance(self.session.auth, hmac.HMAC)
+    def test_default_secure(self, session):
+        assert isinstance(session.key, bytes)
+        assert isinstance(session.auth, hmac.HMAC)
 
-    def test_send(self):
+    def test_send_sync(self, session):
         ctx = zmq.Context()
         A = ctx.socket(zmq.PAIR)
         B = ctx.socket(zmq.PAIR)
         A.bind("inproc://test")
         B.connect("inproc://test")
 
-        msg = self.session.msg("execute", content=dict(a=10))
-        self.session.send(A, msg, ident=b"foo", buffers=[b"bar"])
+        msg = session.msg("execute", content=dict(a=10))
+        session.send(A, msg, ident=b"foo", buffers=[b"bar"])
 
-        ident, msg_list = self.session.feed_identities(B.recv_multipart())
-        new_msg = self.session.deserialize(msg_list)
+        ident, msg_list = session.feed_identities(B.recv_multipart())
+        new_msg = session.deserialize(msg_list)
         self.assertEqual(ident[0], b"foo")
         self.assertEqual(new_msg["msg_id"], msg["msg_id"])
         self.assertEqual(new_msg["msg_type"], msg["msg_type"])
@@ -98,11 +102,11 @@ class TestSession(SessionTestCase):
 
         content = msg["content"]
         header = msg["header"]
-        header["msg_id"] = self.session.msg_id
+        header["msg_id"] = session.msg_id
         parent = msg["parent_header"]
         metadata = msg["metadata"]
         header["msg_type"]
-        self.session.send(
+        session.send(
             A,
             None,
             content=content,
@@ -112,8 +116,8 @@ class TestSession(SessionTestCase):
             ident=b"foo",
             buffers=[b"bar"],
         )
-        ident, msg_list = self.session.feed_identities(B.recv_multipart())
-        new_msg = self.session.deserialize(msg_list)
+        ident, msg_list = session.feed_identities(B.recv_multipart())
+        new_msg = session.deserialize(msg_list)
         self.assertEqual(ident[0], b"foo")
         self.assertEqual(new_msg["msg_id"], header["msg_id"])
         self.assertEqual(new_msg["msg_type"], msg["msg_type"])
@@ -123,10 +127,10 @@ class TestSession(SessionTestCase):
         self.assertEqual(new_msg["parent_header"], msg["parent_header"])
         self.assertEqual(new_msg["buffers"], [b"bar"])
 
-        header["msg_id"] = self.session.msg_id
+        header["msg_id"] = session.msg_id
 
-        self.session.send(A, msg, ident=b"foo", buffers=[b"bar"])
-        ident, new_msg = self.session.recv(B)
+        session.send(A, msg, ident=b"foo", buffers=[b"bar"])
+        ident, new_msg = session.recv(B)
         self.assertEqual(ident[0], b"foo")
         self.assertEqual(new_msg["msg_id"], header["msg_id"])
         self.assertEqual(new_msg["msg_type"], msg["msg_type"])
@@ -137,21 +141,95 @@ class TestSession(SessionTestCase):
         self.assertEqual(new_msg["buffers"], [b"bar"])
 
         # buffers must support the buffer protocol
-        with self.assertRaises(TypeError):
-            self.session.send(A, msg, ident=b"foo", buffers=[1])
+        with pytest.raises(TypeError):
+            session.send(A, msg, ident=b"foo", buffers=[1])
 
         # buffers must be contiguous
         buf = memoryview(os.urandom(16))
-        with self.assertRaises(ValueError):
-            self.session.send(A, msg, ident=b"foo", buffers=[buf[::2]])
+        with pytest.raises(ValueError):
+            session.send(A, msg, ident=b"foo", buffers=[buf[::2]])
 
         A.close()
         B.close()
         ctx.term()
 
-    def test_args(self):
+    async def test_send(self, session):
+        ctx = zmq.asyncio.Context()
+        A = ctx.socket(zmq.PAIR)
+        B = ctx.socket(zmq.PAIR)
+        A.bind("inproc://test")
+        B.connect("inproc://test")
+
+        msg = session.msg("execute", content=dict(a=10))
+        session.send(A, msg, ident=b"foo", buffers=[b"bar"])
+
+        ident, msg_list = session.feed_identities(await B.recv_multipart())
+        new_msg = session.deserialize(msg_list)
+        self.assertEqual(ident[0], b"foo")
+        self.assertEqual(new_msg["msg_id"], msg["msg_id"])
+        self.assertEqual(new_msg["msg_type"], msg["msg_type"])
+        self.assertEqual(new_msg["header"], msg["header"])
+        self.assertEqual(new_msg["content"], msg["content"])
+        self.assertEqual(new_msg["parent_header"], msg["parent_header"])
+        self.assertEqual(new_msg["metadata"], msg["metadata"])
+        self.assertEqual(new_msg["buffers"], [b"bar"])
+
+        content = msg["content"]
+        header = msg["header"]
+        header["msg_id"] = session.msg_id
+        parent = msg["parent_header"]
+        metadata = msg["metadata"]
+        header["msg_type"]
+        session.send(
+            A,
+            None,
+            content=content,
+            parent=parent,
+            header=header,
+            metadata=metadata,
+            ident=b"foo",
+            buffers=[b"bar"],
+        )
+        ident, msg_list = session.feed_identities(await B.recv_multipart())
+        new_msg = session.deserialize(msg_list)
+        self.assertEqual(ident[0], b"foo")
+        self.assertEqual(new_msg["msg_id"], header["msg_id"])
+        self.assertEqual(new_msg["msg_type"], msg["msg_type"])
+        self.assertEqual(new_msg["header"], msg["header"])
+        self.assertEqual(new_msg["content"], msg["content"])
+        self.assertEqual(new_msg["metadata"], msg["metadata"])
+        self.assertEqual(new_msg["parent_header"], msg["parent_header"])
+        self.assertEqual(new_msg["buffers"], [b"bar"])
+
+        header["msg_id"] = session.msg_id
+
+        session.send(A, msg, ident=b"foo", buffers=[b"bar"])
+        ident, new_msg = session.recv(B)
+        self.assertEqual(ident[0], b"foo")
+        self.assertEqual(new_msg["msg_id"], header["msg_id"])
+        self.assertEqual(new_msg["msg_type"], msg["msg_type"])
+        self.assertEqual(new_msg["header"], msg["header"])
+        self.assertEqual(new_msg["content"], msg["content"])
+        self.assertEqual(new_msg["metadata"], msg["metadata"])
+        self.assertEqual(new_msg["parent_header"], msg["parent_header"])
+        self.assertEqual(new_msg["buffers"], [b"bar"])
+
+        # buffers must support the buffer protocol
+        with pytest.raises(TypeError):
+            session.send(A, msg, ident=b"foo", buffers=[1])
+
+        # buffers must be contiguous
+        buf = memoryview(os.urandom(16))
+        with pytest.raises(ValueError):
+            session.send(A, msg, ident=b"foo", buffers=[buf[::2]])
+
+        A.close()
+        B.close()
+        ctx.term()
+
+    def test_args(self, session):
         """initialization arguments for Session"""
-        s = self.session
+        s = session
         self.assertTrue(s.pack is ss.default_packer)
         self.assertTrue(s.unpack is ss.default_unpacker)
         self.assertEqual(s.username, os.environ.get("USER", "username"))
@@ -159,18 +237,24 @@ class TestSession(SessionTestCase):
         s = ss.Session()
         self.assertEqual(s.username, os.environ.get("USER", "username"))
 
-        self.assertRaises(TypeError, ss.Session, pack="hi")
-        self.assertRaises(TypeError, ss.Session, unpack="hi")
+        with pytest.raises(TypeError):
+            ss.Session(pack="hi")
+        with pytest.raises(TypeError):
+            ss.Session(unpack="hi")
         u = str(uuid.uuid4())
         s = ss.Session(username="carrot", session=u)
         self.assertEqual(s.session, u)
         self.assertEqual(s.username, "carrot")
 
     @pytest.mark.skipif(platform.python_implementation() == 'PyPy', reason='Test fails on PyPy')
-    def test_tracking(self):
+    def test_tracking_sync(self, session):
         """test tracking messages"""
-        a, b = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
-        s = self.session
+        ctx = zmq.Context()
+        a = ctx.socket(zmq.PAIR)
+        b = ctx.socket(zmq.PAIR)
+        a.bind("inproc://test")
+        b.connect("inproc://test")
+        s = session
         s.copy_threshold = 1
         loop = ioloop.IOLoop(make_current=False)
         ZMQStream(a, io_loop=loop)
@@ -182,23 +266,58 @@ class TestSession(SessionTestCase):
         msg = s.send(a, "hello", buffers=[M], track=True)
         t = msg["tracker"]
         self.assertTrue(isinstance(t, zmq.MessageTracker))
-        self.assertRaises(zmq.NotDone, t.wait, 0.1)
+        with pytest.raises(zmq.NotDone):
+            t.wait(0.1)
         del M
-        t.wait(1)  # this will raise
+        with pytest.raises(zmq.NotDone):
+            t.wait(1)  # this will raise
+        a.close()
+        b.close()
+        ctx.term()
 
-    def test_unique_msg_ids(self):
+    @pytest.mark.skipif(platform.python_implementation() == 'PyPy', reason='Test fails on PyPy')
+    async def test_tracking(self, session):
+        """test tracking messages"""
+        ctx = zmq.asyncio.Context()
+        a = ctx.socket(zmq.PAIR)
+        b = ctx.socket(zmq.PAIR)
+        a.bind("inproc://test")
+        b.connect("inproc://test")
+        s = session
+        s.copy_threshold = 1
+        loop = ioloop.IOLoop(make_current=False)
+        ZMQStream(a, io_loop=loop)
+
+        msg = s.send(a, "hello", track=False)
+        self.assertTrue(msg["tracker"] is ss.DONE)
+        msg = s.send(a, "hello", track=True)
+        self.assertTrue(isinstance(msg["tracker"], zmq.MessageTracker))
+        M = zmq.Message(b"hi there", track=True)
+        msg = s.send(a, "hello", buffers=[M], track=True)
+        t = msg["tracker"]
+        self.assertTrue(isinstance(t, zmq.MessageTracker))
+        with pytest.raises(zmq.NotDone):
+            t.wait(0.1)
+        del M
+        with pytest.raises(zmq.NotDone):
+            t.wait(1)  # this will raise
+        a.close()
+        b.close()
+        ctx.term()
+
+    def test_unique_msg_ids(self, session):
         """test that messages receive unique ids"""
         ids = set()
         for i in range(2**12):
-            h = self.session.msg_header("test")
+            h = session.msg_header("test")
             msg_id = h["msg_id"]
             self.assertTrue(msg_id not in ids)
             ids.add(msg_id)
 
-    def test_feed_identities(self):
+    def test_feed_identities(self, session):
         """scrub the front for zmq IDENTITIES"""
         content = dict(code="whoda", stuff=object())
-        self.session.msg("execute", content=content)
+        session.msg("execute", content=content)
 
     def test_session_id(self):
         session = ss.Session()
@@ -240,6 +359,9 @@ class TestSession(SessionTestCase):
         session._add_digest(uuid.uuid4().bytes)
         self.assertTrue(len(session.digest_history) == 91)
 
+    def assertIn(self, a, b):
+        assert a in b
+
     def test_bad_pack(self):
         try:
             ss.Session(pack=_bad_packer)
@@ -247,7 +369,7 @@ class TestSession(SessionTestCase):
             self.assertIn("could not serialize", str(e))
             self.assertIn("don't work", str(e))
         else:
-            self.fail("Should have raised ValueError")
+            raise ValueError("Should have raised ValueError")
 
     def test_bad_unpack(self):
         try:
@@ -256,7 +378,7 @@ class TestSession(SessionTestCase):
             self.assertIn("could not handle output", str(e))
             self.assertIn("don't work either", str(e))
         else:
-            self.fail("Should have raised ValueError")
+            raise ValueError("Should have raised ValueError")
 
     def test_bad_packer(self):
         try:
@@ -265,7 +387,7 @@ class TestSession(SessionTestCase):
             self.assertIn("could not serialize", str(e))
             self.assertIn("don't work", str(e))
         else:
-            self.fail("Should have raised ValueError")
+            raise ValueError("Should have raised ValueError")
 
     def test_bad_unpacker(self):
         try:
@@ -274,10 +396,10 @@ class TestSession(SessionTestCase):
             self.assertIn("could not handle output", str(e))
             self.assertIn("don't work either", str(e))
         else:
-            self.fail("Should have raised ValueError")
+            raise ValueError("Should have raised ValueError")
 
     def test_bad_roundtrip(self):
-        with self.assertRaises(ValueError):
+        with pytest.raises(ValueError):
             ss.Session(unpack=lambda b: 5)
 
     def _datetime_test(self, session):
@@ -298,8 +420,8 @@ class TestSession(SessionTestCase):
         self.assertEqual(msg["content"], jsonutil.extract_dates(msg2["content"]))
         self.assertEqual(msg["content"], jsonutil.extract_dates(msg2["content"]))
 
-    def test_datetimes(self):
-        self._datetime_test(self.session)
+    def test_datetimes(self, session):
+        self._datetime_test(session)
 
     def test_datetimes_pickle(self):
         session = ss.Session(packer="pickle")
@@ -314,22 +436,47 @@ class TestSession(SessionTestCase):
         )
         self._datetime_test(session)
 
-    def test_send_raw(self):
+    def test_send_raw_sync(self, session):
         ctx = zmq.Context()
         A = ctx.socket(zmq.PAIR)
         B = ctx.socket(zmq.PAIR)
         A.bind("inproc://test")
         B.connect("inproc://test")
 
-        msg = self.session.msg("execute", content=dict(a=10))
+        msg = session.msg("execute", content=dict(a=10))
+        msg_list = [
+            session.pack(msg[part]) for part in ["header", "parent_header", "metadata", "content"]
+        ]
+        session.send_raw(A, msg_list, ident=b"foo")
+
+        ident, new_msg_list = session.feed_identities(B.recv_multipart())
+        new_msg = session.deserialize(new_msg_list)
+        self.assertEqual(ident[0], b"foo")
+        self.assertEqual(new_msg["msg_type"], msg["msg_type"])
+        self.assertEqual(new_msg["header"], msg["header"])
+        self.assertEqual(new_msg["parent_header"], msg["parent_header"])
+        self.assertEqual(new_msg["content"], msg["content"])
+        self.assertEqual(new_msg["metadata"], msg["metadata"])
+
+        A.close()
+        B.close()
+        ctx.term()
+
+    async def test_send_raw(self, session):
+        ctx = zmq.asyncio.Context()
+        A = ctx.socket(zmq.PAIR)
+        B = ctx.socket(zmq.PAIR)
+        A.bind("inproc://test")
+        B.connect("inproc://test")
+
+        msg = session.msg("execute", content=dict(a=10))
         msg_list = [
-            self.session.pack(msg[part])
-            for part in ["header", "parent_header", "metadata", "content"]
+            session.pack(msg[part]) for part in ["header", "parent_header", "metadata", "content"]
         ]
-        self.session.send_raw(A, msg_list, ident=b"foo")
+        session.send_raw(A, msg_list, ident=b"foo")
 
-        ident, new_msg_list = self.session.feed_identities(B.recv_multipart())
-        new_msg = self.session.deserialize(new_msg_list)
+        ident, new_msg_list = session.feed_identities(B.recv_multipart().result())
+        new_msg = session.deserialize(new_msg_list)
         self.assertEqual(ident[0], b"foo")
         self.assertEqual(new_msg["msg_type"], msg["msg_type"])
         self.assertEqual(new_msg["header"], msg["header"])
@@ -341,8 +488,8 @@ class TestSession(SessionTestCase):
         B.close()
         ctx.term()
 
-    def test_clone(self):
-        s = self.session
+    def test_clone(self, session):
+        s = session
         s._add_digest("initial")
         s2 = s.clone()
         assert s2.session == s.session
diff --git a/tests/utils.py b/tests/utils.py
index 10f5495..317fdce 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -142,6 +142,10 @@ class KMSubclass(RecordCallMixin):
     def shutdown_kernel(self, now=False, restart=False):
         """Record call and defer to superclass"""
 
+    @subclass_recorder
+    def _async_shutdown_kernel(self, now=False, restart=False):
+        """Record call and defer to superclass"""
+
     @subclass_recorder
     def restart_kernel(self, now=False, **kw):
         """Record call and defer to superclass"""
@@ -150,6 +154,10 @@ class KMSubclass(RecordCallMixin):
     def interrupt_kernel(self):
         """Record call and defer to superclass"""
 
+    @subclass_recorder
+    def _async_interrupt_kernel(self):
+        """Record call and defer to superclass"""
+
     @subclass_recorder
     def request_shutdown(self, restart=False):
         """Record call and defer to superclass"""
@@ -159,27 +167,39 @@ class KMSubclass(RecordCallMixin):
         """Record call and defer to superclass"""
 
     @subclass_recorder
-    def _launch_kernel(self, kernel_cmd, **kw):
+    def _async_launch_kernel(self, kernel_cmd, **kw):
         """Record call and defer to superclass"""
 
     @subclass_recorder
-    def _kill_kernel(self):
+    def _async_kill_kernel(self):
         """Record call and defer to superclass"""
 
     @subclass_recorder
     def cleanup_resources(self, restart=False):
         """Record call and defer to superclass"""
 
+    @subclass_recorder
+    def _async_cleanup_resources(self, restart=False):
+        """Record call and defer to superclass"""
+
     @subclass_recorder
     def signal_kernel(self, signum: int):
         """Record call and defer to superclass"""
 
+    @subclass_recorder
+    def _async_signal_kernel(self, signum: int):
+        """Record call and defer to superclass"""
+
     @subclass_recorder
     def is_alive(self):
         """Record call and defer to superclass"""
 
     @subclass_recorder
-    def _send_kernel_sigterm(self, restart: bool = False):
+    def _async_is_alive(self):
+        """Record call and defer to superclass"""
+
+    @subclass_recorder
+    def _async_send_kernel_sigterm(self, restart: bool = False):
         """Record call and defer to superclass"""
 
 
@@ -211,6 +231,10 @@ class MKMSubclass(RecordCallMixin):
     def start_kernel(self, kernel_name=None, **kwargs):
         """Record call and defer to superclass"""
 
+    @subclass_recorder
+    def _async_start_kernel(self, kernel_name=None, **kwargs):
+        """Record call and defer to superclass"""
+
     @subclass_recorder
     def shutdown_kernel(self, kernel_id, now=False, restart=False):
         """Record call and defer to superclass"""
@@ -227,10 +251,18 @@ class MKMSubclass(RecordCallMixin):
     def request_shutdown(self, kernel_id, restart=False):
         """Record call and defer to superclass"""
 
+    @subclass_recorder
+    def _async_request_shutdown(self, kernel_id, restart=False):
+        """Record call and defer to superclass"""
+
     @subclass_recorder
     def finish_shutdown(self, kernel_id, waittime=None, pollinterval=0.1, restart=False):
         """Record call and defer to superclass"""
 
+    @subclass_recorder
+    def _async_finish_shutdown(self, kernel_id, waittime=None, pollinterval=0.1, restart=False):
+        """Record call and defer to superclass"""
+
     @subclass_recorder
     def cleanup_resources(self, kernel_id, restart=False):
         """Record call and defer to superclass"""