#14 · Track B · Stage 进程与调度 · 书 p.65

P10 · EngineCore busy loop

每个 stage 进程里，IO 线程把 ZMQ 消息塞进 input_queue，主线程跑 schedule→execute→update；要求空闲时零 CPU、ADD/ABORT 严格 FIFO、信号能叫醒阻塞的 loop、step 异常是致命的。

⏱ 45 min 难度中实现 mine/p10_engine_core_loop.py 验收 ./check.sh p10 提示 A-14 · 参考解 C-14 solutions/p10_engine_core_loop.py 标记完成

题面速览

在系统中的位置：stage 子进程 = 上游 vLLM 的 EngineCoreProc：输入 IO 线程（ZMQ → input_queue）、主 busy loop、输出 IO 线程（output_queue → ZMQ）。vllm-omni 没改 busy loop 本身，但很多正确性依赖它的语义。

while not shutdown:
    if not engine.has_work():  msg = input_queue.get()      # 阻塞：空闲时零 CPU
    抽干 input_queue（FIFO）：ADD / ABORT / UTILITY / WAKEUP
    if shutdown: break
    if engine.has_work():      for out in engine.step(): emit(("output", out))

FIFO：ADD r1 → ABORT r1 → ADD r2 必须按此顺序作用于引擎；被 abort 的请求一个输出都不能有。
可唤醒的关停：signal handler 或其他线程调用 request_shutdown()：设置标志并往队列放 WAKEUP，阻塞在 get() 的 loop 立即醒来退出，发出 ("shutdown",)。
UTILITY：结果按 call_id 返回；utility 抛异常只影响这次调用，引擎继续。
致命错误：step() 抛异常 → 发出 ("fatal", repr(e)) 并退出，由外层进程监控报告为 stage 死亡（P04）。

验收要点：空闲 0.3s：step 0 次、has_work 调用 < 10 次 · ADD 后跑到完成 · ADD→ABORT→ADD 预先排队：引擎调用顺序严格 FIFO、被 abort 的请求零输出 · 另一线程 request_shutdown：0.5s 内退出并只发出 ('shutdown',) · step 抛异常 → ('fatal', ...) 并退出 · utility 结果/异常按 call_id 返回且不杀引擎。

直觉与类比

主厨和订单夹

主厨（busy loop）每做一“轮”菜（step()）就给锅里所有在做的菜各推进一步；服务员（输入 IO 线程）把单子按顺序夹在订单夹上（input_queue）。① 每轮开火前把订单夹上已有的单子全部按顺序处理掉。② 锅里没菜就坐在打单机旁打盹（阻塞 get()），不来回查看（不轮询）。③ 打烊：经理既要挂“打烊”牌（shutdown 标志），还要打一张空白单（WAKEUP），否则打盹的主厨永远看不到牌子。④ 顾客问库存答不上来（utility 抛异常）——告诉这位顾客即可；灶台着火（step 抛异常）——拉警报（fatal）撤离。

退菜单必须排在下一轮开火之前

订单夹上依次是“加宫保鸡丁（ADD q2）”“退掉宫保鸡丁（ABORT q2）”。主厨一轮只看一张单子的话，会先炒一轮宫保鸡丁，下一轮才看到退单——炒出来的东西（一个输出）已经端出去了。先抽干订单夹再开火，被退的菜一口都不会出锅。

框架视角 · T · SBWC · LIFE

	P10
T	一个 stage 进程里：输入 IO 线程、busy loop 线程、输出 IO 线程，外加信号 / 任意线程调用 request_shutdown。没有事件循环：只用 `queue.Queue` 和 `threading.Event`
S	引擎（scheduler + KV + 模型）只由 busy loop 一个线程碰：单写者。跨线程共享的只有 input_queue（自带锁）和 shutdown 标志
B	唯一允许阻塞的地方：没活时的 `input_queue.get()`（不带 timeout）。有活时只用 `get_nowait()` 抽干
W	busy loop：取消息、按顺序作用于引擎、step、emit。IO 线程只搬运。request_shutdown 只设标志 + 放 WAKEUP，不 emit、不 join
C	入：一条 FIFO input_queue（ADD / ABORT / UTILITY / WAKEUP 共用，顺序天然保持）；出：emit。UTILITY 结果靠 call_id 关联（→ P05）
L	启动 → 运行 → 两种结局：shutdown → `("shutdown",)`；step 抛异常 → `("fatal", ...)` 后返回。二者互斥，恰好一个
I	emit 的元组形状是协议，要精确（`("shutdown",)` 是单元素元组）
F	① 每次 step 之前，已入队的消息全部处理完；② 消息按入队顺序作用于引擎；③ 有活 ⇒ 不阻塞，没活 ⇒ 不自旋；④ 看到 shutdown 就不再 step
E	utility 异常 → 只回这一次调用的错误；step / add 异常 → 致命；处理完消息后引擎可能仍没活（只来了 WAKEUP / UTILITY / ABORT）→ 不 step；阻塞 get 拿到的那条也是正常消息，必须处理

易错点

bug	症状
`get(timeout=...)` 轮询	空闲时 has_work 被调用几十上百次/秒（自旋）
只设 flag 不放 WAKEUP	空闲时 request_shutdown 后 loop 永远卡在 get()（或要等 timeout）
阻塞 get 拿到的消息被丢弃	第一条 ADD 丢失，请求永远没有输出
每步只处理一条消息	ABORT 晚于 step 生效，被 abort 的请求仍产出输出
step 异常没捕获	线程静默死亡，外部只看到“没输出了”，没有 fatal
shutdown 后仍 step 一次	关停时多出 output，或在已开始拆除的资源上运行
先把消息分类（先所有 ADD 再所有 ABORT）	打乱 ADD→ABORT→ADD 的语义

逐步图：例 1 的 loop 每一轮

▶ 预排队 ADD a(2), ADD b(1), UTILITY 7 div(1,0), ABORT [b], UTILITY 8 div(6,3)；200ms 后 ADD c(1)；再 200ms 后 request_shutdown

1. 第 1 轮

① has_work = False → 阻塞 get 拿到 ADD a（必须处理！）

② 抽干：ADD b、UTILITY 7、ABORT [b]、UTILITY 8 → emit ("utility", 7, None, "ZeroDivisionError('division by zero')")、("utility", 8, 2.0, None)

③ shutdown? 否 ④ has_work = True → step 1 → ("output", ("a", 1))

b 在被 step 之前就被 abort，一个输出都没有。

2. 第 2 轮

① has_work = True（不阻塞）② 抽干：空 ③ 否 ④ step 2 → ("output", ("a", 0))

3. 第 3 轮

先自己填：has_work？阻塞拿到什么？④ 做什么？

has_work = False，阻塞约 200ms 直到拿到 ADD c；抽干为空；step 3 → ("output", ("c", 0))。

4. 第 4 轮

先自己填

has_work = False，阻塞直到 request_shutdown 放进来的 WAKEUP；shutdown = 是 → break → emit ("shutdown",)。

5. has_work 一共被调用几次？

答案

①处 4 次 + ④处 3 次 = 7 次（按参考骨架的循环结构；你的结构不同次数可能不同，但必须是个位数）。

emit 收到的完整序列：utility 7 错误、utility 8 结果、(a,1)、(a,0)、(c,0)、("shutdown",)。

三级提示

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

例 1 见逐步图（玩具引擎：add(rid, k) 表示还要跑 k 步；step() 让每个请求 k -= 1，返回 [(rid, 剩余k), ...]，k 归零的移除；utility("div", (a, b)) 返回 a / b）。

例 2：step 抛异常是致命的（同样的预排队，引擎第 2 次 step 抛 RuntimeError("NCCL timeout")）：

引擎调用顺序: add a, add b, abort (b,), step 1, step 2
emit: ("utility", 7, None, "ZeroDivisionError('division by zero')")
      ("utility", 8, 2.0, None)
      ("output", ("a", 1))
      ("fatal", "RuntimeError('NCCL timeout')")

发出 fatal 后直接返回：没有 ("shutdown",)，之后 put 进来的 ADD c 永远不会被处理。对比 utility 7 的 ZeroDivisionError：只影响那一次调用。

例 3：空闲 + 从别的线程关停：启动后 sleep 1 秒，期间 step 0 次、has_work 只被调用 1~2 次；另一线程 request_shutdown，t.join(0.5) 立即返回，emitted == [("shutdown",)]。

关键问题

没有工作时，get()、get(timeout=0.01)、while q.empty(): time.sleep(0.001) 哪个满足“BLOCK, no polling”？
阻塞 get() 拿到的消息是“只是用来唤醒”还是必须处理？丢掉的话例 1 少处理哪条？
loop 阻塞在 get() 时，另一个线程怎样让它立刻醒来？只设 flag 够不够？
醒来并抽干之后发现已请求 shutdown，这一步还要 step 吗？
为什么必须“先抽干全部消息再 step”？每步只取一条，例 1 的 b 会怎样？
utility 的异常和 step 的异常，try/except 分别包在哪？哪个之后继续、哪个之后 return？
fatal 退出后还要 emit ("shutdown",) 吗？正常 shutdown 时最后一条是什么？

细节

kind	payload	动作
ADD	`(req_id, data)`	`engine.add(req_id, data)`
ABORT	`[req_ids]`	`engine.abort(req_ids)`
UTILITY	`(call_id, name, args)`	`engine.utility(name, args)`——args 作为一个 tuple 整体传入，不是 `*args`
WAKEUP	无	什么都不做

emit 形状：("output", out)（step 返回列表的每个元素各一次）、("utility", call_id, result, None)、("utility", call_id, None, repr(exc))、("shutdown",)、("fatal", repr(exc))。
抽干用 get_nowait() 循环直到 queue.Empty。
request_shutdown 要简单不阻塞：设 threading.Event，再 put_nowait 一条 WAKEUP。

L2思路与常见坑思路拆分与常见坑（中度剧透）

状态只多一个 self._shutdown = threading.Event()。
拆一个 _handle(msg) 按 kind 分派；UTILITY 分支内部自己 try/except 并 emit——“阻塞拿到的第一条”和“抽干的其余几条”走同一个函数。
循环结构：while not shutdown: ① 若 not has_work()：阻塞 get 一条并处理；② get_nowait 抽干并处理；③ 若 shutdown → break；④ 若 has_work：step 并逐个 emit。
④ 为什么还要再问 has_work：①② 处理的可能只是 ABORT/UTILITY/WAKEUP，引擎仍没有工作，step 没意义（某些引擎会断言）。
异常边界：整个 while 包在 try: ... except Exception as e: emit(("fatal", repr(e))); return；正常退出循环后 emit ("shutdown",)。utility 的异常在 _handle 里就被吃掉。

L3详解：答案详解：答案、逐步追踪、骨架提示（重度剧透）

关键问题答案

只有不带 timeout 的 get()；另外两种都是轮询。
必须处理——它就是队列里最早的那条；丢掉的话 ADD a 就没了。
往队列里放一条消息（WAKEUP）；只设 flag 不够，阻塞中的 get() 看不到 flag。
不要 step，直接 break 然后 emit shutdown。
只取一条的话：第一步只处理 ADD a 就 step，b 还没加进来；之后 ABORT 可能晚于 b 的某次 step，导致 b 产生输出。
utility：包在单次调用周围，emit 错误后继续。step：包在整个 loop 外面，emit fatal 后 return。
不要。正常 shutdown 的最后一条是 ("shutdown",)。

自检：① 空闲 1 秒，用计数器包装 has_work，看调用次数是否只有个位数；② 另一线程调 request_shutdown，计时 join；③ 构造 ADD→ABORT→ADD 预排队，记录引擎调用顺序；④ 让 step 抛异常，确认最后一条是 fatal 且线程已退出。

面试官追问 & 对照源码

vllm-omni 为什么要在 ABORT 走完 FIFO 之后才释放 overflowed 请求？
max_concurrent_batches > 1 时 step_with_batch_queue 在做什么？

对照源码：engine/stage_engine_core_proc.py:133-177 · :147-154 · :311-345 · vllm/v1/engine/core.py