Merge pull request #518 from minrk/async-def

remove some use of old deprecated coroutine syntax

Author: minrk

ipyparallel/client/magics.py

@@ -43,17 +43,6 @@ def nullcontext():
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 # -----------------------------------------------------------------------------
-# -----------------------------------------------------------------------------
-# Imports
-# -----------------------------------------------------------------------------
-
-try:
-    from asyncio import coroutine
-except ImportError:  # py2
-
-    def coroutine(f):
-        raise NotImplementedError()
-
 
 import inspect
 import re
@@ -79,6 +68,20 @@ def _iscoroutinefunction(f):
     return False
 
 
+def _asyncify(f):
+    """Wrap a blocking call in a coroutine
+
+    Does not make the call non-blocking,
+    just conforms to the API assuming it is awaitable.
+    For use when patching-in replacement methods.
+    """
+
+    async def async_f(*args, **kwargs):
+        return f(*args, **kwargs)
+
+    return async_f
+
+
 NO_LAST_RESULT = "%pxresult recalls last %px result, which has not yet been used."
 
 
@@ -444,14 +447,14 @@ def _enable_autopx(self):
         if _iscoroutinefunction(self.shell.run_cell):
             # original is a coroutine,
             # wrap ours in a coroutine
-            pxrun_cell = coroutine(pxrun_cell)
+            pxrun_cell = _asyncify(pxrun_cell)
         self.shell.run_cell = pxrun_cell
 
         pxrun_nodes = self.pxrun_nodes
         if _iscoroutinefunction(self.shell.run_ast_nodes):
             # original is a coroutine,
             # wrap ours in a coroutine
-            pxrun_nodes = coroutine(pxrun_nodes)
+            pxrun_nodes = _asyncify(pxrun_nodes)
         self.shell.run_ast_nodes = pxrun_nodes
 
         self._autopx = True

ipyparallel/controller/hub.py

@@ -7,6 +7,7 @@
 # Distributed under the terms of the Modified BSD License.
 from __future__ import print_function
 
+import inspect
 import json
 import os
 import sys
@@ -20,8 +21,6 @@
 from jupyter_client.jsonutil import parse_date
 from jupyter_client.session import Session
 from tornado import ioloop
-from tornado.gen import coroutine
-from tornado.gen import maybe_future
 from traitlets import Any
 from traitlets import Bytes
 from traitlets import Dict
@@ -319,8 +318,7 @@ def dispatch_monitor_traffic(self, msg):
             self.log.error("Unrecognized monitor topic: %r", switch)
 
     @util.log_errors
-    @coroutine
-    def dispatch_query(self, msg):
+    async def dispatch_query(self, msg):
         """Route registration requests and queries from clients."""
         try:
             idents, msg = self.session.feed_identities(msg)
@@ -357,8 +355,8 @@ def dispatch_query(self, msg):
 
         try:
             f = handler(idents, msg)
-            if f:
-                yield maybe_future(f)
+            if f and inspect.isawaitable(f):
+                await f
         except Exception:
             content = error.wrap_exception()
             self.log.error("Error handling request: %r", msg_type, exc_info=True)
@@ -1450,16 +1448,15 @@ def db_query(self, client_id, msg):
             buffers=buffers,
         )
 
-    @coroutine
-    def become_dask(self, client_id, msg):
+    async def become_dask(self, client_id, msg):
         """Start a dask.distributed Scheduler."""
         if self.distributed_scheduler is None:
             kwargs = msg['content'].get('scheduler_args', {})
             self.log.info("Becoming dask.distributed scheduler: %s", kwargs)
             from distributed import Scheduler
 
             self.distributed_scheduler = scheduler = Scheduler(**kwargs)
-            yield scheduler.start()
+            await scheduler.start()
         content = {
             'status': 'ok',
             'ip': self.distributed_scheduler.ip,

ipyparallel/util.py

@@ -2,6 +2,9 @@
 """Some generic utilities for dealing with classes, urls, and serialization."""
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
+import asyncio
+import functools
+import inspect
 import logging
 import os
 import re
@@ -23,7 +26,6 @@
 import zmq
 from dateutil.parser import parse as dateutil_parse
 from dateutil.tz import tzlocal
-from decorator import decorator
 from IPython import get_ipython
 from IPython.core.profiledir import ProfileDir
 from IPython.core.profiledir import ProfileDirError
@@ -101,17 +103,32 @@ def get(self, key, default=None):
 # -----------------------------------------------------------------------------
 
 
-@decorator
-def log_errors(f, self, *args, **kwargs):
+def log_errors(f):
     """decorator to log unhandled exceptions raised in a method.
 
     For use wrapping on_recv callbacks, so that exceptions
     do not cause the stream to be closed.
     """
-    try:
-        return f(self, *args, **kwargs)
-    except Exception as e:
-        self.log.error("Uncaught exception in %r" % f, exc_info=True)
+
+    @functools.wraps(f)
+    def logs_errors(self, *args, **kwargs):
+        try:
+            result = f(self, *args, **kwargs)
+        except Exception:
+            self.log.error("Uncaught exception in {f}: {future.exception()}")
+            return
+
+        if inspect.isawaitable(result):
+            # if it's async, schedule logging for when the future resolves
+            future = asyncio.ensure_future(result)
+
+            def _log_error(future):
+                if future.exception():
+                    self.log.error("Uncaught exception in {f}: {future.exception()}")
+
+            future.add_done_callback(_log_error)
+
+    return logs_errors
 
 
 def is_url(url):