gh-145342: asyncio: Add guest mode for running inside external event loops

Add asyncio.start_guest_run() which allows asyncio to run cooperatively
inside a host event loop (e.g. Tkinter, Qt, GTK).  The host loop stays in
control of the main thread while asyncio I/O polling runs in a background
daemon thread.

Implementation:

- Add three public methods to BaseEventLoop -- poll_events(),
  process_events(), and process_ready() -- that decompose _run_once()
  into independently callable steps.
- Refactor _run_once() to delegate to these three methods (zero behaviour
  change for existing code).
- Add Lib/asyncio/guest.py with start_guest_run().
- Add comprehensive tests using a mock host loop (no GUI dependency).
- Add a Tkinter demo in Doc/includes/.

Inspired by Trio start_guest_run() and the asyncio-guest project.

Author: congzhangzh

Doc/includes/asyncio_guest_tkinter.py

@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""Minimal demo: asyncio running as a guest inside Tkinter's mainloop.
+
+A progress bar counts from 0 to MAX_COUNT using ``asyncio.sleep()``.
+The Tk GUI stays fully responsive throughout.  Closing the window or
+pressing the Cancel button cancels the async task cleanly.
+
+Usage::
+
+    python asyncio_guest_tkinter.py
+"""
+
+import asyncio
+import collections
+import tkinter as tk
+import tkinter.ttk as ttk
+import traceback
+
+
+# -- Host adapter for Tkinter ------------------------------------------
+
+class TkHost:
+    """Bridge between asyncio guest mode and the Tk event loop."""
+
+    def __init__(self, root):
+        self.root = root
+        self._tk_func_name = root.register(self._dispatch)
+        self._q = collections.deque()
+
+    def _dispatch(self):
+        self._q.popleft()()
+
+    def run_sync_soon_threadsafe(self, fn):
+        """Schedule *fn* on the Tk thread.
+
+        ``Tkapp_ThreadSend`` (the C layer behind ``root.call`` from a
+        non-Tcl thread) posts the command to the Tcl event queue, making
+        this safe to call from any thread.
+        """
+        self._q.append(fn)
+        self.root.call('after', 'idle', self._tk_func_name)
+
+    def done_callback(self, task):
+        """Called when the async task finishes."""
+        if task.cancelled():
+            print("Task was cancelled.")
+        elif task.exception() is not None:
+            exc = task.exception()
+            traceback.print_exception(type(exc), exc, exc.__traceback__)
+        else:
+            print(f"Task returned: {task.result()}")
+        self.root.destroy()
+
+
+# -- Async workload ----------------------------------------------------
+
+MAX_COUNT = 20
+PERIOD = 0.5  # seconds between increments
+
+
+async def count(progress, root):
+    """Increment a progress bar, updating the Tk GUI each step."""
+    root.wm_title(f"Counting every {PERIOD}s ...")
+    progress.configure(maximum=MAX_COUNT)
+
+    task = asyncio.current_task()
+    loop = asyncio.get_event_loop()
+
+    # Wire the Cancel button and window close to task.cancel().
+    # Use call_soon_threadsafe so the I/O thread's selector is woken.
+    def request_cancel():
+        loop.call_soon_threadsafe(task.cancel)
+
+    cancel_btn = root.nametowidget('cancel')
+    cancel_btn.configure(command=request_cancel)
+    root.protocol("WM_DELETE_WINDOW", request_cancel)
+
+    for i in range(1, MAX_COUNT + 1):
+        await asyncio.sleep(PERIOD)
+        progress.step(1)
+        root.wm_title(f"Count: {i}/{MAX_COUNT}")
+
+    return i
+
+
+# -- Main ---------------------------------------------------------------
+
+def main():
+    root = tk.Tk()
+    root.wm_title("asyncio guest + Tkinter")
+
+    progress = ttk.Progressbar(root, length='6i')
+    progress.pack(fill=tk.BOTH, expand=True, padx=8, pady=(8, 4))
+
+    cancel_btn = tk.Button(root, text='Cancel', name='cancel')
+    cancel_btn.pack(pady=(0, 8))
+
+    host = TkHost(root)
+
+    asyncio.start_guest_run(
+        count, progress, root,
+        run_sync_soon_threadsafe=host.run_sync_soon_threadsafe,
+        done_callback=host.done_callback,
+    )
+
+    root.mainloop()
+
+
+if __name__ == '__main__':
+    main()

Lib/asyncio/__init__.py

@@ -11,6 +11,7 @@
 from .exceptions import *
 from .futures import *
 from .graph import *
+from .guest import *
 from .locks import *
 from .protocols import *
 from .runners import *
@@ -29,6 +30,7 @@
            exceptions.__all__ +
            futures.__all__ +
            graph.__all__ +
+           guest.__all__ +
            locks.__all__ +
            protocols.__all__ +
            runners.__all__ +

Lib/asyncio/base_events.py

@@ -1963,14 +1963,20 @@ def _timer_handle_cancelled(self, handle):
         if handle._scheduled:
             self._timer_cancelled_count += 1
 
-    def _run_once(self):
-        """Run one full iteration of the event loop.
+    def poll_events(self):
+        """Poll for I/O events without processing them.
 
-        This calls all currently ready callbacks, polls for I/O,
-        schedules the resulting callbacks, and finally schedules
-        'call_later' callbacks.
-        """
+        Cleans up cancelled scheduled handles, computes an appropriate
+        timeout from the scheduled callbacks, and calls
+        ``self._selector.select(timeout)``.  Returns the raw event list.
 
+        This method, together with :meth:`process_events` and
+        :meth:`process_ready`, decomposes :meth:`_run_once` into
+        independently callable steps so that an external event loop can
+        drive asyncio (see :func:`asyncio.start_guest_run`).
+
+        .. versionadded:: 3.15
+        """
         sched_count = len(self._scheduled)
         if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
             self._timer_cancelled_count / sched_count >
@@ -2005,11 +2011,29 @@ def _run_once(self):
             elif timeout < 0:
                 timeout = 0
 
-        event_list = self._selector.select(timeout)
+        return self._selector.select(timeout)
+
+    def process_events(self, event_list):
+        """Process I/O events returned by :meth:`poll_events`.
+
+        Delegates to the selector-specific :meth:`_process_events`
+        implementation which turns raw selector events into ready
+        callbacks.
+
+        .. versionadded:: 3.15
+        """
         self._process_events(event_list)
-        # Needed to break cycles when an exception occurs.
-        event_list = None
 
+    def process_ready(self):
+        """Process expired timers and execute ready callbacks.
+
+        Moves scheduled callbacks whose deadline has passed into the
+        ready queue, then runs all callbacks that were ready at call
+        time.  Callbacks enqueued *by* running callbacks are left for
+        the next iteration.
+
+        .. versionadded:: 3.15
+        """
         # Handle 'later' callbacks that are ready.
         end_time = self.time() + self._clock_resolution
         while self._scheduled:
@@ -2044,6 +2068,18 @@ def _run_once(self):
                     self._current_handle = None
             else:
                 handle._run()
+
+    def _run_once(self):
+        """Run one full iteration of the event loop.
+
+        This calls all currently ready callbacks, polls for I/O,
+        schedules the resulting callbacks, and finally schedules
+        'call_later' callbacks.
+        """
+        event_list = self.poll_events()
+        self.process_events(event_list)
+        event_list = None       # Needed to break cycles on exception.
+        self.process_ready()
         handle = None  # Needed to break cycles when an exception occurs.
 
     def _set_coroutine_origin_tracking(self, enabled):

Lib/asyncio/guest.py

@@ -0,0 +1,122 @@
+"""Support for running asyncio as a guest inside another event loop.
+
+This module provides start_guest_run(), which allows asyncio to run
+cooperatively inside a host event loop such as a GUI toolkit's main loop.
+The host loop stays in control of the main thread while asyncio tasks
+execute through a dual-thread architecture:
+
+  Host thread:    process_events() + process_ready() -> sem.release()
+  Backend thread: sem.acquire() -> poll_events() -> notify host
+
+Inspired by Trio's guest mode (trio.lowlevel.start_guest_run).
+"""
+
+__all__ = ('start_guest_run',)
+
+import threading
+from functools import partial
+
+from . import events
+
+
+def start_guest_run(async_fn, *args,
+                    run_sync_soon_threadsafe,
+                    done_callback):
+    """Run an async function as a guest inside another event loop.
+
+    The host event loop (e.g. Tkinter mainloop) remains in control of the
+    main thread.  asyncio I/O polling runs in a daemon background thread
+    and dispatches work back to the host thread via *run_sync_soon_threadsafe*.
+
+    Parameters
+    ----------
+    async_fn : coroutine function
+        The async function to run.
+    *args :
+        Positional arguments passed to *async_fn*.
+    run_sync_soon_threadsafe : callable
+        A callback that schedules a zero-argument callable to run on the
+        host thread.  Must be safe to call from any thread.
+    done_callback : callable
+        Called on the host thread when *async_fn* finishes.  Receives the
+        completed ``asyncio.Task`` as its sole argument.  Callers can
+        inspect the task with ``task.result()``, ``task.exception()``,
+        or ``task.cancelled()``.
+
+    Returns
+    -------
+    asyncio.Task
+        The task wrapping *async_fn*.  To cancel from the host thread,
+        use ``loop.call_soon_threadsafe(task.cancel)`` so that the I/O
+        thread is woken from its selector wait.
+    """
+    loop = events.new_event_loop()
+    events._set_running_loop(loop)
+
+    _shutdown = threading.Event()
+    _sem = threading.Semaphore(0)
+    _done_called = False
+
+    # -- helpers ------------------------------------------------------
+
+    def _finish(task):
+        """Clean up and forward completion to the host."""
+        nonlocal _done_called
+        if _done_called:
+            return
+        _done_called = True
+        events._set_running_loop(None)
+        try:
+            done_callback(task)
+        finally:
+            if not loop.is_closed():
+                loop.close()
+
+    def _process_on_host(event_list):
+        """Run on the host thread: process one batch of asyncio work."""
+        if _shutdown.is_set():
+            return
+        loop.process_events(event_list)
+        loop.process_ready()
+        if not _shutdown.is_set():
+            _sem.release()
+
+    # -- threads -------------------------------------------------------
+
+    def _backend():
+        """Daemon thread: poll for I/O and wake the host."""
+        try:
+            while not _shutdown.is_set():
+                _sem.acquire()
+                if _shutdown.is_set():
+                    break
+                event_list = loop.poll_events()
+                run_sync_soon_threadsafe(
+                    partial(_process_on_host, event_list)
+                )
+        except Exception as exc:
+            _shutdown.set()
+            main_task.cancel(
+                msg=f"asyncio guest I/O thread failed: {exc!r}"
+            )
+            run_sync_soon_threadsafe(lambda: _finish(main_task))
+
+    # -- task setup ----------------------------------------------------
+
+    main_task = loop.create_task(async_fn(*args))
+
+    def _on_task_done(task):
+        _shutdown.set()
+        _sem.release()          # wake backend so it can exit
+        run_sync_soon_threadsafe(lambda: _finish(task))
+
+    main_task.add_done_callback(_on_task_done)
+
+    # Kick off: process the initial callbacks enqueued by create_task.
+    _process_on_host([])
+
+    threading.Thread(
+        target=_backend, daemon=True, name='asyncio-guest-io'
+    ).start()
+
+    return main_task