Harness 内部任务如何排队而不失控：asyncio.Queue、Producer / Consumer 与 Race

前置：异步任务怎样超时、取消和限流：Cancellation、Timeout 与 Semaphore
核心问题：Producer 比 Consumer 快时，Queue 应怎样反馈容量，关闭阶段又怎样避免丢任务、重复消费或共享状态 race？

先看 Harness 里的真实流量

一个 Run 可能同时产生多种短生命周期工作：

Model stream events ─┐
Tool results ────────┼→ event formatter → SSE writer
Trace spans ─────────┘

Planner fan-out → local Tool workers
File reader → chunk producer → embedding workers

如果 producer 无限 put_nowait()，consumer 稍慢就会把内存变成“隐藏队列”。因此第一原则不是“多开 Task”，而是给 in-flight work 一个明确上限。

Bounded Queue 把压力反馈给 Producer

queue: asyncio.Queue[ToolJob] = asyncio.Queue(maxsize=32)

async def produce(job: ToolJob) -> None:
    await queue.put(job)   # 满了就在这里等待

maxsize 的意义是把消费能力不足变成 producer 可观察的等待，而不是继续占内存。

但要分清三种 Queue：

Runtime Ready Queue     → 哪些 Node 逻辑上可调度（Module 09）
asyncio.Queue           → 单进程 coroutine 之间传递临时工作（本文）
Durable Broker Queue    → 跨进程 delivery/ack/redelivery（Module 14）

它们名字都叫 Queue，可靠性语义完全不同。

Producer / Consumer 需要一起设计 Shutdown

Worker loop 不是只有 get → process：

async def worker(q: asyncio.Queue[ToolJob]) -> None:
    while True:
        job = await q.get()
        try:
            await execute(job)
        finally:
            q.task_done()

真正要定义的是：

• 停止接收新 work 的时点；
• 已在 queue 中的 work 是 drain 还是 cancel；
• consumer 失败时父作用域是否一起取消；
• task_done() 是否与“处理完内存消息”还是“业务完成”对应；
• shutdown deadline 到达后还剩什么。

对于 Harness，业务成功必须来自 Runtime/Tool execution record，不能把 queue.task_done() 当成 Node 已成功。

单线程 asyncio 也会 Race

下面没有多线程，仍然有 race：

current = run.progress
await publish_event(current)
run.progress = current + 1

两个 coroutine 都可能在 await 前读到同一个 current。Race 的本质不是 CPU 是否同时执行，而是一个逻辑 read-modify-write 被多个调度单元交错。

处理顺序通常是：

1. 优先让一个 owner 管理可变状态；
2. 用 message passing 避免共享写；
3. 必须共享时缩小临界区；
4. durable state 用 DB version/CAS，不要指望 asyncio.Lock 跨进程生效。

Harness 中适合用进程内 Queue 的地方

适合：

• Trace/Event exporter 的有界缓冲；
• 单进程 ToolExecutor 的待处理批次；
• SSE event formatting pipeline；
• 本地文件 ingestion pipeline。

不适合：

• “服务重启后必须继续”的 Run Dispatch；
• 需要 ack/redelivery 的 Tool job；
• 多实例之间的 ownership；
• 数小时等待的 approval/timer。

这些进入 Module 09/14 的 durable boundary。

Failure Modes

Unbounded Queue 把过载藏进内存

处理： bounded queue + admission/backpressure；监控 queue depth 和 wait time。

把 asyncio.Queue 当 Durable Queue

处理： 进程内 Queue 只优化并发；重要 work identity 先落 durable state，再 dispatch。

如何验证

场景	必须成立
consumer 人为放慢 10 倍	queue depth 有上限，producer 被反压
worker 抛异常	父 TaskGroup 能观察失败，不留下孤儿 consumer
进程强制退出	内存 Queue 丢失不导致 durable Run 消失

自测

1. Producer 速度持续高于 Consumer，但 Queue 无上限。系统最先出现的风险是吞吐下降、内存膨胀还是数据丢失？为什么？
2. 两个 Consumer 同时更新同一内存状态时，即使都来自同一个 asyncio.Queue，为什么仍可能有 race？

Sources

• Python — asyncio queues
• Python — asyncio synchronization primitives
• Python — Coroutines and Tasks