Rope and Jedi race working with async functions

pyls/_utils.py

@@ -4,6 +4,8 @@
 import logging
 import os
 import threading
+import trio
+import sys
 
 log = logging.getLogger(__name__)
 
@@ -163,7 +165,7 @@ def is_process_alive(pid):
             return True
 
 
-def race_hooks(hook_caller, pool, **kwargs):
+def race_hooks(hook_caller, **kwargs):
     """Given a pluggy hook spec, execute impls in parallel returning the first non-None result.
 
     Note this does not support a lot of pluggy functionality, e.g. hook wrappers.
@@ -174,15 +176,30 @@ def race_hooks(hook_caller, pool, **kwargs):
     if not impls:
         return None
 
-    def _apply(impl):
-        return impl, impl.function(**kwargs)
+    async def race(impls):
+        send_channel, receive_channel = trio.open_memory_channel(0)
+
+        async def _apply(impl):
+            return impl, impl.function(**kwargs)
+
+        async def jockey(impl):
+            await send_channel.send(await _apply(impl))
+
+        async with trio.open_nursery() as nursery:
+            for impl in impls:
+                nursery.start_soon(jockey, impl)
+
+            winner = await receive_channel.receive()
+            nursery.cancel_scope.cancel()
+            return winner
 
     # imap unordered gives us an iterator over the items in the order they finish.
     # We have to be careful to set chunksize to 1 to ensure hooks each get their own thread.
     # Unfortunately, there's no way to interrupt these threads, so we just have to leave them be.
     result = None
     while result is None:
-        first_impl, result = next(pool.imap_unordered(_apply, impls, chunksize=1))
+        #  first_impl, result = next(pool.imap_unordered(_apply, impls, chunksize=1))
+        first_impl, result = trio.run(race, impls)
     log.debug("Hook from plugin %s returned: %s", first_impl.plugin_name,
               result)
     return result

pyls/python_ls.py

@@ -2,7 +2,7 @@
 import logging
 import socketserver
 import threading
-from multiprocessing import dummy as multiprocessing
+import sys
 
 from pyls_jsonrpc.dispatchers import MethodDispatcher
 from pyls_jsonrpc.endpoint import Endpoint
@@ -17,7 +17,6 @@
 LINT_DEBOUNCE_S = 0.5  # 500 ms
 PARENT_PROCESS_WATCH_INTERVAL = 10  # 10 s
 MAX_WORKERS = 64
-PLUGGY_RACE_POOL_SIZE = 5
 PYTHON_FILE_EXTENSIONS = ('.py', '.pyi')
 CONFIG_FILEs = ('pycodestyle.cfg', 'setup.cfg', 'tox.ini', '.flake8')
 
@@ -76,7 +75,6 @@ class PythonLanguageServer(MethodDispatcher):
     def __init__(self, rx, tx, check_parent_process=False):
         self.workspace = None
         self.config = None
-        self._pool = None
 
         self._jsonrpc_stream_reader = JsonRpcStreamReader(rx)
         self._jsonrpc_stream_writer = JsonRpcStreamWriter(tx)
@@ -165,7 +163,6 @@ def m_initialize(self, processId=None, rootUri=None, rootPath=None, initializati
         self.config = config.Config(rootUri, initializationOptions or {},
                                     processId, _kwargs.get('capabilities', {}))
         self._dispatchers = self._hook('pyls_dispatchers')
-        self._pool = multiprocessing.Pool(PLUGGY_RACE_POOL_SIZE)
         self._hook('pyls_initialize')
 
         if self._check_parent_process and processId is not None:
@@ -195,10 +192,10 @@ def code_lens(self, doc_uri):
 
     def completions(self, doc_uri, position):
         rope_enabled = self.config.settings()['plugins']['rope_completion']['enabled']
-        if rope_enabled:
+        python_version = sys.version_info[0]
+        if rope_enabled and python_version == 3:
             completions = _utils.race_hooks(
                 self._hook_caller('pyls_completions'),
-                self._pool,
                 document=self.workspace.get_document(doc_uri) if doc_uri else None,
                 position=position,
                 config=self.config,