vLLM Code Reading (Section 4): EngineCore 到 Worker 的执行链路

EngineCore 到 Worker 是 vLLM 推理控制面的核心路径。请求进入 EngineCore 以后，会先经过 scheduler 和 KV cache manager 做调度与逻辑 block 分配，再由 executor 把本轮 SchedulerOutput 广播给 worker，最终触发 GPU 上的模型执行。

这条链路的关键结论是：EngineCore 是调度和控制中心，Worker 是执行模型分片的地方，二者之间采用“广播输入、按需单播输出”的控制面通信模型。 EngineCore 统一管理 request 状态和逻辑 KV block，Worker 只根据 EngineCore 下发的 SchedulerOutput 更新自己的本地状态，并执行本 rank 负责的那部分模型计算。

1. EngineCore 不是模型执行者

理解这条链路的第一步，是把 EngineCore 和 Worker 的职责分开。EngineCore 内部包含 scheduler、KV cache manager 和 executor，但它本身不直接跑 Transformer forward；真正执行模型的是 executor 管理的 worker 进程或 Ray actor。

从请求生命周期看，这条路径可以压缩成下面这条线：

flowchart TD
    A["EngineCore busy loop"] --> B["_process_input_queue"]
    B --> C["scheduler.add_request"]
    C --> D["EngineCore.step"]
    D --> E["scheduler.schedule"]
    E --> F["KVCacheManager.allocate_slots"]
    F --> G["executor.execute_model"]
    G --> H["collective_rpc: broadcast SchedulerOutput"]
    H --> I["Worker busy loop"]
    I --> J["WorkerWrapperBase.execute_model"]
    J --> K["GPUWorker.execute_model"]
    K --> L["GPUModelRunner.execute_model"]
    L --> M["Model forward / sample"]
    M --> N["response_mq"]
    N --> O["scheduler.update_from_output"]

这张图里的重点是：scheduler.schedule() 之前发生的是请求状态和资源决策，executor.execute_model() 之后发生的是多进程或多 actor 的执行控制。EngineCore 把一次 step 需要执行的工作打包成 SchedulerOutput，然后让所有 worker 按同一份调度结果推进。

2. EngineCore 的 busy loop：先收输入，再执行 step

EngineCore 的主循环可以理解成一个高频事件循环。它不断从输入队列中取新请求、abort 请求和 utility command，只要系统里还有活跃请求，就不断调用 step() 推进模型执行。

核心结构是：

EngineCore busy loop
  -> _process_input_queue
  -> scheduler.add_request
  -> step
     -> scheduler.schedule
     -> executor.execute_model
     -> scheduler.update_from_output
  -> output_queue

这里最重要的设计是输入处理和模型执行没有被粗暴地拆成两个完全独立阶段。_process_input_queue 会在合适时机 drain 输入队列，把新请求及时放入 scheduler；而 step() 则负责根据当前所有 waiting/running 请求做一次调度和执行。

_process_input_queue 看起来复杂，是因为它需要兼顾吞吐和延迟。系统空闲时，它可以阻塞等待新请求；系统已有工作时，它不能长期阻塞在输入队列上，而是要尽快回到 step()，否则 GPU 会被控制面拖慢。

3. Scheduler 和 KVCacheManager：逻辑 block 由 EngineCore 统一管理

Scheduler 的核心职责是决定“这一轮哪些 request 执行多少 token”。它不是简单地把 request 放进 batch，而是要综合 token budget、KV cache 空间、prefix cache、preemption、LoRA、encoder budget、chunked prefill 等条件，最终生成一个 SchedulerOutput。

KV block 的管理也发生在 EngineCore 这一侧。Worker 会在初始化时 profile 可用显存，根据模型的 KV cache spec 计算出每个 rank 能承载多少 block；但真正的逻辑 block 分配由 EngineCore 里的 KVCacheManager 负责。这样做的好处是所有 worker 都按照一致的 block id 理解 KV cache，EngineCore 可以用一个逻辑 block 池管理请求生命周期。

KV block 的顶层模型可以画成：

flowchart LR
    subgraph Init["初始化阶段"]
        P["Worker profile memory"]
        S["get_kv_cache_spec"]
        C["compute num blocks"]
        I["initialize_from_config"]
    end

    subgraph Core["EngineCore"]
        KM["KVCacheManager"]
        Pool["logical block pool"]
        Sch["Scheduler"]
    end

    subgraph Worker["Each Worker"]
        Local["local KV cache tensors"]
        Map["block_id -> local cache slot"]
    end

    P --> S --> C --> I
    I --> Local
    Sch --> KM
    KM --> Pool
    Pool -- "block ids in SchedulerOutput" --> Map

这里需要特别注意“逻辑 block”和“物理 GPU 内存”的区别。EngineCore 管的是逻辑 block id 和 request 到 block 的映射；Worker 侧根据同一份 block id，在自己的本地 KV cache tensor 中读写对应位置。

4. RayExecutorV2：Ray 管 actor，vLLM 管每 step 的通信

在 Ray 后端的 V2 路径里，executor 使用的是 RayExecutorV2。它继承自 MultiprocExecutor，因此 execute_model() 和 collective_rpc() 的主体逻辑仍然复用 MultiprocExecutor 的实现。

RayExecutorV2 的特殊之处在初始化阶段。它用 Ray placement group 把每个 rank 放到对应节点和 GPU 上，然后创建 RayWorkerProc actor。actor 初始化好以后，会调用 run.remote() 进入一个长期运行的 busy loop。

初始化链路可以概括为：

sequenceDiagram
    participant EC as EngineCore / RayExecutorV2
    participant Ray as Ray runtime
    participant A as RayWorkerProc actor
    participant W as WorkerWrapperBase / GPUWorker

    EC->>Ray: initialize_ray_cluster / placement group
    EC->>Ray: ray.remote(RayWorkerProc).remote(...)
    Ray->>A: create actor on assigned node/GPU
    EC->>A: get_node_and_physical_gpu_ids.remote()
    EC->>A: initialize_worker.remote(...)
    A->>W: WorkerProc.__init__
    W->>W: init_worker / init_device / load_model
    EC->>A: wait_for_init.remote()
    A-->>EC: response_mq handle
    EC->>A: run.remote()
    A->>A: worker_busy_loop()

这说明 Ray 不是每个 step 都负责执行模型 RPC。Ray 主要负责 actor 的资源放置、生命周期和故障监控；真正高频的 step 级命令，是 vLLM 通过自己的 MessageQueue 广播给所有 worker 的。

5. 控制面通信：广播 input，单播 output

EngineCore 到 Worker 的通信模式可以总结成一句话：输入命令广播给所有 rank，输出结果只从需要返回的 rank 读取。 这个设计很适合 TP/PP 场景，因为所有 rank 都必须收到相同的调度信息，但最终 token 输出通常只需要最后一个 PP stage 的 TP rank 0 返回。

下面这个 SVG 图把这套通信关系画得更直接。假设 world size 为 4，rpc_broadcast_mq 是一条一对多输入通道，而 response_mq 是每个 worker 各有一条的一对一返回通道。

collective_rpc("execute_model", ...) 发送的不是函数地址，而是方法名和参数。普通推理路径里，method 是字符串 "execute_model"；worker 收到后在本地执行 getattr(self.worker, "execute_model")。因此这个 RPC 更像命令分发，不是把本机函数指针传到远端。

6. Worker busy loop：actor 长驻，循环取命令

Ray actor 的 run() 进入后基本不会返回。它先完成 message queue 的 ready handshake，然后进入 worker_busy_loop()，不断从 rpc_broadcast_mq 里取命令。

worker 侧循环可以抽象为：

while True:
    method, args, kwargs, output_rank = rpc_broadcast_mq.dequeue(indefinite=True)

    if isinstance(method, str):
        func = getattr(self.worker, method)
    else:
        func = partial(cloudpickle.loads(method), self.worker)

    output = func(*args, **kwargs)

    if output_rank is None or self.rank == output_rank:
        worker_response_mq.enqueue(output)

这段代码说明 Worker busy loop 的职责非常单纯：它不做调度决策，也不管理全局请求队列，只负责执行 EngineCore 广播下来的命令。调度决策已经在 SchedulerOutput 中完成，Worker 只是按这份调度结果更新本地 batch 状态并执行模型。

7. WorkerWrapperBase：RPC facade，不是计算核心

WorkerWrapperBase 是 Worker 侧最容易误解的对象。它不是 GPU worker 本身，而是 executor RPC 层看到的 worker facade，负责初始化真实 worker、维护生命周期、处理多模态缓存，并把方法转发给真实的 GPUWorker。

对象关系是：

RayWorkerProc actor
  └── self.worker = WorkerWrapperBase
        └── self.worker = GPUWorker
              └── model_runner = GPUModelRunner

WorkerWrapperBase.execute_model() 很薄：

def execute_model(self, scheduler_output):
    self._apply_mm_cache(scheduler_output)
    return self.worker.execute_model(scheduler_output)

所以真正开始推理的路径是：

flowchart TD
    A["worker_busy_loop receives method='execute_model'"] --> B["WorkerWrapperBase.execute_model"]
    B --> C["_apply_mm_cache"]
    C --> D["GPUWorker.execute_model"]
    D --> E["PP receive if not first stage"]
    E --> F["GPUModelRunner.execute_model"]
    F --> G["_update_states"]
    G --> H["_prepare_inputs / _preprocess"]
    H --> I["set_forward_context"]
    I --> J["self.model(...)"]
    J --> K["sample / return IntermediateTensors"]
    K --> L["PP send if not last stage"]

这里的总分关系很清楚：Wrapper 负责进入真实 worker，GPUWorker 负责 PP 通信和调用 model runner，GPUModelRunner 负责构造输入、准备 attention metadata、执行 forward 和后处理。

8. SchedulerOutput：新请求广播，后续 step 发 diff

SchedulerOutput 是 EngineCore 和 Worker 之间最重要的数据结构。它包含本 step 新增请求、已缓存请求的增量信息、每个请求调度多少 token、分配到的 block id、需要释放的 request、KV connector metadata 等内容。

新请求的数据在 scheduled_new_reqs 里，其中 NewRequestData 会包含 prompt_token_ids、sampling_params、block_ids、num_computed_tokens 等字段。Worker 第一次收到新请求后，会把这些信息写入本地 CachedRequestState，后续 step 不需要反复发送完整 prompt，只需要在 scheduled_cached_reqs 里发送必要 diff。

这个设计可以理解为：

flowchart LR
    A["first scheduled step"] --> B["scheduled_new_reqs"]
    B --> C["prompt_token_ids / block_ids / params"]
    C --> D["Worker local CachedRequestState"]
    D --> E["later steps"]
    E --> F["scheduled_cached_reqs"]
    F --> G["new block ids / num_computed_tokens / output token diff"]

这也是为什么 SchedulerOutput 要广播给所有 worker。每个 rank 都需要维护一致的 request 状态和 block table，即使某些 rank 在 PP 中并不直接消费原始 token id，它仍然需要知道当前 request 的序列长度、positions、KV block 映射和 finished/preempted 状态。

9. PP=2、TP=8：prompt token 广播给谁，谁真正消费

对于 PP=2, TP=8，一个 EngineCore 下面不是两个 worker，而是 16 个 worker。每个 PP stage 内有 8 个 TP rank，总 world size 是 2 * 8 = 16。

可以画成：

flowchart TB
    subgraph Core["EngineCore / RayExecutorV2"]
        S["SchedulerOutput"]
    end

    subgraph PP0["PP stage 0"]
        W00["rank 0 / TP0"]
        W01["rank 1 / TP1"]
        W07["rank 7 / TP7"]
    end

    subgraph PP1["PP stage 1"]
        W10["rank 8 / TP0"]
        W11["rank 9 / TP1"]
        W17["rank 15 / TP7"]
    end

    S --> W00
    S --> W01
    S --> W07
    S --> W10
    S --> W11
    S --> W17

    W00 -- "hidden states" --> W10
    W01 -- "hidden states" --> W11
    W07 -- "hidden states" --> W17

    W10 --> O["ModelRunnerOutput from output_rank"]

从控制面看，SchedulerOutput 会广播到所有 16 个 worker，因此新请求里的 prompt_token_ids 也会随 scheduled_new_reqs 被所有 worker 收到。从计算面看，原始 token id 主要由第一个 PP stage 用来做 embedding 和前半段 layer 计算；第二个 PP stage 主要接收前一段 pipeline 发来的 intermediate tensors，再执行后半段 layers。

这解释了一个容易混淆的点：“收到 prompt token”不等于“每个 PP stage 都用 prompt token 作为模型输入”。 后续 PP stage 需要这份调度信息维护本地状态和 KV block 映射，但它的 forward 输入主要是上一段发来的 hidden states。

10. 总结

EngineCore 到 Worker 的顶层链路可以抽象为一个清晰的控制面模型。EngineCore 负责请求调度、KV block 逻辑分配和执行命令广播；Worker 负责按 rank 执行模型分片、维护本地 request cache 和 KV cache tensor；RayExecutorV2 负责 actor 放置和生命周期，但不负责高频 step 的模型 RPC。

可以把这套设计压缩成三句话：

EngineCore 管全局状态：request、scheduler、逻辑 KV block。
Executor 管执行入口：collective_rpc 广播 SchedulerOutput。
Worker 管本地执行：更新本地 request state，执行 GPUModelRunner，按 output_rank 返回结果。

再往下一层，重点就不再是控制面，而是具体推理如何完成：_update_states 如何构造 batch，_prepare_inputs 如何生成 input ids 和 positions，set_forward_context 如何把 attention metadata 传进模型，以及 TP/PP 在 layer 内部具体怎么切权重和通信。