Search code examples
pythonpython-3.xtkinterpython-asyncioevent-loop

How exactly does tkinter's mainloop work?


I was interested in trying to merge Python's tkinter with asyncio, and after having read this answer I was largely successful. For reference, you can recreate the mainloop as follows:

import asyncio
import tkinter as tk


class AsyncTk(tk.Tk):

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.is_running = True

    async def async_loop(self):
        """Asynchronous equivalent of `Tk.mainloop`."""
        while self.is_running:
            self.update()
            await asyncio.sleep(0)

    def destroy(self):
        super().destroy()
        self.is_running = False


async def main():
    root = AsyncTk()
    await asyncio.gather(
        root.async_loop(),
        other_async_functions(),
    )

However, this comment pointed out that in some cases, the GUI may freeze during an self.update() call. One may instead use self.update_idletasks() to prevent this, but since root.async_loop is supposed to simulate the root.mainloop, I fear never running some tasks may cause other problems.

I couldn't find the source code for how root.mainloop works, though I did discover that replacing self.update() with

self.tk.dooneevent(tk._tkinter.DONT_WAIT)

should produce more fine-grained concurrency by only doing one event instead of flushing all events (I'm assuming that's what it does, I couldn't find the documentation for this either but it seemed to work).


So my two questions are:

  1. Is it fine to use just self.update_idletasks() only, and never run whatever else self.update() is supposed to run?

  2. How exactly does root.mainloop() work?


For some code that can be ran and experimented with:

"""Example integrating `tkinter`'s `mainloop` with `asyncio`."""
import asyncio
from random import randrange
from time import time
import tkinter as tk


class AsyncTk(tk.Tk):
    """
    An asynchronous Tk class.

    Use `await root.async_loop()` instead of `root.mainloop()`.

    Schedule asynchronous tasks using `asyncio.create_task(...)`.
    """
    is_running: bool

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.is_running = True

    async def async_loop(self):
        """An asynchronous version of `root.mainloop()`."""
        # For threaded calls.
        self.tk.willdispatch()
        # Run until `self.destroy` is called.
        while self.is_running:
            #self.update_idletasks()   # NOTE: using `update_idletasks`
                                       # prevents the window from freezing
                                       # when you try to resize the window.
            self.update()
            #self.tk.dooneevent(tk._tkinter.DONT_WAIT)
            await asyncio.sleep(0)

    def destroy(self):
        """
        Destroy this and all descendants widgets. This will
        end the application of this Tcl interpreter.
        """
        super().destroy()
        # Mark the Tk as not running.
        self.is_running = False

    async def rotator(self, interval, d_per_tick):
        """
        An example custom method for running code asynchronously
        instead of using `tkinter.Tk.after`.

        NOTE: Code that can use `tkinter.Tk.after` is likely
              preferable, but this may not fit all use-cases and
              may sometimes require more complicated code.
        """
        canvas = tk.Canvas(self, height=600, width=600)
        canvas.pack()
        deg = 0
        color = 'black'
        arc = canvas.create_arc(
            100,
            100,
            500,
            500,
            style=tk.CHORD,
            start=0,
            extent=deg,
            fill=color,
        )
        while self.is_running:
            deg, color = deg_color(deg, d_per_tick, color)
            canvas.itemconfigure(arc, extent=deg, fill=color)
            await asyncio.sleep(interval)


def deg_color(deg, d_per_tick, color):
    """Helper function for updating the degree and color."""
    deg += d_per_tick
    if 360 <= deg:
        deg %= 360
        color = f"#{randrange(256):02x}{randrange(256):02x}{randrange(256):02x}"
    return deg, color

async def main():
    root = AsyncTk()
    await asyncio.gather(root.async_loop(), root.rotator(1/60, 2))

if __name__ == "__main__":
    asyncio.run(main())

Solution

  • Although trying to rewrite the tkinter loop seems troublesome, it seems rewriting the asyncio loop is quite easy, given tkinter's after function. The main gist of it is this:

    """Example integrating `tkinter`'s `mainloop` with `asyncio`."""
    import asyncio
    import tkinter as tk
    from typing import Any, Awaitable, TypeVar
    
    T = TypeVar("T")
    
    
    class AsyncTk(tk.Tk):
        """
        A Tk class that can run asyncio awaitables alongside the tkinter application.
    
        Use `root.run_with_mainloop(awaitable)` instead of `root.mainloop()` as a way to run
        coroutines alongside it. It functions similarly to using `asyncio.run(awaitable)`.
    
        Alternatively use `await root.async_loop()` if you need to run this in an asynchronous
        context. Because this doesn't run `root.mainloop()` directly, it may not behave exactly
        the same as using `root.run_with_mainloop(awaitable)`.
        """
        is_running: bool
    
        def __init__(self, /, *args: Any, **kwargs: Any) -> None:
            super().__init__(*args, **kwargs)
            self.is_running = True
    
        def __advance_loop(self, loop: asyncio.AbstractEventLoop, timeout, /) -> None:
            """Helper method for advancing the asyncio event loop."""
            # Stop soon i.e. only advance the event loop a little bit.
            loop.call_soon(loop.stop)
            loop.run_forever()
            # If tkinter is still running, repeat this method.
            if self.is_running:
                self.after(timeout, self.__advance_loop, loop, timeout)
    
        async def async_loop(self, /) -> None:
            """
            An asynchronous variant of `root.mainloop()`.
    
            Because this doesn't run `root.mainloop()` directly, it may not behave exactly
            the same as using `root.run_with_mainloop(awaitable)`.
            """
            # For threading.
            self.tk.willdispatch()
            # Run initial update.
            self.update()
            # Run until `self.destroy()` is called.
            while self.is_running:
                # Let other code run.
                # Uses a non-zero sleep time because tkinter should be expected to be slow.
                # This decreases the busy wait time.
                await asyncio.sleep(tk._tkinter.getbusywaitinterval() / 10_000)
                # Run one event.
                self.tk.dooneevent(tk._tkinter.DONT_WAIT)
    
        def run_with_mainloop(self, awaitable: Awaitable[T], /, *, timeout: float = 0.001) -> T:
            """
            Run an awaitable alongside the tkinter application.
    
            Similar to using `asyncio.run(awaitable)`.
    
            Use `root.run_with_mainloop(awaitable, timeout=...)` to
            customize the frequency the asyncio event loop is updated.
            """
            if not isinstance(awaitable, Awaitable):
                raise TypeError(f"awaitable must be an Awaitable, got {awaitable!r}")
            elif not isinstance(timeout, (float, int)):
                raise TypeError(f"timeout must be a float or integer, got {timeout!r}")
            # Start a new event loop with the awaitable in it.
            loop = asyncio.new_event_loop()
            task = loop.create_task(awaitable)
            # Use tkinter's `.after` to run the asyncio event loop.
            self.after(0, self.__advance_loop, loop, max(1, int(timeout * 1000)))
            # Run tkinter, which periodically checks
            self.mainloop()
            # After tkinter is done, wait until `asyncio` is done.
            try:
                return loop.run_until_complete(task)
            finally:
                loop.run_until_complete(loop.shutdown_asyncgens())
                loop.close()
    
        def destroy(self, /) -> None:
            super().destroy()
            self.is_running = False
    

    The example application may be fixed up like this:

    import asyncio
    from random import randrange
    import tkinter as tk
    
    def deg_color(deg, d_per_tick, color):
        """Helper function for updating the degree and color."""
        deg += d_per_tick
        if 360 <= deg:
            deg %= 360
            color = f"#{randrange(256):02x}{randrange(256):02x}{randrange(256):02x}"
        return deg, color
    
    async def rotator(root, interval, d_per_tick):
        """
        An example custom method for running code asynchronously
        instead of using `tkinter.Tk.after`.
    
        NOTE: Code that can use `tkinter.Tk.after` is likely
              preferable, but this may not fit all use-cases and
              may sometimes require more complicated code.
        """
        canvas = tk.Canvas(root, height=600, width=600)
        canvas.pack()
        deg = 0
        color = 'black'
        arc = canvas.create_arc(
            100,
            100,
            500,
            500,
            style=tk.CHORD,
            start=0,
            extent=deg,
            fill=color,
        )
        while root.is_running:
            deg, color = deg_color(deg, d_per_tick, color)
            canvas.itemconfigure(arc, extent=deg, fill=color)
            await asyncio.sleep(interval)
    
    def main():
        root = AsyncTk()
        root.run_with_mainloop(rotator(root, 1/60, 2))
    
    if __name__ == "__main__":
        main()