03-asyncio 与并发模型盲区梳理

本文以 Python 3.12、默认 Task 工厂为基线。示例是独立演示或注明前置条件的教学片段，不是某个项目提交的源码摘录；项目里的共享对象、限流和退出策略仍要回查对应版本。

闭包、nonlocal 等纯语法见 01 Python 语法；接口和消息验证见 02 类型系统与注解。这里专注异步工作的推进与收尾。

一 两个 Run 的网络等待能否重叠

1 先创建了哪些工作

用同一个 ask 比较，差别不在最后写了几个 await，而在等待前已经安排了哪些工作。

import asyncio

async def ask(name):
    print(name, "start")
    await asyncio.sleep(0)
    print(name, "end")
    return name

async def main():
    await ask("A")
    await ask("B")
    c = asyncio.create_task(ask("C"))
    d = asyncio.create_task(ask("D"))
    print(await c, await d)

asyncio.run(main())

结果：先是 A start → A end → B start → B end；默认调度下再是 C start → D start → C end → D end，最后 C D。C、D 已经都成为 Task，所以先 await C 不会推迟 D 的开始。真实 I/O 谁先完成取决于就绪时间；gather(c, d) 的结果列表则按传入顺序。

如果借菜谱类比，函数是菜谱，对象是一次具体制作的状态，Task 是被调度和跟踪的工作；Task 不对应一条独占线程。只创建协程对象再丢弃，还可能收到 never-awaited 警告。

asyncio.run(main()) 先求值 main() 得到对象，再创建并管理事件循环与主 Task，然后执行 main 的函数体。主任务结束后，还要清理剩余任务、异步生成器和默认执行器，关闭循环，才返回结果或抛出异常。它既不反复从头跑 main，也不启动后台线程后立即返回。同线程已有运行中的循环时不能嵌套调用 run；协程内用 get_running_loop()，不用自己拼几行 new_event_loop() 冒充完整清理。Runners

事件循环处理就绪回调、定时器和 I/O 事件，通常每次在循环线程里推进一个任务片段。实际挂起后，其他工作才有机会执行。await 的对象已就绪时可能直接继续，不保证切换；上例 sleep(0) 明确主动挂起。3.12 可选的 eager Task 工厂会改变开始执行的时机，不能把这里的默认时序照搬过去。

协程让多个 I/O 等待区间重叠，也保留顺序写法、try/finally 和 async with 的清理结构。它通常比每项等待各用一条线程更轻，但内存差距要测量，不能固定说“每个几 KB、数量差三个量级”。并发是管理多项未完成工作，并行才涉及同一时刻执行计算；一个循环的调度模型也不能证明整个应用没有线程池、其他进程或释放 GIL 的计算库。

2 单线程为什么仍会丢一次加法

下面只演示同一循环里的普通整数；锁保护的是跨等待点的整次读改写。

import asyncio

async def count_runs(locked):
    count = 0
    lock = asyncio.Lock()
    async def change():
        nonlocal count
        old = count
        await asyncio.sleep(0)
        count = old + 1
    async def increment():
        if locked:
            async with lock:
                await change()
        else:
            await change()
    await asyncio.gather(increment(), increment())
    return count

print(asyncio.run(count_runs(False)))  # 1
print(asyncio.run(count_runs(True)))   # 2

无锁时两项任务都先读到 0，挂起后各写回 1。若只是无挂起点的普通整数 count += 1，同循环其他普通 Task 不会插进这段连续代码；这个结论不能推广到外部线程、数据库或跨 await 的资源操作。asyncio.Lock 是任务间同步工具，不是线程锁或数据库事务；也别把一整次模型网络请求套进全局锁，意外把所有 Run 串行化。

3 请求上下文跟随谁

连接标识、日志请求 ID 可以显式传参；调用链很深时，也能用 ContextVar。它保存当前 Context 的绑定，不是“每调一个协程就复制一个变量”。

import asyncio
from contextvars import ContextVar

rid = ContextVar("rid", default="unset")

async def read_id():
    return rid.get()

async def change_id():
    rid.set("direct")

async def main():
    token = rid.set("before")
    try:
        child = asyncio.create_task(read_id())
        rid.set("after")
        print(await child, rid.get())  # before after
        await change_id()
        print(rid.get())               # direct
    finally:
        rid.reset(token)
    print(rid.get())                   # unset

asyncio.run(main())

新 Task 默认复制创建瞬间的上下文绑定，创建者之后 set 不会跟过去；显式传 context= 时遵循指定上下文。直接 await 通常仍在当前 Task 的上下文，修改会留在这里。复制不是深拷贝，绑定的列表等可变对象仍可能共享。token/reset 用于恢复请求范围；threading.local() 只按线程隔离，区分不了同线程上的 Task。ContextVar

MiniClaude 的 _writer_var 可以用同一方式传当前 writer，但仍要判断子任务是否该继承它、连接是否活到任务结束。上下文绑定不替你管理连接寿命。

二 把响应交回等待者并确认日志完成

1 Future 负责配对 不负责发请求

一条连接可以传多条请求和事件推送，下一条消息未必就是刚才请求的答案。发送方先建 Future，用唯一 ID 登记，再发送；接收循环独立读消息，按 ID 完成对应 Future，等待者恢复。这接近 Java CompletableFuture 的“未来结果”抽象，但不能照搬其 API 或线程安全规则。

以下是同一循环里的配对骨架，不是完整客户端。已导入 asyncio；pending 是连接专属字典，rid 在该连接寿命内从未用过，send(rid) 是已定义的异步发送函数，外围还负责拒绝已关闭连接的新请求。

future = asyncio.get_running_loop().create_future()
pending[rid] = future
try:
    await send(rid)  # 先登记；发送失败也会清理
    result = await asyncio.wait_for(future, 3)
finally:
    if pending.get(rid) is future:
        pending.pop(rid, None)
    if not future.done():
        future.cancel()
    elif not future.cancelled():
        future.exception()  # 发送失败时，也收走已到达的响应异常

接收侧只处理已经校验过的响应；事件推送另走分支。下面保留缺字段的失败路径，而不是悄悄给一个默认结果：

future = pending.get(message["id"])
if future is not None and not future.done():
    if "error" in message:
        future.set_exception(RuntimeError(str(message["error"])))
    elif "result" in message:
        future.set_result(message["result"])
    else:
        future.set_exception(ValueError("响应缺少结果或错误"))

结果：等待者拿到结果，或收到远端错误。0、False、[] 都是合法结果，不能写 message.get("result") or {}。成功、远端错误、发送失败、超时和调用方取消都走同一清理出口；这里的 3 秒仅限制等响应，真实客户端还应给发送等阶段定义期限。

断连时先禁止新请求，再取出并清空 pending，对尚未完成的 Future 设置 ConnectionError，唤醒所有等待者。重复、迟到或未知 ID 的响应应忽略或记录；同连接内不复用 ID，避免旧响应配到新请求，可集中生成递增 ID 或 UUID。清理时检查仍是原 Future，防止误删其他登记。pop(id, None) 缺键不报错是因为提供了默认值，裸 pop(id) 仍可能抛 KeyError。JSON 编码、报文边界、重连和协议验证不由 Future 自动完成。

2 done 不只表示成功

结束后再次 set_result() 或 set_exception() 会抛 InvalidStateError，因此接收侧先查 done。在同循环里，从检查到设置之间没有 await，另一个普通 Task 不会在中间插入；这不是线程同步保障。asyncio Future 非线程安全，外部线程要用 loop.call_soon_threadsafe(...) 把“检查并完成”一起安排回循环线程。Future

3 Queue 记着两种数量

以下片段放进协程运行，已导入 asyncio。它把“取走”和“确认处理结束”拆开：

queue = asyncio.Queue(maxsize=1)
await queue.put("trace-1")
item = await queue.get()
joined = asyncio.create_task(queue.join())
await asyncio.sleep(0)
print(queue.empty(), joined.done())  # True False
queue.task_done()
await joined

订单被取走，菜还没做好。每个成功 get 的条目应恰好 task_done 一次；多调抛 ValueError，少调会让 join 一直等。消费者常用 try/finally 结算已取条目，但结算不代表业务成功：写文件失败或取消后，仍要记录、传播失败或决定重试。Queue

TraceWriter 把生产与消费暂时解耦，关闭顺序应是：停止接收新记录 → 排空已接受工作并检查消费者失败 → 停止并等待消费者 → 关闭文件。不能只 await join：消费者已崩溃、尚有未完成项时，计数再也到不了零，需同时监督消费者并定义排空失败策略。

有界队列满时 await put() 会等待，这就是背压；无界队列不因容量等待，却可能吃光内存。后台消费者若仍同步写慢磁盘，照样阻塞循环。Queue 本身也非线程安全，更不保证通过它连接的整个业务没有共享状态。

三 工具命令超时后不能只停止等待

1 先判断工作会卡在哪里

run_in_executor(None, fn, ...) 的 None 表示默认执行器，通常是线程池。移出循环和多核提速是两个目标，不能把线程池叫作所有 CPU 重任务的最优解。进程池需可传递的函数和参数，通常把函数定义在模块顶层，脚本保留 if __name__ == "__main__": 入口保护；小任务未必更快。

取消等待结果的 Task 不保证已经运行的线程或进程工作停止，关闭执行器也可能需要等待；长工作常需停止标记、分块计算或进程管理。to_thread() 会传播当前 Context，不能推广到所有执行器接口。

2 停止直接子进程之后还要回收

BashTool 一类工具可以用 asyncio 子进程接口异步等待。下面是协程内片段，已导入 asyncio、sys；只运行当前 Python 解释器并输出 hello，不依赖 shell：

proc = await asyncio.create_subprocess_exec(
    sys.executable, "-c", "print('hello')",
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE,
)
output = asyncio.create_task(proc.communicate())
try:
    stdout, stderr = await asyncio.wait_for(asyncio.shield(output), 1)
except (TimeoutError, asyncio.CancelledError):
    if proc.returncode is None:
        try:
            proc.kill()
        except ProcessLookupError:
            pass  # 进程可能恰好已经退出
    await output
    raise
if proc.returncode != 0:
    raise RuntimeError(stderr.decode())
print(stdout.decode().strip())  # hello

正常取得输出并检查退出码；超时或一次取消时先 kill，再等待 communicate 排空管道并回收进程，最后继续抛出异常。shield() 只避免本次外层取消传给 output Task，不替我们持有、取消或等待它。这个片段针对已经取得 proc 的阶段和一次取消请求，不覆盖启动阶段取消、反复取消或清理本身失败；生产代码应为这些路径定义监督和退出期限。

create_subprocess_exec(program, *args) 直接传程序和参数；create_subprocess_shell(command) 把字符串交给 shell，才能解析管道、重定向等语法，也增加注入与后代进程管理风险。同步 subprocess.run() 则会阻塞调用线程。

管道大量输出不读、只 await wait，可能互相等待。communicate() 同时读 stdout 和 stderr，不必靠合并两路才能避免这种死锁；stderr=STDOUT 只是另一种输出组织方式。但 communicate 把输出攒在内存，不适合无界输出。杀掉直接子进程也未必终止整棵进程树。asyncio 子进程

3 异步执行和沙箱分开检查

评估 BashTool 时，还要查命令是否含外部输入、是否真的需要 shell；运行用户、目录和环境变量如何限制；能访问哪些文件与网络；stdout/stderr 是否限量；超时是否回收直接进程及后代；Unix／Windows 的停止方式是否不同；部署是否另有容器或权限隔离。

一个 create_subprocess_shell() 片段本身没有提供沙箱，但仅凭它也不能断言整个 MiniClaude 部署毫无隔离。

四 服务退出时收回连接和后台 Run

1 Task 要有结果和异常的负责人

长时间的 agent.run 可以先返回 run_id，再通过事件推送进度；这个响应只表示“任务已接受”。学习记录中的调用关系是 handler 调 SessionManager 的 send_message，runner 在 SessionManager 内创建；不能仅凭这一片段推定当前项目实现。短操作可以直接 await，长操作是否后台化取决于协议与业务，而不只是耗时。

以下是管理片段，在运行中的循环里使用；已导入 asyncio，run(goal) 是已定义的协程，report(error) 是快速且不抛错的同步报告函数，running 由服务持有：

running = set()

def on_done(task):
    running.discard(task)
    if not task.cancelled():
        error = task.exception()
        if error is not None:
            report(error)

task = asyncio.create_task(run(goal))
running.add(task)
task.add_done_callback(on_done)

循环只保留 Task 的弱引用，需要可靠完成的后台工作应有强引用所有者；不保存可能在完成前回收，也可能无人观察异常。完成后由循环安排同步的 callback(task)，不用轮询；回调不能直接 await 或执行慢阻塞操作，再创建异步清理 Task 也要管新 Task 的寿命。

discard 只移除引用，缺项也不报错；remove 缺项会报错。cancel 本身不会把 Task 从集合删除，不能用它解释为什么必须 discard。取得结果或异常仍要通过 await、result、exception 等途径；仅移除登记不能算已处理失败。

2 取消是请求 还需要等待确认

await task 返回结果或传播异常。task.cancel() 请求任务在下一次可处理取消的机会收到 CancelledError，不会当场硬停 Python 代码，更不会自动停掉线程函数。3.12 中 CancelledError 继承 BaseException，普通 except Exception 接不住。

工作协程用 finally 清理；若捕获取消，通常清理后继续 raise，吞掉它可能破坏超时和 TaskGroup。调用方若主动取消自己管理的任务，应再 await 并确认那次取消，不能取消完就丢弃；也不应随意吞掉自身从上层收到的取消。

开始前取消和执行中取消不同。Task 尚未进入函数体时被取消，函数体里的 try/finally 没机会建立。需要比较执行中清理时，可让 worker 在进入 try 后 started.set()，调用方先 await started.wait() 再 cancel；连接等外部资源则必须在创建 Task 时就登记，不能只靠 handler 的 finally。

TaskGroup 自 3.11 起提供结构化并发，但仅管理加入组的任务，也不能硬停拒绝协作取消的代码。Task、取消与 TaskGroup

3 Server 和 Run 有两条不同的关闭链

start_server() 建立监听并把 reader、writer 交给每次连接的回调；回调是协程函数时才自动调度为 Task，不能把所有同步回调也解释成协程。并发来自循环，不是每连接分配线程。默认已经开始接入，但 Server 对象返回并不负责让主协程永远活着；直接 asyncio.run(asyncio.start_server(...)) 会在得到 Server 后收尾。

可以让主协程持续等待 serve_forever() 或退出事件。连接写入用 writer.write() 配合 await writer.drain() 处理发送背压，结束用 close/wait_closed。测试服务器可绑定 127.0.0.1、端口 0，由系统分配端口，避免意外开放到外网。

Server 链的顺序不能颠倒。 在 CPython 3.12.1 起的修正行为中，Server.wait_closed() 要等监听关闭且活动连接结束；async with server 退出会等待它。已有客户端连上却不发数据时，handler 可能一直等读取；只关监听、等主协程结束后才指望 run 取消 handler，可能卡住。Server 生命周期、CPython 3.12.14 实现

一种明确所有权的做法是：在同步 on_connect 回调里创建 handler Task 的同时登记 Task → writer，完成回调移除登记、关闭遗漏的 writer 并观察异常。关闭标记已设时立即拒绝新连接；这样尚未开始运行的 Task 所对应的 writer 也能被找到。

若采用 serve_forever，可把它放进独立 serving Task，主协程用 shield 等待。主协程被取消后，先在 finally 设置关闭标记、关闭监听、关闭已登记 writer、取消并等待连接 Task，再取消并等待 serving，最后 await Server.wait_closed 或离开 async with。直接 await serve_forever 时，取消会先进入它自己的 Server 关闭等待，可能还没轮到外层关闭连接。shield 的用途是让外层有机会按顺序收尾，不是免除回收责任。

这仍有版本与竞态边界：原审阅稿在 CPython 3.12.14 的“新连接接入与关闭重叠”测试中记录过标准库内部断言，不应据一段教学代码宣称竞态已解决。真实服务还需覆盖这类重叠、空闲连接、反复取消，给清理失败与退出耗时定策略。

Run 链先确定作用域。 单个 _current_run_task 并拒绝第二次请求，表达“最多一个 Run”；改为 set 只解决记录多项任务和持有引用，不自动限制模型 API、数据库连接或内存。各 Run 即使内部顺序调用模型，它们的网络等待仍可能重叠；是否允许重叠要看全局限流器、信号量和 provider 实现。独立 Run 状态也不等于 EventBus、TraceWriter、会话和文件完全不共享。

退出先停止接收新 Run，再取稳定任务快照，决定排空还是取消，等待全部结束，最后关它们依赖的资源。若用 return_exceptions 收尾，必须有回调或调用方处理失败；错误记录本身也要有保留策略，不能无限积累。wait_idle() 只说明观察到当前已登记任务结束，生产者还能创建任务时不保证从此永远空闲。

回查项目时问清：Run 属于连接、Session 还是应用？断连后继续还是取消？run_id 返回前是否已登记？后台失败从哪条通道通知？谁限制并发，谁处理排空超时？这些答案决定关闭链如何组合。

附录 信号 平台与执行单元的边界

A 不必给所有程序重写信号处理

普通命令行入口通常先利用 asyncio.run() 的 Ctrl+C 行为：在主线程且原 SIGINT 处理器允许接管等条件下，Runner 取消主任务，让 finally 有机会清理。服务框架可能已经管理信号，不应随意覆盖。

需要在 Unix 主线程接 SIGTERM 时，可以让同步回调设置 Event，再由主协程异步退出。以下片段在主协程内使用，已导入 asyncio、signal；它不适用于 Windows，且假设本程序拥有该信号处理权：

loop = asyncio.get_running_loop()
stop = asyncio.Event()
loop.add_signal_handler(signal.SIGTERM, stop.set)
try:
    await stop.wait()
    # 在这里 await 应用的异步关闭函数
finally:
    loop.remove_signal_handler(signal.SIGTERM)

不要直接传 async 回调，也不能在普通回调里直接 await；只打印“准备退出”不会结束等待。若回调创建清理 Task，仍须保存并等待它。Windows 不支持 add_signal_handler；3.12 默认 ProactorEventLoop 支持异步子进程，不能把 Unix 的全部写法照搬。平台支持

signal.signal() 并非一概禁用。Python 处理函数由解释器稍后在主解释器的主线程执行，不等同于底层 C 信号处理器；不能用“必然数据竞争”或“只能执行 C 层 async-signal-safe 操作”概括它。这里选择循环接口，是为把退出通知接入任务生命周期。Python 信号规则

B 创建关系不等于自动回收关系

Thread.join() 不能等待当前线程自身，也不能等待尚未启动的线程；进程的信号权限、平台与监管方式同样影响终止和回收。与其套“父协程杀子协程”的类比，不如沿代码确认：谁驱动这项工作，它在哪个等待点交出控制权，谁负责最后的结果和清理。

后续路线：04 Agent 架构与设计模式 把任务管理放回模块职责；05 IPC 与网络通信 继续看报文边界与请求协议；06 工程实践与工具链 回到开发与验证流程。