Replace Session.post by Session.threadsafe_call

Session.post is a hack to allow communication from other threads, but it is much
simpler and cleaner to allow callbacks be invoked from the main thread.

Also support an optional "setup callback" in the Session.run.

Author: tarruda

neovim/api/common.py

@@ -204,22 +204,23 @@ def __init__(self, session, hook):
         self._in = self._hook.from_nvim
         self._out = self._hook.to_nvim
 
-    def post(self, name, *args):
-        """Wrapper for Session.post."""
-        self._session.post(name, *args)
+    def threadsafe_call(self, fn, *args, **kwargs):
+        """Wrapper for Session.threadsafe_call."""
+        self._session.threadsafe_call(fn, *args, **kwargs)
 
     def next_message(self):
         """Wrapper for Session.next_message."""
         msg = self._session.next_message()
-        return walk(self._in, msg, self, msg[1], msg[0])
+        if msg:
+            return walk(self._in, msg, self, msg[1], msg[0])
 
     def request(self, name, *args):
         """Wrapper for Session.request."""
         args = walk(self._out, args, self, name, 'out-request')
         return walk(self._in, self._session.request(name, *args), self, name,
                     'out-request')
 
-    def run(self, request_cb, notification_cb):
+    def run(self, request_cb, notification_cb, setup_cb=None):
         """Wrapper for Session.run."""
         def filter_request_cb(name, args):
             result = request_cb(self._in(name, self, name, 'request'),
@@ -230,7 +231,7 @@ def filter_notification_cb(name, args):
             notification_cb(self._in(name, self, name, 'notification'),
                             walk(self._in, args, self, name, 'notification'))
 
-        self._session.run(filter_request_cb, filter_notification_cb)
+        self._session.run(filter_request_cb, filter_notification_cb, setup_cb)
 
     def stop(self):
         """Wrapper for Session.stop."""

neovim/msgpack_rpc/async_session.py

@@ -28,16 +28,9 @@ def __init__(self, msgpack_stream):
             2: self._on_notification
         }
 
-    def post(self, method, args):
-        """Post a notification to the queue from another thread.
-
-        A msgpack-rpc notification with method `method` and argument `args` is
-        posted to the notification queue. This can be used to send messages
-        from other threads.
-        """
-        # We encode method names to be consitent with names coming from Nvim,
-        # which always come as byte strings
-        self._msgpack_stream.post((2, method.encode('utf-8'), args,))
+    def threadsafe_call(self, fn):
+        """Wrapper around `MsgpackStream.threadsafe_call`."""
+        self._msgpack_stream.threadsafe_call(fn)
 
     def request(self, method, args, response_cb):
         """Send a msgpack-rpc request to Nvim.

neovim/msgpack_rpc/event_loop/asyncio.py

@@ -107,8 +107,8 @@ def _run(self):
     def _stop(self):
         self._loop.stop()
 
-    def _interrupt(self):
-        self._loop.call_soon_threadsafe(lambda: self.stop())
+    def _threadsafe_call(self, fn):
+        self._loop.call_soon_threadsafe(fn)
 
     def _setup_signals(self, signals):
         self._signals = list(signals)

neovim/msgpack_rpc/event_loop/base.py

@@ -1,7 +1,6 @@
 """Common code for event loop implementations."""
 import logging
 import signal
-import threading
 
 
 logger = logging.getLogger(__name__)
@@ -112,11 +111,13 @@ def send(self, data):
         debug("Sending '%s'", data)
         self._send(data)
 
-    def interrupt(self):
-        """Stop the event loop from another thread."""
-        debug('Interrupted event loop from thread %s',
-              threading.current_thread())
-        self._interrupt()
+    def threadsafe_call(self, fn):
+        """Call a function in the event loop thread.
+
+        This is the only safe way to interact with a session from other
+        threads.
+        """
+        self._threadsafe_call(fn)
 
     def run(self, data_cb):
         """Run the event loop."""

neovim/msgpack_rpc/event_loop/uv.py

@@ -1,5 +1,6 @@
 """Event loop implementation that uses pyuv(libuv-python bindings)."""
 import sys
+from collections import deque
 
 import pyuv
 
@@ -12,9 +13,10 @@ class UvEventLoop(BaseEventLoop):
 
     def _init(self):
         self._loop = pyuv.Loop()
-        self._async = pyuv.Async(self._loop, lambda h: self.stop())
+        self._async = pyuv.Async(self._loop, self._on_async)
         self._connection_error = None
         self._error_stream = None
+        self._callbacks = deque()
 
     def _on_connect(self, stream, error):
         self.stop()
@@ -96,9 +98,14 @@ def _run(self):
     def _stop(self):
         self._loop.stop()
 
-    def _interrupt(self):
+    def _threadsafe_call(self, fn):
+        self._callbacks.append(fn)
         self._async.send()
 
+    def _on_async(self, handle):
+        while self._callbacks:
+            self._callbacks.popleft()()
+
     def _setup_signals(self, signals):
         self._signal_handles = []
         handler = lambda h, signum: self._on_signal(signum)

neovim/msgpack_rpc/msgpack_stream.py

@@ -1,6 +1,5 @@
 """Msgpack handling in the event loop pipeline."""
 import logging
-from collections import deque
 
 from msgpack import Packer, Unpacker
 
@@ -20,24 +19,17 @@ class MsgpackStream(object):
     def __init__(self, event_loop):
         """Wrap `event_loop` on a msgpack-aware interface."""
         self._event_loop = event_loop
-        self._posted = deque()
         self._packer = Packer(use_bin_type=True, encoding=None)
         self._unpacker = Unpacker()
         self._message_cb = None
-        self._stopped = False
 
     def set_packer_encoding(self, encoding):
         """Switch encoding for Unicode strings."""
         self._packer = Packer(use_bin_type=True, encoding=encoding)
 
-    def post(self, msg):
-        """Post `msg` to the read queue of the `MsgpackStream` instance.
-
-        Use the event loop `interrupt()` method to push msgpack objects from
-        other threads.
-        """
-        self._posted.append(msg)
-        self._event_loop.interrupt()
+    def threadsafe_call(self, fn):
+        """Wrapper around `BaseEventLoop.threadsafe_call`."""
+        self._event_loop.threadsafe_call(fn)
 
     def send(self, msg):
         """Queue `msg` for sending to Nvim."""
@@ -51,22 +43,13 @@ def run(self, message_cb):
         a message has been successfully parsed from the input stream.
         """
         self._message_cb = message_cb
-        self._run()
+        self._event_loop.run(self._on_data)
         self._message_cb = None
 
     def stop(self):
         """Stop the event loop."""
-        self._stopped = True
         self._event_loop.stop()
 
-    def _run(self):
-        self._stopped = False
-        while not self._stopped:
-            if self._posted:
-                self._message_cb(self._posted.popleft())
-                continue
-            self._event_loop.run(self._on_data)
-
     def _on_data(self, data):
         self._unpacker.feed(data)
         while True:

neovim/msgpack_rpc/session.py

@@ -25,10 +25,18 @@ def __init__(self, async_session):
         self._request_cb = self._notification_cb = None
         self._pending_messages = deque()
         self._is_running = False
+        self._setup_exception = None
 
-    def post(self, name, *args):
-        """Simple wrapper around `AsyncSession.post`."""
-        self._async_session.post(name, args)
+    def threadsafe_call(self, fn, *args, **kwargs):
+        """Wrapper around `AsyncSession.threadsafe_call`."""
+        def handler():
+            fn(*args, **kwargs)
+
+        def greenlet_wrapper():
+            gr = greenlet.greenlet(handler)
+            gr.switch()
+
+        self._async_session.threadsafe_call(greenlet_wrapper)
 
     def next_message(self):
         """Block until a message(request or notification) is available.
@@ -42,7 +50,8 @@ def next_message(self):
             return self._pending_messages.popleft()
         self._async_session.run(self._enqueue_request_and_stop,
                                 self._enqueue_notification_and_stop)
-        return self._pending_messages.popleft()
+        if self._pending_messages:
+            return self._pending_messages.popleft()
 
     def request(self, method, *args):
         """Send a msgpack-rpc request and block until as response is received.
@@ -67,7 +76,7 @@ def request(self, method, *args):
             raise self.error_wrapper(err)
         return rv
 
-    def run(self, request_cb, notification_cb):
+    def run(self, request_cb, notification_cb, setup_cb=None):
         """Run the event loop to receive requests and notifications from Nvim.
 
         Like `AsyncSession.run()`, but `request_cb` and `notification_cb` are
@@ -76,6 +85,23 @@ def run(self, request_cb, notification_cb):
         self._request_cb = request_cb
         self._notification_cb = notification_cb
         self._is_running = True
+        self._setup_exception = None
+
+        def on_setup():
+            try:
+                setup_cb()
+            except Exception as e:
+                self._setup_exception = e
+                self.stop()
+
+        if setup_cb:
+            # Create a new greenlet to handle the setup function
+            gr = greenlet.greenlet(on_setup)
+            gr.switch()
+
+        if self._setup_exception:
+            raise self._setup_exception
+
         # Process all pending requests and notifications
         while self._pending_messages:
             msg = self._pending_messages.popleft()
@@ -85,6 +111,9 @@ def run(self, request_cb, notification_cb):
         self._request_cb = None
         self._notification_cb = None
 
+        if self._setup_exception:
+            raise self._setup_exception
+
     def stop(self):
         """Stop the event loop."""
         self._async_session.stop()

setup.py

@@ -24,6 +24,6 @@
       author_email='[email protected]',
       license='MIT',
       packages=['neovim', 'neovim.api', 'neovim.msgpack_rpc',
-                'neovim.msgpack_rpc.event_loop', 'neovim.plugins'],
+                'neovim.msgpack_rpc.event_loop', 'neovim.plugin'],
       install_requires=install_requires,
       zip_safe=False)

test/test_client_rpc.py

@@ -7,8 +7,7 @@
 
 @with_setup(setup=cleanup)
 def test_call_and_reply():
-    def notification_cb(name, args):
-        eq(name, 'setup')
+    def setup_cb():
         cmd = 'let g:result = rpcrequest(%d, "client-call", 1, 2, 3)' % cid
         vim.command(cmd)
         eq(vim.vars['result'], [4, 5, 6])
@@ -19,33 +18,27 @@ def request_cb(name, args):
         eq(args, [1, 2, 3])
         return [4, 5, 6]
 
-    vim.session.post('setup')
-    vim.session.run(request_cb, notification_cb)
+    vim.session.run(request_cb, None, setup_cb)
 
 
 @with_setup(setup=cleanup)
 def test_call_api_before_reply():
-    def notification_cb(name, args):
-        eq(name, 'setup2')
+    def setup_cb():
         cmd = 'let g:result = rpcrequest(%d, "client-call2", 1, 2, 3)' % cid
         vim.command(cmd)
         eq(vim.vars['result'], [7, 8, 9])
         vim.session.stop()
 
     def request_cb(name, args):
-        eq(name, 'client-call2')
-        eq(args, [1, 2, 3])
         vim.command('let g:result2 = [7, 8, 9]')
         return vim.vars['result2']
 
-    vim.session.post('setup2')
-    vim.session.run(request_cb, notification_cb)
+    vim.session.run(request_cb, None, setup_cb)
 
 
 @with_setup(setup=cleanup)
 def test_recursion():
-    def notification_cb(name, args):
-        eq(name, 'setup3')
+    def setup_cb():
         vim.vars['result1'] = 0
         vim.vars['result2'] = 0
         vim.vars['result3'] = 0
@@ -71,5 +64,4 @@ def request_cb(name, args):
             vim.command(cmd)
         return n
 
-    vim.session.post('setup3')
-    vim.session.run(request_cb, notification_cb)
+    vim.session.run(request_cb, None, setup_cb)

test/test_concurrency.py

@@ -1,33 +1,11 @@
-from time import sleep
-from random import random
 from nose.tools import with_setup, eq_ as eq
 from common import vim, cleanup
-from threading import Thread, Timer
-
-
-@with_setup(setup=cleanup)
-def test_custom_messages():
-    def produce(i):
-        vim.session.post('custom-message', None)
-        sleep(0.05 * random())
-
-    count = 50
-    for i in range(count):
-        t = Thread(target=produce, args=(i,))
-        t.daemon = True
-        t.start()
-
-    custom_messages = []
-    while len(custom_messages) < 50:
-        custom_messages.append(vim.session.next_message())
-
-    eq(len(custom_messages), 50)
+from threading import Timer
 
 
 @with_setup(setup=cleanup)
 def test_interrupt_from_another_thread():
-    timer = Timer(0.5, lambda: vim.session.post('timeout'))
+    session = vim.session
+    timer = Timer(0.5, lambda: session.threadsafe_call(lambda: session.stop()))
     timer.start()
-    msg = vim.session.next_message()
-    eq(msg[0], 'notification')
-    eq(msg[1], 'timeout')
+    eq(vim.session.next_message(), None)