How exactly does tkinter's mainloop work?-CodePudding

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:

Is it fine to use just self.update_idletasks() only, and never run whatever else self.update() is supposed to run?
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())

CodePudding user response：

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

If you do that, only events in the idle queue will get run. There are many other sorts of events that would never get processed, such as any user interaction.

How exactly does root.mainloop() work?

At its core it's a C-based function inside the embedded tcl interpreter that waits for events to arrive on a queue and then processes them. A good starting point for doing more research is the Tcl_DoOneEvent man page or the actual code in tclEvent.c

How exactly it works might be different on each platform since it has to have hooks into the native windowing system.

The page Event Loop on Tkdocs.com has one of the best high-level overviews of the event loop.

CodePudding user response：

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.
            await asyncio.sleep(0)
            # 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.
        # For some reason, directly using `loop.run_forever()` didn't work for me.
        while not task.done():
            loop.call_soon(loop.stop)
            loop.run_forever()
        loop.close()
        return task.result()

    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()