Let the GSource dispatch python callbacks

Turn things around and rather than iterating the GLib main context from
python iterate the python mainloop from GLib.

In order to do this, we need to be able to calculate the timeout (and
whether anything can be dispatched) from inside the prepare and check
functions of the source.
We can do that easily by looking at the _ready and _schedule attributes
of the loop. Once we have that, we can dispatch everything scheduled by
python by calling _run_once and relying on our select() implementation
to return immediately.

Author: Benjamin Berg

asyncio_glib/glib_events.py

@@ -1,10 +1,19 @@
+import sys
 import asyncio
+from asyncio import events
 import threading
 
 from gi.repository import GLib
 
 from . import glib_selector
 
+# The override for GLib.MainLoop.run installs a signal wakeup fd,
+# which interferes with asyncio signal handlers.  Try to get the
+# direct version.
+try:
+    g_main_loop_run = super(GLib.MainLoop, GLib.MainLoop).run
+except AttributeError:
+    g_main_loop_run = GLib.MainLoop.run
 
 __all__ = (
     'GLibEventLoop',
@@ -13,20 +22,136 @@
 
 
 class GLibEventLoop(asyncio.SelectorEventLoop):
-    """An asyncio event loop that runs the GLib main loop"""
+    """An asyncio event loop that runs the GLib context using a main loop"""
 
-    def __init__(self, main_context=None):
-        if main_context is None:
-            main_context = GLib.MainContext.default()
-        selector = glib_selector.GLibSelector(main_context)
+    # This is based on the selector event loop, but never actually runs select()
+    # in the strict sense.
+    # We use the selector to register all FDs with the main context using our
+    # own GSource. For python timeouts/idle equivalent, we directly query them
+    # from the context by providing the _get_timeout_ms function that the
+    # GSource uses. This in turn access _ready and _scheduled to calculate
+    # the timeout and whether python can dispatch anything non-FD based yet.
+    #
+    # To simplify matters, we call the normal _run_once method of the base
+    # class which will call select(). As we know that we are ready at the time
+    # that select() will return immediately with the FD information we have
+    # gathered already.
+    #
+    # With that, we just need to override and slightly modify the run_forever
+    # method so that it calls g_main_loop_run instead of looping _run_once.
+
+    def __init__(self, main_context):
+        # A mainloop in case we want to run our context
+        assert main_context is not None
+        self._context = main_context
+        self._main_loop = GLib.MainLoop.new(self._context, False)
+
+        selector = glib_selector.GLibSelector(self._context, self)
         super().__init__(selector)
 
+        # This is used by run_once to not busy loop if the timeout is floor'ed to zero
+        self._clock_resolution = 1e-3
+
+    def run_forever(self):
+        self._check_closed()
+        self._check_running()
+        self._set_coroutine_origin_tracking(self._debug)
+        self._thread_id = threading.get_ident()
+
+        old_agen_hooks = sys.get_asyncgen_hooks()
+        sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook,
+                               finalizer=self._asyncgen_finalizer_hook)
+        try:
+            events._set_running_loop(self)
+            g_main_loop_run(self._main_loop)
+        finally:
+            self._thread_id = None
+            events._set_running_loop(None)
+            self._set_coroutine_origin_tracking(False)
+            sys.set_asyncgen_hooks(*old_agen_hooks)
+
+    def time(self):
+        return GLib.get_monotonic_time() / 1000000
+
+    def _get_timeout_ms(self):
+        if self._ready:
+            return 0
+
+        if self._scheduled:
+            timeout = (self._scheduled[0]._when - self.time()) * 1000
+            return timeout if timeout >= 0 else 0
+
+        return -1
+
+    def is_running(self):
+        # If we are currently the owner, then the context is running
+        # (and we are being dispatched by it)
+        if self._context.is_owner():
+            return True
+
+        # Otherwise, it might (but shouldn't) be running in a different thread
+        # Try aquiring it, if that fails, another thread is owning it
+        if not self._context.acquire():
+            return True
+        self._context.release()
+
+        return False
+
+    def stop(self):
+        self._main_loop.quit()
 
-class GLibEventLoopPolicy(asyncio.DefaultEventLoopPolicy):
+class GLibEventLoopPolicy(events.AbstractEventLoopPolicy):
     """An asyncio event loop policy that runs the GLib main loop"""
 
+    _loops = {}
+
+    def get_event_loop(self):
+        """Get the event loop for the current context.
+
+        Returns an event loop object implementing the BaseEventLoop interface,
+        or raises an exception in case no event loop has been set for the
+        current context and the current policy does not specify to create one.
+
+        It should never return None."""
+        # Get the thread default main context
+        ctx = GLib.MainContext.get_thread_default()
+        # If there is none, and we are on the main thread, then use the default context
+        if ctx is None and threading.current_thread() is threading.main_thread():
+            ctx = GLib.MainContext.default()
+        # Otherwise, if there is still none, create a new one for this thread and push it
+        if ctx is None:
+            ctx = GLib.MainContext.new()
+            ctx.push_thread_default()
+
+        # Note: We cannot attach it to ctx, as getting the default will always
+        #       return a new python wrapper. But, we can use hash() as that returns
+        #       the pointer to the C structure.
+        if ctx in self._loops:
+            loop = self._loops[ctx]
+            # If the loop is already closed, then return a new one instead
+            if not loop._closed:
+                return loop
+
+        self._loops[ctx] = GLibEventLoop(ctx)
+        return self._loops[ctx]
+
+    def set_event_loop(self, loop):
+        """Set the event loop for the current context to loop."""
+        raise NotImplementedError
+
     def new_event_loop(self):
-        if threading.current_thread() != threading.main_thread():
-            raise RuntimeError("GLibEventLoopPolicy only allows the main "
-                               "thread to create event loops")
-        return GLibEventLoop()
+        """Create and return a new event loop object according to this
+        policy's rules. If there's need to set this loop as the event loop for
+        the current context, set_event_loop must be called explicitly."""
+        raise NotImplementedError
+
+    # Child processes handling (Unix only).
+
+    def get_child_watcher(self):
+        "Get the watcher for child processes."
+        raise NotImplementedError
+
+    def set_child_watcher(self, watcher):
+        """Set the watcher for child processes."""
+        raise NotImplementedError
+

asyncio_glib/glib_selector.py

@@ -17,28 +17,41 @@ def __init__(self, selector):
         self._selector = selector
 
     def prepare(self):
-        return False, self._selector._get_timeout_ms()
+        timeout = self._selector._loop._get_timeout_ms()
+
+        # NOTE: Always return False as we query the FDs in only in check
+        return False, timeout
 
     def check(self):
-        return False
+        has_events = False
+        timeout = self._selector._loop._get_timeout_ms()
+        if timeout == 0:
+            has_events = True
 
-    def dispatch(self, callback, args):
         for (fd, tag) in self._fd_to_tag.items():
             condition = self.query_unix_fd(tag)
             events = self._fd_to_events.setdefault(fd, 0)
-            if condition & GLib.IOCondition.IN:
+            if condition & (GLib.IOCondition.IN | GLib.IOCondition.HUP):
                 events |= selectors.EVENT_READ
+                has_events = True
             if condition & GLib.IOCondition.OUT:
                 events |= selectors.EVENT_WRITE
+                has_events = True
             self._fd_to_events[fd] = events
+
+        return has_events
+
+    def dispatch(self, callback, args):
+        # Now, wag the dog by its tail
+        self._selector._loop._run_once()
         return GLib.SOURCE_CONTINUE
 
     def register(self, fd, events):
         assert fd not in self._fd_to_tag
 
         condition = GLib.IOCondition(0)
         if events & selectors.EVENT_READ:
-            condition |= GLib.IOCondition.IN
+            condition |= GLib.IOCondition.IN | GLib.IOCondition.HUP
         if events & selectors.EVENT_WRITE:
             condition |= GLib.IOCondition.OUT
         self._fd_to_tag[fd] = self.add_unix_fd(fd, condition)
@@ -56,13 +69,15 @@ def clear(self):
 
 class GLibSelector(selectors._BaseSelectorImpl):
 
-    def __init__(self, context):
+    def __init__(self, context, loop):
         super().__init__()
         self._context = context
+        self._loop = loop
         self._source = _SelectorSource(self)
         self._source.attach(self._context)
 
     def close(self):
+        self._source.clear()
         self._source.destroy()
         super().close()
 
@@ -76,23 +91,13 @@ def unregister(self, fileobj):
         self._source.unregister(key.fd)
         return key
 
-    def _get_timeout_ms(self):
-        """Return the timeout for the current select/iteration"""
-        return self._timeout
-
     def select(self, timeout=None):
-        # Calling .set_ready_time() always causes a mainloop iteration to finish
-        if timeout is not None and timeout >= 0:
-            self._timeout = int(timeout * 1000)
-        else:
-            self._timeout = -1
-
-        self._source.clear()
-        self._context.iteration(True)
-
+        # Dummy select that just returns immediately with what we already know.
         ready = []
         for key in self.get_map().values():
             events = self._source.get_events(key.fd) & key.events
             if events != 0:
                 ready.append((key, events))
+        self._source.clear()
+
         return ready

tests/test_glib_events.py

@@ -7,12 +7,13 @@
     from asyncio import test_utils
 
 from asyncio_glib import glib_events
-
+import gi
+from gi.repository import GLib
 
 class GLibEventLoopTests(UnixEventLoopTestsMixin, test_utils.TestCase):
 
     def create_event_loop(self):
-        return glib_events.GLibEventLoop()
+        return glib_events.GLibEventLoop(main_context=GLib.MainContext.default())
 
     def test_read_pipe(self):
         raise unittest.SkipTest("TODO")
@@ -37,8 +38,4 @@ def test_new_event_loop(self):
         loop.close()
 
     def test_set_event_loop(self):
-        policy = self.create_policy()
-        loop = policy.new_event_loop()
-        policy.set_event_loop(loop)
-        self.assertIs(loop, policy.get_event_loop())
-        loop.close()
+        raise unittest.SkipTest("Not compatible with GLib")