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，键是事件循环对象、值是该循环当前正在执行的任务，用于回答“当前任务是什么”；
• 一个WeakSet（scheduled_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 = {}

原文档指出该设计存在三类缺陷：

1. 性能（Performance）：WeakSet需要维护一套完整弱引用以及相应的弱引用回调，在任务被回收时执行清理。这让 GC 负担加重；在任务数量很大的应用中，会形成瓶颈，内存占用上升、性能下降。而“查当前任务”需要在该字典上做一次字典查找，也偏慢。
2. 线程安全（Thread safety）：3.14 之前对WeakSet的并发迭代并不安全，多线程下调用asyncio.all_tasks()可能得到不一致结果，甚至抛出RuntimeError。相关回溯见 gh-123089 与 gh-80788。
3. 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存储，原因在文档与代码中都很明确：

1. 外部内省工具（如pstree）无法在事件循环对象上访问任意属性，因此不能把任务挂在 loop 上供外部读取；而线程状态是运行时内建结构，可由 C API 稳定访问；
2. per-thread 存储天然支持第三方事件循环实现（如 uvloop），不依赖 loop 对象的内部属性；
3. 对最常见（单线程）的 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.h：

typedef 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 state's list. struct llist_node asyncio_tasks_head; // `asyncio_tasks_lock` is used when tasks are moved // from thread's list to interpreter's 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.c：

static 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-threading（Py_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 interpreter's // 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-world

asyncio.all_tasks()现在遍历所有线程的 per-thread 任务链表 + 解释器级任务链表来收集全部任务。在 free-threading 下，为保证遍历期间没有线程正在增删任务，会先stop-the-world暂停所有线程，从而获得跨线程一致且线程安全的快照视图。相关实现在 Modules/_asynciomodule.c：

• add_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.c：

static 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_task（Modules/_asynciomodule.c）支持“换出旧任务、换入新任务”，并采取转移所有权的方式减少冗余引用计数。

任务在事件循环中被调度执行时，会在task_step附近调用enter_task/leave_task（L3419-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）：

1. 快路径（一般情况）：若loop就是当前线程正在运行的事件循环（ts->asyncio_running_loop == loop），则直接返回ts->asyncio_running_task（不存在则返回None），完全无需加锁；
2. 慢路径（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 函数名）：

流程要点：

1. 任务创建即被register_task登记进当前线程链表；
2. 处于 pending 的任务反复被task_step推进；一旦 done/cancelled 即触发unregister_task；
3. 非 free-threading 构建直接unregister_task_safe无锁移除；free-threading 构建则先判断是否同线程，同线程无锁，跨线程需 stop-the-world 后移除；
4. 线程销毁时若任务链表非空，则把剩余任务整体并入解释器级任务链表后再释放线程状态。

整体设计实现了无锁执行，在多个事件循环运行于不同线程的 free-threading 场景下扩展良好。

内省工具：python -m asyncio pstree

per-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.py）：

try: 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 hooks

asyncio依据 PEP 525 定义的sys.set_asyncgen_hooks设置两类终结钩子：

由于_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 = False

hook 实现位于 Lib/asyncio/base_events.py：

def _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执行有两条互补路径：

1. 对象回收路径：异步生成器因失去引用被终结时，运行时（_PyGen_Finalize）触发 finalizer hook，事件循环创建aclose()任务，使finally在事件循环中被执行；
2. 循环关闭路径：事件循环关闭（shutdown_asyncgens）时，检查是否仍有活跃异步生成器，若有则逐个await agen.aclose()（用asyncio.gather并行调度）并等待其完成，之后才真正loop.close()。

文档给出的可运行示例（使用asyncio.run，最终会打印executing finally block）：

import 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_asyncgens

shutdown_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-threading（Py_GIL_DISABLED）构建下用多线程多事件循环压测任务创建/销毁，观察all_tasks()与current_task()的无锁行为差异。

【免费下载链接】cpythonThe Python programming language项目地址: https://gitcode.com/GitHub_Trending/cp/cpython