Merge pull request #835 from blink1073/synchronous_managers

Author: blink1073

.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

.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

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 _(self, *args, **kwargs):
         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,18 +52,19 @@ 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
 
     # replies come on the shell channel
     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)

jupyter_client/blocking/client.py

@@ -20,7 +20,7 @@ def _(self, *args, **kwargs):
         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 _
 

jupyter_client/channels.py

@@ -14,6 +14,7 @@
 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,15 +50,15 @@ 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] = "",
     ):
         """Create the heartbeat monitor 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 @@ def _create_socket(self) -> None:
 
         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,16 +122,16 @@ async def _async_run(self) -> None:
 
             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)
             # poll(0) means return immediately (see http://api.zeromq.org/2-1:zmq-poll)
             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 @@ async def _async_run(self) -> None:
                 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 @@ def call_handlers(self, since_last_heartbeat: float) -> None:
 
 
 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 @@ def __init__(self, socket: zmq.asyncio.Socket, session: Session, loop: t.Any = N
         """
         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 @@ def send(self, msg: t.Dict[str, t.Any]) -> None:
 
     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))