读书笔记 · 约 26,677 字 · 预计推送 3 部分

Python asyncio 实战：取消、超时、背压、资源清理和常见死锁 —— 主题精读笔记 未开始

目录

题记与说明

0. 来源清单与可信度说明

#来源深读程度定位
1Nathaniel J. Smith, *Timeouts and cancellation for humans*（vorpus.org, 2018）<br>https://vorpus.org/blog/timeouts-and-cancellation-for-humans/全文精读（WebFetch 结构化提取）取消/超时的设计语义第一性原理；Trio cancel scope 与 shield 思想的出处
2Python 官方文档 *Coroutines and Tasks*<br>https://docs.python.org/3/library/asyncio-task.html全文结构化精读（Task.cancel/uncancel、timeout、TaskGroup、shield、wait_for、gather、create_task）语义与契约的权威定义
3Python 官方文档 *Developing with asyncio*<br>https://docs.python.org/3/library/asyncio-dev.html全文结构化精读阻塞事件循环、线程安全、未消费异常、异步生成器清理等工程坑
4*Bound the queue as well as concurrency*（yemingming.wiki）<br>https://yemingming.wiki/en/articles/bounded-concurrency-needs-a-bounded-queue全文精读背压：有界队列 vs 只限并发数
5CPython Issue gh-90155 *asyncio.Semaphore waiters deque doesn't work*<br>https://github.com/python/cpython/issues/90155全文结构化精读（含维护者评论）真实死锁案例：Semaphore 丢失唤醒与取消竞态

辅助/未深读来源（仅经搜索摘要确认存在，未逐字精读，笔记中不作事实依据）：

1. 七维提取

1.1 核心问题

asyncio 把"取消"实现为通过 await 点注入的异常（CancelledError），把"超时"实现为到点后的取消，把"背压"留给用户用有界队列/信号量搭建。这带来四类生产事故：

  1. 取消丢失与取消滥用：吞掉 CancelledError、清理代码自身被取消、gather() 出错时兄弟任务失控泄漏；
  2. 超时失真：wait_for 总耗时可能超过设定值、事件循环被阻塞时超时集体失效、requests.get 这类库"不可能保证在有限时间内返回"；
  3. 背压缺失：无界队列 + "每任务一信号量"让内存先于第一个结果爆掉；
  4. 死锁：Semaphore 唤醒竞态（真实 CPython bug）、哨兵顺序错误、阻塞事件循环、跨线程误用非线程安全原语。

一句话：asyncio 的难点从来不是"怎么并发"，而是"怎么停下来、怎么不满溢、怎么善后"。

1.2 主要主张

各来源交叉印证出的核心主张：

1.3 方法（技术路线与论证方式）

1.4 数据与证据

文献证据（原文引述，均给出处）：

  1. 超时缺失=潜伏 bug（NJS）：*"Pretty much every place your program interacts with another program or person or system, it needs a timeout, and if you don't have one, that's a latent bug."*
  2. requests.get 无界（NJS）：*"it is currently not possible to guarantee that requests.get will return in any finite time."*
  3. cancel 只是请求（docs.task）：*"Therefore, unlike Future.cancel(), Task.cancel() does not guarantee that the Task will be cancelled, although suppressing cancellation completely is not common and is actively discouraged."*
  4. 吞异常会破坏结构化并发（docs.task）：*"The asyncio components that enable structured concurrency, like asyncio.TaskGroup and asyncio.timeout(), are implemented using cancellation internally and might misbehave if a coroutine swallows asyncio.CancelledError. Similarly, user code should not generally call uncancel()."*
  5. 清理规范（docs.task）：*"It is recommended that coroutines use try/finally blocks to robustly perform clean-up logic. In case asyncio.CancelledError is explicitly caught, it should generally be propagated when clean-up is complete."*
  6. wait_for 总时长上溢（docs.task）：*"will wait until the future is actually cancelled, so the total wait time may exceed the timeout."*
  7. TaskGroup vs gather（docs.task）：*"TaskGroup provides stronger safety guarantees than gather for scheduling a nesting of subtasks: if a task (or a subtask ...) raises an exception, TaskGroup will, while gather will not, cancel the remaining scheduled tasks."*；gather 默认下 *"Other awaitables in the aws sequence won't be cancelled and will continue to run."*
  8. fire-and-forget 陷阱（docs.task）：*"Save a reference to the result of this function, to avoid a task disappearing mid-execution. The event loop only keeps weak references to tasks. A task that isn't referenced elsewhere may get garbage collected at any time, even before it's done."*
  9. shield 契约（docs.task）：*"if the coroutine containing it is cancelled, the Task running in something() is not cancelled… Although its caller is still cancelled, so the 'await' expression still raises a CancelledError."*
  10. shield 应克制（NJS）：*"if you think you need to use shielding then you almost certainly should rethink what you're trying to do."*；shield *"only blocks outer cancel scopes from applying, and has no effect on inner scopes."*
  11. 有界并发≠有界内存（yemingming）：*"A semaphore limits active work, not necessarily the number of allocated tasks."*、*"Ten permits can still leave a hundred thousand tasks allocated"*、*"task_done is queue accounting, not proof of successful business processing."*、*"An in-memory queue does not survive process restart; resumable work needs durable pending records and acknowledgements."*
  12. Semaphore 竞态（gh-90155）：Guido——*"Result set, then task cancelled, but not yet running"*；*"after the future's result is set the future cannot be cancelled, but the task can"*；*"the cancellation is what makes it so challenging. :-("*；asvetlov——locks *"don't provide a strong FIFO guarantee"*，彻底修复 *"requires async release method"*。修复：PR #93222，commit 24e03796248a（2022-09-22），并回溯 3.10/3.9 等。
  13. 阻塞循环（docs.dev）：*"Blocking (CPU-bound) code should not be called directly."*（1 秒 CPU 计算会让所有 Task 和 I/O 延迟 1 秒）；debug 模式下 *"Callbacks taking longer than 100 milliseconds are logged."*
  14. 未取回异常（docs.dev）：Future 的异常无人 await 时，GC 时打日志 *"Task exception was never retrieved"*。
  15. 异步生成器清理（docs.dev）：*"It is recommended to manually close the asynchronous generator."*（aclose()/contextlib.aclosing()）；避免跨任务并发迭代。

实证证据（本次运行，Python 3.12.12 / Windows 11，脚本与输出见 §3）：

1.5 结论

  1. 取消的正确姿势是一个契约，不是一行代码：try/finally 清理 → 捕获 CancelledError 只做清理然后重抛（确要吞掉必须配对 uncancel()，且仅限极特殊场景）→ 不在清理里放可挂死/可被再次取消的长操作（必要时 shield+重试，并理解 shield 只保护"内部任务"不保护"外层 await"）。
  2. 超时应该用 asyncio.timeout()/TaskGroup（3.11+），而不是裸 wait_for：前者把 CancelledError 定点转译为 TimeoutError、可 reschedule、可嵌套；后者的总时长会被清理工作拖长（实测 0.1s 超时实占 0.62s），且超时精度受限于事件循环活性（被同步阻塞时实测 0.2s 超时延迟到 1.065s）。
  3. 背压 = 三个独立阀门：活跃并发（信号量/worker 数）、排队缓冲（Queue(maxsize=N)，让 await put() 真的阻塞）、结果消费速度。只限并发不限队列，内存照样先爆；只限队列不限单源速率，一个慢源占满 worker。有界队列的"阻塞"是特性不是缺陷——它把压力暴露给上游。
  4. 死锁多来自四个根因：(a) 同步原语唤醒竞态/无 FIFO 保证（gh-90155 真实案例，2022 年才修）；(b) 哨兵/关闭顺序错误（worker 永远等不到结束信号）；(c) 阻塞事件循环（一切 await 全体冻结）；(d) 吞掉 CancelledError 或 fire-and-forget 任务让管线某环节"静默死亡"，下游永远等待。
  5. 结构化并发（TaskGroup）是当前 asyncio 最接近 Trio 主张的形态："首个失败取消全部兄弟 + ExceptionGroup 汇总 + 不丢外部取消"，比 gather() 的"失败即孤儿"安全得多。

1.6 局限

作者/来源层面承认的局限：

我（笔记作者）推断的局限：

1.7 对我的启发

  1. 写"可取消的安全代码"清单要进 code review 模板：① CancelledError 必须重抛；② 清理放 finally/async with；③ 清理里的 await 要么短、要么 shield+重试、要么同步化（把状态写入队列交给别的任务落盘）；④ 禁止裸 except Exception 兜住 CancelledError（它继承 BaseException，普通 except Exception 本来就接不住——反过来说，except: 裸捕获是取消杀手）。
  2. 超时一律用 async with asyncio.timeout(s) 或 TaskGroup，超时值要"按整条链路"设定：requests 式逐操作 timeout 不保证整体返回；deadline（绝对时间）才能跨层传递不衰减——NJS 的 Deadline.after() 桥接模式值得抄。
  3. "信号量 + 无界任务列表"反模式要替换为"固定 worker + 有界队列"：10 个 permit 仍可能堆 10 万个 task 对象；Queue(maxsize=N) 的阻塞 put 才是真背压。估算 maxsize 从"消费端速率 × 可容忍延迟"出发，而不是拍脑袋。
  4. 事件循环健康即超时健康：任何同步阻塞（CPU、文件、time.sleep、阻塞日志）都会让所有超时同时失效——debug 模式（>100ms 回调告警）应常开；日志必须异步化。
  5. 对"取消×唤醒"类竞态保持敬畏：连 Guido 都写 *"the cancellation is what makes it so challenging. :-("*——自己实现同步原语/条件变量时优先复用标准库，并避免在"已拿到结果但未运行"的窗口里叠加 cancel。
  6. Windows 上 asyncio.sleep(0.001) 可能实际睡 ~15.6ms（本次实测背压基准中两种运行分别呈现 1ms 与 15.6ms 粒度）——细粒度超时/限速在 Windows 上不可信，基准测试要用足够大的间隔（本次用 50ms 才得到干净数据）。
  7. 结构化并发的迁移成本是值得的：把 gather() 换成 TaskGroup 消除"孤儿任务"这一整类泄漏/死锁；3.15 的 TaskGroup.cancel() 进一步提供"非异常性提前退出"，语义更完整。

2. 逐来源精读摘记

2.1 NJS《Timeouts and cancellation for humans》(vorpus.org, 2018)

论证主线：程序与外界交互有三态——成功、失败、挂死；挂死必须靠超时兜底，因此"每个交互点都需要超时，否则是潜伏 bug"。但主流超时 API 全是坏的：

与 asyncio 的关系（2018 时点）：*"some impedence mismatch between the cancel scope model … and asyncio's current task-oriented, edge-triggered cancellation"*；没有 nursery 时*"there's no reliable way to propagate cancellation across tasks"*。（注：3.11 TaskGroup + asyncio.timeout 正是对这一批评的制度化回应。）

对"资源清理"的直接启示：边沿触发取消下，*"if a Cancelled exception were to suddenly materialize in a place the user wasn't prepared for it … it could corrupt internal state."*；经典翻车：取消打断 send_message 后，finally 里的 ws.close() 自己也会挂死（close 握手永远发不出去）——清理代码是取消风暴中最脆弱的一段。shield 是最后的绳子：*"you can set a cancel scope to 'shield' its contents from outside cancellations"*，但 *"only blocks outer cancel scopes from applying, and has no effect on inner scopes"*，且 *"if you think you need to use shielding then you almost certainly should rethink what you're trying to do."*

2.2 官方文档《Coroutines and Tasks》(docs.python.org/3/library/asyncio-task.html)

取消语义（逐条契约）：

超时：

TaskGroup（结构化并发的官方实现）：

其他关键契约：

2.3 官方文档《Developing with asyncio》(docs.python.org/3/library/asyncio-dev.html)

2.4 《Bound the queue as well as concurrency》(yemingming.wiki)

核心洞察：信号量限制的是"开工数"，不是"存在的任务数"——*"A semaphore limits active work, not necessarily the number of allocated tasks."*、*"Ten permits can still leave a hundred thousand tasks allocated"*。正确结构把三个量分开：活跃工作 / 排队工作 / 待消费结果，各自设界：固定 worker + 有界队列，*"the producer waits for space, workers remove items, and results are persisted promptly."*

背压即反馈：*"Waiting on a full queue is useful feedback. Replacing it with an unbounded buffer simply hides pressure from the producer."*

故障模式清单：

容量设定：示例 4 worker / maxsize 8；原则是*"concurrency, connections, request rate, and result-write capacity around measured bottlenecks rather than one fashionable number"*。验收方式：放慢消费端 + 注入必现失败，验证生产者被阻塞、内存有界、无 worker 滞留。

2.5 CPython gh-90155《asyncio.Semaphore waiters deque doesn't work》

bug 本体（两层）：

  1. 饥饿：紧循环 while True: async with sem: 的任务释放后立刻重取，waiters deque 里的等待者永远轮不上（20 任务、值 5 的信号量，实测只有 0–4 号反复获取；前面插一个 await asyncio.sleep(0) 就能掩盖）。
  2. 丢失唤醒/取消竞态：多个 waiter 时，若队首 waiter 已被唤醒或已取消，release() 不再去唤醒下一个 → 永久卡死。Guido 概括的难点窗口：*"Result set, then task cancelled, but not yet running"*——*"after the future's result is set the future cannot be cancelled, but the task can"*；并感叹 *"the cancellation is what makes it so challenging. :-("*

修复史：2021-12 立案，多次 PR 反复（#30052、#31910（被指未修好）、#32047/#32049 回溯），有效修复为 PR #93222，commit 24e03796248a（2022-09-22）+ 一串 miss-islington 回溯。语义要点（kumaraditya303）：只有当 *"not self.locked() and not self._waiters"* 才立即获取，且新获取者必须 yield 一次，否则 *"it can cause starvation on the waiting tasks"*。

留下的设计坦白（asvetlov）：asyncio locks *"don't provide a strong FIFO guarantee"*；正确语义 *"requires async release method"*（如 await sem.release_and_wait()），意味着彻底修复要动 API。对使用者的推论：不要把公平性/顺序假设建立在 asyncio 同步原语上；不要在锁/信号量的关键区里叠加复杂取消逻辑。

3. 本机实证记录（Python 3.12.12 / Windows 11 x64）

运行方式：python -I <script>（-I 隔离模式）。脚本位于工作区 verify/ 目录（asyncio_verify.py、backpressure2.py、bp_debug.py、bp_decisive.py、cleanup_edge.py）。

#演示命令关键输出结论
A吞掉取消python -I verify/asyncio_verify.py swallow[A] swallow: result=done t.cancelled()=False<br>[A] re-raise: result=<raised CancelledError> t.cancelled()=True吞 CancelledError = 取消丢失；重抛才传播
B1无界队列python -I verify/backpressure2.py（maxsize=0, n=2000）qsize@0.1s=1992, producer_done@0.1s=True, total=31.08s生产者瞬间倒满队列，1992 条堆积
B2有界队列同上（maxsize=10）；另跑 bp_decisive.py（maxsize=10, 消费 50ms/条, n=30）t=0.12s qsize=9 producer_done=False puts_completed=12；total=1.86s；backpressured: 18/30队列恒 ≤10，put 真阻塞，总耗时≈n×消费延迟，背压生效
C阻塞循环×超时python -I verify/asyncio_verify.py block_timeoutasyncio.timeout(0.2) fired, but elapsed = 1.065s事件循环被 time.sleep(1.0) 阻塞时超时整体失灵
DTaskGroup 首败取消python -I verify/asyncio_verify.py taskgroupslow sibling cancelled after ~0.105s of its 10s sleep，抛 ExceptionGroup首个失败立即取消兄弟并汇总异常
F1单次取消 vs finally 清理python -I verify/cleanup_edge.py[F1] await t took 0.31s，cleanup finished 打印单次 cancel 只投递一次异常，finally 内 await 可完成
F2双重取消打断清理同上[F2] await t took 0.00s，"cleanup finished" 未打印第二次 cancel 落在 finally 内 → 清理拦腰折断
F3wait_for 超时上溢同上[F3] TimeoutError after 0.62s (timeout was 0.1s)证实文档"总耗时可能超过 timeout"
F4shield 契约同上shield hit by second cancel, retrying cleanup → cleanup completed外层 await 仍抛 CancelledError，被 shield 任务存活

Windows 定时器粒度警告（实证中发现）：同一批实验里 asyncio.sleep(0.001) 有时 ~1ms、有时 ~15.6ms 才返回（B1 的 31.08s 总耗时即 2000×15.6ms 所致；把消费间隔放大到 50ms 后数据立即变干净）。推论：在 Windows 上做 asyncio 基准或依赖亚 10ms 超时/限速时，必须先测量本机定时器粒度；中间一度怀疑"maxsize=10 时 put 不阻塞"，用 50ms 慢消费者重测后才排除基准伪影（期间还揪出一个被误杀后台任务遗留的孤儿 python.exe 进程，taskkill //F //IM python.exe 清理后复测一致）。

未能在本机复现的历史 bug：gh-90155 的丢失唤醒/饥饿 bug 在 3.12.12 上已修复（该修复于 2022-09 落地），本次仅凭 issue 文字转述，未做旧版本复现实验。

4. 实战速查（由上述证据综合）

4.1 取消与清理

# 模板：可安全取消的任务
async def worker(q: asyncio.Queue):
    try:
        item = await q.get()
        ...
    except asyncio.CancelledError:
        raise                      # 清理完必须重抛；吞掉=取消丢失（实测A）
    finally:
        await short_cleanup()      # 单次 cancel 下可完成（实测F1）
        # 若清理较长/可能被叠加取消：shield + except + 重试（实测F4）
        # try: await asyncio.shield(save_state())
        # except asyncio.CancelledError: await save_state()

4.2 超时

try:
    async with asyncio.timeout(10):        # 3.11+；可 reschedule、可嵌套
        await do_request()
except TimeoutError:                        # 只能在 with 外面接
    ...

4.3 背压

queue: asyncio.Queue = asyncio.Queue(maxsize=8)      # 排队工作设界
SEM = asyncio.Semaphore(4)                            # 活跃工作设界
async def producer():
    for item in make_items():
        await queue.put(item)                         # 队列满则阻塞——这是特性
for _ in range(4):
    tg.create_task(consumer(queue))                   # worker 数=消费端容量

4.4 死锁排查顺序

  1. 循环里有没有同步阻塞（time.sleep、CPU、阻塞日志）？→ 开 debug 模式看 >100ms 告警。
  2. 有没有人吞了 CancelledError / 裸 except:？→ 取消传不下去，上层超时全部变形。
  3. 队列/信号量/哨兵的关闭顺序对不对？worker 在等谁？
  4. 有没有 fire-and-forget 任务被 GC 或异常未取（"Task exception was never retrieved"）？
  5. 是否跨线程直接摸了 asyncio 对象（几乎全部非线程安全）？→ 只能 call_soon_threadsafe/run_coroutine_threadsafe。
  6. 自定义同步原语？→ 回忆 gh-90155：取消×唤醒竞态连标准库都修了一年；没有 FIFO 保证的假设不要写进代码。

5. 原文引述存档（便于后续引用）

"Pretty much every place your program interacts with another program or person or system, it needs a timeout, and if you don't have one, that's a latent bug."
—— Nathaniel J. Smith, *Timeouts and cancellation for humans*
"it is currently not possible to guarantee that requests.get will return in any finite time."
—— 同上
"Cancel scopes can be nested; Cancelled exceptions know which scope triggered them, and will keep propagating until they reach the corresponding with block."
—— 同上
"if you think you need to use shielding then you almost certainly should rethink what you're trying to do."
—— 同上
"that's something the computer should handle, rather than making flaky humans do the work."
—— 同上（关于把作用域栈交给运行时管理）
"Therefore, unlike Future.cancel(), Task.cancel() does not guarantee that the Task will be cancelled, although suppressing cancellation completely is not common and is actively discouraged."
—— Python docs, Coroutines and Tasks
"The asyncio components that enable structured concurrency, like asyncio.TaskGroup and asyncio.timeout(), are implemented using cancellation internally and might misbehave if a coroutine swallows asyncio.CancelledError."
—— 同上
"The asyncio.timeout() context manager is what transforms the asyncio.CancelledError into a TimeoutError, which means the TimeoutError can only be caught outside of the context manager."
—— 同上
"will wait until the future is actually cancelled, so the total wait time may exceed the timeout."
—— 同上（wait_for）
"TaskGroup provides stronger safety guarantees than gather for scheduling a nesting of subtasks: if a task (or a subtask, a task scheduled by a task) raises an exception, TaskGroup will, while gather will not, cancel the remaining scheduled tasks."
—— 同上
"Save a reference to the result of this function, to avoid a task disappearing mid-execution. The event loop only keeps weak references to tasks."
—— 同上（create_task）
"Almost all asyncio objects are not thread safe."
—— Python docs, Developing with asyncio
"Callbacks taking longer than 100 milliseconds are logged."
—— 同上（debug 模式）
"A semaphore limits active work, not necessarily the number of allocated tasks."
—— yemingming.wiki, Bound the queue as well as concurrency
"Ten permits can still leave a hundred thousand tasks allocated"
—— 同上
"Waiting on a full queue is useful feedback. Replacing it with an unbounded buffer simply hides pressure from the producer."
—— 同上
"An in-memory queue does not survive process restart; resumable work needs durable pending records and acknowledgements."
—— 同上
"Result set, then task cancelled, but not yet running" / "the cancellation is what makes it so challenging. :-("
—— gvanrossum, CPython gh-90155 评论
asyncio locks "don't provide a strong FIFO guarantee"; proper fix "requires async release method"
—— asvetlov, CPython gh-90155 评论

*笔记完。检索与实证均于 2026-11-03 完成；未修改 matrix.csv、backlog、nightly-log.md 或其他条目笔记。*

← 资料库今日播报 →