#9 · P02 · 交互时序图

P02 Orchestrator 时序图

一步一步看例 1（S0→S1→S2，final={0,2}）里每个 worker 在做什么、消息走哪条通道、账本怎么变——再对比两个经典反例：先 submit 后注册、dispatcher 自己轮流等。

架构图（SBWC 填完之后）

橙色虚线 = 向 stage 提交（submit）；灰色实线 = 队列 put / get。只有 dispatcher 和（注册时的）request_handler 会碰账本 request_states。点下面“下一步”，图上正在动的部分会高亮。

▶ 例 1：add_request("r1", "hi there") 的一生

1. 用户调用 add_request

add_request 只做一件事：await _req_q.put(("add", "r1", "hi there"))。它不碰账本、不 submit——公开 API 要薄（I）。

2. handler 先注册，再 submit S0

_handle_add：request_states["r1"] = ReqState(...)、submitted = {0} ——全是同步代码，没有 await，不会被打断；然后才 await S0.submit("r1", "hi there")。

从这一刻起，无论 S0 多快吐输出、dispatcher 何时运行，都能查到 r1。

3. S0 吐出 "hi"，reader 0 搬运

reader 0 阻塞在 S0.get_output()，拿到 [StageOutput("r1","hi",False)]，打上 sid 放进 _ready_q：(0, [..])。reader 不看内容、不碰账本。

4. dispatcher 处理 "hi"

stage_data = {0: ['hi']}；0 ∈ final → done = {0,2} ⊆ {} 为 False → put OrchOutput("r1", 0, "hi", False)。S0 未完成，不转发。

5. 用户拿到第一条

next_output() = wait_for(_out_q.get(), timeout)：(0, 'hi', False)。

6. S0 吐出 "there"(fin) → 发给用户 + 转发 S1

finished = {0}，{0,2} ⊄ {0} → 发 (0, 'there', False)。

同一条输出又满足 “o.finished 且 0 < last_stage” → forward_fn(1, prompt, ['hi','there'])，先 submitted.add(1) 再 await S1.submit(...)。两件事是两个独立的 if。

7. S1 完成 → 只转发，不给用户

S1 ∉ final：不发给用户。1 完成且 1 < 2 → forward_fn(2, …, [['HI','THERE']]) → submit S2。

8. S2 吐出 "wav:HI"

发 (2, 'wav:HI', False)。

9. S2 吐出 "wav:THERE"(fin) → 终态 + 清理

finished = {0,1,2} ⊇ {0,2} → 发 (2, 'wav:THERE', True) → _cleanup("r1") → return。账本里 r1 没了。

10. 迟到的 "wav:LATE"

dispatcher 查不到 r1 → dropped_outputs += 1 → return。绝不复活状态。

反例 1：先 submit 后注册

顺序	时间线	结果
❌ 先 submit 后注册	`await submit()` 交出控制权 → S0 很快吐出 "hi" → reader 搬进 `_ready_q` → dispatcher 查 request_states：查不到 r1 → 当成未知请求丢弃 → 回到 handler，这时才注册	用户永远收不到 "hi"，`dropped_outputs = 1`
✅ 先注册后 submit	注册（无 await，不会被打断）→ `await submit()` → 之后无论谁先跑，dispatcher 都能查到 r1	正确

反例 2：dispatcher 自己轮流等（违反 B 原则）

while True:
    for sid, client in enumerate(self.stages):
        outs = await client.get_output()     # ← 一次只能等一个
        ...

时刻	dispatcher 在做什么	S0	S1
t1	`await S0.get_output()`，拿到 "hi"	已吐出 "hi"	还没有任务
t2	`await S1.get_output()`	"there" 已吐出，没人读	没有任务，永远不会有输出
t3…	一直卡在 S1	还在等	S1 的任务要等 S0 的 "there" 被处理后才会提交，可 "there" 没人读

结果是死锁：dispatcher 等 S1，S1 等 dispatcher 转发，而转发需要先读到 S0 的 "there"。规则：每个独立的阻塞源配一个专门守着它的 worker。

生命周期（L）

async def start(self):
    for sid, client in enumerate(self.stages):          # 每个阻塞源一个 worker（B → W）
        self._tasks.append(asyncio.create_task(self._reader(sid, client)))
    self._tasks.append(asyncio.create_task(self._request_handler()))
    self._tasks.append(asyncio.create_task(self._dispatcher()))

async def shutdown(self):
    for t in self._tasks:
        t.cancel()                                       # 在各自的 await 处抛 CancelledError
    await asyncio.gather(*self._tasks, return_exceptions=True)   # 等它们真正结束

常见坑：① 不保存 task 引用 → 可能被 GC，shutdown 也找不到；② 只 cancel 不 gather → “Task was destroyed but it is pending”；③ start() 里 await 了永远不返回的东西。