use tornado for better compatiblity than raw asyncio

when we are running our own loops for sync apis

Author: minrk

ipyparallel/_async.py

@@ -1,9 +1,21 @@
 """Async utilities"""
 import asyncio
+import concurrent.futures
 import inspect
-from concurrent.futures import ThreadPoolExecutor
+import threading
 from functools import partial
 
+from tornado.ioloop import IOLoop
+
+
+def _asyncio_run(coro):
+    """Like asyncio.run, but works when there's no event loop"""
+    # for now: using tornado for broader compatibility with FDs,
+    # e.g. when using the only partially functional default
+    # Proactor on windows
+    loop = IOLoop()
+    return loop.run_sync(lambda: asyncio.ensure_future(coro))
+
 
 class AsyncFirst:
     """Wrapper class that defines synchronous `_sync` method wrappers
@@ -16,13 +28,30 @@ class AsyncFirst:
 
     _async_thread = None
 
+    def _thread_main(self):
+        asyncio_loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(asyncio_loop)
+        loop = self._thread_loop = IOLoop.current()
+        loop.add_callback(self._loop_started.set)
+        loop.start()
+
     def _in_thread(self, async_f, *args, **kwargs):
         """Run an async function in a background thread"""
         if self._async_thread is None:
-            self._async_thread = ThreadPoolExecutor(1)
-        future = self._async_thread.submit(
-            lambda: asyncio.run(async_f(*args, **kwargs))
-        )
+            self._loop_started = threading.Event()
+            self._async_thread = threading.Thread(target=self._thread_main, daemon=True)
+            self._async_thread.start()
+            self._loop_started.wait(timeout=5)
+
+        future = concurrent.futures.Future()
+
+        async def thread_callback():
+            try:
+                future.set_result(await async_f(*args, **kwargs))
+            except Exception as e:
+                future.set_exception(e)
+
+        self._thread_loop.add_callback(thread_callback)
         return future.result()
 
     def _synchronize(self, async_f, *args, **kwargs):
@@ -31,10 +60,16 @@ def _synchronize(self, async_f, *args, **kwargs):
         Uses asyncio.run if asyncio is not running,
         otherwise puts it in a background thread
         """
-        if asyncio.get_event_loop().is_running():
+        try:
+            loop = asyncio.get_event_loop()
+        except RuntimeError:
+            # sometimes get returns a RuntimeError
+            # if there's no current loop under certain policies
+            loop = None
+        if loop and loop.is_running():
             return self._in_thread(async_f, *args, **kwargs)
         else:
-            return asyncio.run(async_f(*args, **kwargs))
+            return _asyncio_run(async_f(*args, **kwargs))
 
     def __getattr__(self, name):
         if name.endswith("_sync"):

ipyparallel/tests/test_async.py

@@ -32,13 +32,13 @@ def test_getattr():
         a.sync_method_sync
 
 
-def test_asyncio_run():
+def test_sync_no_asyncio():
     a = A()
     assert a.async_method_sync() == 'async'
     assert a._async_thread is None
 
 
-async def test_asyncio_sync():
+async def test_sync_asyncio():
     a = A()
     assert a.async_method_sync() == 'async'
     assert a._async_thread is not None