EngineCoreProc执行推理流程

EngineCoreProc代码

┌─────────────────────────────────────────────────────────────────────┐
│                     EngineCoreProc 进程                              │
├─────────────────────────────────────────────────────────────────────┤
│ 输入线程 (process_input_sockets)                                    │
│   ├─ 接收 ZMQ 消息 → 反序列化 → 放入 input_queue                     │
│   └─ 特殊处理：ADD 预预处理，ABORT 双队列                            │
├─────────────────────────────────────────────────────────────────────┤
│ 主线程 (run_busy_loop) ─────────────────────────────────────────────┤
│   while running:                                                    │
│     1. _process_input_queue(): 从 input_queue 取请求 → 更新 scheduler│
│     2. _process_engine_step():                                      │
│          ├─ scheduler.schedule()                                    │
│          ├─ model_executor.execute_model() (异步)                   │
│          ├─ model_executor.sample_tokens() (必要时)                 │
│          ├─ scheduler.update_from_output()                          │
│          └─ 结果放入 output_queue                                   │
├─────────────────────────────────────────────────────────────────────┤
│ 输出线程 (process_output_sockets)                                    │
│   └─ 从 output_queue 取结果 → 序列化 → ZMQ 发送回客户端              │
└─────────────────────────────────────────────────────────────────────┘

 后续的分析假定model_executor的类型为MultiprocExecutor。
MultiprocExecutor.execute_model 的实现：

# https://github.com/vllm-project/vllm/blob/v0.20.1/vllm/v1/executor/multiproc_executor.py#L306
class MultiprocExecutor
    def execute_model(  # type: ignore[override]
        self, scheduler_output: SchedulerOutput, non_block: bool = False
    ) -> ModelRunnerOutput | None | Future[ModelRunnerOutput | None]:
        return self.collective_rpc(
            "execute_model",
            args=(scheduler_output,),
            unique_reply_rank=self.output_rank,
            non_block=non_block,
            timeout=envs.VLLM_EXECUTE_MODEL_TIMEOUT_SECONDS,
            kv_output_aggregator=self.kv_output_aggregator,
        )

execute_model的执行过程

EngineCoreProc
    │
    └─> step()
            │
            ├─> scheduler_output = scheduler.schedule()
            │
            └─> future = model_executor.execute_model(scheduler_output, non_block=True)
                    │
                    │   (model_executor 实际是 MultiprocExecutor 实例)
                    ▼
MultiprocExecutor.execute_model()
    │
    └─> collective_rpc("execute_model", args=(scheduler_output,), unique_reply_rank=output_rank)
            │
            ├─> 序列化调用信息 (method="execute_model", args, kwargs, output_rank)
            │
            ├─> self.rpc_broadcast_mq.enqueue(...)   // 将请求放入广播队列
            │
            ├─> 根据 output_rank 确定需要等待哪些 response_mq
            │
            ├─> get_response(): 对每个 response_mq.dequeue() 阻塞等待结果
            │
            └─> 返回结果 (或 Future)
                    │
                    │   (Worker 进程侧)
                    ▼
WorkerProc.worker_busy_loop()   // 每个 worker 进程的主循环
    │
    ├─> method, args, kwargs, output_rank = self.rpc_broadcast_mq.dequeue(indefinite=True)
    │
    ├─> func = getattr(self.worker, method)   // method = "execute_model"
    │
    ├─> output = func(*args, **kwargs)        // 调用 Worker.execute_model
    │
    └─> if output_rank is None or self.rank == output_rank:
            self.enqueue_output(output)       // 结果放入 worker_response_mq
                    │
                    │   (结果返回)
                    ▼
MultiprocExecutor 的 get_response() 从 response_mq 收到结果
    │
    └─> 返回给 future.result() 的调用者 (EngineCoreProc)

EngineCore初始化流程图

EngineCore.__init__
    │
    └─> executor_class = MultiprocExecutor
            │
            └─> MultiprocExecutor._init_executor
                    │
                    ├─> 创建 MessageQueue（DP leader）
                    │
                    ├─> for each local_rank:
                    │       WorkerProc.make_worker_process
                    │           │
                    │           ├─> 创建 Process (target=worker_main)
                    │           └─> 启动子进程
                    │
                    └─> WorkerProc.wait_for_ready
                            │
                            └─> 等待每个 worker 的 ready_pipe

每个子进程执行 worker_main:
    │
    ├─> WorkerProc.__init__
    │       │
    │       ├─> WorkerWrapperBase.init_worker
    │       │       │
    │       │       ├─> 解析 worker_cls = "vllm.v1.worker.gpu_worker.Worker"
    │       │       └─> worker = Worker(**kwargs)
    │       │
    │       ├─> worker.init_device()
    │       │       ├─> 设置 CUDA 设备、分布式环境
    │       │       ├─> 拍摄内存快照
    │       │       └─> 创建 GPUModelRunner
    │       │
    │       ├─> worker.load_model()
    │       │
    │       └─> _init_message_queues()
    │
    ├─> 通过 ready_pipe 发送 READY + MQ 句柄
    │
    └─> worker.worker_busy_loop()   // 等待并执行推理请求

 executor_class按照vllm_config获取：executor_class = Executor.get_class(vllm_config)。流程图直接以MultiprocExecutor分析。

# https://github.com/vllm-project/vllm/blob/v0.20.1/vllm/entrypoints/cli/serve.py#L266
# https://github.com/vllm-project/vllm/blob/v0.20.1/vllm/v1/executor/abstract.py#L48
class Executor(ABC):
    """Abstract base class for vLLM executors."

    An executor is responsible for executing the model on one device,
    or it can be a distributed executor that can execute the model on multiple devices.
    """

    uses_ray: bool = False  # whether the executor uses Ray for orchestration.
    supports_pp: bool = False  # whether the executor supports PP

    @staticmethod
    def get_class(vllm_config: VllmConfig) -> type["Executor"]:
        executor_class: type[Executor]
        parallel_config = vllm_config.parallel_config
        distributed_executor_backend = parallel_config.distributed_executor_backend
        # distributed_executor_backend must be set in VllmConfig.__post_init__
        if isinstance(distributed_executor_backend, type):
            if not issubclass(distributed_executor_backend, Executor):
                raise TypeError(
                    "distributed_executor_backend must be a subclass of "
                    f"Executor. Got {distributed_executor_backend}."
                )
            executor_class = distributed_executor_backend
        elif distributed_executor_backend == "ray":
            if envs.VLLM_USE_RAY_V2_EXECUTOR_BACKEND:
                from vllm.v1.executor.ray_executor_v2 import RayExecutorV2

                executor_class = RayExecutorV2
            else:
                from vllm.v1.executor.ray_executor import RayDistributedExecutor

                executor_class = RayDistributedExecutor
        elif distributed_executor_backend == "mp":
            from vllm.v1.executor.multiproc_executor import MultiprocExecutor

            executor_class = MultiprocExecutor
        elif distributed_executor_backend == "uni":
            from vllm.v1.executor.uniproc_executor import UniProcExecutor

            executor_class = UniProcExecutor
        elif distributed_executor_backend == "external_launcher":
            # TODO: make v1 scheduling deterministic
            # to support external launcher
            executor_class = ExecutorWithExternalLauncher
        elif isinstance(distributed_executor_backend, str):
            executor_class = resolve_obj_by_qualname(distributed_executor_backend)
            if not issubclass(executor_class, Executor):
                raise TypeError(
                    "distributed_executor_backend must be a subclass of "
                    f"Executor. Got {executor_class}."
                )
        else:
            raise ValueError(
                f"Unknown distributed executor backend: {distributed_executor_backend}"
            )
        return executor_class

MultiprocExecutor的核心功能

MultiprocExecutor源码

┌─────────────────────────────────────────────────────────────────┐
│                    MultiprocExecutor                             │
├─────────────────────────────────────────────────────────────────┤
│ 1. 初始化阶段 (_init_executor)                                   │
│    • 校验 TP×PP×PCP = world_size                                 │
│    • 设置多进程环境 (OMP_NUM_THREADS=1)                          │
│    • 创建分布式初始化地址 (loopback + 随机端口)                   │
│    • 若为 DP leader → 创建广播 MessageQueue (SchedulerOutput)    │
│    • 创建共享锁 shared_worker_lock                               │
├─────────────────────────────────────────────────────────────────┤
│ 2. 启动 Worker 进程 (循环 local_rank)                            │
│    • 计算 global_rank = local_world_size * node_rank_within_dp   │
│    • 创建 ready_pipe / death_pipe                                │
│    • 启动 Process(target=WorkerProc.worker_main)                 │
│    • 收集 UnreadyWorkerProcHandle                                │
├─────────────────────────────────────────────────────────────────┤
│ 3. 等待所有 Worker 就绪 (wait_for_ready)                         │
│    • 监听 ready_pipe，接收子进程 READY 消息                      │
│    • 提取 worker_response_mq 句柄                                │
│    • 返回 WorkerProcHandle 列表                                  │
├─────────────────────────────────────────────────────────────────┤
│ 4. 后续就绪 & 监控                                               │
│    • 收集 response_mqs（本地/远程输出队列）                      │
│    • 等待所有 MQ 就绪 (wait_until_ready)                         │
│    • 启动后台监控线程 (MultiprocWorkerMonitor)                   │
│    • 初始化 futures 队列                                         │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                    子进程 (WorkerProc)                           │
├─────────────────────────────────────────────────────────────────┤
│ • 继承 WorkerWrapperBase → 实例化具体 Worker (GPU/CPU)          │
│ • 调用 worker.init_device() → 设置设备、分布式环境              │
│ • 调用 worker.load_model() → 加载模型权重                       │
│ • 初始化消息队列 (_init_message_queues)                         │
│   - 单节点: 创建本地 worker_response_mq                          │
│   - 多节点: 通过分布式组创建跨节点 MQ                            │
│ • 启动死亡监控线程 (death_pipe)                                 │
│ • 发送 READY + MQ 句柄给父进程                                  │
│ • 进入 worker_busy_loop()                                       │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                    运行时 RPC 调用流程                           │
├─────────────────────────────────────────────────────────────────┤
│  collective_rpc(method, args, unique_reply_rank)                │
│         │                                                       │
│         ├─> 将 (method, args, kwargs, output_rank) 放入广播MQ   │
│         │                                                       │
│         ▼                                                       │
│  WorkerProc.worker_busy_loop: 从广播MQ 取出请求                  │
│         │                                                       │
│         ├─> 执行 self.worker.method(*args, **kwargs)            │
│         │                                                       │
│         ├─> 若 self.rank == output_rank → 结果放入 response_mq  │
│         │                                                       │
│         ▼                                                       │
│  Executor 从 response_mqs 读取结果 → 聚合/返回                   │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                     健康监控 & 关闭                              │
├─────────────────────────────────────────────────────────────────┤
│  监控线程: wait(proc.sentinel) → 进程死亡 → 触发 failure_callback│
│  shutdown():                                                     │
│    • 关闭 death_writer 通知子进程退出                            │
│    • 等待进程终止 (4s grace → SIGTERM → SIGKILL)                │
│    • 关闭所有 MessageQueue                                       │
└─────────────────────────────────────────────────────────────────┘

worker_busy_loop

# https://github.com/vllm-project/vllm/blob/v0.20.1/vllm/v1/executor/multiproc_executor.py#L944
class WorkerProc:
    def worker_busy_loop(self):
        """Main busy loop for Multiprocessing Workers"""
        assert self.rpc_broadcast_mq is not None
        while True:
            method, args, kwargs, output_rank = self.rpc_broadcast_mq.dequeue(
                indefinite=True
            )
            try:
                if isinstance(method, str):
                    func = getattr(self.worker, method)
                elif isinstance(method, bytes):
                    func = partial(cloudpickle.loads(method), self.worker)

                output = func(*args, **kwargs)
            except Exception as e:
                # Notes have been introduced in python 3.11
                if hasattr(e, "add_note"):
                    e.add_note(traceback.format_exc())
                logger.exception("WorkerProc hit an exception.")
                # exception might not be serializable, so we convert it to
                # string, only for logging purpose.
                if output_rank is None or self.rank == output_rank:
                    self.handle_output(e)
                continue

            if output_rank is None or self.rank == output_rank:
                self.handle_output(output)
 
    def enqueue_output(self, output: Any):
        """Prepares output from the worker and enqueues it to the
        worker_response_mq. If the output is an Exception, it is
        converted to a FAILURE response.
        """
        if isinstance(output, AsyncModelRunnerOutput):
            output = output.get_output()

        if isinstance(output, Exception):
            result = (WorkerProc.ResponseStatus.FAILURE, str(output))
        else:
            result = (WorkerProc.ResponseStatus.SUCCESS, output)
        if (response_mq := self.worker_response_mq) is not None:
            response_mq.enqueue(result)

    def handle_output(self, output: Any):
        """Handles output from the worker. If async scheduling is enabled,
        it is passed to the async_output_busy_loop thread. Otherwise, it is
        enqueued directly to the worker_response_mq.
        """
        if self.use_async_scheduling:
            self.async_output_queue.put(output)
        else:
            self.enqueue_output(output)

 MultiprocExecutor和WorkerProc使用两组MessageQueue传递数据。
 在 MultiprocExecutor 和 RayExecutorV2 中，MessageQueue 被用作控制平面（Control Plane），与负责数据传输的 NCCL 数据平面（Data Plane） 相分离。这种解耦设计使得控制消息的传递不会干扰高速的数据传输。广播调度输出：Executor会创建一个 MessageQueue 实例（如 self.rpc_broadcast_mq），用于将调度器产生的 SchedulerOutputs 广播给所有的 Worker 进程。，

┌─────────────┐  enqueue(RPC)   ┌────────────────────────────┐
│  Executor   │ ──────────────> │   rpc_broadcast_mq (Writer) │
│             │                  │  - Shared Ring Buffer      │
└─────────────┘                  │  - PUB socket              │
       ▲                         └────────────┬───────────────┘
       │                                      │ dequeue
       │  response_mqs[rank].dequeue()        │ (blocking)
       │                              ┌────────▼───────────────┐
       │                              │ WorkerProc.busy_loop   │
       │                              │  - Get (method,args)   │
       │                              │  - self.worker.method()│
       │                              │  - result → response_mq│
       │                              └────────┬───────────────┘
       │                                       │ enqueue(result)
       └───────────────────────────────────────┘

response_mqs的创建流程：

Worker 进程                               Executor 进程
─────────────────────────────────────────────────────────────────
_init_message_queues()
  │
  ├─ 单节点: worker_response_mq = MessageQueue(1,1)  [writer]
  └─ 多节点: 通过分布式组创建 writer + peer_handles
  │
worker_main()
  │
  └─ 发送 READY:
       handle = worker_response_mq.export_handle()
       peer_handles = ...
       ready_writer.send({handle, peer_handles})
                                          │
                                          ▼
                              wait_for_ready()
                                │
                                └─ wait_for_response_handle_ready()
                                      │
                                      ├─ 对每个 handle:
                                      │    MessageQueue.create_from_handle(handle, rank)
                                      │    → 创建 reader 模式的 MQ
                                      │
                                      └─ 返回 WorkerProcHandle(reader_mq, peer_mqs)
                                          │
                                          ▼
                              _init_executor()
                                │
                                └─ 收集 self.response_mqs[rank] = reader_mq

WorkerWrapperBase.init_worker

WorkerWrapperBase.init_worker代码
parallel_config.worker_extension_cls的配置与平台相关。

# https://github.com/vllm-project/vllm/blob/v0.20.1/vllm/platforms/cuda.py#L217
class CudaPlatformBase(Platform):
    _enum = PlatformEnum.CUDA

    @classmethod
    def check_and_update_config(cls, vllm_config: VllmConfig) -> None:
        parallel_config = vllm_config.parallel_config
        model_config = vllm_config.model_config

        if parallel_config.worker_cls == "auto":
            parallel_config.worker_cls = "vllm.v1.worker.gpu_worker.Worker"

gpu_worker

gpu_worker.py

┌─────────────────────────────────────────────────────────────────────────────┐
│                           Worker 初始化 (__init__)                          │
│  - 保存 vllm_config, rank, local_rank                                       │
│  - 设置 float32 matmul 精度                                                 │
│  - 创建 ElasticEP 执行器，初始化休眠 buffer 字典，可选权重传输引擎           │
│  - 校验 profiler 类型（torch/cuda/None），懒创建                             │
└─────────────────────────────────────────────────────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                            init_device()                                    │
│  1. 处理 NCCL 环境变量                                                      │
│  2. 调整 local_rank（支持数据并行 DP）→ 设置 cuda 设备                      │
│  3. 初始化分布式环境（init_worker_distributed_environment）                 │
│  4. 设置随机种子，清理缓存，拍摄内存快照 init_snapshot                      │
│  5. 计算请求内存 requested_memory（基于 gpu_memory_utilization）            │
│  6. 初始化 workspace 管理器（双缓冲）                                       │
│  7. 创建 GPUModelRunner（V1 或 V2，根据 env）                               │
└─────────────────────────────────────────────────────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                              load_model()                                   │
│  - 是否启用 sleep mode？                                                    │
│    ├─ 是 → 进入 weights 内存池上下文 (CuMemAllocator)                       │
│    └─ 否 → nullcontext()                                                   │
│  - 调用 model_runner.load_model() 加载真实或 dummy 权重                    │
└─────────────────────────────────────────────────────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                      determine_available_memory()                          │
│  是否已指定 kv_cache_memory_bytes？                                         │
│  ├─ 是 → 直接返回该值，跳过 profiling                                      │
│  └─ 否 → 执行 memory profiling：                                           │
│      - 记录初始内存（权重 + 空闲）                                          │
│      - 运行 profile_run() 获取峰值激活内存和非 torch 增量                   │
│      - 若需估算 CUDA Graph 内存 → profile_cudagraph_memory()               │
│      - 计算 non_kv_cache_memory                                            │
│      - 计算 available_kv_cache_memory_bytes                                │
│      - 打印建议调整 gpu_memory_utilization 的日志                          │
└─────────────────────────────────────────────────────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                       initialize_from_config()                             │
│  1. 更新 cache_config.num_gpu_blocks                                       │
│  2. 初始化 KV 传输组（ensure_kv_transfer_initialized）                     │
│  3. 分配 KV Cache：                                                         │
│     - 若 sleep mode 启用 → 进入 kv_cache 内存池                             │
│     - 否则直接分配                                                         │
│  4. 若模型需 routed experts → 初始化捕获器                                 │
│  5. 若 KV zeroing 需要 → 在普通分配器中构建元数据                           │
└─────────────────────────────────────────────────────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                      compile_or_warm_up_model()                            │
│  1. 收集需要编译/预热的 batch size 列表                                     │
│     - 来自 compile_sizes 中未被 CUDA Graph 捕获的大小                       │
│     - 补充未覆盖的 compile_range 终点                                       │
│  2. 对每个 size 执行 _dummy_run() 触发编译                                  │
│  3. kernel_warmup() 预热各算子                                             │
│  4. 若未强制 eager → capture_model() 捕获 CUDA Graph，返回实际内存         │
│  5. 比对实际与估算的 CUDA Graph 内存                                       │
│  6. 若未指定 kv_cache_memory_bytes → 记录详细内存分解并建议使用固定配置     │
│  7. V2 runner：warmup_kernels (execute_model + sample_tokens)              │
│     V1 & PP last rank：预分配 sampler 内存                                  │
│  8. 重置随机种子                                                            │
└─────────────────────────────────────────────────────────────────────────────┘
                                       │
                                       ▼
                        ┌──────────────────────┐
                        │   推理执行循环        │
                        │ (execute_model +     │
                        │  sample_tokens)      │
                        └──────────────────────┘
                                       │
                ┌──────────────────────┴──────────────────────┐
                │                                             │
                ▼                                             ▼
┌───────────────────────────────┐          ┌───────────────────────────────┐
│      execute_model 执行流程    │          │        sample_tokens          │
│ 1. 等待之前 PP 发送完成         │          │ 直接委托 model_runner         │
│ 2. 若非第一个 PP rank：        │          └───────────────────────────────┘
│    异步接收 intermediate_tensors│
│ 3. 通过 annotate_profile 标注  │
│    当前 iteration 的请求类型    │
│ 4. model_runner.execute_model()│
│ 5. 若输出为 IntermediateTensors│
│    且非最后 PP rank：           │
│    → 异步发送到下一级 PP        │
│    → 返回 None                  │
│ 6. 否则返回 ModelRunnerOutput   │
└───────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         可选：休眠 / 唤醒                                   │
│  sleep(level):                                                             │
│    - level=2 → 保存模型所有 buffer 到 CPU                                  │
│    - 调用 CuMemAllocator.sleep(offload_tags)                               │
│  wake_up(tags):                                                            │
│    - 调用 allocator.wake_up(tags) 恢复内存池                               │
│    - 若保存的 buffer 非空 → 拷回模型                                       │
│    - 若 tags 含 kv_cache 且为量化 → 重新初始化 FP8 scales                  │
└─────────────────────────────────────────────────────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                         可选：在线权重更新                                  │
│  init_weight_transfer_engine(init_info) → 建立后端连接（如 NCCL group）    │
│  update_weights(update_info):                                              │
│    - 解析更新信息（checkpoint 格式或直接 tensor）                          │
│    - 若 checkpoint：layerwise reload 模式                                 │
│    - 否则直接 copy_ 到模型参数                                            │
│    - torch.accelerator.synchronize()                                      │
└─────────────────────────────────────────────────────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                              shutdown()                                    │
│  - 关闭 KV transfer group                                                 │
│  - 关闭 profiler                                                          │
│  - 关闭 weight_transfer_engine                                            │
│  - model_runner.shutdown() 释放 GPU 资源                                  │
└─────────────────────────────────────────────────────────────────────────────┘

Reference

[1] Model Runner V2 架构（三）：异步调度
[2] vllm异步调度