P02 ↔ vllm-omni 源码对照

一句话结论与拓扑差别

练习：mine/p02_orchestrator.py · 源码：vllm-project/vllm-omni @ fa8b419 · vllm_omni/engine/orchestrator.py（行号固定在该 commit）。

主线程（serving loop）              "orchestrator" 线程 · 自己的事件循环              stage 进程 × N（可多 replica）
add_request() ─sync_q.put─► janus request_queue ─► _request_handler ─submit_initial─► StagePool / client
                                                     request_states                       │ get_output
output handler ◄─call_soon_threadsafe─ OutputQueueSink ◄─ dispatcher loop ◄─ ready_q ◄─ reader(s,r) × k
                                                     forward: process_engine_inputs → submit ─► EngineCore (P10)

逐项对照表

建议阅读路线（约 300 行）

① 线程与事件循环的边界（T）

主线程（HTTP / serving loop）调用 add_request()，把消息 sync_q.put 进 janus 队列；一个名叫 "orchestrator" 的后台线程里 new_event_loop()，在那里跑 orchestrator.run()。这是 P02 替你省掉的一层：P02 的调用方和 orchestrator 在同一个事件循环里。

# omni_engine_base.py:303-330 · janus 队列 + orchestrator 线程
self.request_queue: janus.Queue[EngineQueueMessage] = janus.Queue(maxsize=_REQUEST_QUEUE_MAXSIZE)
self.output_queue: janus.Queue[EngineQueueMessage] = janus.Queue()
self._output_sink = OutputQueueSink(self.output_queue.sync_q)
self.rpc_output_queue: janus.Queue[EngineQueueMessage] = janus.Queue()
...
self.orchestrator_thread = threading.Thread(
    target=self._bootstrap_orchestrator,
    args=(stage_init_timeout, startup_future),
    daemon=True,
    name="orchestrator",

# omni_engine_base.py:440-452 · 线程里新建事件循环
def _bootstrap_orchestrator(self, stage_init_timeout, startup_future) -> None:
    """Create loop, initialize stages, then run Orchestrator."""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

② 先注册，再 submit

在 _handle_add_request 里找这两句的先后：注册 request_states 发生在任何 await 之前。

# orchestrator.py:3107-3138
req_state = OrchestratorRequestState(request_id=request_id, prompt=original_prompt, ...)
self.request_states[request_id] = req_state                     # 3117：注册
self._maybe_attach_native_kv_transfer_params(req_state, prompt)
...
if not await self._dispatch_or_fail_request(                     # 3128：第一个 await
    lambda: self.stage_pools[stage_id].submit_initial(request_id, req_state, prompt, ...),
    req_id=request_id, stage_id=stage_id, operation="add_request",
):

③ reader：只搬运，连错误也只是搬运

注意 except BaseException：reader 自己出错时不改任何状态，而是把错误包装成一条消息交给 dispatcher——单写者原则的延续。

# orchestrator.py:1117-1136
async def _llm_replica_reader(stage_id: int, replica_id: int, client: Any) -> None:
    try:
        pool = self.stage_pools[stage_id]
        while not self._shutdown_event.is_set():
            raw_outputs = await pool.read_llm_raw_output(client)
            if raw_outputs is None:
                await asyncio.sleep(0.001)
                continue
            await ready_q.put(("llm", stage_id, replica_id, raw_outputs))
            self._orch_monitor.set_dispatch_queue_size(ready_q.qsize())
    except asyncio.CancelledError:
        raise
    except BaseException as e:  # noqa: BLE001 - routed to the dispatcher
        await ready_q.put(("error", stage_id, replica_id, e))

④ dispatcher：串行消费 ready_q

真实代码同时等三件事：新输出、关闭信号、replica 变化，所以用 asyncio.wait(FIRST_COMPLETED)，每轮重建 pending_get。

# orchestrator.py:1264-1272, 1291-1293
while not self._shutdown_event.is_set():
    if pending_get is None:
        pending_get = asyncio.create_task(ready_q.get(), name="orch-dispatch-get")
    done, _ = await asyncio.wait(
        {pending_get, shutdown_task, reconcile_task},
        timeout=max(0.0, next_reconcile - _time.monotonic()),
        return_when=asyncio.FIRST_COMPLETED,
    )
    if shutdown_task in done:
        ...
    kind, stage_id, replica_id, payload = pending_get.result()
    pending_get = None

⑤ _route_output：P02 _handle_output 的真身

顺序：标记 finished → 计算 request_finished → final stage 就发 OutputMessage → 若 finished 且 stage_id < final_stage_id 就转发 → request_finished 则清理。

# orchestrator.py:1933-1967 · 终态判断 + 发给用户
if finished and not segment_finished:
    req_state.finished_stage_ids.add(stage_id)
request_finished = False
if finished and self.stage_pools[stage_id].final_output and not segment_finished:
    req_state.finished_final_output_stage_ids.add(stage_id)
    final_output_stage_ids = req_state.final_output_stage_ids or {req_state.final_stage_id}
    request_finished = final_output_stage_ids.issubset(req_state.finished_final_output_stage_ids)
...
elif self.stage_pools[stage_id].final_output:
    message = OutputMessage(request_id=req_id, stage_id=stage_id, ..., finished=request_finished)
    if (request_finished and self.async_chunk and not req_state.streaming.enabled
            and not set(req_state.stage_submit_ts).issubset(req_state.finished_stage_ids)):
        # Direct chunk delivery can finish the decoder before the upstream engine's
        # terminal output/metrics reaches this loop. Keep request state until routed.
        req_state.pending_final_output = message
        request_finished = False
    else:
        await self.output_async_queue.put(message)

# orchestrator.py:1988-1993 · 转发条件
if ((finished or segment_finished)
        and stage_id < req_state.final_stage_id
        and not req_state.skip_legacy_stage_forward
        and (not self.async_chunk or not self._stage_receives_async_chunks(stage_id + 1))
        and (not self._next_stage_already_submitted(stage_id, req_state) or req_state.streaming.enabled)

# orchestrator.py:2036-2043 · pending_final 放行 + 清理
pending = req_state.pending_final_output
if pending is not None and set(req_state.stage_submit_ts).issubset(req_state.finished_stage_ids):
    req_state.pending_final_output = None
    await self.output_async_queue.put(pending)
    request_finished = True
if request_finished and not req_state.session_owned:
    await self._cleanup_request_ids([req_id, *self._cfg_tracker.cleanup_parent(req_id)])

⑥ run() 的 finally：cancel + gather

外加最后一遍兜底：扫 asyncio.all_tasks(loop)，把漏网的 task 也取消并等待。

# orchestrator.py:530-555
finally:
    self._shutdown_event.set()
    for task in tasks:
        if not task.done():
            task.cancel()
    try:
        await asyncio.gather(*tasks, return_exceptions=True)
    except Exception:
        pass
    await self._shutdown_extensions()
    ...
    self._shutdown_stages()
    loop = asyncio.get_running_loop()
    pending = [t for t in asyncio.all_tasks(loop) if t is not asyncio.current_task() and not t.done()]
    for task in pending:
        task.cancel()
    if pending:
        await asyncio.gather(*pending, return_exceptions=True)

真实代码比 P02 多出来的东西

读完之后想一想

1. 真实 orchestrator 在另一个线程里。主线程调用 add_request 时，为什么必须走 janus 的 sync_q.put，而不能直接调用 orchestrator 循环里某个 asyncio.Queue.put_nowait？
2. _route_output 里，发给用户的 OutputMessage 是在转发给下一个 stage 之前放进队列的。如果反过来，先 await _forward_to_next_stage（一次跨进程通信）再发给用户，用户体感上会有什么差别？
3. reader 出错时为什么不直接调用 _handle_dead_replica，而要把错误包成消息交给 dispatcher？