Add loop.call_soon_threadsafe() and re-implement precise sleep (#146)

Author: almarklein

rendercanvas/_coreutils.py

@@ -6,10 +6,19 @@
 import re
 import sys
 import time
+import queue
 import weakref
 import logging
+import threading
 import ctypes.util
 from contextlib import contextmanager
+from collections import namedtuple
+
+
+# %% Constants
+
+
+IS_WIN = sys.platform.startswith("win")  # Note that IS_WIN is false on Pyodide
 
 
 # %% Logging
@@ -93,6 +102,132 @@ def proxy(*args, **kwargs):
     return proxy
 
 
+# %% Helper for scheduling call-laters
+
+
+class CallLaterThread(threading.Thread):
+    """An object that can be used to do "call later" from a dedicated thread.
+
+    This is helpful to implement a call-later mechanism on some backends, and
+    serves as an alternative timeout mechanism in Windows (to overcome its
+    notorious 15.6ms ticks).
+
+    Windows historically uses ticks that go at 64 ticks per second, i.e. 15.625
+    ms each. Other platforms are "tickless" and (in theory) have microsecond
+    resolution.
+
+    Care is taken to realize precise timing, in the order of 1 ms. Nevertheless,
+    on OS's other than Windows, the native timers are more accurate than this
+    threaded approach. I suspect that this is related to the GIL; two threads
+    cannot run at the same time.
+    """
+
+    Item = namedtuple("Item", ["time", "index", "callback", "args"])
+
+    def __init__(self):
+        super().__init__()
+        self._queue = queue.SimpleQueue()
+        self._count = 0
+        self.daemon = True  # don't let this thread prevent shutdown
+        self.start()
+
+    def call_later_from_thread(self, delay, callback, *args):
+        """In delay seconds, call the callback from the scheduling thread."""
+        self._count += 1
+        item = CallLaterThread.Item(
+            time.perf_counter() + float(delay), self._count, callback, args
+        )
+        self._queue.put(item)
+
+    def run(self):
+        perf_counter = time.perf_counter
+        Empty = queue.Empty  # noqa: N806
+        q = self._queue
+        priority = []
+        is_win = IS_WIN
+
+        wait_until = None
+        timestep = 0.001  # for doing small sleeps
+        leeway = timestep / 2  # a little offset so waiting exactly right on average
+        leeway += 0.0005  # extra offset to account for GIL etc. (0.5ms seems ok)
+
+        while True:
+            # == Wait for input
+
+            if wait_until is None:
+                # Nothing to do but wait
+                new_item = q.get(True, None)
+            else:
+                # We wait for the queue with a timeout. But because the timeout is not very precise,
+                # we wait shorter, and then go in a loop with some hard sleeps.
+                # Windows has 15.6 ms resolution ticks. But also on other OSes,
+                # it benefits precision to do the last bit with hard sleeps.
+                offset = 0.016 if is_win else timestep
+                try:
+                    new_item = q.get(True, max(0, wait_until - perf_counter() - offset))
+                except Empty:
+                    new_item = None
+                    while perf_counter() < wait_until:
+                        time.sleep(timestep)
+                        try:
+                            new_item = q.get_nowait()
+                            break
+                        except Empty:
+                            pass
+
+            # Put it in our priority queue
+            if new_item is not None:
+                priority.append(new_item)
+                priority.sort(reverse=True)
+
+            del new_item
+
+            # == Process items until we have to wait
+
+            item = None
+            while True:
+                # Get item that is up next
+                try:
+                    item = priority.pop(-1)
+                except IndexError:
+                    wait_until = None
+                    break
+
+                # If it's not yet time for the item, put it back, and go wait
+                item_time_threshold = item.time - leeway
+                if perf_counter() < item_time_threshold:
+                    priority.append(item)
+                    wait_until = item_time_threshold
+                    break
+
+                # Otherwise, handle the callback
+                try:
+                    item.callback(*item.args)
+                except Exception as err:
+                    logger.error(f"Error in CallLaterThread callback: {err}")
+
+            del item
+
+
+_call_later_thread = None
+
+
+def call_later_from_thread(delay: float, callback: object, *args: object):
+    """Utility that calls a callback after a specified delay, from a separate thread.
+
+    The caller is responsible for the given callback to be thread-safe.
+    There is one global thread that handles all callbacks. This thread is spawned the first time
+    that this function is called.
+
+    Note that this function should only be used in environments where threading is available.
+    E.g. on Pyodide this will raise ``RuntimeError: can't start new thread``.
+    """
+    global _call_later_thread
+    if _call_later_thread is None:
+        _call_later_thread = CallLaterThread()
+    return _call_later_thread.call_later_from_thread(delay, callback, *args)
+
+
 # %% lib support
 
 

rendercanvas/_loop.py

@@ -8,7 +8,7 @@
 from inspect import iscoroutinefunction
 from typing import TYPE_CHECKING
 
-from ._coreutils import logger, log_exception
+from ._coreutils import logger, log_exception, call_later_from_thread
 from .utils.asyncs import sleep
 from .utils import asyncadapter
 
@@ -29,7 +29,7 @@ class BaseLoop:
     """The base class for an event-loop object.
 
     Canvas backends can implement their own loop subclass (like qt and wx do), but a
-    canvas backend can also rely on one of muliple loop implementations (like glfw
+    canvas backend can also rely on one of multiple loop implementations (like glfw
     running on asyncio or trio).
 
     The lifecycle states of a loop are:
@@ -46,7 +46,7 @@ class BaseLoop:
     * Stopping the loop (via ``.stop()``) closes the canvases, which will then stop the loop.
     * From there it can go back to the ready state (which would call ``_rc_init()`` again).
     * In backends like Qt, the native loop can be started without us knowing: state "active".
-    * In interactive settings like an IDE that runs an syncio or Qt loop, the
+    * In interactive settings like an IDE that runs an asyncio or Qt loop, the
       loop can become "active" as soon as the first canvas is created.
 
     """
@@ -176,8 +176,11 @@ async def wrapper():
     def call_soon(self, callback: CallbackFunction, *args: Any) -> None:
         """Arrange for a callback to be called as soon as possible.
 
-        The callback will be called in the next iteration of the event-loop,
-        but other pending events/callbacks may be handled first. Returns None.
+        The callback will be called in the next iteration of the event-loop, but
+        other pending events/callbacks may be handled first. Returns None.
+
+        Not thread-safe; use ``call_soon_threadsafe()`` for scheduling callbacks
+        from another thread.
         """
         if not callable(callback):
             raise TypeError("call_soon() expects a callable.")
@@ -190,6 +193,22 @@ async def wrapper():
 
         self._rc_add_task(wrapper, "call_soon")
 
+    def call_soon_threadsafe(self, callback: CallbackFunction, *args: Any) -> None:
+        """A thread-safe variant of ``call_soon()``."""
+
+        if not callable(callback):
+            raise TypeError("call_soon_threadsafe() expects a callable.")
+        elif iscoroutinefunction(callback):
+            raise TypeError(
+                "call_soon_threadsafe() expects a normal callable, not an async one."
+            )
+
+        def wrapper():
+            with log_exception("Callback error:"):
+                callback(*args)
+
+        self._rc_call_soon_threadsafe(wrapper)
+
     def call_later(self, delay: float, callback: CallbackFunction, *args: Any) -> None:
         """Arrange for a callback to be called after the given delay (in seconds)."""
         if delay <= 0:
@@ -214,7 +233,7 @@ def run(self) -> None:
         its fine to start the loop in the normal way.
 
         This call usually blocks, but it can also return immediately, e.g. when there are no
-        canvases, or when the loop is already active (e.g. interactve via IDE).
+        canvases, or when the loop is already active (e.g. interactive via IDE).
         """
 
         # Can we enter the loop?
@@ -360,8 +379,13 @@ def _rc_stop(self):
     def _rc_add_task(self, async_func, name):
         """Add an async task to the running loop.
 
-        This method is optional. A subclass must either implement ``_rc_add_task`` or ``_rc_call_later``.
+        True async loop-backends (like asyncio and trio) should implement this.
+        When they do, ``_rc_call_later`` is not used.
+
+        Other loop-backends can use the default implementation, which uses the
+        ``asyncadapter`` which runs coroutines using ``_rc_call_later``.
 
+        * If you implement this, make ``_rc_call_later()`` raise an exception.
         * Schedule running the task defined by the given co-routine function.
         * The name is for debugging purposes only.
         * The subclass is responsible for cancelling remaining tasks in _rc_stop.
@@ -374,11 +398,23 @@ def _rc_add_task(self, async_func, name):
     def _rc_call_later(self, delay, callback):
         """Method to call a callback in delay number of seconds.
 
-        This method is optional. A subclass must either implement ``_rc_add_task`` or ``_rc_call_later``.
+        Backends that implement ``_rc_add_task`` should not implement this.
+        Other backends can use the default implementation, which uses a
+        scheduler thread and ``_rc_call_soon_threadsafe``. But they can also
+        implement this using the loop-backend's own mechanics.
 
-        * If you implememt this, make ``_rc_add_task()`` call ``super()._rc_add_task()``.
+        * If you implement this, make ``_rc_add_task()`` call ``super()._rc_add_task()``.
+        * Take into account that on Windows, timers are usually inaccurate.
         * If delay is zero, this should behave like "call_soon".
-        * No need to catch errors from the callback; that's dealt with internally.
+        * No need to catch errors from the callback; that's dealt with
+          internally.
         * Return None.
         """
+        call_later_from_thread(delay, self._rc_call_soon_threadsafe, callback)
+
+    def _rc_call_soon_threadsafe(self, callback):
+        """Method to schedule a callback in the loop's thread.
+
+        Must be thread-safe; this may be called from a different thread.
+        """
         raise NotImplementedError()

rendercanvas/_scheduler.py

@@ -2,17 +2,13 @@
 The scheduler class/loop.
 """
 
-import sys
 import time
 import weakref
 
 from ._enums import UpdateMode
 from .utils.asyncs import sleep, Event
 
 
-IS_WIN = sys.platform.startswith("win")
-
-
 class Scheduler:
     """Helper class to schedule event processing and drawing."""
 
@@ -121,20 +117,9 @@ async def __scheduler_task(self):
             # Determine amount of sleep
             sleep_time = delay - (time.perf_counter() - last_tick_time)
 
-            if IS_WIN:
-                # On Windows OS-level timers have an in accuracy of 15.6 ms.
-                # This can cause sleep to take longer than intended. So we sleep
-                # less, and then do a few small sync-sleeps that have high accuracy.
-                await sleep(max(0, sleep_time - 0.0156))
-                sleep_time = delay - (time.perf_counter() - last_tick_time)
-                while sleep_time > 0:
-                    time.sleep(min(sleep_time, 0.001))  # sleep hard for at most 1ms
-                    await sleep(0)  # Allow other tasks to run but don't wait
-                    sleep_time = delay - (time.perf_counter() - last_tick_time)
-            else:
-                # Wait. Even if delay is zero, it gives control back to the loop,
-                # allowing other tasks to do work.
-                await sleep(max(0, sleep_time))
+            # Wait. Even if delay is zero, it gives control back to the loop,
+            # allowing other tasks to do work.
+            await sleep(max(0, sleep_time))
 
             # Below is the "tick"
 

rendercanvas/asyncio.py

@@ -88,8 +88,12 @@ def _rc_add_task(self, func, name):
             self.__tasks.add(task)
             task.add_done_callback(self.__tasks.discard)
 
-    def _rc_call_later(self, *args):
+    def _rc_call_later(self, delay, callback):
         raise NotImplementedError()  # we implement _rc_add_task instead
 
+    def _rc_call_soon_threadsafe(self, callback):
+        loop = self._interactive_loop or self._run_loop
+        loop.call_soon_threadsafe(callback)
+
 
 loop = AsyncioLoop()

rendercanvas/offscreen.py

@@ -172,5 +172,8 @@ def _rc_add_task(self, async_func, name):
     def _rc_call_later(self, delay, callback):
         self._callbacks.append((time.perf_counter() + delay, callback))
 
+    def _rc_call_soon_threadsafe(self, callback):
+        self._callbacks.append((0, callback))
+
 
 loop = StubLoop()