tkinter 的主循环究竟是如何工作的?

How exactly does tkinter's mainloop work?

我有兴趣尝试将 Python 的 tkinterasyncio 合并,在阅读 之后,我大获成功。作为参考,您可以按如下方式重新创建 mainloop

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

但是, 指出,在某些情况下,GUI 可能会在 self.update() 调用期间冻结。人们可能会使用 self.update_idletasks() 来防止这种情况,但由于 root.async_loop 应该模拟 root.mainloop,我担心永远不会 运行 一些任务可能会导致其他问题。

我找不到 root.mainloop 工作原理的源代码,尽管我确实发现用

替换 self.update()
self.tk.dooneevent(tk._tkinter.DONT_WAIT)

应该通过只执行一个事件而不是刷新所有事件来产生更细粒度的并发(我假设这就是它所做的,我也找不到这方面的文档,但它似乎有效)。


所以我的两个问题是:

  1. 只使用 self.update_idletasks() 并且永远不要使用 运行 其他 self.update() 应该 运行 的东西吗?

  2. root.mainloop() 究竟是如何工作的?


对于一些可以 运行 并试验过的代码:

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

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?

它的核心是嵌入式 tcl 解释器中的一个 C-based 函数,它等待事件到达队列然后处理它们。进行更多研究的一个很好的起点是 Tcl_DoOneEvent man page or the actual code in tclEvent.c

具体它的工作方式在每个平台上可能有所不同,因为它必须与本机窗口系统挂钩。

Tkdocs.com 上的页面 Event Loop 是对事件循环最好的 high-level 概述之一。

虽然尝试重写 tkinter 循环似乎很麻烦,但考虑到 tkinterafter 函数,重写 asyncio 循环似乎很容易。它的主要要点是:

"""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

示例应用程序可以这样修复:

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