同步 I/O 拖死 Agent

这篇来自我在一个 Agent 平台里处理过的一类生产问题：文件上传变慢时，正在流式输出的 Agent、后台任务和心跳也会一起停顿。

文中的代码全部是为公开复盘重新写的最小实现，不是公司源码；实验数字只来自读者当前浏览器，也不是生产容量。真实经历在这里承担的是问题来源、判断过程和我负责过的改造边界，而不是拿内部数据给文章增加戏剧性。

问题最后并不复杂：同步 SDK 被直接放进了 async def。调用它的协程在等网络，该进程或 Worker 上负责这些协程的 event loop 也跟着一起等。

先不用记结论。下面的两个按钮执行相同数量、相同耗时的工作。左边把工作直接放在共享调度器里，右边把它移到独立 Worker。这里用浏览器主线程类比服务端 event loop，用 Worker 类比线程卸载；忙循环只是为了稳定制造共享线程停顿，不模拟 I/O 类型、Python GIL 或服务端吞吐。这个实验只能说明调度关系，不是生产压测。

Clean-room 实验 · 共享调度器

同一份工作，放在哪里执行

每个圆点是一次实际回调；被浏览器跳过的理论心跳会计入右侧差额。

浏览器主线程最大定时器延迟

—

理想间隔为 20ms；这里只比较两种调度方式。

实际 / 理论心跳

—

setInterval 不会补发漏掉的回调，因此按经过时间计算理论次数。

选择一种执行方式，观察无关心跳是否被拖住。

症状为什么像“模型变慢”

Agent Runtime 天然有很多长连接和长任务：模型流式输出、工具调用、文件处理、后台自动化、频道消息、心跳。它们经常共用一个进程里的 event loop。

当某条链路卡住时，用户看到的是“Agent 没反应”；监控可能看到 WebSocket 心跳超时、首个事件变慢、后台任务堆积。最容易被怀疑的是模型服务，因为模型本来就慢、延迟也有波动。

我当时真正用来区分两者的信号不是单次请求耗时，而是无关请求是否同时停顿：

一个文件操作发生时，原本稳定的心跳一起迟到。
模型替换成固定延迟的 mock 后，停顿仍然存在。
CPU 没有被持续打满，但 event-loop lag 突然抬高。
采样堆栈或代码调用链最终进入同步文件、同步 HTTP 或对象存储 SDK，而不是模型推理。

在没有共享资源耗尽或重试风暴时，一次异步模型等待主要影响等待它的任务；共享 event loop 被阻塞，则会直接暂停该进程里依赖它调度的任务。

`async def` 不会改写调用栈

问题代码通常没有明显的 sleep，看起来甚至很“异步”：

async def persist_artifact(source, store):
    payload = source.read()          # 同步文件 I/O
    digest = sha256(payload).hexdigest()
    store.put_object(digest, payload)  # 同步网络 I/O
    return digest

async def 只说明这个函数可以挂起，并不保证里面的每次调用都会主动让出执行权。只要 read()、哈希、压缩或 SDK 的 HTTP 请求没有真正异步，event loop 就会一直执行到它们返回。

最小改法是把不可替换的同步边界整体移出共享线程：

def persist_artifact_sync(source, store):
    payload = source.read()
    digest = sha256(payload).hexdigest()
    store.put_object(digest, payload)
    return digest

async def persist_artifact(source, store):
    return await asyncio.to_thread(persist_artifact_sync, source, store)

我在生产改造里处理过的不是一个孤立调用点，而是一类边界：对象存储、同步 HTTP 客户端、媒体读写、文件同步和部分 CPU 处理。真正有效的做法，是在 adapter 层把“会阻塞”标出来，而不是在每个上层调用者里凭感觉补线程。

线程卸载还不够

把同步调用交给线程池，只解决“不要冻结 event loop”，没有解决容量。

如果每个 Agent 都能无限提交上传任务，阻塞会从 event loop 转移到线程池队列；内存、文件描述符和远端连接仍会被耗尽。公开示例里我会把适配器写成这样：

class BlockingStoreAdapter:
    def __init__(self, client, max_in_flight: int, on_background_error):
        self.client = client
        self.slots = asyncio.Semaphore(max_in_flight)
        self._running: set[asyncio.Task] = set()
        self._on_background_error = on_background_error

    def _release_slot(self, operation: asyncio.Task) -> None:
        self._running.discard(operation)
        if operation.cancelled():
            self._on_background_error(
                RuntimeError("offloaded operation Task was cancelled")
            )
            return  # fail closed：底层线程是否结束已经不可证明
        self.slots.release()
        if error := operation.exception():
            self._on_background_error(error)

    async def put(self, key: str, payload: bytes) -> None:
        await self.slots.acquire()
        try:
            operation = asyncio.create_task(asyncio.to_thread(
                self.client.put_object,
                key,
                payload,
            ))
        except BaseException:
            self.slots.release()
            raise

        self._running.add(operation)
        operation.add_done_callback(self._release_slot)
        await asyncio.shield(operation)

这里先建立三条边界：

卸载位置固定。 上层只调用异步 adapter，不需要知道底下是同步 SDK。
执行中并发有上限。 在普通调用方取消 put() 的路径里，shield 让槽位继续跟随底层 operation，调用方退出也不会提前放进第二个同步调用。内部 Task 不对外暴露且不得直接 cancel()；adapter 关闭还要先停止新准入，再 shield 并 drain 全部 operation，这段生命周期代码没有塞进示例。这个 Semaphore 只约束单进程 adapter，跨 Worker 需要共享准入机制；等待协程也可能无限增长，生产入口还需要有界队列、排队超时和过载拒绝。
完成语义不被夸大。 await to_thread(...) 只表示同步 SDK 已经返回；是否已持久化服从存储 API 的契约，必要时还要校验或异步对账。

共享 client 还必须明确支持多线程并发。若 SDK 没有这个保证，就要使用每线程或每次调用的 client、受控连接池，或者在 adapter 内串行化访问，不能因为加了 to_thread 就默认它线程安全。

取消也不是免费的。取消等待 to_thread 的协程，并不会强行终止已经开始的同步调用。因此外部写入需要幂等键、超时和事后对账；不能假设协程收到 CancelledError，远端操作就一定没有发生。

我最后看的四类信号

单看接口 P95 很难定位 Agent Runtime。一次任务至少应该拆成：

信号	口径	常见问题
queue	获得执行槽之前的区间	并发准入、线程池或连接池饱和
dispatch	进入协程到第一次真正让出的区间	同步 I/O、CPU 密集处理
external	可重复出现的外部调用子区间	模型、存储、数据库的延迟与超时
first event	从请求开始到首个可见事件的累计结果	上述区间叠加以及缓冲策略

除此之外，我会单独采 event-loop lag，而不是从请求耗时反推。lag 高能证明共享调度器没有按时运行，但不能单独证明原因就是同步 I/O：CPU 或 GIL 占用、GC、宿主调度和容器节流都会抬高它，仍要与调用链、CPU 和系统指标交叉验证。

验证时也不直接上真实模型。先用固定延迟、固定响应的 mock 把模型方差拿掉，再组合短任务、长上传、取消、超时和后台任务。需要同时观察：

event-loop lag 的分位数和最大值；
心跳、首事件与普通 API 是否互相影响；
线程池排队、拒绝和远端错误率；
取消之后是否出现迟到的副作用；
服务恢复后是否能对账出半完成的写入。

多开 Worker 不是修复

增加进程 Worker 能缩小单个 event loop 的故障半径，也能暂时提高吞吐，但它不会让同步调用变异步。流量重新分布、长连接粘在某个 Worker 上，或者每个 Worker 都走到同一段同步代码时，问题仍会回来。

CPU 密集任务也不适合一律丢线程。Python 代码如果长期占用 GIL，应当进入进程池、独立任务服务或原生实现；线程池更适合已有同步 SDK 和会释放 GIL 的阻塞 I/O。选择的依据始终是工作负载，而不是“异步代码里不能出现线程”。

这次复盘真正留下的东西

我负责过这类阻塞链路的定位、同步边界清理和可观测性补齐。最重要的产物不是若干个 to_thread，而是把一条工程规则变成了可检查的契约：

大范围调用点扫描和一部分机械迁移使用了代码 Agent 辅助；我负责定义问题、建立证据链、确定并发边界、审查改动，并设计和验收改造验证。这比把每一行都说成手工完成，更接近这次工作的真实责任。

运行在 Agent event loop 上的路径，不得直接进入未证明非阻塞的文件、网络、数据库或子进程调用。

代码评审可以检查它，监控可以用 lag 发现它，压测可以用无关心跳验证它。这样下一次 SDK、工具或媒体链路进入系统时，不需要等到所有 Agent 一起停顿，才知道共享调度器又被占住了。