nano-vllm-omni · 提示 ← P02P01 →
#10 · Track A · 前端与 Orchestrator · 书 p.51

P03 · Orchestrator v2：async_chunk 预热 + 终态守门

开启 async_chunk 后 stage 之间走 connector 直传，orchestrator 不再转发数据；但控制面和数据面是两条路，完成信号可能乱序——音频 stage 先说“我完了”，talker 还没报告完成。

⏱ 45 min 难度 中 实现 mine/p03_async_chunk_orchestrator.py 验收 ./check.sh p03 提示 A-10 · 参考解 C-10 solutions/p03_async_chunk_orchestrator.py

题面速览

时间 →
stage2 (audio, final):  ... wav5(fin)           ← 控制面上先到
stage1 (talker):        ............... (fin)   ← 晚到（它的 IPC 慢了一拍）
如果在 wav5(fin) 时就发 finished 并清理：
 - 清理会回收 stage1 仍在使用的资源（SHM、绑定）
 - stage1 的完成消息成了“未知请求”的输出

验收要点：每个 stage 只被提交一次、下游拿到 prewarm 占位输入 · 非 final 的 stage1 迟 200ms 才报告完成时终态必须等它 · 20 个请求 + 迟到完成：每个请求恰好一个终态且是最后一条 · 清理后到达的输出被丢弃、状态不复活。

直觉与类比

车间主任和传送带

P02：每道工序做完一整批零件交给车间主任，主任搬到下一道工序。P03：工序之间装了传送带（connector），上一道每做好一个零件就直接放上去——流水线并行起来。车间主任不再搬零件，只剩两件事：① 开工时通知所有工序“有新订单，准备接传送带”（prewarm）；② 收工时确认每一道工序都停机了，才给客户签收单（终态守门）。

快递和签收短信

货物走物流（数据面），“已发货/已完成”走短信（控制面）。最后一站的签收短信可能比中转站的“已出库”短信先到。你一看到“已签收”就把订单归档删除，几秒后中转站的短信到了你却找不到这张订单；更糟的是归档时你可能已经把中转站还在用的托盘（共享内存、绑定）收回了。

框架视角 · 只有两处变了

和 P02 相比
T 不变：同一个事件循环里的 task。新增的 connector 是 stage 进程之间的边界，orchestrator 不碰它
S ReqState 多用一个字段 pending_final；submitted 的含义变成“预热过的所有 stage”；stage_data 不再需要
B / W / C 完全不变：reader × N、request_handler、dispatcher、三个队列——所以 P03 能直接继承 P02
L / I 不变
F 变了：add 时一次性标记 submitted；终态条件多了 submitted ⊆ finished；非 final stage 的完成事件也可能触发发送（放行 pending_final）
E 清理后再到的输出照样丢弃计数，绝不能因为 S1 的迟到完成消息而重新建出状态

易错点

逐步图：例 1（talker 的完成报告迟到）

▶ final={0,2}，prewarm_fn = lambda sid, p: f"warm{sid}:{len(p)}"，add_request("r1", "hi there")
1. add：一次性标记、依次 submit

注册 → submitted = {0,1,2}（在第一次 await 之前）→ S0.submit("r1", "hi there") → S1.submit("r1", "warm1:8") → S2.submit("r1", "warm2:8")。每个 stage 只 submit 一次。

2. S0 ("hi", False)

finished = {}；all_final = F；all_done = F → 发 (0,'hi',F)。

3. S0 ("there", True)

finished = {0}；all_final = F → 发 (0,'there',F)。不转发。

4. S2 ("wav:HI", False)

finished = {0} → 发 (2,'wav:HI',F)。

5. S2 ("wav:THERE", True)：扣下终态

finished = {0,2}；all_final = T；all_done = {0,1,2} ⊆ {0,2} = F

→ 构造 (2,'wav:THERE',T) 存进 pending_final，什么都不发，return。

6. S1 (None, True)：最后一块拼图

finished = {0,1,2}；S1 不是 final，但 all_done 成立且 pending 不为空 → 发出 (2,'wav:THERE',T)，清理。

7. S2 ("wav:EXTRA", False)

状态已清理 → dropped_outputs == 1，不复活。

三级提示

L1I/O 示例与关键问题I/O 示例 + 关键问题 + 细节（轻度剧透）

例 1 见上方逐步图。

例 2：正常顺序（S1 在 S2 之前完成）：S0 hi → (0,'hi',F)；S0 there(T) → (0,'there',F)；S1 tok(T) → 无；S2 wav:HI → (2,'wav:HI',F)；S2 wav:THERE(T) → (2,'wav:THERE',T)，随即清理。不需要 pending_final：最后一个 final stage 完成的那一刻所有已提交的 stage 也都完成了。

例 3：最后完成的 final stage 是 S0：

到达 用户收到 finished
S0 ("hi", False) (0,'hi',False) {}
S2 ("wav:HI", True) (2,'wav:HI',False) {2}
S1 (None, True) （无） {1,2}
S0 ("there", True) (0,'there',True)，清理 —

终态消息的 stage_id 不一定是 last_stage——它是“让条件第一次成立的那条输出”所在的 stage。

关键问题

  1. P02 里 submitted 是随转发一点点增长的；P03 里 add 时就要包含 0..last。沿用 P02 只标 {0}，例 1 在哪一步出错、错成什么样？在第一次 await 之前标完有什么好处？
  2. 例 1 第 4 行：最后一个 final stage 完成但 S1 还没完成。用户应该收到什么？要存什么、存在哪？
  3. 非 final stage 报告完成时 handler 要做什么？什么条件下什么都不用做？
  4. “所有 final stage 都完成”和“所有已提交的 stage 都完成”分别写出来。各自控制什么？
  5. 非 final stage 的 data 会出现在用户侧吗？stage_data 在本题还有用吗？
  6. 继承 P02 时哪些可以原封不动复用？
  7. pending_final 已存好之后，又来一条非 finished 的 S1 输出，怎么处理？

细节

  • 构造函数里 forward_fn=lambda *_: None 只是为了满足父类签名。
  • 提交顺序：先 S0（真实 prompt），再 S1、S2……（占位），范围 0..last_stage。
  • 两道门：OrchOutput.finished 的取值仍是 P02 的“所有 final stage 都完成”；但 finished=True 的消息何时发出多了一个条件 submitted ⊆ finished。
  • pending_final 存的是已构造好的 finished=True 的 OrchOutput，之后原样发出。
  • 如果你的 P02 把这两步写死在 dispatcher 里而不是独立方法，先回去重构 P02。
L2思路与常见坑思路拆分与常见坑（中度剧透）
  • add 路径：建 ReqState → 注册 → submitted = {0..last}（全部同步）→ await S0.submit(prompt) → 依次 await S_k.submit(prewarm_fn(k, prompt))。
  • 输出路径决策表（先查状态、未知就 drop；若 finished 就加入 st.finished）：
这条输出来自 all_final（final ⊆ finished） all_done（submitted ⊆ finished） 动作
final stage 否 — 发 OrchOutput(..., finished=False)
final stage 是 是 发 finished=True 的消息，清理
final stage 是 否 构造 finished=True 的消息，存进 pending_final，不发
非 final stage — 是，且有 pending_final 发 pending_final，清理
非 final stage — 其他 什么都不做
  • submitted 是守门的依据：少一个 stage，门就少一道。在第一次 await 之前一次性标完，和 P02“先注册再 submit”是同一个原则。
L3详解：答案详解：答案、逐步追踪、骨架提示（重度剧透）
关键问题答案
  1. 只标 {0} 时，S2 的 "wav:THERE" 到达那一刻 finished={0,2}，{0} ⊆ {0,2} 成立，终态立即发出并清理——正是本题要防止的“提前终态”；随后 S1 的完成报告变成未知 id，dropped_outputs == 1。在第一次 await 前标完，则无论哪个 await 点插进来的输出，看到的都是完整的 submitted。
  2. 什么都不发。构造 finished=True 的 OrchOutput，存进 st.pending_final，直接 return。
  3. 把 sid 加进 finished 后检查 submitted ⊆ finished 且 pending_final is not None：都成立就发出 pending 并清理。pending 为空（还有 final stage 没完成，例 2 的 S1）时什么都不做。
  4. all_final = final_stage_ids ⊆ finished：决定 OrchOutput.finished 的值。all_done = submitted ⊆ finished：决定 finished=True 的消息立即发出（并清理）还是先扣在 pending_final。
  5. 不会；非 final stage 的输出只用来更新 finished。stage_data 不需要。
  6. start / shutdown / add_request / next_output / reader / dispatcher / request handler / _cleanup 全部复用；只换 _handle_add 和 _handle_output。
  7. 它不是 finished，不改变任何集合，落入“什么都不做”。

例 1 状态演化（✓ = finished=True）：

事件 submitted finished all_final all_done pending 发给用户
add {0,1,2} {} None
S0 hi {0,1,2} {} F F None (0,hi,F)
S0 there ✓ {0,1,2} {0} F F None (0,there,F)
S2 wav:HI {0,1,2} {0} F F None (2,wav:HI,F)
S2 wav:THERE ✓ {0,1,2} {0,2} T F (2,wav:THERE,T) —
S1 ✓ {0,1,2} {0,1,2} T T → 发出 (2,wav:THERE,T)，清理
S2 wav:EXTRA — — dropped += 1

骨架注意：_cleanup 在 await self._out_q.put 之后——无界队列的 put 不会真正挂起，即使挂起，状态也已一致（pending 已取出）。

面试官追问 & 对照源码

对照源码：engine/orchestrator.py:1954-1967 · :2036-2040 · :2711-2870