CPython asyncio 实现深度解析Python 3.14 任务管理重构与异步生成器终结机制【免费下载链接】cpythonThe Python programming language项目地址: https://gitcode.com/GitHub_Trending/cp/cpython本文以 CPython 源码仓库中的实现说明文档 InternalDocs/asyncio.md 为骨架系统梳理 CPython 3.14 在asyncio底层做的两项核心改造其一把任务管理任务登记、当前任务查询从“基于事件循环的全局WeakSet 全局字典”重构为“基于线程状态的内建双向链表与字段”显著改善性能、线程安全与 free-threading 下的扩展性其二利用 PEP 525 的异步生成器 hook 机制保证未被完整迭代的异步生成器也能在事件循环中安全执行finally块。读完本文你将掌握这两部分设计背后的数据结构、关键代码路径、锁与 stop-the-world 策略的取舍并能在仓库源码中快速定位对应实现。本文全部实现细节均对应仓库真实代码文中所涉相对路径均以仓库根目录为基准两大部分分别为 C 实现_asyncio与 Python 实现Lib/asyncio。第一部分任务管理C 实现Pre-3.14全局 WeakSet 与全局字典的旧设计及其痛点在 Python 3.14 之前asyncio的 C 实现用两种容器管理任务生命周期一个全局字典current_tasks键是事件循环对象、值是该循环当前正在执行的任务用于回答“当前任务是什么”一个WeakSetscheduled_tasks存放所有被调度到事件循环上运行的任务。选用弱引用集合是为了让事件循环不持有任务的强引用从而在任务不再被引用时能被垃圾回收。/* Dictionary containing tasks that are currently active in all running event loops. {EventLoop: Task} */ PyObject *current_tasks; /* WeakSet containing all tasks scheduled to run on event loops. */ PyObject *scheduled_tasks;纯 Python 的降级实现至今仍保留在 Lib/asyncio/tasks.py 中作为无法导入_asyncio时的回退路径可作对照_scheduled_tasks weakref.WeakSet() _eager_tasks set() # Dictionary containing tasks that are currently active in # all running event loops. {EventLoop: Task} _current_tasks {}原文档指出该设计存在三类缺陷性能PerformanceWeakSet需要维护一套完整弱引用以及相应的弱引用回调在任务被回收时执行清理。这让 GC 负担加重在任务数量很大的应用中会形成瓶颈内存占用上升、性能下降。而“查当前任务”需要在该字典上做一次字典查找也偏慢。线程安全Thread safety3.14 之前对WeakSet的并发迭代并不安全多线程下调用asyncio.all_tasks()可能得到不一致结果甚至抛出RuntimeError。相关回溯见 gh-123089 与 gh-80788。free-threading 下扩展性差Poor scaling全局共享的WeakSet横跨所有线程。任务入集合、出集合属于高频操作各线程会对同一容器争用同样的多个线程访问“当前任务”也会因争用全局current_tasks字典而无法随线程数扩展。3.14 新设计概览按线程per-thread存储针对上述问题Python 3.14 引入了两项核心变更按线程维护任务的循环双向链表每个线程维护自己的任务链表任务的加入与移除无需加锁既高效又线程安全在 free-threading 下能随线程数良好扩展同时它允许外部内省工具如python -m asyncio pstree检视运行在所有线程中的任务。该能力随“Audit asyncio thread safety”工作gh-128002落地。按线程保存当前任务当前任务不再存于全局字典而是直接存进当前线程状态PyThreadState省去字典查找每个线程各自维护自己的当前任务相关工作见 gh-129898。选择per-thread 而非 per-loop存储原因在文档与代码中都很明确外部内省工具如pstree无法在事件循环对象上访问任意属性因此不能把任务挂在 loop 上供外部读取而线程状态是运行时内建结构可由 C API 稳定访问per-thread 存储天然支持第三方事件循环实现如 uvloop不依赖 loop 对象的内部属性对最常见单线程的 asyncio 使用场景它避免了在性能关键的“任务加入/移除链表”路径上做 loop 属性查找等一系列额外调用更高效。数据结构链表节点嵌入任务对象与线程/解释器状态新方案的核心是通用llist_node结构定义见 Include/internal/pycore_llist.h提供llist_insert_tail、llist_remove、llist_for_each_safe、llist_concat等原语。链表服务于所有asyncio.Task及其子类实例第三方自定义任务类型仍回退到WeakSet实现。链表节点被直接嵌入到任务对象内部避免了为链表节点再做一次内存分配typedef struct TaskObj { ... struct llist_node asyncio_node; // 文档中亦写作 task_node / asyncio_node } TaskObj;实际的TaskObj定义位于 Modules/_asynciomodule.c。PyThreadState准确说是内部扩展结构_PyThreadStateImpl新增字段asyncio_running_loop、asyncio_running_task两个强引用指针以及一个循环链表头asyncio_tasks_head见 Include/internal/pycore_tstate.htypedef struct _PyThreadStateImpl { ... PyObject *asyncio_running_loop; // Strong reference PyObject *asyncio_running_task; // Strong reference ... /* Head of circular linked-list of all tasks which are instances of asyncio.Task or subclasses of it used in asyncio.all_tasks. */ struct llist_node asyncio_tasks_head; ... } _PyThreadStateImpl;说明内部文档中给出的结构体示意将字段写作asyncio_current_loop/asyncio_current_task实际仓库源码中这两个字段的正式命名为asyncio_running_loop/asyncio_running_task语义一致“当前正在运行的事件循环/任务”。后续叙述均以源码实际命名为准。PyInterpreterState也新增字段用于承接线程状态释放后残留的任务可能发生其他线程仍持有本线程任务的引用见 Include/internal/pycore_interp_structs.h// Per-interpreter list of tasks, any lingering tasks from thread // states gets added here and removed from the corresponding // thread states list. struct llist_node asyncio_tasks_head; // asyncio_tasks_lock is used when tasks are moved // from threads list to interpreters list. PyMutex asyncio_tasks_lock;两者合并起来即为文档所给的整体结构示意typedef struct TaskObj { ... struct llist_node asyncio_node; } TaskObj; typedef struct PyThreadState { ... struct llist_node asyncio_tasks_head; } PyThreadState; typedef struct PyInterpreterState { ... struct llist_node asyncio_tasks_head; PyMutex asyncio_tasks_lock; } PyInterpreterState;asyncio_tasks_lock只用于保护解释器级列表免受并发修改例如线程销毁时把残留任务并入解释器列表的操作。任务的登记与注销register_task / unregister_task任务创建后通过register_task加入当前线程链表任务结束done/cancelled后由unregister_task移出链表。两者实现于 Modules/_asynciomodule.cstatic void register_task(_PyThreadStateImpl *ts, TaskObj *task) { if (task-task_node.next ! NULL) { // already registered assert(task-task_node.prev ! NULL); return; } struct llist_node *head ts-asyncio_tasks_head; llist_insert_tail(head, task-task_node); } static inline void unregister_task_safe(TaskObj *task) { if (task-task_node.next NULL) { // not registered assert(task-task_node.prev NULL); return; } llist_remove(task-task_node); } static void unregister_task(TaskObj *task) { #ifdef Py_GIL_DISABLED // check if we are in the same thread // if so, we can avoid locking if (task-task_tid _Py_ThreadId()) { unregister_task_safe(task); } else { // we are in a different thread // stop the world then check and remove the task PyThreadState *tstate _PyThreadState_GET(); _PyEval_StopTheWorld(tstate-interp); unregister_task_safe(task); _PyEval_StartTheWorld(tstate-interp); } #else unregister_task_safe(task); #endif }值得注意的实现细节register_task/unregister_task_safe首先检查task_node.next是否非空以判断是否已在链表中避免重复登记/移除链表本身存的是借用引用borrowed reference且加入/移出都是单纯的指针操作因此单线程内完全无锁lock-free在free-threadingPy_GIL_DISABLED构建下创建任务的线程 id 会存入TaskObj的task_tid字段。注销时先比对task-task_tid _Py_ThreadId()若注销发生在创建它的同一线程直接无锁移除即可否则跨线程注销需要_PyEval_StopTheWorld暂停解释器内所有线程待安全移除后再_PyEval_StartTheWorld恢复。这保证了链表不被并发撕裂。调用点包括任务创建L2373 附近、任务生命周期终结L2966 附近以及任务开始/结束执行时的进出场逻辑L3419-L3454。线程状态销毁时残留任务迁移到解释器级列表当线程状态被销毁时其任务链表可能仍有“残留任务”——比如另一个线程还持有该线程任务的引用导致这些任务尚未完成。因此 Python/pystate.c 中的PyThreadState_Clear会先清掉当前循环/任务强引用再在asyncio_tasks_lock保护下用llist_concat把线程链表整体并入解释器级任务链表Py_CLEAR(((_PyThreadStateImpl *)tstate)-asyncio_running_loop); Py_CLEAR(((_PyThreadStateImpl *)tstate)-asyncio_running_task); PyMutex_Lock(tstate-interp-asyncio_tasks_lock); // merge any lingering tasks from thread state to interpreters // tasks list llist_concat(tstate-interp-asyncio_tasks_head, ((_PyThreadStateImpl *)tstate)-asyncio_tasks_head); PyMutex_Unlock(tstate-interp-asyncio_tasks_lock);线程状态释放后解释器级列表仍继续持有这些任务保证它们不会被“丢”并且仍可被all_tasks()枚举。两个列表的初始化分别在interpreter与tstate创建路径中完成Python/pystate.c 与 L1640 附近且asyncio_tasks_lock初始化为零值PyMutex。all_tasks() 的一致性遍历与 stop-the-worldasyncio.all_tasks()现在遍历所有线程的 per-thread 任务链表 解释器级任务链表来收集全部任务。在 free-threading 下为保证遍历期间没有线程正在增删任务会先stop-the-world暂停所有线程从而获得跨线程一致且线程安全的快照视图。相关实现在 Modules/_asynciomodule.cadd_tasks_llist(head, tasks)遍历某个链表头因为链表持有借用引用为防止任务在遍历时被其他线程并发释放先用_Py_TryIncref尝试提升引用计数若对象正被并发释放则失败跳过成功后追加进结果列表add_tasks_interp(interp, tasks)先遍历解释器级链表free-threading 下断言interp-stoptheworld.world_stopped已成立再通过_Py_FOR_EACH_TSTATE_BEGIN/END遍历所有线程状态的任务链表。all_tasks的入口在 Modules/_asynciomodule.c 附近其逻辑为先加入 eager 任务再执行上述两条链表的遍历。当前任务的进出场与快/慢路径查询任务开始执行、暂停、结束时分别通过enter_task/leave_task更新线程状态上的当前任务字段定义于 Modules/_asynciomodule.cstatic int enter_task(_PyThreadStateImpl *ts, PyObject *loop, PyObject *task) { if (ts-asyncio_running_loop ! loop) { PyErr_Format(PyExc_RuntimeError, loop %R is not the running loop, loop); return -1; } if (ts-asyncio_running_task ! NULL) { PyErr_Format(PyExc_RuntimeError, Cannot enter into task %R while another task %R is being executed., task, ts-asyncio_running_task); return -1; } ts-asyncio_running_task Py_NewRef(task); return 0; } static int leave_task(_PyThreadStateImpl *ts, PyObject *loop, PyObject *task) { if (ts-asyncio_running_loop ! loop) { PyErr_Format(PyExc_RuntimeError, loop %R is not the running loop, loop); return -1; } if (ts-asyncio_running_task ! task) { PyErr_Format(PyExc_RuntimeError, Invalid attempt to leave task %R while task %R is entered., task, ts-asyncio_running_task ? ts-asyncio_running_task : Py_None); return -1; } Py_CLEAR(ts-asyncio_running_task); return 0; }enter_task/leave_task都做了严格校验事件循环必须是本线程“正在运行的循环”且当前任务状态一致进入前不得已有正在执行的任务离开时任务必须匹配否则抛出RuntimeError——这与纯 Python 回退实现 Lib/asyncio/tasks.py 的语义保持一致。另有swap_current_taskModules/_asynciomodule.c支持“换出旧任务、换入新任务”并采取转移所有权的方式减少冗余引用计数。任务在事件循环中被调度执行时会在task_step附近调用enter_task/leave_taskL3419-L3434 附近保证整条协程执行期间asyncio.current_task()能取到正确对象。这些函数同时也以_asyncio._register_task、_asyncio._unregister_task、_asyncio._enter_task、_asyncio._leave_task、_asyncio._swap_current_task的形式对 Python 层暴露Modules/_asynciomodule.c。查询当前任务current_task(loop)则分为快慢两条路径Modules/_asynciomodule.c快路径一般情况若loop就是当前线程正在运行的事件循环ts-asyncio_running_loop loop则直接返回ts-asyncio_running_task不存在则返回None完全无需加锁慢路径free-threading、跨循环查询若目标loop不是当前线程的运行中循环则需要stop-the-world暂停解释器内所有线程遍历各线程状态、比对asyncio_running_loop loop找到匹配线程后返回其asyncio_running_task没有任何匹配线程状态时返回None。_PyThreadStateImpl *ts (_PyThreadStateImpl *)_PyThreadState_GET(); // Fast path for the current running loop of current thread // no locking or stop the world pause is required if (ts-asyncio_running_loop loop) { if (ts-asyncio_running_task ! NULL) { Py_DECREF(loop); return Py_NewRef(ts-asyncio_running_task); } Py_DECREF(loop); Py_RETURN_NONE; } // ... otherwise: _PyEval_StopTheWorld(interp), iterate all tstates ...这样的设计保证在 free-threading 下各线程访问“自己运行循环的当前任务”互不争用全局字典从根上消除了旧方案在current_tasks全局字典上的竞争。端到端流程一图流综合上述代码路径可得到如下完整生命周期图与文档流程图一致节点对应实际 C 函数名流程要点任务创建即被register_task登记进当前线程链表处于 pending 的任务反复被task_step推进一旦 done/cancelled 即触发unregister_task非 free-threading 构建直接unregister_task_safe无锁移除free-threading 构建则先判断是否同线程同线程无锁跨线程需 stop-the-world 后移除线程销毁时若任务链表非空则把剩余任务整体并入解释器级任务链表后再释放线程状态。整体设计实现了无锁执行在多个事件循环运行于不同线程的 free-threading 场景下扩展良好。内省工具python -m asyncio pstreeper-thread 存储的动机之一是支持跨线程任务检视。仓库中的 Lib/asyncio/main.py 提供了asyncio模块的 CLI 入口内部import asyncio.tools提供工具实现其中的任务树检视即通过python -m asyncio pstree触发可以借此观察多个线程/事件循环中登记的全部任务状态。Python 层如何选择 C/Python 两套实现纯 Python 版任务管理函数_register_task、_unregister_task、_enter_task、_leave_task、_swap_current_task、current_task、all_tasks等定义在 Lib/asyncio/tasks.py。随后模块尾部会尝试导入_asyncio扩展一旦成功即用 C 实现覆盖这些名字Lib/asyncio/tasks.pytry: from _asyncio import (_register_task, _register_eager_task, _unregister_task, _unregister_eager_task, _enter_task, _leave_task, _swap_current_task, ...) ... _c_current_task current_task _c_register_task _register_task ... except ImportError: ...这意味着标准 CPython 分发默认走上述 C 快速实现只有在_asyncio不可用的受限环境才退化到全局字典 WeakSet的 Python 回退实现即 3.14 前的旧语义。若在回退实现中重编译为 free-threading全局字典/WeakSet 旧缺陷依旧存在这反衬出新方案价值主要体现在默认的 C 路径上。第二部分异步生成器的终结Python 实现asyncio的异步生成器async generator终结逻辑主要在纯 Python 层实现Lib/asyncio。它要解决一个根本矛盾异步生成器必须由协程驱动因此其终结执行finally块也必须发生在事件循环运行期间。问题未被完整迭代的异步生成器可能不执行 finally大多数异步生成器在“被完整迭代直到耗尽”后会被自动关闭但若它在耗尽前就被放弃例如async for中途break且未手动await agen.aclose()则不会被正确关闭finally块可能永远不执行。文档给出了如下示例import asyncio async def agen(): try: yield 1 finally: await asyncio.sleep(1) print(finally executed) async def main(): async for i in agen(): break loop asyncio.EventLoop() loop.run_until_complete(main())该代码不会打印finally executed——因为异步生成器agen未被完整迭代也没有被手动await agen.aclose()关闭。注意示例为示意写法实际事件循环通过asyncio.run()/loop.run_until_complete()获取。解决方案PEP 525 的 asyncgen hooksasyncio依据 PEP 525 定义的sys.set_asyncgen_hooks设置两类终结钩子Hook触发时机asyncio 侧行为firstiter hook异步生成器第一次被迭代时将其加入loop._asyncgens一个weakref.WeakSet事件循环由此跟踪所有活跃异步生成器finalizer hook异步生成器**即将被终结对象不再被引用**时从loop._asyncgens移除它并通过self.create_task(agen.aclose())调度一个任务去关闭它保证finally块在事件循环运行期间执行由于_asyncgens是弱引用集合事件循环不会阻止异步生成器被回收同时借助 hooks回收前的“最后机会”被用于把aclose()重新调度到事件循环中。源码对照BaseEventLoop 中的 hooks 与关闭流程两个 hook 及_asyncgens集合定义于 Lib/asyncio/base_events.py# A weak set of all asynchronous generators that are # being iterated by the loop. self._asyncgens weakref.WeakSet() # Set to True when loop.shutdown_asyncgens is called. self._asyncgens_shutdown_called Falsehook 实现位于 Lib/asyncio/base_events.pydef _asyncgen_finalizer_hook(self, agen): self._asyncgens.discard(agen) if not self.is_closed(): self.create_task(agen.aclose()) def _asyncgen_firstiter_hook(self, agen): if self._asyncgens_shutdown_called: raise RuntimeError(...) self._asyncgens.add(agen)在BaseEventLoop初始化/运行路径中事件循环会把这两个方法通过sys.set_asyncgen_hooks(firstiter..., finalizer...)注册Lib/asyncio/base_events.py并支持保存/恢复旧 hooks避免覆盖外层配置。完整工作流结合上述实现文档给出如下流程节点与base_events.py中方法一一对应完整地看保证finally执行有两条互补路径对象回收路径异步生成器因失去引用被终结时运行时_PyGen_Finalize触发 finalizer hook事件循环创建aclose()任务使finally在事件循环中被执行循环关闭路径事件循环关闭shutdown_asyncgens时检查是否仍有活跃异步生成器若有则逐个await agen.aclose()用asyncio.gather并行调度并等待其完成之后才真正loop.close()。文档给出的可运行示例使用asyncio.run最终会打印executing finally blockimport asyncio async def agen(): try: yield 1 yield 2 finally: print(executing finally block) async def main(): async for item in agen(): print(item) break # not fully iterated asyncio.run(main())循环关闭时的兜底shutdown_asyncgensshutdown_asyncgens是上面的“循环关闭路径”的公开入口位于 Lib/asyncio/base_events.py。其行为要点置self._asyncgens_shutdown_called True此后新创建的异步生成器若还想注册 firstiter hook 会得到错误提示提示需要显式调用loop.shutdown_asyncgens()而非在循环内部若当前没有活跃异步生成器直接返回否则把集合转为列表、清空_asyncgens再逐个await ag.aclose()并gather等待全部完成。asyncio.run()的配套实现Lib/asyncio/runners.py会在主任务结束后、关闭循环前调用shutdown_asyncgens确保即使在break等“未完整迭代”场景下异步生成器的finally块也不会被吞掉。小结CPython 3.14 的这两块改动分别回答了 asyncio 的两个经典工程问题任务管理C 层用“嵌入任务对象的链表节点 线程状态链表头 解释器级残留列表”替代全局WeakSet/字典用task_tid比对 stop-the-world 处理跨线程注销换取无锁高频路径与 free-threading 扩展性asyncio.all_tasks()/asyncio.current_task()因而既快又线程安全还支撑起python -m asyncio pstree这类跨线程内省工具。异步生成器终结Python 层通过 PEP 525 的 firstiter/finalizer hooks 把“生成器对象回收”与“事件循环仍可驱动其finally”两者桥接起来配合shutdown_asyncgens在循环关闭时兜底最终保证用户写在try/finally中的清理逻辑不会因提前break而丢失。本文所有论断均可回溯到 InternalDocs/asyncio.md 及以下源码_asynciomodule.c、pycore_tstate.h、pycore_interp_structs.h、pycore_llist.h、pystate.c、base_events.py、tasks.py、runners.py 与main.py。若想进一步做实验可基于本仓库构建 CPython 3.14并在 free-threadingPy_GIL_DISABLED构建下用多线程多事件循环压测任务创建/销毁观察all_tasks()与current_task()的无锁行为差异。【免费下载链接】cpythonThe Python programming language项目地址: https://gitcode.com/GitHub_Trending/cp/cpython创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 CPython asyncio 实现深度解析：Python 3.14 任务管理重构与异步生成器终结机制
CPython asyncio 实现深度解析Python 3.14 任务管理重构与异步生成器终结机制【免费下载链接】cpythonThe Python programming language项目地址: https://gitcode.com/GitHub_Trending/cp/cpython本文以 CPython 源码仓库中的实现说明文档 InternalDocs/asyncio.md 为骨架系统梳理 CPython 3.14 在asyncio底层做的两项核心改造其一把任务管理任务登记、当前任务查询从“基于事件循环的全局WeakSet 全局字典”重构为“基于线程状态的内建双向链表与字段”显著改善性能、线程安全与 free-threading 下的扩展性其二利用 PEP 525 的异步生成器 hook 机制保证未被完整迭代的异步生成器也能在事件循环中安全执行finally块。读完本文你将掌握这两部分设计背后的数据结构、关键代码路径、锁与 stop-the-world 策略的取舍并能在仓库源码中快速定位对应实现。本文全部实现细节均对应仓库真实代码文中所涉相对路径均以仓库根目录为基准两大部分分别为 C 实现_asyncio与 Python 实现Lib/asyncio。第一部分任务管理C 实现Pre-3.14全局 WeakSet 与全局字典的旧设计及其痛点在 Python 3.14 之前asyncio的 C 实现用两种容器管理任务生命周期一个全局字典current_tasks键是事件循环对象、值是该循环当前正在执行的任务用于回答“当前任务是什么”一个WeakSetscheduled_tasks存放所有被调度到事件循环上运行的任务。选用弱引用集合是为了让事件循环不持有任务的强引用从而在任务不再被引用时能被垃圾回收。/* Dictionary containing tasks that are currently active in all running event loops. {EventLoop: Task} */ PyObject *current_tasks; /* WeakSet containing all tasks scheduled to run on event loops. */ PyObject *scheduled_tasks;纯 Python 的降级实现至今仍保留在 Lib/asyncio/tasks.py 中作为无法导入_asyncio时的回退路径可作对照_scheduled_tasks weakref.WeakSet() _eager_tasks set() # Dictionary containing tasks that are currently active in # all running event loops. {EventLoop: Task} _current_tasks {}原文档指出该设计存在三类缺陷性能PerformanceWeakSet需要维护一套完整弱引用以及相应的弱引用回调在任务被回收时执行清理。这让 GC 负担加重在任务数量很大的应用中会形成瓶颈内存占用上升、性能下降。而“查当前任务”需要在该字典上做一次字典查找也偏慢。线程安全Thread safety3.14 之前对WeakSet的并发迭代并不安全多线程下调用asyncio.all_tasks()可能得到不一致结果甚至抛出RuntimeError。相关回溯见 gh-123089 与 gh-80788。free-threading 下扩展性差Poor scaling全局共享的WeakSet横跨所有线程。任务入集合、出集合属于高频操作各线程会对同一容器争用同样的多个线程访问“当前任务”也会因争用全局current_tasks字典而无法随线程数扩展。3.14 新设计概览按线程per-thread存储针对上述问题Python 3.14 引入了两项核心变更按线程维护任务的循环双向链表每个线程维护自己的任务链表任务的加入与移除无需加锁既高效又线程安全在 free-threading 下能随线程数良好扩展同时它允许外部内省工具如python -m asyncio pstree检视运行在所有线程中的任务。该能力随“Audit asyncio thread safety”工作gh-128002落地。按线程保存当前任务当前任务不再存于全局字典而是直接存进当前线程状态PyThreadState省去字典查找每个线程各自维护自己的当前任务相关工作见 gh-129898。选择per-thread 而非 per-loop存储原因在文档与代码中都很明确外部内省工具如pstree无法在事件循环对象上访问任意属性因此不能把任务挂在 loop 上供外部读取而线程状态是运行时内建结构可由 C API 稳定访问per-thread 存储天然支持第三方事件循环实现如 uvloop不依赖 loop 对象的内部属性对最常见单线程的 asyncio 使用场景它避免了在性能关键的“任务加入/移除链表”路径上做 loop 属性查找等一系列额外调用更高效。数据结构链表节点嵌入任务对象与线程/解释器状态新方案的核心是通用llist_node结构定义见 Include/internal/pycore_llist.h提供llist_insert_tail、llist_remove、llist_for_each_safe、llist_concat等原语。链表服务于所有asyncio.Task及其子类实例第三方自定义任务类型仍回退到WeakSet实现。链表节点被直接嵌入到任务对象内部避免了为链表节点再做一次内存分配typedef struct TaskObj { ... struct llist_node asyncio_node; // 文档中亦写作 task_node / asyncio_node } TaskObj;实际的TaskObj定义位于 Modules/_asynciomodule.c。PyThreadState准确说是内部扩展结构_PyThreadStateImpl新增字段asyncio_running_loop、asyncio_running_task两个强引用指针以及一个循环链表头asyncio_tasks_head见 Include/internal/pycore_tstate.htypedef struct _PyThreadStateImpl { ... PyObject *asyncio_running_loop; // Strong reference PyObject *asyncio_running_task; // Strong reference ... /* Head of circular linked-list of all tasks which are instances of asyncio.Task or subclasses of it used in asyncio.all_tasks. */ struct llist_node asyncio_tasks_head; ... } _PyThreadStateImpl;说明内部文档中给出的结构体示意将字段写作asyncio_current_loop/asyncio_current_task实际仓库源码中这两个字段的正式命名为asyncio_running_loop/asyncio_running_task语义一致“当前正在运行的事件循环/任务”。后续叙述均以源码实际命名为准。PyInterpreterState也新增字段用于承接线程状态释放后残留的任务可能发生其他线程仍持有本线程任务的引用见 Include/internal/pycore_interp_structs.h// Per-interpreter list of tasks, any lingering tasks from thread // states gets added here and removed from the corresponding // thread states list. struct llist_node asyncio_tasks_head; // asyncio_tasks_lock is used when tasks are moved // from threads list to interpreters list. PyMutex asyncio_tasks_lock;两者合并起来即为文档所给的整体结构示意typedef struct TaskObj { ... struct llist_node asyncio_node; } TaskObj; typedef struct PyThreadState { ... struct llist_node asyncio_tasks_head; } PyThreadState; typedef struct PyInterpreterState { ... struct llist_node asyncio_tasks_head; PyMutex asyncio_tasks_lock; } PyInterpreterState;asyncio_tasks_lock只用于保护解释器级列表免受并发修改例如线程销毁时把残留任务并入解释器列表的操作。任务的登记与注销register_task / unregister_task任务创建后通过register_task加入当前线程链表任务结束done/cancelled后由unregister_task移出链表。两者实现于 Modules/_asynciomodule.cstatic void register_task(_PyThreadStateImpl *ts, TaskObj *task) { if (task-task_node.next ! NULL) { // already registered assert(task-task_node.prev ! NULL); return; } struct llist_node *head ts-asyncio_tasks_head; llist_insert_tail(head, task-task_node); } static inline void unregister_task_safe(TaskObj *task) { if (task-task_node.next NULL) { // not registered assert(task-task_node.prev NULL); return; } llist_remove(task-task_node); } static void unregister_task(TaskObj *task) { #ifdef Py_GIL_DISABLED // check if we are in the same thread // if so, we can avoid locking if (task-task_tid _Py_ThreadId()) { unregister_task_safe(task); } else { // we are in a different thread // stop the world then check and remove the task PyThreadState *tstate _PyThreadState_GET(); _PyEval_StopTheWorld(tstate-interp); unregister_task_safe(task); _PyEval_StartTheWorld(tstate-interp); } #else unregister_task_safe(task); #endif }值得注意的实现细节register_task/unregister_task_safe首先检查task_node.next是否非空以判断是否已在链表中避免重复登记/移除链表本身存的是借用引用borrowed reference且加入/移出都是单纯的指针操作因此单线程内完全无锁lock-free在free-threadingPy_GIL_DISABLED构建下创建任务的线程 id 会存入TaskObj的task_tid字段。注销时先比对task-task_tid _Py_ThreadId()若注销发生在创建它的同一线程直接无锁移除即可否则跨线程注销需要_PyEval_StopTheWorld暂停解释器内所有线程待安全移除后再_PyEval_StartTheWorld恢复。这保证了链表不被并发撕裂。调用点包括任务创建L2373 附近、任务生命周期终结L2966 附近以及任务开始/结束执行时的进出场逻辑L3419-L3454。线程状态销毁时残留任务迁移到解释器级列表当线程状态被销毁时其任务链表可能仍有“残留任务”——比如另一个线程还持有该线程任务的引用导致这些任务尚未完成。因此 Python/pystate.c 中的PyThreadState_Clear会先清掉当前循环/任务强引用再在asyncio_tasks_lock保护下用llist_concat把线程链表整体并入解释器级任务链表Py_CLEAR(((_PyThreadStateImpl *)tstate)-asyncio_running_loop); Py_CLEAR(((_PyThreadStateImpl *)tstate)-asyncio_running_task); PyMutex_Lock(tstate-interp-asyncio_tasks_lock); // merge any lingering tasks from thread state to interpreters // tasks list llist_concat(tstate-interp-asyncio_tasks_head, ((_PyThreadStateImpl *)tstate)-asyncio_tasks_head); PyMutex_Unlock(tstate-interp-asyncio_tasks_lock);线程状态释放后解释器级列表仍继续持有这些任务保证它们不会被“丢”并且仍可被all_tasks()枚举。两个列表的初始化分别在interpreter与tstate创建路径中完成Python/pystate.c 与 L1640 附近且asyncio_tasks_lock初始化为零值PyMutex。all_tasks() 的一致性遍历与 stop-the-worldasyncio.all_tasks()现在遍历所有线程的 per-thread 任务链表 解释器级任务链表来收集全部任务。在 free-threading 下为保证遍历期间没有线程正在增删任务会先stop-the-world暂停所有线程从而获得跨线程一致且线程安全的快照视图。相关实现在 Modules/_asynciomodule.cadd_tasks_llist(head, tasks)遍历某个链表头因为链表持有借用引用为防止任务在遍历时被其他线程并发释放先用_Py_TryIncref尝试提升引用计数若对象正被并发释放则失败跳过成功后追加进结果列表add_tasks_interp(interp, tasks)先遍历解释器级链表free-threading 下断言interp-stoptheworld.world_stopped已成立再通过_Py_FOR_EACH_TSTATE_BEGIN/END遍历所有线程状态的任务链表。all_tasks的入口在 Modules/_asynciomodule.c 附近其逻辑为先加入 eager 任务再执行上述两条链表的遍历。当前任务的进出场与快/慢路径查询任务开始执行、暂停、结束时分别通过enter_task/leave_task更新线程状态上的当前任务字段定义于 Modules/_asynciomodule.cstatic int enter_task(_PyThreadStateImpl *ts, PyObject *loop, PyObject *task) { if (ts-asyncio_running_loop ! loop) { PyErr_Format(PyExc_RuntimeError, loop %R is not the running loop, loop); return -1; } if (ts-asyncio_running_task ! NULL) { PyErr_Format(PyExc_RuntimeError, Cannot enter into task %R while another task %R is being executed., task, ts-asyncio_running_task); return -1; } ts-asyncio_running_task Py_NewRef(task); return 0; } static int leave_task(_PyThreadStateImpl *ts, PyObject *loop, PyObject *task) { if (ts-asyncio_running_loop ! loop) { PyErr_Format(PyExc_RuntimeError, loop %R is not the running loop, loop); return -1; } if (ts-asyncio_running_task ! task) { PyErr_Format(PyExc_RuntimeError, Invalid attempt to leave task %R while task %R is entered., task, ts-asyncio_running_task ? ts-asyncio_running_task : Py_None); return -1; } Py_CLEAR(ts-asyncio_running_task); return 0; }enter_task/leave_task都做了严格校验事件循环必须是本线程“正在运行的循环”且当前任务状态一致进入前不得已有正在执行的任务离开时任务必须匹配否则抛出RuntimeError——这与纯 Python 回退实现 Lib/asyncio/tasks.py 的语义保持一致。另有swap_current_taskModules/_asynciomodule.c支持“换出旧任务、换入新任务”并采取转移所有权的方式减少冗余引用计数。任务在事件循环中被调度执行时会在task_step附近调用enter_task/leave_taskL3419-L3434 附近保证整条协程执行期间asyncio.current_task()能取到正确对象。这些函数同时也以_asyncio._register_task、_asyncio._unregister_task、_asyncio._enter_task、_asyncio._leave_task、_asyncio._swap_current_task的形式对 Python 层暴露Modules/_asynciomodule.c。查询当前任务current_task(loop)则分为快慢两条路径Modules/_asynciomodule.c快路径一般情况若loop就是当前线程正在运行的事件循环ts-asyncio_running_loop loop则直接返回ts-asyncio_running_task不存在则返回None完全无需加锁慢路径free-threading、跨循环查询若目标loop不是当前线程的运行中循环则需要stop-the-world暂停解释器内所有线程遍历各线程状态、比对asyncio_running_loop loop找到匹配线程后返回其asyncio_running_task没有任何匹配线程状态时返回None。_PyThreadStateImpl *ts (_PyThreadStateImpl *)_PyThreadState_GET(); // Fast path for the current running loop of current thread // no locking or stop the world pause is required if (ts-asyncio_running_loop loop) { if (ts-asyncio_running_task ! NULL) { Py_DECREF(loop); return Py_NewRef(ts-asyncio_running_task); } Py_DECREF(loop); Py_RETURN_NONE; } // ... otherwise: _PyEval_StopTheWorld(interp), iterate all tstates ...这样的设计保证在 free-threading 下各线程访问“自己运行循环的当前任务”互不争用全局字典从根上消除了旧方案在current_tasks全局字典上的竞争。端到端流程一图流综合上述代码路径可得到如下完整生命周期图与文档流程图一致节点对应实际 C 函数名流程要点任务创建即被register_task登记进当前线程链表处于 pending 的任务反复被task_step推进一旦 done/cancelled 即触发unregister_task非 free-threading 构建直接unregister_task_safe无锁移除free-threading 构建则先判断是否同线程同线程无锁跨线程需 stop-the-world 后移除线程销毁时若任务链表非空则把剩余任务整体并入解释器级任务链表后再释放线程状态。整体设计实现了无锁执行在多个事件循环运行于不同线程的 free-threading 场景下扩展良好。内省工具python -m asyncio pstreeper-thread 存储的动机之一是支持跨线程任务检视。仓库中的 Lib/asyncio/main.py 提供了asyncio模块的 CLI 入口内部import asyncio.tools提供工具实现其中的任务树检视即通过python -m asyncio pstree触发可以借此观察多个线程/事件循环中登记的全部任务状态。Python 层如何选择 C/Python 两套实现纯 Python 版任务管理函数_register_task、_unregister_task、_enter_task、_leave_task、_swap_current_task、current_task、all_tasks等定义在 Lib/asyncio/tasks.py。随后模块尾部会尝试导入_asyncio扩展一旦成功即用 C 实现覆盖这些名字Lib/asyncio/tasks.pytry: from _asyncio import (_register_task, _register_eager_task, _unregister_task, _unregister_eager_task, _enter_task, _leave_task, _swap_current_task, ...) ... _c_current_task current_task _c_register_task _register_task ... except ImportError: ...这意味着标准 CPython 分发默认走上述 C 快速实现只有在_asyncio不可用的受限环境才退化到全局字典 WeakSet的 Python 回退实现即 3.14 前的旧语义。若在回退实现中重编译为 free-threading全局字典/WeakSet 旧缺陷依旧存在这反衬出新方案价值主要体现在默认的 C 路径上。第二部分异步生成器的终结Python 实现asyncio的异步生成器async generator终结逻辑主要在纯 Python 层实现Lib/asyncio。它要解决一个根本矛盾异步生成器必须由协程驱动因此其终结执行finally块也必须发生在事件循环运行期间。问题未被完整迭代的异步生成器可能不执行 finally大多数异步生成器在“被完整迭代直到耗尽”后会被自动关闭但若它在耗尽前就被放弃例如async for中途break且未手动await agen.aclose()则不会被正确关闭finally块可能永远不执行。文档给出了如下示例import asyncio async def agen(): try: yield 1 finally: await asyncio.sleep(1) print(finally executed) async def main(): async for i in agen(): break loop asyncio.EventLoop() loop.run_until_complete(main())该代码不会打印finally executed——因为异步生成器agen未被完整迭代也没有被手动await agen.aclose()关闭。注意示例为示意写法实际事件循环通过asyncio.run()/loop.run_until_complete()获取。解决方案PEP 525 的 asyncgen hooksasyncio依据 PEP 525 定义的sys.set_asyncgen_hooks设置两类终结钩子Hook触发时机asyncio 侧行为firstiter hook异步生成器第一次被迭代时将其加入loop._asyncgens一个weakref.WeakSet事件循环由此跟踪所有活跃异步生成器finalizer hook异步生成器**即将被终结对象不再被引用**时从loop._asyncgens移除它并通过self.create_task(agen.aclose())调度一个任务去关闭它保证finally块在事件循环运行期间执行由于_asyncgens是弱引用集合事件循环不会阻止异步生成器被回收同时借助 hooks回收前的“最后机会”被用于把aclose()重新调度到事件循环中。源码对照BaseEventLoop 中的 hooks 与关闭流程两个 hook 及_asyncgens集合定义于 Lib/asyncio/base_events.py# A weak set of all asynchronous generators that are # being iterated by the loop. self._asyncgens weakref.WeakSet() # Set to True when loop.shutdown_asyncgens is called. self._asyncgens_shutdown_called Falsehook 实现位于 Lib/asyncio/base_events.pydef _asyncgen_finalizer_hook(self, agen): self._asyncgens.discard(agen) if not self.is_closed(): self.create_task(agen.aclose()) def _asyncgen_firstiter_hook(self, agen): if self._asyncgens_shutdown_called: raise RuntimeError(...) self._asyncgens.add(agen)在BaseEventLoop初始化/运行路径中事件循环会把这两个方法通过sys.set_asyncgen_hooks(firstiter..., finalizer...)注册Lib/asyncio/base_events.py并支持保存/恢复旧 hooks避免覆盖外层配置。完整工作流结合上述实现文档给出如下流程节点与base_events.py中方法一一对应完整地看保证finally执行有两条互补路径对象回收路径异步生成器因失去引用被终结时运行时_PyGen_Finalize触发 finalizer hook事件循环创建aclose()任务使finally在事件循环中被执行循环关闭路径事件循环关闭shutdown_asyncgens时检查是否仍有活跃异步生成器若有则逐个await agen.aclose()用asyncio.gather并行调度并等待其完成之后才真正loop.close()。文档给出的可运行示例使用asyncio.run最终会打印executing finally blockimport asyncio async def agen(): try: yield 1 yield 2 finally: print(executing finally block) async def main(): async for item in agen(): print(item) break # not fully iterated asyncio.run(main())循环关闭时的兜底shutdown_asyncgensshutdown_asyncgens是上面的“循环关闭路径”的公开入口位于 Lib/asyncio/base_events.py。其行为要点置self._asyncgens_shutdown_called True此后新创建的异步生成器若还想注册 firstiter hook 会得到错误提示提示需要显式调用loop.shutdown_asyncgens()而非在循环内部若当前没有活跃异步生成器直接返回否则把集合转为列表、清空_asyncgens再逐个await ag.aclose()并gather等待全部完成。asyncio.run()的配套实现Lib/asyncio/runners.py会在主任务结束后、关闭循环前调用shutdown_asyncgens确保即使在break等“未完整迭代”场景下异步生成器的finally块也不会被吞掉。小结CPython 3.14 的这两块改动分别回答了 asyncio 的两个经典工程问题任务管理C 层用“嵌入任务对象的链表节点 线程状态链表头 解释器级残留列表”替代全局WeakSet/字典用task_tid比对 stop-the-world 处理跨线程注销换取无锁高频路径与 free-threading 扩展性asyncio.all_tasks()/asyncio.current_task()因而既快又线程安全还支撑起python -m asyncio pstree这类跨线程内省工具。异步生成器终结Python 层通过 PEP 525 的 firstiter/finalizer hooks 把“生成器对象回收”与“事件循环仍可驱动其finally”两者桥接起来配合shutdown_asyncgens在循环关闭时兜底最终保证用户写在try/finally中的清理逻辑不会因提前break而丢失。本文所有论断均可回溯到 InternalDocs/asyncio.md 及以下源码_asynciomodule.c、pycore_tstate.h、pycore_interp_structs.h、pycore_llist.h、pystate.c、base_events.py、tasks.py、runners.py 与main.py。若想进一步做实验可基于本仓库构建 CPython 3.14并在 free-threadingPy_GIL_DISABLED构建下用多线程多事件循环压测任务创建/销毁观察all_tasks()与current_task()的无锁行为差异。【免费下载链接】cpythonThe Python programming language项目地址: https://gitcode.com/GitHub_Trending/cp/cpython创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考