io handler task

longer but more explicit: only one message is handled at a time

rplugin/python/nvim_ipy.py

@@ -75,17 +75,37 @@ def __init__(self, wraps):
     def __getattr__(self, name):
         return partial(self.wraps.vim.session.threadsafe_call, getattr(self.wraps, name))
 
+# FIXME: un-reinvent this wheel
+class MsgQueue(object):
+    def __init__(self):
+        self.msgs = deque()
+        self.waiters = deque()
+
+    def get(self):
+        while not self.msgs:
+            gr = greenlet.getcurrent()
+            self.waiters.append(gr)
+            gr.parent.switch()
+        return self.msgs.popleft()
+
+    def put(self, msg):
+        self.msgs.append(msg)
+        if self.waiters:
+            handler = self.waiters.popleft()
+            handler.parent = greenlet.getcurrent()
+            handler.switch()
 
 @neovim.plugin
 class IPythonPlugin(object):
     def __init__(self, vim):
         self.vim = vim
         self.buf = None
         self.has_connection = False
+
         self.pending_shell_msgs = {}
+        self.io_msgs = MsgQueue()
+        self.handle_io()
 
-        self.io_msgs = deque()
-        self.handling_io = False
 
     def configure(self):
         #FIXME: rethink the entire configuration interface thing
@@ -270,15 +290,13 @@ def on_ipy_interrupt(self, args):
     def on_ipy_terminate(self, args):
         self.km.shutdown_kernel()
 
-    # this will be simpler when we switch to single-thread IO
     def on_iopub_msg(self, m):
-        self.io_msgs.append(m)
-        if self.handling_io:
-            return
-        self.handling_io = True
-        while self.io_msgs:
-            self._on_iopub_msg(self.io_msgs.popleft())
-        self.handling_io = False
+        self.io_msgs.put(m)
+
+    @ipy_async
+    def handle_io(self):
+        while True:
+            self._on_iopub_msg(self.io_msgs.get())
 
     def _on_iopub_msg(self, m):
         #FIXME: figure out the smoothest way to to matchaddpos