From daa917b65648145b9e5ed168eb8ae24aa02e1724 Mon Sep 17 00:00:00 2001 From: Cong Zhang <13283869+congzhangzh@users.noreply.github.com> Date: Sat, 28 Feb 2026 12:55:35 +0800 Subject: [PATCH 01/10] 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. --- Doc/includes/asyncio_guest_tkinter.py | 110 +++++++++++++ Lib/asyncio/__init__.py | 2 + Lib/asyncio/base_events.py | 54 +++++-- Lib/asyncio/guest.py | 122 +++++++++++++++ Lib/test/test_asyncio/test_guest.py | 212 ++++++++++++++++++++++++++ 5 files changed, 491 insertions(+), 9 deletions(-) create mode 100644 Doc/includes/asyncio_guest_tkinter.py create mode 100644 Lib/asyncio/guest.py create mode 100644 Lib/test/test_asyncio/test_guest.py diff --git a/Doc/includes/asyncio_guest_tkinter.py b/Doc/includes/asyncio_guest_tkinter.py new file mode 100644 index 00000000000000..719a76e72d1159 --- /dev/null +++ b/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() diff --git a/Lib/asyncio/__init__.py b/Lib/asyncio/__init__.py index 32a5dbae03af21..9f5eae4c170a8d 100644 --- a/Lib/asyncio/__init__.py +++ b/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__ + diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py index 6619c87bcf5b93..fcca65c816d591 100644 --- a/Lib/asyncio/base_events.py +++ b/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): diff --git a/Lib/asyncio/guest.py b/Lib/asyncio/guest.py new file mode 100644 index 00000000000000..71cf207ad6f70a --- /dev/null +++ b/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 diff --git a/Lib/test/test_asyncio/test_guest.py b/Lib/test/test_asyncio/test_guest.py new file mode 100644 index 00000000000000..b9c92e76df2c5a --- /dev/null +++ b/Lib/test/test_asyncio/test_guest.py @@ -0,0 +1,212 @@ +"""Tests for asyncio.start_guest_run().""" + +import asyncio +import queue +import threading +import time +import unittest + + +class MockHost: + """A minimal host event loop that uses a thread-safe queue. + + Simulates a GUI toolkit main loop without any actual GUI dependency. + Callbacks are collected in a queue and drained by :meth:`run`. + """ + + def __init__(self): + self._queue = queue.Queue() + self._done = threading.Event() + self._task = None + + def run_sync_soon_threadsafe(self, fn): + self._queue.put(fn) + + def done_callback(self, task): + self._task = task + self._done.set() + + def run(self, timeout=10.0): + """Drain callbacks until *done_callback* fires or *timeout* expires.""" + deadline = time.monotonic() + timeout + while not self._done.is_set(): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("MockHost.run() timed out") + try: + fn = self._queue.get(timeout=min(remaining, 0.05)) + fn() + except queue.Empty: + pass + # Drain any trailing callbacks. + while True: + try: + fn = self._queue.get_nowait() + fn() + except queue.Empty: + break + return self._task + + +class TestGuestRun(unittest.TestCase): + """Test asyncio.start_guest_run with a mock host loop.""" + + def _run_guest(self, async_fn, *args, timeout=10.0): + """Helper: run *async_fn* in guest mode and return the completed task.""" + host = MockHost() + asyncio.start_guest_run( + async_fn, *args, + run_sync_soon_threadsafe=host.run_sync_soon_threadsafe, + done_callback=host.done_callback, + ) + return host.run(timeout=timeout) + + # -- basic lifecycle ----------------------------------------------- + + def test_simple_return(self): + async def coro(): + return 42 + + task = self._run_guest(coro) + self.assertTrue(task.done()) + self.assertEqual(task.result(), 42) + + def test_return_none(self): + async def coro(): + pass + + task = self._run_guest(coro) + self.assertIsNone(task.result()) + + def test_with_args(self): + async def add(a, b): + return a + b + + task = self._run_guest(add, 3, 7) + self.assertEqual(task.result(), 10) + + # -- exception propagation ----------------------------------------- + + def test_exception(self): + async def coro(): + raise ValueError("boom") + + task = self._run_guest(coro) + self.assertTrue(task.done()) + with self.assertRaises(ValueError) as cm: + task.result() + self.assertEqual(str(cm.exception), "boom") + + # -- cancellation -------------------------------------------------- + + def test_cancel_from_host(self): + started = threading.Event() + + async def coro(): + started.set() + await asyncio.sleep(3600) + + host = MockHost() + task = asyncio.start_guest_run( + coro, + run_sync_soon_threadsafe=host.run_sync_soon_threadsafe, + done_callback=host.done_callback, + ) + # Wait for the coroutine to start, then cancel. + # Use call_soon_threadsafe to wake the I/O thread's selector. + started.wait(timeout=5) + loop = task.get_loop() + loop.call_soon_threadsafe(task.cancel) + host.run(timeout=5) + self.assertTrue(task.cancelled()) + + # -- asyncio primitives work inside guest -------------------------- + + def test_sleep(self): + async def coro(): + t0 = asyncio.get_event_loop().time() + await asyncio.sleep(0.1) + elapsed = asyncio.get_event_loop().time() - t0 + return elapsed + + task = self._run_guest(coro) + elapsed = task.result() + self.assertGreaterEqual(elapsed, 0.05) + + def test_create_task(self): + async def helper(): + await asyncio.sleep(0.01) + return "helper" + + async def coro(): + t = asyncio.ensure_future(helper()) + result = await t + return result + + task = self._run_guest(coro) + self.assertEqual(task.result(), "helper") + + def test_gather(self): + async def sleeper(n): + await asyncio.sleep(0.01 * n) + return n + + async def coro(): + results = await asyncio.gather( + sleeper(1), sleeper(2), sleeper(3) + ) + return results + + task = self._run_guest(coro) + self.assertEqual(task.result(), [1, 2, 3]) + + def test_call_later(self): + async def coro(): + loop = asyncio.get_event_loop() + fut = loop.create_future() + loop.call_later(0.05, fut.set_result, "later") + return await fut + + task = self._run_guest(coro) + self.assertEqual(task.result(), "later") + + def test_call_soon_threadsafe(self): + async def coro(): + loop = asyncio.get_event_loop() + fut = loop.create_future() + + def setter(): + loop.call_soon_threadsafe(fut.set_result, "safe") + threading.Timer(0.05, setter).start() + return await fut + + task = self._run_guest(coro) + self.assertEqual(task.result(), "safe") + + +class TestBaseEventLoopDecomposition(unittest.TestCase): + """Verify that poll_events / process_events / process_ready exist + and compose correctly (i.e. _run_once still works).""" + + def test_methods_exist(self): + loop = asyncio.new_event_loop() + try: + self.assertTrue(hasattr(loop, 'poll_events')) + self.assertTrue(hasattr(loop, 'process_events')) + self.assertTrue(hasattr(loop, 'process_ready')) + finally: + loop.close() + + def test_run_once_still_works(self): + """asyncio.run() exercises _run_once(); ensure it still functions + after the refactor.""" + async def coro(): + await asyncio.sleep(0) + return "ok" + + result = asyncio.run(coro()) + self.assertEqual(result, "ok") + + +if __name__ == '__main__': + unittest.main() From 8e7f0ee2dc4c1034e99db2f2e7dfa4da97f50a76 Mon Sep 17 00:00:00 2001 From: Cong Zhang <13283869+congzhangzh@users.noreply.github.com> Date: Sat, 28 Feb 2026 14:31:19 +0800 Subject: [PATCH 02/10] gh-145342: asyncio: Add docs and NEWS entry for guest mode --- Doc/library/asyncio-eventloop.rst | 25 ++++ Doc/library/asyncio-guest.rst | 113 ++++++++++++++++++ Doc/library/asyncio.rst | 1 + ...-28-14-00-00.gh-issue-145342.GuestMode.rst | 8 ++ 4 files changed, 147 insertions(+) create mode 100644 Doc/library/asyncio-guest.rst create mode 100644 Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst diff --git a/Doc/library/asyncio-eventloop.rst b/Doc/library/asyncio-eventloop.rst index bdb24b3a58c267..54456b80167567 100644 --- a/Doc/library/asyncio-eventloop.rst +++ b/Doc/library/asyncio-eventloop.rst @@ -218,6 +218,31 @@ Running and stopping the loop .. versionchanged:: 3.12 Added the *timeout* parameter. +Decomposing event loop iteration +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +The following methods decompose a single :meth:`~asyncio.BaseEventLoop._run_once` +iteration into independently callable steps. They are used internally by +:func:`asyncio.start_guest_run`; see :ref:`asyncio-guest` for full documentation. + +.. method:: loop.poll_events() + + Poll for I/O events and return the raw event list. + + .. versionadded:: 3.15 + +.. method:: loop.process_events(event_list) + + Process I/O events returned by :meth:`poll_events`. + + .. versionadded:: 3.15 + +.. method:: loop.process_ready() + + Process expired timers and execute ready callbacks. + + .. versionadded:: 3.15 + Scheduling callbacks ^^^^^^^^^^^^^^^^^^^^ diff --git a/Doc/library/asyncio-guest.rst b/Doc/library/asyncio-guest.rst new file mode 100644 index 00000000000000..9340df9bedd18a --- /dev/null +++ b/Doc/library/asyncio-guest.rst @@ -0,0 +1,113 @@ +.. currentmodule:: asyncio + +.. _asyncio-guest: + +========== +Guest Mode +========== + +**Source code:** :source:`Lib/asyncio/guest.py` + +---- + +Running asyncio as a Guest in Another Event Loop +================================================= + +*Guest mode* allows asyncio to run cooperatively inside a *host* event loop +such as a GUI toolkit's main loop (Tkinter, Qt, GTK, etc.). Instead of +replacing the host loop, asyncio piggybacks on it: + +* The **host thread** keeps running its own main loop as usual. +* A **background daemon thread** blocks on the selector (I/O polling). + When I/O events arrive it hands them back to the host thread via a + thread-safe callback. +* The host thread then runs :meth:`~asyncio.BaseEventLoop.process_events` + and :meth:`~asyncio.BaseEventLoop.process_ready` to advance the asyncio + event loop by one step, then signals the background thread to poll again. + +This dual-thread architecture means neither the host loop nor the asyncio +loop starves the other. + +Typical use cases: + +* Incrementally migrating a Tkinter/Qt/GTK application to ``async/await`` + without replacing the native event loop. +* Embedding asyncio I/O (HTTP clients, websockets, …) inside a GUI app. +* Running asyncio alongside a framework that owns the main thread. + +.. rubric:: Example + +See :source:`Doc/includes/asyncio_guest_tkinter.py` for a complete Tkinter +example that embeds asyncio inside ``tkinter.mainloop()`` using +:func:`start_guest_run`. + +.. rubric:: API + +.. function:: start_guest_run(async_fn, *args, run_sync_soon_threadsafe, done_callback) + + Run *async_fn* 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*. + + :param async_fn: The async function to run as the top-level coroutine. + :param args: Positional arguments forwarded to *async_fn*. + :param run_sync_soon_threadsafe: A callable that schedules a zero-argument + callable on the host event loop's thread in a thread-safe manner. + For Tkinter use ``widget.after(0, fn)``; for Qt use a + ``QMetaObject.invokeMethod`` wrapper; etc. + :param done_callback: Called on the host thread when *async_fn* finishes. + Receives the completed :class:`Task` as its sole argument. Inspect + the outcome with :meth:`Task.result`, :meth:`Task.exception`, or + :meth:`Task.cancelled`. + :returns: The :class:`Task` wrapping *async_fn*. + + To cancel the task from the host thread, use:: + + loop.call_soon_threadsafe(task.cancel) + + This wakes the I/O thread from its selector wait so cancellation is + processed promptly. + + .. versionadded:: 3.15 + +.. rubric:: Low-level Event Loop Methods + +The following three methods on :class:`BaseEventLoop` are used internally by +:func:`start_guest_run`. They decompose :meth:`~BaseEventLoop._run_once` +into independently callable steps and are documented here for completeness. + +.. method:: loop.poll_events() + + Poll for I/O events without processing them. + + Cleans up cancelled scheduled handles, computes an appropriate timeout + from the scheduled callbacks, and calls the underlying selector. Returns + the raw event list. + + Together with :meth:`~BaseEventLoop.process_events` and + :meth:`~BaseEventLoop.process_ready`, this method decomposes + :meth:`~BaseEventLoop._run_once` into independently callable steps so that + an external event loop can drive asyncio (see :func:`start_guest_run`). + + .. versionadded:: 3.15 + +.. method:: loop.process_events(event_list) + + Process I/O events returned by :meth:`~BaseEventLoop.poll_events`. + + Delegates to the selector-specific ``_process_events`` implementation + which turns raw selector events into ready callbacks. + + .. versionadded:: 3.15 + +.. method:: loop.process_ready() + + 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 diff --git a/Doc/library/asyncio.rst b/Doc/library/asyncio.rst index 0f72e31dee5f1d..561bd338cc321e 100644 --- a/Doc/library/asyncio.rst +++ b/Doc/library/asyncio.rst @@ -120,6 +120,7 @@ for full functionality and the latest features. asyncio-policy.rst asyncio-platforms.rst asyncio-extending.rst + asyncio-guest.rst .. toctree:: :caption: Guides and Tutorials diff --git a/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst b/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst new file mode 100644 index 00000000000000..948489958e7000 --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst @@ -0,0 +1,8 @@ +Add :func:`asyncio.start_guest_run` to allow asyncio to run cooperatively +inside a host event loop (e.g. Tkinter, Qt, GTK). The host loop retains +control of the main thread while asyncio I/O polling runs in a background +daemon thread. Also adds three low-level :class:`~asyncio.BaseEventLoop` +methods -- :meth:`~asyncio.BaseEventLoop.poll_events`, +:meth:`~asyncio.BaseEventLoop.process_events`, and +:meth:`~asyncio.BaseEventLoop.process_ready` -- that decompose +:meth:`~asyncio.BaseEventLoop._run_once` into independently callable steps. From 2c0170bcda9f213a2dc6219c236b7d13e06bbf95 Mon Sep 17 00:00:00 2001 From: Cong Zhang <13283869+congzhangzh@users.noreply.github.com> Date: Fri, 14 Aug 2026 20:44:34 -0400 Subject: [PATCH 03/10] gh-145342: Add _guest_mode flag and guest-mode signal-handler guards In guest mode the host event loop owns signal handling: add_signal_handler() and remove_signal_handler() raise RuntimeError so that asyncio never touches the process-global signal wakeup fd. --- Lib/asyncio/base_events.py | 4 ++++ Lib/asyncio/unix_events.py | 9 +++++++++ 2 files changed, 13 insertions(+) diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py index fdcde8fe2f6cb2..30f4ba32e42e0f 100644 --- a/Lib/asyncio/base_events.py +++ b/Lib/asyncio/base_events.py @@ -432,6 +432,10 @@ def __init__(self): # Identifier of the thread running the event loop, or None if the # event loop is not running self._thread_id = None + # True while the loop is driven as a guest inside a host event + # loop (see asyncio.start_guest_run); signal handling is then + # delegated to the host. + self._guest_mode = False self._clock_resolution = time.get_clock_info('monotonic').resolution self._exception_handler = None self.set_debug(coroutines._is_debug_mode()) diff --git a/Lib/asyncio/unix_events.py b/Lib/asyncio/unix_events.py index 646ae71bbf5919..8a44c9d06276f0 100644 --- a/Lib/asyncio/unix_events.py +++ b/Lib/asyncio/unix_events.py @@ -100,6 +100,11 @@ def add_signal_handler(self, sig, callback, *args): "with add_signal_handler()") self._check_signal(sig) self._check_closed() + if self._guest_mode: + raise RuntimeError( + "add_signal_handler() is not supported in asyncio guest " + "mode; the host event loop owns signal handling -- " + "forward signals to the loop with call_soon_threadsafe()") try: # set_wakeup_fd() raises ValueError if this is not the # main thread. By calling it early we ensure that an @@ -149,6 +154,10 @@ def remove_signal_handler(self, sig): Return True if a signal handler was removed, False if not. """ self._check_signal(sig) + if self._guest_mode: + raise RuntimeError( + "remove_signal_handler() is not supported in asyncio " + "guest mode; the host event loop owns signal handling") try: del self._signal_handlers[sig] except KeyError: From aa4259c42bd97f8de47db739d59cf3b53b16b365 Mon Sep 17 00:00:00 2001 From: Cong Zhang <13283869+congzhangzh@users.noreply.github.com> Date: Fri, 14 Aug 2026 20:44:34 -0400 Subject: [PATCH 04/10] gh-145342: Make the guest I/O thread non-daemonic with graceful shutdown Address review feedback on the guest-mode runtime: - The I/O thread is no longer a daemon thread; it is joined when the run finishes, and a threading._register_atexit() hook (the concurrent.futures pattern) wakes it out of its selector wait so an unfinished run cannot hang interpreter shutdown. - Reuse loop._run_forever_setup()/_run_forever_cleanup() so the whole guest run counts as running: get_running_loop() works, is_running() is true, nested run_forever()/run_until_complete() raise, asyncgen hooks and coroutine origin tracking are installed and restored. On Windows this also establishes the proactor self-reading loop needed for call_soon_threadsafe() to wake IocpProactor.select(). - Preserve the host's signal wakeup fd across loop creation and close (the proactor installs/resets it in __init__/close on the main thread). - On completion, run the same cleanup as asyncio.run() -- cancel remaining tasks, shutdown asyncgens and the default executor, close the loop -- before invoking done_callback. - Harden the host/IO-thread handshake: exception-safe token handoff, and an abort path that unwinds the main task if the I/O thread dies. --- Lib/asyncio/guest.py | 380 ++++++++++++++++++++++++++++++++++--------- 1 file changed, 304 insertions(+), 76 deletions(-) diff --git a/Lib/asyncio/guest.py b/Lib/asyncio/guest.py index 71cf207ad6f70a..2b2021c04a8832 100644 --- a/Lib/asyncio/guest.py +++ b/Lib/asyncio/guest.py @@ -2,121 +2,349 @@ 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: +The host loop stays in control of its 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 + Host thread: process_events() + process_ready() -> hand token to I/O thread + I/O thread: wait for token -> poll_events() -> hand events to host + +A single "token" (semaphore permit) ping-pongs between the two threads, +so exactly one of them touches the event loop at any moment. Inspired by Trio's guest mode (trio.lowlevel.start_guest_run). """ __all__ = ('start_guest_run',) +import signal import threading from functools import partial +from . import constants from . import events +from . import runners +from . import tasks +from .log import logger + +# Stop callbacks of unfinished guest runs, so interpreter shutdown can +# unblock and join their (non-daemon) I/O threads. +_lock = threading.Lock() +_stoppers = set() +_shutting_down = False +_atexit_registered = False + + +def _python_exit(): + """Wake and join the I/O threads of all unfinished guest runs. + + Registered via threading._register_atexit() so it runs before the + interpreter joins non-daemon threads (the concurrent.futures + pattern). Each run is abandoned: pending tasks are not cancelled + and done_callback is not called. + """ + global _shutting_down + with _lock: + _shutting_down = True + stoppers = list(_stoppers) + for stop in stoppers: + stop() + + +def _ensure_atexit(): + global _atexit_registered + with _lock: + if _shutting_down: + raise RuntimeError( + 'cannot start an asyncio guest run at interpreter shutdown') + if not _atexit_registered: + threading._register_atexit(_python_exit) + _atexit_registered = True + + +def _save_wakeup_fd(): + """Return the current signal wakeup fd, or None if it cannot be read. + + There is no getter for the wakeup fd, so briefly swapping it out is + the only way to read it. A signal arriving between the two calls + loses its wakeup byte (its Python-level handler still runs); the + window is a few instructions wide. + """ + if (not hasattr(signal, 'set_wakeup_fd') + or threading.current_thread() is not threading.main_thread()): + return None + fd = signal.set_wakeup_fd(-1) + if fd != -1: + signal.set_wakeup_fd(fd) + return fd + + +def _restore_wakeup_fd(fd): + if fd is not None: + signal.set_wakeup_fd(fd) 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. + """Run async_fn(*args) as a guest inside another event loop. + + Must be called from the host event loop's thread. The host loop + (e.g. Tkinter's mainloop) remains in control of that thread; asyncio + I/O polling runs in a background non-daemon thread that is joined + when the run finishes. Returns the Task wrapping *async_fn*; to + cancel it from the host, use loop.call_soon_threadsafe(task.cancel) + so that the I/O thread is woken from its selector wait. + + run_sync_soon_threadsafe is a callable that schedules a + zero-argument callable to run on the host loop's thread. It must be + thread-safe, must not block, and must not raise; it need not + preserve ordering. + + done_callback is called on the host thread after the run has fully + finished: remaining tasks cancelled, asynchronous generators and the + default executor shut down, and the loop closed (the same cleanup as + asyncio.run()). It receives the completed Task as its sole + argument. + + For the whole run the guest loop is the thread's running loop: + get_running_loop() works inside guest tasks, loop.is_running() is + true, and starting another loop on the thread raises RuntimeError. + Signal handling stays with the host: the guest loop never touches + signal.set_wakeup_fd(), and loop.add_signal_handler() raises + RuntimeError. The host may forward signals into the loop with + loop.call_soon_threadsafe(). + + If the interpreter exits while the run is unfinished, the run is + abandoned: the I/O thread is woken and joined, pending tasks are not + cancelled, and done_callback is not called. """ + _ensure_atexit() + + # Create the loop without letting it capture the signal wakeup fd: + # on Windows, BaseProactorEventLoop.__init__ installs its self-pipe + # as the wakeup fd when on the main thread, which would silently + # break the host's signal handling. + host_wakeup_fd = _save_wakeup_fd() loop = events.new_event_loop() - events._set_running_loop(loop) + _restore_wakeup_fd(host_wakeup_fd) + + # The host owns signal handling (see the docstring). Never reset: + # 'finally' blocks of tasks cancelled during the final cleanup must + # not be able to install signal handlers either. + loop._guest_mode = True - _shutdown = threading.Event() - _sem = threading.Semaphore(0) - _done_called = False + def _close_loop(): + # BaseProactorEventLoop.close() resets the wakeup fd to -1 on + # the main thread; preserve the host's fd across it. + fd = _save_wakeup_fd() + try: + loop.close() + finally: + _restore_wakeup_fd(fd) - # -- helpers ------------------------------------------------------ + # Mark the loop as running for the whole guest run. On Windows this + # also starts the proactor's self-reading loop, so that + # call_soon_threadsafe() can wake the I/O thread's poll. + try: + loop._run_forever_setup() + except BaseException: + _close_loop() + raise - def _finish(task): - """Clean up and forward completion to the host.""" - nonlocal _done_called - if _done_called: + shutdown = threading.Event() + wakeup = threading.Semaphore(0) + finished = False + cleaned_up = False + + try: + main_task = loop.create_task(async_fn(*args)) + except BaseException: + loop._run_forever_cleanup() + _close_loop() + raise + + # -- helpers (host thread unless noted) ---------------------------- + + def _cleanup_running_state(): + nonlocal cleaned_up + if cleaned_up: return - _done_called = True - events._set_running_loop(None) + cleaned_up = True + loop._run_forever_cleanup() + + def _join_backend(): + # By the token invariant the I/O thread is parked in + # wakeup.acquire() here, never in the selector; the + # _write_to_self() wake-up is defensive insurance. + shutdown.set() + wakeup.release() try: - done_callback(task) + loop._write_to_self() + except Exception: + pass + if backend_thread.ident is not None: + backend_thread.join() + + def _deliver(task, *, graceful): + with _lock: + _stoppers.discard(_stop_at_exit) + try: + if graceful: + # Same cleanup as asyncio.run(). + runners._cancel_all_tasks(loop) + loop.run_until_complete(loop.shutdown_asyncgens()) + loop.run_until_complete( + loop.shutdown_default_executor( + constants.THREAD_JOIN_TIMEOUT)) finally: - if not loop.is_closed(): - loop.close() + _close_loop() + done_callback(task) + + def _finish(task): + # Always scheduled via run_sync_soon_threadsafe, never called + # inline from a loop callback: the cleanup below drives the loop + # with run_until_complete(), which would re-enter the + # process_ready() iteration such a callback runs in. + nonlocal finished + if finished: + return + finished = True + _join_backend() + _cleanup_running_state() + _deliver(task, graceful=True) + + def _abort(exc): + # The I/O thread died; the loop can no longer be driven through + # the guest handshake. Unwind the main task directly. + nonlocal finished + if finished: + return + finished = True + _join_backend() + _cleanup_running_state() + if not main_task.done(): + main_task.cancel( + msg=f'asyncio guest I/O thread failed: {exc!r}') + try: + loop.run_until_complete( + tasks.gather(main_task, return_exceptions=True)) + except Exception: + # The loop itself is broken (e.g. the selector raised); + # abandon any pending work. + pass + loop.call_exception_handler({ + 'message': 'asyncio guest I/O thread failed', + 'exception': exc, + 'task': main_task, + }) + _deliver(main_task, graceful=False) def _process_on_host(event_list): - """Run on the host thread: process one batch of asyncio work.""" - if _shutdown.is_set(): + """Process one batch of asyncio work on the host thread.""" + if shutdown.is_set() or loop.is_closed(): return - loop.process_events(event_list) - loop.process_ready() - if not _shutdown.is_set(): - _sem.release() + try: + loop.process_events(event_list) + loop.process_ready() + except BaseException: + # Internal loop failure (user callback exceptions are routed + # to the exception handler by Handle._run). Unblock the I/O + # thread so it can exit and be joined. + shutdown.set() + wakeup.release() + raise + # Hand the polling token back to the I/O thread. Exactly one + # token circulates: while the host runs a batch, the I/O thread + # is parked in wakeup.acquire(), so it can never poll (or touch + # the loop at all) concurrently with this function. + if not shutdown.is_set(): + wakeup.release() - # -- threads ------------------------------------------------------- + def _on_task_done(task): + # Runs inside process_ready() while the host holds the token, + # so the I/O thread is parked in wakeup.acquire(): the released + # token wakes it, it observes 'shutdown', and exits without + # re-entering the selector. + shutdown.set() + wakeup.release() + try: + run_sync_soon_threadsafe(partial(_finish, task)) + except Exception as exc: + loop.call_exception_handler({ + 'message': ('asyncio guest run could not schedule its ' + 'completion callback on the host'), + 'exception': exc, + 'task': task, + }) def _backend(): - """Daemon thread: poll for I/O and wake the host.""" + """I/O thread: wait for the token, poll, hand events to the host.""" try: - while not _shutdown.is_set(): - _sem.acquire() - if _shutdown.is_set(): - break + while True: + wakeup.acquire() + if shutdown.is_set(): + return event_list = loop.poll_events() + if shutdown.is_set(): + # Interpreter exit woke the selector; the host may + # be gone, so do not call into it. + return run_sync_soon_threadsafe( - partial(_process_on_host, event_list) - ) + 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)) + if shutdown.is_set(): + return + shutdown.set() + try: + run_sync_soon_threadsafe(partial(_abort, exc)) + except Exception: + logger.error( + 'asyncio guest run abandoned: host is unreachable ' + 'after an I/O thread failure', exc_info=True) - # -- task setup ---------------------------------------------------- + def _stop_at_exit(): + # Interpreter-exit hook: unblock and join the I/O thread so + # shutdown does not hang on a non-daemon thread. The run is + # abandoned -- no cancellation, no callbacks, no loop.close(). + shutdown.set() + wakeup.release() + try: + loop._write_to_self() + except Exception: + pass + if backend_thread.ident is not None: + backend_thread.join() - 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)) + # -- start --------------------------------------------------------- main_task.add_done_callback(_on_task_done) - # Kick off: process the initial callbacks enqueued by create_task. - _process_on_host([]) + backend_thread = threading.Thread( + target=_backend, name='asyncio-guest-io') + + with _lock: + shutting_down = _shutting_down + if not shutting_down: + _stoppers.add(_stop_at_exit) + if shutting_down: + loop._run_forever_cleanup() + _close_loop() + raise RuntimeError( + 'cannot start an asyncio guest run at interpreter shutdown') - threading.Thread( - target=_backend, daemon=True, name='asyncio-guest-io' - ).start() + try: + # Process the callbacks enqueued by create_task(), then let the + # I/O thread take over polling. The thread is started even if + # the task already finished: it consumes the shutdown token and + # exits, keeping the join logic uniform. + _process_on_host([]) + backend_thread.start() + except BaseException: + with _lock: + _stoppers.discard(_stop_at_exit) + shutdown.set() + _cleanup_running_state() + _close_loop() + raise return main_task From 6b461aa8ff2eae532baf82b2a52807a4ab8e6bea Mon Sep 17 00:00:00 2001 From: Cong Zhang <13283869+congzhangzh@users.noreply.github.com> Date: Fri, 14 Aug 2026 20:44:35 -0400 Subject: [PATCH 05/10] gh-145342: Expand asyncio guest mode tests Follow test_asyncio conventions (threading_helper, policy reset) and cover the new behavior: non-daemon I/O thread joined after the run, running-loop semantics, signal-handler RuntimeError, restored asyncgen hooks, asyncio.run()-equivalent cleanup, wakeup fd preservation, and clean interpreter exit with an unfinished run. --- Lib/test/test_asyncio/test_guest.py | 256 ++++++++++++++++++++++++++-- 1 file changed, 239 insertions(+), 17 deletions(-) diff --git a/Lib/test/test_asyncio/test_guest.py b/Lib/test/test_asyncio/test_guest.py index b9c92e76df2c5a..a5c19ce0f054cc 100644 --- a/Lib/test/test_asyncio/test_guest.py +++ b/Lib/test/test_asyncio/test_guest.py @@ -2,9 +2,20 @@ import asyncio import queue +import signal +import socket +import sys import threading import time import unittest +from test.support import threading_helper +from test.support.script_helper import assert_python_ok + +threading_helper.requires_working_threading(module=True) + + +def tearDownModule(): + asyncio.events._set_event_loop_policy(None) class MockHost: @@ -48,8 +59,13 @@ def run(self, timeout=10.0): return self._task -class TestGuestRun(unittest.TestCase): - """Test asyncio.start_guest_run with a mock host loop.""" +class GuestTestCase(unittest.TestCase): + + def setUp(self): + self._thread_key = threading_helper.threading_setup() + + def tearDown(self): + threading_helper.threading_cleanup(*self._thread_key) def _run_guest(self, async_fn, *args, timeout=10.0): """Helper: run *async_fn* in guest mode and return the completed task.""" @@ -61,6 +77,10 @@ def _run_guest(self, async_fn, *args, timeout=10.0): ) return host.run(timeout=timeout) + +class TestGuestRun(GuestTestCase): + """Test asyncio.start_guest_run with a mock host loop.""" + # -- basic lifecycle ----------------------------------------------- def test_simple_return(self): @@ -85,6 +105,20 @@ async def add(a, b): task = self._run_guest(add, 3, 7) self.assertEqual(task.result(), 10) + def test_early_sync_completion(self): + # The task can already be done when the I/O thread starts. + async def coro(): + return 'early' + + host = MockHost() + task = asyncio.start_guest_run( + coro, + run_sync_soon_threadsafe=host.run_sync_soon_threadsafe, + done_callback=host.done_callback, + ) + self.assertIs(host.run(), task) + self.assertEqual(task.result(), 'early') + # -- exception propagation ----------------------------------------- def test_exception(self): @@ -124,14 +158,13 @@ async def coro(): def test_sleep(self): async def coro(): - t0 = asyncio.get_event_loop().time() + loop = asyncio.get_running_loop() + t0 = loop.time() await asyncio.sleep(0.1) - elapsed = asyncio.get_event_loop().time() - t0 - return elapsed + return loop.time() - t0 task = self._run_guest(coro) - elapsed = task.result() - self.assertGreaterEqual(elapsed, 0.05) + self.assertGreaterEqual(task.result(), 0.05) def test_create_task(self): async def helper(): @@ -140,8 +173,7 @@ async def helper(): async def coro(): t = asyncio.ensure_future(helper()) - result = await t - return result + return await t task = self._run_guest(coro) self.assertEqual(task.result(), "helper") @@ -152,17 +184,14 @@ async def sleeper(n): return n async def coro(): - results = await asyncio.gather( - sleeper(1), sleeper(2), sleeper(3) - ) - return results + return await asyncio.gather(sleeper(1), sleeper(2), sleeper(3)) task = self._run_guest(coro) self.assertEqual(task.result(), [1, 2, 3]) def test_call_later(self): async def coro(): - loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() fut = loop.create_future() loop.call_later(0.05, fut.set_result, "later") return await fut @@ -171,20 +200,213 @@ async def coro(): self.assertEqual(task.result(), "later") def test_call_soon_threadsafe(self): + timer = None + async def coro(): - loop = asyncio.get_event_loop() + nonlocal timer + loop = asyncio.get_running_loop() fut = loop.create_future() def setter(): loop.call_soon_threadsafe(fut.set_result, "safe") - threading.Timer(0.05, setter).start() + timer = threading.Timer(0.05, setter) + timer.start() return await fut task = self._run_guest(coro) self.assertEqual(task.result(), "safe") + timer.join() + + # -- thread lifecycle ---------------------------------------------- + + def test_io_thread_nondaemon_and_joined(self): + seen = {} + + async def coro(): + # The I/O thread is started after the initial batch; a sleep + # guarantees it is up and polling by the time we look. + await asyncio.sleep(0.01) + for thread in threading.enumerate(): + if thread.name == 'asyncio-guest-io': + seen['thread'] = thread + + task = self._run_guest(coro) + self.assertIsNone(task.exception()) + self.assertIn('thread', seen) + self.assertFalse(seen['thread'].daemon) + self.assertFalse(seen['thread'].is_alive()) + + def test_interpreter_exit_with_pending_run(self): + # Exiting with an unfinished guest run must not hang: the atexit + # hook wakes the non-daemon I/O thread out of its selector wait + # and joins it. + code = ( + 'import asyncio, collections\n' + 'q = collections.deque()\n' + 'async def coro():\n' + ' await asyncio.sleep(3600)\n' + 'asyncio.start_guest_run(\n' + ' coro,\n' + ' run_sync_soon_threadsafe=q.append,\n' + ' done_callback=lambda task: None,\n' + ')\n' + ) + assert_python_ok('-c', code) + + # -- running-loop semantics ---------------------------------------- + + def test_is_running_inside(self): + async def coro(): + return asyncio.get_running_loop().is_running() + + task = self._run_guest(coro) + self.assertTrue(task.result()) + + def test_nested_run_raises(self): + test = self + + async def coro(): + loop = asyncio.get_running_loop() + inner = asyncio.sleep(0) + try: + with test.assertRaises(RuntimeError): + loop.run_until_complete(inner) + finally: + inner.close() + inner = asyncio.sleep(0) + try: + with test.assertRaises(RuntimeError): + asyncio.run(inner) + finally: + inner.close() + + task = self._run_guest(coro) + self.assertIsNone(task.exception()) + + def test_state_restored_after_run(self): + old_hooks = sys.get_asyncgen_hooks() + + async def coro(): + pass + + task = self._run_guest(coro) + self.assertEqual(sys.get_asyncgen_hooks(), old_hooks) + self.assertIsNone(asyncio._get_running_loop()) + self.assertFalse(task.get_loop().is_running()) + + # -- signal handling ----------------------------------------------- + + @unittest.skipUnless(hasattr(signal, 'SIGUSR1'), + 'requires UNIX signal handling') + def test_add_signal_handler_raises(self): + async def coro(): + loop = asyncio.get_running_loop() + loop.add_signal_handler(signal.SIGUSR1, lambda: None) + + task = self._run_guest(coro) + with self.assertRaisesRegex(RuntimeError, 'guest mode'): + task.result() + + @unittest.skipUnless(hasattr(signal, 'SIGUSR1'), + 'requires UNIX signal handling') + def test_remove_signal_handler_raises(self): + async def coro(): + loop = asyncio.get_running_loop() + loop.remove_signal_handler(signal.SIGUSR1) + + task = self._run_guest(coro) + with self.assertRaisesRegex(RuntimeError, 'guest mode'): + task.result() + + @unittest.skipUnless(hasattr(signal, 'set_wakeup_fd'), + 'requires signal.set_wakeup_fd') + def test_wakeup_fd_preserved(self): + if threading.current_thread() is not threading.main_thread(): + self.skipTest('requires the main thread') + rsock, wsock = socket.socketpair() + self.addCleanup(rsock.close) + self.addCleanup(wsock.close) + wsock.setblocking(False) + old_fd = signal.set_wakeup_fd(wsock.fileno()) + self.addCleanup(signal.set_wakeup_fd, old_fd) + + async def coro(): + await asyncio.sleep(0.01) + + self._run_guest(coro) + + fd = signal.set_wakeup_fd(-1) + if fd != -1: + signal.set_wakeup_fd(fd) + self.assertEqual(fd, wsock.fileno()) + + # -- final cleanup matches asyncio.run() --------------------------- + + def test_background_task_cancelled_on_finish(self): + state = {} + + async def background(): + await asyncio.sleep(3600) + + async def coro(): + state['bg'] = asyncio.get_running_loop().create_task(background()) + await asyncio.sleep(0.01) + + task = self._run_guest(coro) + self.assertIsNone(task.exception()) + self.assertTrue(state['bg'].cancelled()) + + def test_abandoned_asyncgen_finalized(self): + finalized = False + holder = [] + + async def agen(): + nonlocal finalized + try: + yield 1 + finally: + finalized = True + + async def coro(): + it = agen() + holder.append(it) # keep it alive until shutdown_asyncgens() + await anext(it) + + task = self._run_guest(coro) + self.assertIsNone(task.exception()) + self.assertTrue(finalized) + + def test_loop_closed_in_done_callback(self): + # Cleanup (cancel remaining tasks, close the loop) happens + # before done_callback, like asyncio.run(). + seen = {} + host = MockHost() + original = host.done_callback + + def done_callback(task): + seen['closed'] = task.get_loop().is_closed() + original(task) + + async def coro(): + pass + + asyncio.start_guest_run( + coro, + run_sync_soon_threadsafe=host.run_sync_soon_threadsafe, + done_callback=done_callback, + ) + host.run() + self.assertTrue(seen['closed']) + + def test_loop_closed_after_run(self): + async def coro(): + pass + + task = self._run_guest(coro) + self.assertTrue(task.get_loop().is_closed()) -class TestBaseEventLoopDecomposition(unittest.TestCase): +class TestBaseEventLoopDecomposition(GuestTestCase): """Verify that poll_events / process_events / process_ready exist and compose correctly (i.e. _run_once still works).""" From 5f11a0ad1212db21f4b669956b06e5b81a029237 Mon Sep 17 00:00:00 2001 From: Cong Zhang <13283869+congzhangzh@users.noreply.github.com> Date: Fri, 14 Aug 2026 20:44:35 -0400 Subject: [PATCH 06/10] gh-145342: Update asyncio guest mode docs and NEWS Document the non-daemon I/O thread, lifecycle and cleanup semantics, signal-handling delegation to the host, and the host requirements. --- Doc/includes/asyncio_guest_tkinter.py | 2 +- Doc/library/asyncio-guest.rst | 84 +++++++++++++++---- ...-28-14-00-00.gh-issue-145342.GuestMode.rst | 10 ++- 3 files changed, 77 insertions(+), 19 deletions(-) diff --git a/Doc/includes/asyncio_guest_tkinter.py b/Doc/includes/asyncio_guest_tkinter.py index 719a76e72d1159..356fbe46b5a0a6 100644 --- a/Doc/includes/asyncio_guest_tkinter.py +++ b/Doc/includes/asyncio_guest_tkinter.py @@ -64,7 +64,7 @@ async def count(progress, root): progress.configure(maximum=MAX_COUNT) task = asyncio.current_task() - loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() # Wire the Cancel button and window close to task.cancel(). # Use call_soon_threadsafe so the I/O thread's selector is woken. diff --git a/Doc/library/asyncio-guest.rst b/Doc/library/asyncio-guest.rst index 9340df9bedd18a..c803e7570c17d7 100644 --- a/Doc/library/asyncio-guest.rst +++ b/Doc/library/asyncio-guest.rst @@ -18,15 +18,16 @@ such as a GUI toolkit's main loop (Tkinter, Qt, GTK, etc.). Instead of replacing the host loop, asyncio piggybacks on it: * The **host thread** keeps running its own main loop as usual. -* A **background daemon thread** blocks on the selector (I/O polling). +* A **background I/O thread** blocks on the selector (I/O polling). When I/O events arrive it hands them back to the host thread via a - thread-safe callback. + thread-safe callback. The thread is not a daemon thread; it is joined + when the guest run finishes. * The host thread then runs :meth:`~asyncio.BaseEventLoop.process_events` and :meth:`~asyncio.BaseEventLoop.process_ready` to advance the asyncio - event loop by one step, then signals the background thread to poll again. + event loop by one step, then signals the I/O thread to poll again. -This dual-thread architecture means neither the host loop nor the asyncio -loop starves the other. +Exactly one of the two threads touches the event loop at any moment, so +neither the host loop nor the asyncio loop starves the other. Typical use cases: @@ -47,23 +48,26 @@ example that embeds asyncio inside ``tkinter.mainloop()`` using Run *async_fn* 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*. + Must be called from the host event loop's thread. The host loop + (e.g. ``tkinter.mainloop()``) remains in control of that thread; + asyncio I/O polling runs in a background non-daemon thread that is + joined when the run finishes. :param async_fn: The async function to run as the top-level coroutine. :param args: Positional arguments forwarded to *async_fn*. :param run_sync_soon_threadsafe: A callable that schedules a zero-argument - callable on the host event loop's thread in a thread-safe manner. - For Tkinter use ``widget.after(0, fn)``; for Qt use a - ``QMetaObject.invokeMethod`` wrapper; etc. - :param done_callback: Called on the host thread when *async_fn* finishes. - Receives the completed :class:`Task` as its sole argument. Inspect - the outcome with :meth:`Task.result`, :meth:`Task.exception`, or + callable on the host event loop's thread. It must be thread-safe, + must not block, and must not raise; it need not preserve ordering. + For Tkinter use a ``root.call('after', 'idle', ...)`` wrapper; for + Qt use a ``QMetaObject.invokeMethod`` wrapper; etc. + :param done_callback: Called on the host thread after the run has fully + finished and the loop is closed (see :ref:`asyncio-guest-lifecycle`). + Receives the :class:`Task` as its sole argument. Inspect the + outcome with :meth:`Task.result`, :meth:`Task.exception`, or :meth:`Task.cancelled`. :returns: The :class:`Task` wrapping *async_fn*. - To cancel the task from the host thread, use:: + To cancel the task from the host, use:: loop.call_soon_threadsafe(task.cancel) @@ -72,6 +76,56 @@ example that embeds asyncio inside ``tkinter.mainloop()`` using .. versionadded:: 3.15 +.. _asyncio-guest-lifecycle: + +Lifecycle and Cleanup +===================== + +For the whole guest run the guest loop is the host thread's running +loop: :func:`get_running_loop` works inside guest tasks, +:meth:`loop.is_running() ` returns ``True``, and +starting another event loop on that thread — including a nested +:func:`asyncio.run` or :meth:`loop.run_until_complete` — raises +:exc:`RuntimeError`. Consequently a thread that is already running an +asyncio event loop cannot start a guest run. + +When the main task finishes, cleanup equivalent to :func:`asyncio.run` +takes place on the host thread: remaining tasks are cancelled, +asynchronous generators and the default executor are shut down, the I/O +thread is joined, and the loop is closed. Only then is *done_callback* +invoked. + +If the interpreter exits while a guest run is unfinished, the run is +abandoned: the I/O thread is woken and joined so that interpreter +shutdown does not hang, pending tasks are not cancelled, and +*done_callback* is not called. + +Signal Handling +=============== + +In guest mode the *host* owns signal handling: + +* The guest loop never touches :func:`signal.set_wakeup_fd`, neither to + install a file descriptor nor to reset it on close, so the host's + signal wake-up pipeline stays intact. +* :meth:`loop.add_signal_handler` and :meth:`loop.remove_signal_handler` + raise :exc:`RuntimeError`. +* To let asyncio code react to a signal, catch it in the host (with + :func:`signal.signal` or the host framework's facilities) and forward + it into the loop with :meth:`loop.call_soon_threadsafe`. + +Host Requirements +================= + +* *run_sync_soon_threadsafe* must be thread-safe, non-blocking, and must + not raise. It may run callbacks in any order. +* Host code running *outside* guest callbacks (for example a GUI button + handler) must interact with the loop exclusively through + :meth:`loop.call_soon_threadsafe`, even though it runs on the loop's + own thread: the I/O thread may be inside the selector, and only + ``call_soon_threadsafe`` wakes it safely. +* :meth:`loop.stop` is not supported in guest mode. + .. rubric:: Low-level Event Loop Methods The following three methods on :class:`BaseEventLoop` are used internally by diff --git a/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst b/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst index 948489958e7000..54f08187397745 100644 --- a/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst +++ b/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst @@ -1,8 +1,12 @@ Add :func:`asyncio.start_guest_run` to allow asyncio to run cooperatively inside a host event loop (e.g. Tkinter, Qt, GTK). The host loop retains -control of the main thread while asyncio I/O polling runs in a background -daemon thread. Also adds three low-level :class:`~asyncio.BaseEventLoop` -methods -- :meth:`~asyncio.BaseEventLoop.poll_events`, +control of its thread while asyncio I/O polling runs in a background +non-daemon thread that is joined when the run finishes. In guest mode the +host owns signal handling: the loop never touches +:func:`signal.set_wakeup_fd` and :meth:`loop.add_signal_handler` raises +:exc:`RuntimeError`. Also adds three low-level +:class:`~asyncio.BaseEventLoop` methods -- +:meth:`~asyncio.BaseEventLoop.poll_events`, :meth:`~asyncio.BaseEventLoop.process_events`, and :meth:`~asyncio.BaseEventLoop.process_ready` -- that decompose :meth:`~asyncio.BaseEventLoop._run_once` into independently callable steps. From e5ecca2a861c8c9ebf20a90241ceb77908a9f327 Mon Sep 17 00:00:00 2001 From: Cong Zhang <13283869+congzhangzh@users.noreply.github.com> Date: Fri, 14 Aug 2026 21:03:36 -0400 Subject: [PATCH 07/10] gh-145342: Adapt guest mode to current main and fix docs CI - test_guest: use asyncio.set_event_loop(None) in tearDownModule; the event loop policy system was removed on main. - Docs: de-duplicate the poll_events/process_events/process_ready reference entries (keep asyncio-eventloop.rst as the canonical location), fix cross-references to the loop.* targets, bump versionadded to 3.16, and add a What's New entry. --- Doc/library/asyncio-eventloop.rst | 27 +++++++--- Doc/library/asyncio-guest.rst | 54 +++++-------------- Doc/whatsnew/3.16.rst | 5 ++ Lib/asyncio/base_events.py | 6 +-- Lib/test/test_asyncio/test_guest.py | 2 +- ...-28-14-00-00.gh-issue-145342.GuestMode.rst | 14 ++--- 6 files changed, 49 insertions(+), 59 deletions(-) diff --git a/Doc/library/asyncio-eventloop.rst b/Doc/library/asyncio-eventloop.rst index 9b5e1f652ab704..ead73390b2ac28 100644 --- a/Doc/library/asyncio-eventloop.rst +++ b/Doc/library/asyncio-eventloop.rst @@ -210,27 +210,40 @@ Running and stopping the loop Decomposing event loop iteration ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -The following methods decompose a single :meth:`~asyncio.BaseEventLoop._run_once` -iteration into independently callable steps. They are used internally by -:func:`asyncio.start_guest_run`; see :ref:`asyncio-guest` for full documentation. +The following methods decompose a single iteration of the event loop +into independently callable steps. They are used internally by +:func:`asyncio.start_guest_run`; see :ref:`asyncio-guest` for full +documentation. .. method:: loop.poll_events() - Poll for I/O events and return the raw event list. + Poll for I/O events without processing them. - .. versionadded:: 3.15 + Cleans up cancelled scheduled handles, computes an appropriate + timeout from the scheduled callbacks, and calls the underlying + selector. Returns the raw event list. + + .. versionadded:: 3.16 .. method:: loop.process_events(event_list) Process I/O events returned by :meth:`poll_events`. - .. versionadded:: 3.15 + Delegates to the selector-specific event processing that turns raw + selector events into ready callbacks. + + .. versionadded:: 3.16 .. method:: loop.process_ready() Process expired timers and execute ready callbacks. - .. versionadded:: 3.15 + 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.16 Scheduling callbacks ^^^^^^^^^^^^^^^^^^^^ diff --git a/Doc/library/asyncio-guest.rst b/Doc/library/asyncio-guest.rst index c803e7570c17d7..e92b8a59325a47 100644 --- a/Doc/library/asyncio-guest.rst +++ b/Doc/library/asyncio-guest.rst @@ -22,9 +22,11 @@ replacing the host loop, asyncio piggybacks on it: When I/O events arrive it hands them back to the host thread via a thread-safe callback. The thread is not a daemon thread; it is joined when the guest run finishes. -* The host thread then runs :meth:`~asyncio.BaseEventLoop.process_events` - and :meth:`~asyncio.BaseEventLoop.process_ready` to advance the asyncio - event loop by one step, then signals the I/O thread to poll again. +* The host thread then runs + :meth:`loop.process_events() ` and + :meth:`loop.process_ready() ` to advance + the asyncio event loop by one step, then signals the I/O thread to + poll again. Exactly one of the two threads touches the event loop at any moment, so neither the host loop nor the asyncio loop starves the other. @@ -74,7 +76,7 @@ example that embeds asyncio inside ``tkinter.mainloop()`` using This wakes the I/O thread from its selector wait so cancellation is processed promptly. - .. versionadded:: 3.15 + .. versionadded:: 3.16 .. _asyncio-guest-lifecycle: @@ -128,40 +130,10 @@ Host Requirements .. rubric:: Low-level Event Loop Methods -The following three methods on :class:`BaseEventLoop` are used internally by -:func:`start_guest_run`. They decompose :meth:`~BaseEventLoop._run_once` -into independently callable steps and are documented here for completeness. - -.. method:: loop.poll_events() - - Poll for I/O events without processing them. - - Cleans up cancelled scheduled handles, computes an appropriate timeout - from the scheduled callbacks, and calls the underlying selector. Returns - the raw event list. - - Together with :meth:`~BaseEventLoop.process_events` and - :meth:`~BaseEventLoop.process_ready`, this method decomposes - :meth:`~BaseEventLoop._run_once` into independently callable steps so that - an external event loop can drive asyncio (see :func:`start_guest_run`). - - .. versionadded:: 3.15 - -.. method:: loop.process_events(event_list) - - Process I/O events returned by :meth:`~BaseEventLoop.poll_events`. - - Delegates to the selector-specific ``_process_events`` implementation - which turns raw selector events into ready callbacks. - - .. versionadded:: 3.15 - -.. method:: loop.process_ready() - - 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 +:func:`start_guest_run` drives the loop through three low-level methods +-- :meth:`loop.poll_events() `, +:meth:`loop.process_events() `, and +:meth:`loop.process_ready() ` -- which +decompose a single iteration of the event loop into independently +callable steps. See :ref:`asyncio-event-loop` for their reference +documentation. diff --git a/Doc/whatsnew/3.16.rst b/Doc/whatsnew/3.16.rst index 063755e1eadcb5..1dfc5f5176f078 100644 --- a/Doc/whatsnew/3.16.rst +++ b/Doc/whatsnew/3.16.rst @@ -109,6 +109,11 @@ asyncio socket file created for *path*. (Contributed by Sam Bull in :gh:`94984`.) +* Add :func:`asyncio.start_guest_run` to run asyncio cooperatively inside + a host event loop, such as a GUI toolkit's main loop, that owns the + thread. See :ref:`asyncio-guest`. + (Contributed by Cong Zhang in :gh:`145342`.) + codecs ------ diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py index be9fc4825b0d00..92a47fca698514 100644 --- a/Lib/asyncio/base_events.py +++ b/Lib/asyncio/base_events.py @@ -2005,7 +2005,7 @@ def poll_events(self): independently callable steps so that an external event loop can drive asyncio (see :func:`asyncio.start_guest_run`). - .. versionadded:: 3.15 + .. versionadded:: 3.16 """ sched_count = len(self._scheduled) if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and @@ -2050,7 +2050,7 @@ def process_events(self, event_list): implementation which turns raw selector events into ready callbacks. - .. versionadded:: 3.15 + .. versionadded:: 3.16 """ self._process_events(event_list) @@ -2062,7 +2062,7 @@ def process_ready(self): time. Callbacks enqueued *by* running callbacks are left for the next iteration. - .. versionadded:: 3.15 + .. versionadded:: 3.16 """ # Handle 'later' callbacks that are ready. now = self.time() diff --git a/Lib/test/test_asyncio/test_guest.py b/Lib/test/test_asyncio/test_guest.py index a5c19ce0f054cc..01bcdbd4567646 100644 --- a/Lib/test/test_asyncio/test_guest.py +++ b/Lib/test/test_asyncio/test_guest.py @@ -15,7 +15,7 @@ def tearDownModule(): - asyncio.events._set_event_loop_policy(None) + asyncio.set_event_loop(None) class MockHost: diff --git a/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst b/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst index 54f08187397745..1023249cd5ea47 100644 --- a/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst +++ b/Misc/NEWS.d/next/Library/2026-02-28-14-00-00.gh-issue-145342.GuestMode.rst @@ -3,10 +3,10 @@ inside a host event loop (e.g. Tkinter, Qt, GTK). The host loop retains control of its thread while asyncio I/O polling runs in a background non-daemon thread that is joined when the run finishes. In guest mode the host owns signal handling: the loop never touches -:func:`signal.set_wakeup_fd` and :meth:`loop.add_signal_handler` raises -:exc:`RuntimeError`. Also adds three low-level -:class:`~asyncio.BaseEventLoop` methods -- -:meth:`~asyncio.BaseEventLoop.poll_events`, -:meth:`~asyncio.BaseEventLoop.process_events`, and -:meth:`~asyncio.BaseEventLoop.process_ready` -- that decompose -:meth:`~asyncio.BaseEventLoop._run_once` into independently callable steps. +:func:`signal.set_wakeup_fd` and :meth:`loop.add_signal_handler +` raises :exc:`RuntimeError`. Also adds +three low-level event loop methods -- :meth:`loop.poll_events +`, :meth:`loop.process_events +`, and :meth:`loop.process_ready +` -- that decompose a single iteration of the +event loop into independently callable steps. From 8173afc49007afb932b33fb48deb92722232a976 Mon Sep 17 00:00:00 2001 From: Cong Zhang <13283869+congzhangzh@users.noreply.github.com> Date: Fri, 14 Aug 2026 22:00:45 -0400 Subject: [PATCH 08/10] gh-145342: Reference the asyncio-guest example collection in the docs The asyncio-guest project (the proof of concept this feature is based on) has runnable guest-mode examples for Tkinter, Qt, GTK, pygame, Win32 and Tornado hosts; point to it and to Trio's guest mode from the guest mode docs and the module docstring. --- Doc/library/asyncio-guest.rst | 11 +++++++++++ Lib/asyncio/guest.py | 7 ++++++- 2 files changed, 17 insertions(+), 1 deletion(-) diff --git a/Doc/library/asyncio-guest.rst b/Doc/library/asyncio-guest.rst index e92b8a59325a47..cfffb85149ca06 100644 --- a/Doc/library/asyncio-guest.rst +++ b/Doc/library/asyncio-guest.rst @@ -44,6 +44,17 @@ See :source:`Doc/includes/asyncio_guest_tkinter.py` for a complete Tkinter example that embeds asyncio inside ``tkinter.mainloop()`` using :func:`start_guest_run`. +.. seealso:: + + The `asyncio-guest `__ + project — the proof of concept this feature is based on — has runnable + examples for many more hosts: Tkinter, Qt (PyQt5/PySide6), GTK, + pygame, Win32 and Tornado. + + `Trio's guest mode + `__, + which pioneered this approach. + .. rubric:: API .. function:: start_guest_run(async_fn, *args, run_sync_soon_threadsafe, done_callback) diff --git a/Lib/asyncio/guest.py b/Lib/asyncio/guest.py index 2b2021c04a8832..bcc9724b5f31cc 100644 --- a/Lib/asyncio/guest.py +++ b/Lib/asyncio/guest.py @@ -11,7 +11,12 @@ A single "token" (semaphore permit) ping-pongs between the two threads, so exactly one of them touches the event loop at any moment. -Inspired by Trio's guest mode (trio.lowlevel.start_guest_run). +Inspired by Trio's guest mode (trio.lowlevel.start_guest_run). The +asyncio-guest project, which this implementation grew out of, has +runnable examples for Tkinter, Qt, GTK, pygame, Win32 and Tornado +hosts: + + https://github.com/congzhangzh/asyncio-guest """ __all__ = ('start_guest_run',) From 42c8f71763d8a0435ac3ae0178451ee11ba283c4 Mon Sep 17 00:00:00 2001 From: Cong Zhang <13283869+congzhangzh@users.noreply.github.com> Date: Tue, 13 Oct 2026 10:35:39 -0400 Subject: [PATCH 09/10] gh-145342: Avoid the I/O thread round trip for non-blocking polls When callbacks are ready or a non-blocking select(0) returns events, run the next guest step on the host thread instead of handing the poll to the I/O thread. The I/O thread is now used only for polls that would block, which removes two thread switches per step for busy tasks. --- Doc/library/asyncio-guest.rst | 4 +- Lib/asyncio/base_events.py | 10 ++++- Lib/asyncio/guest.py | 27 ++++++++++-- Lib/test/test_asyncio/test_guest.py | 67 +++++++++++++++++++++++++++++ 4 files changed, 103 insertions(+), 5 deletions(-) diff --git a/Doc/library/asyncio-guest.rst b/Doc/library/asyncio-guest.rst index cfffb85149ca06..da8f4a46118520 100644 --- a/Doc/library/asyncio-guest.rst +++ b/Doc/library/asyncio-guest.rst @@ -26,7 +26,9 @@ replacing the host loop, asyncio piggybacks on it: :meth:`loop.process_events() ` and :meth:`loop.process_ready() ` to advance the asyncio event loop by one step, then signals the I/O thread to - poll again. + poll again. If callbacks are already ready or I/O events are pending, + the host thread skips the I/O thread and schedules the next step on + itself, so busy tasks do not pay for a thread switch on every step. Exactly one of the two threads touches the event loop at any moment, so neither the host loop nor the asyncio loop starves the other. diff --git a/Lib/asyncio/base_events.py b/Lib/asyncio/base_events.py index 1b8a737933dc9a..00f5bc005870de 100644 --- a/Lib/asyncio/base_events.py +++ b/Lib/asyncio/base_events.py @@ -2022,6 +2022,14 @@ def poll_events(self): .. versionadded:: 3.16 """ + return self._selector.select(self._poll_timeout()) + + def _poll_timeout(self): + """Return the timeout for the next selector poll. + + Cleans up cancelled scheduled handles first. Returns 0 if + callbacks are ready, None if nothing is scheduled. + """ sched_count = len(self._scheduled) if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and self._timer_cancelled_count / sched_count > @@ -2056,7 +2064,7 @@ def poll_events(self): elif timeout < 0: timeout = 0 - return self._selector.select(timeout) + return timeout def process_events(self, event_list): """Process I/O events returned by :meth:`poll_events`. diff --git a/Lib/asyncio/guest.py b/Lib/asyncio/guest.py index bcc9724b5f31cc..34ae0b0c19f7d4 100644 --- a/Lib/asyncio/guest.py +++ b/Lib/asyncio/guest.py @@ -9,7 +9,10 @@ I/O thread: wait for token -> poll_events() -> hand events to host A single "token" (semaphore permit) ping-pongs between the two threads, -so exactly one of them touches the event loop at any moment. +so exactly one of them touches the event loop at any moment. The token +only goes to the I/O thread when the poll would block: if callbacks are +ready or I/O events are pending, the host keeps it and schedules the +next batch on itself. Inspired by Trio's guest mode (trio.lowlevel.start_guest_run). The asyncio-guest project, which this implementation grew out of, has @@ -250,7 +253,9 @@ def _process_on_host(event_list): return try: loop.process_events(event_list) + event_list = None loop.process_ready() + timeout = loop._poll_timeout() except BaseException: # Internal loop failure (user callback exceptions are routed # to the exception handler by Handle._run). Unblock the I/O @@ -258,12 +263,28 @@ def _process_on_host(event_list): shutdown.set() wakeup.release() raise + if shutdown.is_set(): + return + # Fast path: poll without blocking on the host thread. If there + # is work to do right away, run the next batch on the host too; + # a round trip through the I/O thread costs two thread switches. + try: + event_list = loop._selector.select(0) + except Exception as exc: + # Report it like an I/O thread failure. Scheduled rather + # than called inline, so that _abort() can join the I/O + # thread even when this is the initial batch. + shutdown.set() + run_sync_soon_threadsafe(partial(_abort, exc)) + return + if timeout == 0 or event_list: + run_sync_soon_threadsafe(partial(_process_on_host, event_list)) + return # Hand the polling token back to the I/O thread. Exactly one # token circulates: while the host runs a batch, the I/O thread # is parked in wakeup.acquire(), so it can never poll (or touch # the loop at all) concurrently with this function. - if not shutdown.is_set(): - wakeup.release() + wakeup.release() def _on_task_done(task): # Runs inside process_ready() while the host holds the token, diff --git a/Lib/test/test_asyncio/test_guest.py b/Lib/test/test_asyncio/test_guest.py index 01bcdbd4567646..5cd341a44ae9ca 100644 --- a/Lib/test/test_asyncio/test_guest.py +++ b/Lib/test/test_asyncio/test_guest.py @@ -8,6 +8,7 @@ import threading import time import unittest +from unittest import mock from test.support import threading_helper from test.support.script_helper import assert_python_ok @@ -217,6 +218,72 @@ def setter(): self.assertEqual(task.result(), "safe") timer.join() + def test_socket_io(self): + async def coro(): + a, b = socket.socketpair() + ra, wa = await asyncio.open_connection(sock=a) + rb, wb = await asyncio.open_connection(sock=b) + try: + for i in range(20): + wa.write(b'%d\n' % i) + line = await rb.readline() + wb.write(line) + self.assertEqual(await ra.readline(), b'%d\n' % i) + finally: + wa.close() + wb.close() + await wa.wait_closed() + await wb.wait_closed() + return "ok" + + task = self._run_guest(coro) + self.assertEqual(task.result(), "ok") + + # -- polling ------------------------------------------------------- + + def test_ready_callbacks_do_not_go_through_io_thread(self): + # While callbacks are ready the poll timeout is zero; such polls + # run on the host thread instead of being handed to the I/O thread. + orig_poll_events = asyncio.BaseEventLoop.poll_events + io_polls = 0 + + def poll_events(loop): + nonlocal io_polls + io_polls += 1 + return orig_poll_events(loop) + + async def coro(): + for _ in range(1000): + await asyncio.sleep(0) + + with mock.patch.object(asyncio.BaseEventLoop, 'poll_events', + poll_events): + task = self._run_guest(coro) + self.assertIsNone(task.result()) + self.assertLess(io_polls, 10) + + def test_busy_guest_does_not_starve_host(self): + # Host-thread polls reschedule themselves through the host loop, + # so host callbacks still run while guest tasks are busy. + host = MockHost() + host_ran = threading.Event() + + async def coro(): + for i in range(100_000): + if host_ran.is_set(): + return i + await asyncio.sleep(0) + return None + + asyncio.start_guest_run( + coro, + run_sync_soon_threadsafe=host.run_sync_soon_threadsafe, + done_callback=host.done_callback, + ) + host.run_sync_soon_threadsafe(host_ran.set) + task = host.run() + self.assertIsNotNone(task.result()) + # -- thread lifecycle ---------------------------------------------- def test_io_thread_nondaemon_and_joined(self): From ff3800b10af58dcc80cd49cfe2638f8724ea4b9f Mon Sep 17 00:00:00 2001 From: Cong Zhang <13283869+congzhangzh@users.noreply.github.com> Date: Tue, 13 Oct 2026 11:18:06 -0400 Subject: [PATCH 10/10] gh-145342: Swap per-thread loop state around guest batches Install the guest loop's running-loop state, async generator hooks and coroutine origin tracking depth only while guest callbacks run, and restore the host's in between. Host code no longer has its async generators adopted by the guest loop, and the host can itself be an asyncio event loop. The first batch is now scheduled through run_sync_soon_threadsafe instead of running inline, since a guest task cannot start while a host task is running. --- Doc/library/asyncio-guest.rst | 24 ++-- Lib/asyncio/guest.py | 128 ++++++++++++++++------ Lib/test/test_asyncio/test_guest.py | 163 ++++++++++++++++++++++++++-- 3 files changed, 266 insertions(+), 49 deletions(-) diff --git a/Doc/library/asyncio-guest.rst b/Doc/library/asyncio-guest.rst index da8f4a46118520..0b3e7c09297d91 100644 --- a/Doc/library/asyncio-guest.rst +++ b/Doc/library/asyncio-guest.rst @@ -39,6 +39,8 @@ Typical use cases: without replacing the native event loop. * Embedding asyncio I/O (HTTP clients, websockets, …) inside a GUI app. * Running asyncio alongside a framework that owns the main thread. +* Running a second, isolated asyncio loop inside an asyncio program, for + example a server under test whose loop is separate from the test's. .. rubric:: Example @@ -96,13 +98,21 @@ example that embeds asyncio inside ``tkinter.mainloop()`` using Lifecycle and Cleanup ===================== -For the whole guest run the guest loop is the host thread's running -loop: :func:`get_running_loop` works inside guest tasks, -:meth:`loop.is_running() ` returns ``True``, and -starting another event loop on that thread — including a nested -:func:`asyncio.run` or :meth:`loop.run_until_complete` — raises -:exc:`RuntimeError`. Consequently a thread that is already running an -asyncio event loop cannot start a guest run. +*async_fn* starts running in the first callback passed to +*run_sync_soon_threadsafe*, after :func:`start_guest_run` has returned. + +The guest loop is the host thread's running loop only while guest +callbacks run. Around each batch of callbacks, the thread's running +loop, asynchronous generator hooks (see :func:`sys.set_asyncgen_hooks`) +and coroutine origin tracking depth are switched to the guest's, and +the host's are restored afterwards. So inside guest tasks +:func:`get_running_loop` returns the guest loop and starting another +event loop — including a nested :func:`asyncio.run` or +:meth:`loop.run_until_complete` — raises :exc:`RuntimeError`, while host +code keeps its own state: asynchronous generators it iterates are not +finalized by the guest loop, and the host may itself be an asyncio event +loop. :meth:`loop.is_running() ` returns +``True`` for the whole guest run. When the main task finishes, cleanup equivalent to :func:`asyncio.run` takes place on the host thread: remaining tasks are cancelled, diff --git a/Lib/asyncio/guest.py b/Lib/asyncio/guest.py index 34ae0b0c19f7d4..3d7652c72fd5e6 100644 --- a/Lib/asyncio/guest.py +++ b/Lib/asyncio/guest.py @@ -24,7 +24,9 @@ __all__ = ('start_guest_run',) +import contextlib import signal +import sys import threading from functools import partial @@ -91,6 +93,37 @@ def _restore_wakeup_fd(fd): signal.set_wakeup_fd(fd) +def _get_thread_state(): + """Return the per-thread state that a running event loop installs.""" + return (events._get_running_loop(), + sys.get_asyncgen_hooks(), + sys.get_coroutine_origin_tracking_depth()) + + +def _set_thread_state(state): + running_loop, asyncgen_hooks, origin_tracking_depth = state + events._set_running_loop(running_loop) + sys.set_asyncgen_hooks(*asyncgen_hooks) + sys.set_coroutine_origin_tracking_depth(origin_tracking_depth) + + +@contextlib.contextmanager +def _host_loop_hidden(): + """Hide the host's running loop, if any, and restore the thread state. + + The guest loop must not start, or be driven to completion at + shutdown, while another loop is the thread's running loop. An + asyncio host is suspended in one of its callbacks at that point, so + the two loops never actually run nested. + """ + state = _get_thread_state() + events._set_running_loop(None) + try: + yield + finally: + _set_thread_state(state) + + def start_guest_run(async_fn, *args, run_sync_soon_threadsafe, done_callback): @@ -114,9 +147,14 @@ def start_guest_run(async_fn, *args, asyncio.run()). It receives the completed Task as its sole argument. - For the whole run the guest loop is the thread's running loop: - get_running_loop() works inside guest tasks, loop.is_running() is - true, and starting another loop on the thread raises RuntimeError. + The guest loop is the thread's running loop, with its async + generator hooks installed, only while guest callbacks run; the + host's are restored in between. So host code keeps its own asyncio + state, and the host may itself be an asyncio event loop. + loop.is_running() is true for the whole run. async_fn starts + running in the first callback passed to run_sync_soon_threadsafe, + after start_guest_run() has returned. + Signal handling stays with the host: the guest loop never touches signal.set_wakeup_fd(), and loop.add_signal_handler() raises RuntimeError. The host may forward signals into the loop with @@ -150,35 +188,54 @@ def _close_loop(): finally: _restore_wakeup_fd(fd) - # Mark the loop as running for the whole guest run. On Windows this - # also starts the proactor's self-reading loop, so that - # call_soon_threadsafe() can wake the I/O thread's poll. - try: - loop._run_forever_setup() - except BaseException: - _close_loop() - raise - shutdown = threading.Event() wakeup = threading.Semaphore(0) finished = False cleaned_up = False - try: - main_task = loop.create_task(async_fn(*args)) - except BaseException: - loop._run_forever_cleanup() - _close_loop() - raise + # Mark the loop as running for the whole guest run. On Windows this + # also starts the proactor's self-reading loop, so that + # call_soon_threadsafe() can wake the I/O thread's poll. The + # per-thread part of the setup is captured as guest_state and + # swapped in only around guest callbacks (see _enter_guest()). + with _host_loop_hidden(): + try: + loop._run_forever_setup() + except BaseException: + _close_loop() + raise + try: + main_task = loop.create_task(async_fn(*args)) + except BaseException: + loop._run_forever_cleanup() + _close_loop() + raise + guest_state = _get_thread_state() # -- helpers (host thread unless noted) ---------------------------- + def _enter_guest(): + # Install the guest's per-thread state; return the host's. + host_state = _get_thread_state() + _set_thread_state(guest_state) + return host_state + + def _leave_guest(host_state): + # Keep changes made by guest callbacks, e.g. by loop.set_debug(). + nonlocal guest_state + guest_state = _get_thread_state() + _set_thread_state(host_state) + def _cleanup_running_state(): nonlocal cleaned_up if cleaned_up: return cleaned_up = True - loop._run_forever_cleanup() + host_state = _enter_guest() + try: + loop._run_forever_cleanup() + finally: + _set_thread_state(host_state) def _join_backend(): # By the token invariant the I/O thread is parked in @@ -199,11 +256,12 @@ def _deliver(task, *, graceful): try: if graceful: # Same cleanup as asyncio.run(). - runners._cancel_all_tasks(loop) - loop.run_until_complete(loop.shutdown_asyncgens()) - loop.run_until_complete( - loop.shutdown_default_executor( - constants.THREAD_JOIN_TIMEOUT)) + with _host_loop_hidden(): + runners._cancel_all_tasks(loop) + loop.run_until_complete(loop.shutdown_asyncgens()) + loop.run_until_complete( + loop.shutdown_default_executor( + constants.THREAD_JOIN_TIMEOUT)) finally: _close_loop() done_callback(task) @@ -234,8 +292,9 @@ def _abort(exc): main_task.cancel( msg=f'asyncio guest I/O thread failed: {exc!r}') try: - loop.run_until_complete( - tasks.gather(main_task, return_exceptions=True)) + with _host_loop_hidden(): + loop.run_until_complete( + tasks.gather(main_task, return_exceptions=True)) except Exception: # The loop itself is broken (e.g. the selector raised); # abandon any pending work. @@ -251,6 +310,7 @@ def _process_on_host(event_list): """Process one batch of asyncio work on the host thread.""" if shutdown.is_set() or loop.is_closed(): return + host_state = _enter_guest() try: loop.process_events(event_list) event_list = None @@ -263,6 +323,8 @@ def _process_on_host(event_list): shutdown.set() wakeup.release() raise + finally: + _leave_guest(host_state) if shutdown.is_set(): return # Fast path: poll without blocking on the host thread. If there @@ -353,22 +415,22 @@ def _stop_at_exit(): if not shutting_down: _stoppers.add(_stop_at_exit) if shutting_down: - loop._run_forever_cleanup() + _cleanup_running_state() _close_loop() raise RuntimeError( 'cannot start an asyncio guest run at interpreter shutdown') try: - # Process the callbacks enqueued by create_task(), then let the - # I/O thread take over polling. The thread is started even if - # the task already finished: it consumes the shutdown token and - # exits, keeping the join logic uniform. - _process_on_host([]) + # The I/O thread waits for the token, which the host holds until + # the first batch has run. That batch is scheduled rather than + # run inline: with an asyncio host the caller is usually a host + # task, and a guest task cannot start while another task runs. backend_thread.start() + run_sync_soon_threadsafe(partial(_process_on_host, [])) except BaseException: with _lock: _stoppers.discard(_stop_at_exit) - shutdown.set() + _join_backend() _cleanup_running_state() _close_loop() raise diff --git a/Lib/test/test_asyncio/test_guest.py b/Lib/test/test_asyncio/test_guest.py index 5cd341a44ae9ca..6f606f02f49ead 100644 --- a/Lib/test/test_asyncio/test_guest.py +++ b/Lib/test/test_asyncio/test_guest.py @@ -107,7 +107,7 @@ async def add(a, b): self.assertEqual(task.result(), 10) def test_early_sync_completion(self): - # The task can already be done when the I/O thread starts. + # The task can finish in the very first batch. async def coro(): return 'early' @@ -135,26 +135,42 @@ async def coro(): # -- cancellation -------------------------------------------------- def test_cancel_from_host(self): - started = threading.Event() + host = MockHost() + + def cancel_from_host(): + # Host code, e.g. a button handler. Use call_soon_threadsafe + # to wake the I/O thread's selector. + task.get_loop().call_soon_threadsafe(task.cancel) async def coro(): - started.set() + host.run_sync_soon_threadsafe(cancel_from_host) await asyncio.sleep(3600) - host = MockHost() task = asyncio.start_guest_run( coro, run_sync_soon_threadsafe=host.run_sync_soon_threadsafe, done_callback=host.done_callback, ) - # Wait for the coroutine to start, then cancel. - # Use call_soon_threadsafe to wake the I/O thread's selector. - started.wait(timeout=5) - loop = task.get_loop() - loop.call_soon_threadsafe(task.cancel) host.run(timeout=5) self.assertTrue(task.cancelled()) + def test_starts_after_return(self): + started = False + + async def coro(): + nonlocal started + started = True + + host = MockHost() + asyncio.start_guest_run( + coro, + run_sync_soon_threadsafe=host.run_sync_soon_threadsafe, + done_callback=host.done_callback, + ) + self.assertFalse(started) + host.run() + self.assertTrue(started) + # -- asyncio primitives work inside guest -------------------------- def test_sleep(self): @@ -361,6 +377,135 @@ async def coro(): self.assertIsNone(asyncio._get_running_loop()) self.assertFalse(task.get_loop().is_running()) + # -- per-thread state isolation ------------------------------------ + + def test_host_state_between_batches(self): + # Host code sees the host's running loop and asyncgen hooks, not + # the guest's, between guest batches. + def firstiter(agen): + pass + + def finalizer(agen): + pass + + host = MockHost() + seen = {} + + def host_code(): + seen['running_loop'] = asyncio._get_running_loop() + seen['hooks'] = sys.get_asyncgen_hooks() + + async def coro(): + host.run_sync_soon_threadsafe(host_code) + await asyncio.sleep(0.01) + return sys.get_asyncgen_hooks().firstiter + + old_hooks = sys.get_asyncgen_hooks() + sys.set_asyncgen_hooks(firstiter=firstiter, finalizer=finalizer) + try: + task = asyncio.start_guest_run( + coro, + run_sync_soon_threadsafe=host.run_sync_soon_threadsafe, + done_callback=host.done_callback, + ) + host.run() + self.assertEqual(sys.get_asyncgen_hooks(), + (firstiter, finalizer)) + finally: + sys.set_asyncgen_hooks(*old_hooks) + self.assertIsNone(seen['running_loop']) + self.assertEqual(seen['hooks'], (firstiter, finalizer)) + self.assertEqual(task.result(), + task.get_loop()._asyncgen_firstiter_hook) + + def test_set_debug_during_run(self): + # loop.set_debug() changes the guest's coroutine origin tracking + # for later batches, but not the host's. + host = MockHost() + host_depths = [] + + def host_code(): + host_depths.append(sys.get_coroutine_origin_tracking_depth()) + + async def coro(): + loop = asyncio.get_running_loop() + loop.set_debug(True) + await asyncio.sleep(0) + host.run_sync_soon_threadsafe(host_code) + await asyncio.sleep(0.01) + depth = sys.get_coroutine_origin_tracking_depth() + loop.set_debug(False) + await asyncio.sleep(0) + return depth + + old_depth = sys.get_coroutine_origin_tracking_depth() + asyncio.start_guest_run( + coro, + run_sync_soon_threadsafe=host.run_sync_soon_threadsafe, + done_callback=host.done_callback, + ) + task = host.run() + self.assertGreater(task.result(), 0) + self.assertEqual(host_depths, [old_depth]) + self.assertEqual(sys.get_coroutine_origin_tracking_depth(), old_depth) + + def test_host_asyncgen_not_adopted(self): + # An async generator first iterated by host code belongs to the + # host: the guest loop must not finalize it at shutdown. + async def agen(): + yield 1 + yield 2 + + host = MockHost() + host_agen = agen() + + def host_code(): + with self.assertRaises(StopIteration): + host_agen.asend(None).send(None) + + async def coro(): + host.run_sync_soon_threadsafe(host_code) + await asyncio.sleep(0.01) + return host_agen in asyncio.get_running_loop()._asyncgens + + asyncio.start_guest_run( + coro, + run_sync_soon_threadsafe=host.run_sync_soon_threadsafe, + done_callback=host.done_callback, + ) + task = host.run() + self.assertFalse(task.result()) + self.assertIsNotNone(host_agen.ag_frame) # not closed by the guest + host_agen.aclose().close() + + def test_asyncio_host(self): + # The host may itself be an asyncio event loop. + async def guest(): + loop = asyncio.get_running_loop() + await asyncio.sleep(0.01) + return loop, asyncio.current_task() + + async def main(): + host_loop = asyncio.get_running_loop() + done = host_loop.create_future() + task = asyncio.start_guest_run( + guest, + run_sync_soon_threadsafe=host_loop.call_soon_threadsafe, + done_callback=done.set_result, + ) + self.assertIs(asyncio.get_running_loop(), host_loop) + await asyncio.sleep(0.01) + self.assertIs(await done, task) + self.assertIs(asyncio.get_running_loop(), host_loop) + return host_loop, task + + host_loop, task = asyncio.run(main()) + guest_loop, current = task.result() + self.assertIsNot(guest_loop, host_loop) + self.assertIs(guest_loop, task.get_loop()) + self.assertIs(current, task) + self.assertTrue(guest_loop.is_closed()) + # -- signal handling ----------------------------------------------- @unittest.skipUnless(hasattr(signal, 'SIGUSR1'),