0

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())
Simply Beautiful Art
  • 1,284
  • 15
  • 16

2 Answers2

1

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.

Bryan Oakley
  • 370,779
  • 53
  • 539
  • 685
  • I looked at the code and documentation for a bit and came up with replacing `self.update()` with `while self.is_running and 0 != self.tk.dooneevent(tk._tkinter.XXX_EVENTS | tk._tkinter.DONT_WAIT): ...` for each type of event, but it doesn't seem to allow me to have the GUI continue to run while resizing it like with `self.update_idletasks()` does. – Simply Beautiful Art Jan 26 '22 at 01:19
  • Strangely the documentation there says "The TCL_DONT_WAIT flag causes Tcl_DoOneEvent not to put the process to sleep: it will check for events but if none are found then it returns immediately with a return value of 0 to indicate that no work was done. Tcl_DoOneEvent will also return 0 without doing anything if the only alternative is to block forever..." even though when I try print-debugging with it, it clearly stops the GUI from running entirely while the window is being resized. – Simply Beautiful Art Jan 26 '22 at 01:22
  • @Simply Beautiful Art: You don't need to worry that tkinter's event loop will block your asyncio task loop. Tkinter will not consume 100% of the CPU cycles, unless you've made a mistake and written an infinite loop. There might be brief peaks of activity but in general there will be plenty of CPU bandwidth to keep your asyncio code running. I do it all the time and it works fine. You may want to look at this, however: https://stackoverflow.com/questions/55464512/tkinter-and-asyncio-window-drag-resize-blocks-event-loop-single-thread – Paul Cornelius Jan 26 '22 at 02:44
  • @SimplyBeautifulArt I strongly recommend you do a simple self.update() and not try to mess around with doing one event. – Paul Cornelius Jan 26 '22 at 02:46
  • @PaulCornelius That is what I assumed until I ran it myself (see the linked comment in the OP as well as your linked question). I do think the suggested workaround of calling the asyncio's event loop with tkinter interesting, I'll see what I can do with that (I'm interested in making this clean to use). – Simply Beautiful Art Jan 26 '22 at 03:48
  • @PaulCornelius The linked solution ended up working quite well. Admittedly somewhat disappointed that you can't do `await root.loop()` or something similar, but being able to do `root.run(awaitable)` is quite sufficient I think. – Simply Beautiful Art Jan 26 '22 at 04:51
1

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()
Simply Beautiful Art
  • 1,284
  • 15
  • 16
  • I think that in `async def async_loop` the statement `await asyncio.sleep(0)` should be >0, also a `0.001` is enough. – hussic Jan 26 '22 at 18:12
  • @hussic It seemed to work fine for me, but FWIW you can use `tk._tkinter.getbusywaitinterval() / 1000` for what `tkinter` uses. I think it might already use this internally however, because the frame rate visibly drops if you do so. – Simply Beautiful Art Jan 26 '22 at 19:22
  • Yes it works, but a value of 0 is CPU intensive like a polling cycle. At least under Windows. – hussic Jan 27 '22 at 09:52
  • 1
    On my system with a 0 value, loop is executed more than 60.000 times/sec, with a 0.0001 value is executed less than 1000 times/sec. – hussic Jan 27 '22 at 11:25
  • @hussic That's a good point, thanks for pointing it out. – Simply Beautiful Art Jan 28 '22 at 00:14