vLLM V1 源码阅读笔记

本文基于 vLLM V1 (v0.18.x) 源码阅读整理，记录了从请求进入到 token 输出的完整链路。 完整的架构图可下载 vllm-architecture.drawio 在 draw.io 中查看。 调度过程详解图可下载 vllm-scheduler-64videos.drawio 查看 64 路视频的调度全景。

整体架构概览

vLLM V1 采用多进程架构，核心由三类进程组成：

1. 主进程 (Main Process) — API Server + 请求预处理 + 输出后处理
2. EngineCore 子进程 — Scheduler 调度 + KV Cache 管理 + 执行器编排
3. Worker 进程 — 每个 GPU 一个进程，负责模型推理（Tensor Parallel）

进程间通过 ZMQ (input/output socket) 进行异步通信，EngineCore 与 Worker 之间则使用基于 multiprocessing.shared_memory 的 shm_broadcast 环形缓冲 + RPC 机制。

一、主进程 (Main Process)

1.1 启动入口

主进程运行 vllm.entrypoints.openai.api_server，由 Uvicorn + FastAPI 驱动。启动流程：

api_server
  └─ lifespan: build_async_engine_client (协程, contextmanager)
       └─ build_async_engine_client_from_engine_args
            └─ AsyncLLM.from_vllm_config(vllm_config)
                 └─ AsyncLLM.__init__()

1.2 AsyncLLM 初始化

AsyncLLM.__init__ 创建以下核心组件：

# async_llm.py __init__ 核心初始化
self.renderer = renderer = renderer_from_config(self.vllm_config)
self.io_processor = get_io_processor(...)          # 可选插件，多数模型为 None
self.input_processor = InputProcessor(self.vllm_config, renderer)
self.output_processor = OutputProcessor(renderer.tokenizer, ...)
self.engine_core = EngineCoreClient.make_async_mp_client(...)

Renderer 是 V1 新引入的关键抽象层，位于 vllm/renderers/。不同 tokenizer 类型对应不同实现：

• HfRenderer：HuggingFace tokenizer（大多数模型）
• MistralRenderer：Mistral tokenizer
• QwenVLRenderer 等：特定模型变体

Renderer 内部持有 AsyncMicrobatchTokenizer，将阻塞的 tokenizer 调用卸载到 ThreadPoolExecutor，并支持微批量（micro-batch）合并以提升吞吐。

EngineCoreClient 根据 data_parallel_size 决定使用：

• 单 DP：AsyncMPClient
• 多 DP：DPAsyncMPClient / DPLBAsyncMPClient

1.3 请求处理流程（以 OpenAI Chat 请求为例）

一个 Chat Completion 请求的完整链路如下：

HTTP Request
  → OpenAIServingChat.create_chat_completion()
    → render_chat_request()
      → renderer.render_chat_async()          ← ① 异步渲染
        → parse_chat_messages_async()         ← 异步解析消息 + 多模态加载
          → AsyncMultiModalItemTracker        ← fetch_video_async() / fetch_image_async()
            → MediaConnector.load_from_url_async()
              → run_in_executor(global_thread_pool, load_bytes)  ← 线程池解码
        → tokenize_prompts_async()            ← ② 异步 tokenize
          → AsyncMicrobatchTokenizer.encode()
            → run_in_executor(ThreadPoolExecutor, tokenizer)     ← 线程池 tokenize
      → 返回 ProcessorInputs
    → engine_client.generate(engine_prompt, sampling_params)
      → AsyncLLM.add_request()
        → InputProcessor.process_inputs()     ← ③ 封装 EngineCoreRequest
        → engine_core.add_request_async()     ← ④ ZMQ 发送到 EngineCore
      → output_handler 推送结果到 queue       ← ⑤ 异步输出
      → async for chunk in generate(): yield  ← ⑥ SSE 流式返回

① Renderer.render_chat_async() — 异步渲染

• parse_chat_messages_async() 解析 OpenAI messages 格式，提取文本和多模态 URL
• 多模态数据（图像/视频/音频）通过 fetch_*_async() → run_in_executor(global_thread_pool, ...) 在媒体线程池中异步加载和解码
• 多个媒体文件通过 asyncio.gather() 并发加载
• 渲染聊天模板（apply_chat_template）

② tokenize_prompts_async() — 异步 Tokenize

• 使用 AsyncMicrobatchTokenizer，将阻塞的 tokenizer 调用通过 run_in_executor(ThreadPoolExecutor(max_workers=1), ...) 卸载到单独的 worker 线程
• 支持微批量（micro-batch）：在 batch_wait_timeout_s（默认 2ms）内收集多个请求合并 tokenize，减少调用开销

③ InputProcessor.process_inputs() — 封装请求

• 接收已处理好的 ProcessorInputs（已包含 prompt_token_ids、mm_kwargs、mm_placeholders 等）
• 校验参数、计算 max_tokens、构造 EngineCoreRequest

④ engine_core.add_request_async() — 发送到 EngineCore

• 通过 ZMQ input_socket 将 EngineCoreRequest (msgpack 序列化) 发送到 EngineCore 子进程

⑤ output_handler (后台 asyncio Task)

• 持续从 ZMQ output socket 拉取 EngineCoreOutputs
• 调用 OutputProcessor.process_outputs() 进行 detokenize
• 将输出按 chunk 分批处理（VLLM_V1_OUTPUT_PROC_CHUNK_SIZE），每批之间 await asyncio.sleep(0) 让出事件循环，避免长时间阻塞
• 生成 RequestOutput 推送到对应请求的 RequestOutputCollector 队列

关于 GIL：主进程中 tokenizer 和视频解码等 CPU 密集操作不会阻塞 event loop 接收新请求，这依靠两重机制协同工作：

1. run_in_executor + 线程池：所有阻塞操作都被卸载到工作线程，event loop 线程通过 await 等待结果，期间可以处理其他协程
2. C/Rust 扩展自动释放 GIL：HuggingFace tokenizers（Rust 实现）、OpenCV（C++ 实现）、NumPy（C 实现）在执行计算时会主动释放 GIL，工作线程不会阻塞 event loop 线程获取 GIL

即使存在少量纯 Python 代码持有 GIL，CPython 3.2+ 的 GIL 调度机制也会每 5ms（sys.getswitchinterval()）强制切换，确保 event loop 线程有机会运行。

二、EngineCore 子进程

每个 DP rank 对应一个 EngineCoreProc 进程，这是 vLLM 的调度核心。

2.1 初始化

EngineCoreProc.__init__()
  ├─ 连接 ZMQ socket (input/output address)，与主进程握手建立通信
  └─ EngineCore.__init__() — 创建核心组件
       ├─ MultiprocExecutor (model_executor)
       ├─ Scheduler
       └─ StructuredOutputManager

2.2 IO 线程与主循环

EngineCore 进程内部有 3 个线程协同工作，实现 ZMQ 通信与 GPU 计算的重叠：

┌─ EngineCoreProc 进程 ───────────────────────────────────────────────┐
│                                                                     │
│  Input IO Thread ──► input_queue ──► Main Thread ──► output_queue ──► Output IO Thread
│  (ZMQ recv,                         (busy loop)                      (ZMQ send,
│   msgpack 反序列化,                                                    msgpack 序列化,
│   释放 GIL)                                                           释放 GIL)
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

源码注释写道：“Background Threads and Queues for IO. These enable us to overlap ZMQ socket IO with GPU since they release the GIL, and to overlap some serialization/deserialization with the model forward pass.”

Input IO Thread (process_input_sockets):

• 从 ZMQ input socket 接收消息并 msgpack 反序列化
• 将 (request_type, data) 放入 input_queue（Python queue.Queue）

Output IO Thread (process_output_sockets):

• 从 output_queue 取出 EngineCoreOutputs
• msgpack 序列化后通过 ZMQ output socket 发回主进程

Main Thread 运行 run_busy_loop()：

def run_busy_loop(self):
    while self._handle_shutdown():
        self._process_input_queue()    # ① 取请求
        self._process_engine_step()    # ② 调度 + 执行 + 更新 + 输出

① _process_input_queue() — 从 input_queue 取出新请求

• 无工作时阻塞等待（queue.get(block=True)）
• 有工作时非阻塞排空队列（queue.get(block=False)）
• _handle_client_request() 根据请求类型分派：ADD → preprocess_request() → 加入 Scheduler.waiting

② _process_engine_step() — 单次引擎步骤

• scheduler.schedule() → SchedulerOutput（本次 step 跑哪些请求、token budget、block 分配）
• model_executor.execute_model(scheduler_output) → ModelRunnerOutput（含 sampled tokens）
• scheduler.update_from_output() — 更新请求状态，释放已完成请求的 KV blocks
• 构造 EngineCoreOutputs 放入 output_queue → Output IO Thread 异步发送

2.3 Scheduler 与 KV Cache 管理

schedule() 每个 step 调用：

1. 优先调度 running 队列（decode 请求）— 计算 token budget，allocate_slots
2. 调度 waiting 队列（prefill 请求）— get_computed_blocks（prefix cache lookup）→ allocate_slots → 移入 running

KVCacheManager（PagedAttention 核心）：

• free_block_queue（双向链表）管理空闲 block
• req_to_blocks dict 记录每个请求占用的 block
• allocate_slots / free / cache_blocks 操作
• Prefix Caching: hash_request_tokens + find_longest_cache_hit

2.3.1 统一的 Token-Gap 调度模型

V1 Scheduler 最优雅的设计之一是没有”prefill 阶段”和”decode 阶段”的区分。源码注释明确写道：

“There’s no ‘decoding phase’ nor ‘prefill phase’ in the scheduler. Each request just has the num_computed_tokens and num_tokens_with_spec.”

调度器用两个数值统一描述所有请求的状态：

每一步，调度器让 num_computed_tokens 尽量追赶 num_tokens_with_spec：

• “Prefill” 不过是差距很大（需要处理整个 prompt）
• “Decode” 不过是差距为 1（每次生成一个 token）
• “Chunked Prefill” 是差距很大但被 budget 截断

这个统一模型足够通用，可以覆盖 chunked prefill、prefix caching、speculative decoding 等所有场景。

2.3.2 调度循环与 Token Budget

每次 schedule() 的执行流程：

token_budget = max_num_scheduled_tokens (通常 = max_num_batched_tokens)

Step 1: 遍历 running 队列（decode 优先）
  ├── 每个 request: num_new = target - computed (decode 通常 = 1)
  ├── allocate_slots → 分配 KV blocks
  ├── 失败 → 触发 Preemption（抢占低优先级请求）
  └── budget -= num_new

Step 2: 遍历 waiting 队列（仅在无抢占时）
  ├── 前提: budget > 0 且 len(running) < max_num_seqs
  ├── num_new = min(remaining_prompt, budget)
  ├── chunked prefill: 如果 prompt 太长，分多个 step 完成
  ├── 分配 KV blocks → 移入 running
  └── budget -= num_new

断言: sum(num_scheduled_tokens) <= max_num_scheduled_tokens

2.3.3 关键调度参数对比

简单类比：max_model_len 管一个请求能跑多远，max_num_batched_tokens 管一步能干多少活，max_num_seqs 管同时能跑多少个。

2.3.4 抢占机制 (Preemption)

当 KV block 分配失败时，调度器逐步抢占优先级最低的请求来释放空间：

while allocate_slots 失败:
  if PRIORITY 策略:
    preempt max(running, key=(priority, arrival_time))  # 优先级最低的
  else (FCFS):
    preempt running.pop()  # 最晚加入的

被抢占的请求:
  ① 释放所有 KV blocks
  ② num_computed_tokens 重置为 0
  ③ 放回 waiting 队列头部（优先重新调度）

三、Worker 进程

每个 GPU 一个 Worker 进程，通过 Tensor Parallel 协同工作。

3.1 启动与初始化

MultiprocExecutor (EngineCore 进程中)
  └─ spawn WorkerProc × local_world_size (= TP × PP × PCP)
       └─ Worker init (每个 GPU)
            ├─ init_device() — 设置 CUDA device, 分布式通信组 (NCCL)
            ├─ model_runner.load_model() — 实例化模型架构, 加载权重, torch.compile()
            └─ initialize_cache() — profiling forward pass → 分配 KV cache tensors → warmup CUDA graphs

Worker 进程进入 worker_busy_loop，通过 rpc_broadcast_mq.dequeue 等待来自 MultiprocExecutor 的消息（func + 参数），执行后 handle_output。

shm_broadcast: 基于 multiprocessing.shared_memory 的多槽环形缓冲 + 内存栅栏/每读端完成位 + SpinCondition 唤醒。pickle 后小对象整块进环，大对象打 overflow 标并用本机 ZMQ IPC 传 multipart；多机时再配合 TCP XPUB/SUB。

3.2 GPUModelRunner.execute_model()

每次 step 的执行流程：

① _update_states()

• prune 已结束请求，更新 InputBatch metadata
• 更新 block table（paged KV 索引）

② _prepare_inputs()

• CPU → GPU: input_ids, positions, slot_mapping
• 构造 attention metadata（FlashAttention / FlashInfer）

③ model.forward()

• eager 模式或 CUDA Graph replay
• 所有序列拼接为超长序列（Continuous Batching）
• PagedAttention kernel: 读写 KV cache blocks

④ gather last-token hidden states → compute logits

⑤ Sampler: 从 logits 采样 token

• 支持 greedy / temperature / top-p / top-k / guided decoding

返回 SamplerOutput → EngineCoreOutputs

四、多模态输入预处理

V1 将多模态预处理全部异步化，通过线程池避免阻塞 event loop 和 GPU Worker。

4.1 请求解析（异步路径）

HTTP 请求进入 OpenAIServingChat 后，走异步解析链路：

① parse_chat_messages_async() — 解析 OpenAI messages 格式

遍历消息内容，遇到多模态 URL 时不立即加载，而是向 AsyncMultiModalItemTracker 注册一个协程（coroutine）：

# chat_utils.py — AsyncMultiModalContentParser
async def _video_with_uuid_async(self, video_url, uuid):
    video = await self._connector.fetch_video_async(video_url) if video_url else None
    return video, uuid

def parse_video(self, video_url, uuid=None):
    coro = self._video_with_uuid_async(video_url, uuid)  # 注册协程，不立即执行
    placeholder = self._tracker.add("video", coro)

② AsyncMultiModalItemTracker.resolve_items() — 并发加载所有多模态数据

resolved_items_by_modality = {
    modality: await asyncio.gather(*coros)  # 所有 image/video/audio 并发加载
    for modality, coros in self._items_by_modality.items()
}

③ 媒体加载线程池 — 真正的 IO 和解码在线程池中执行

# connector.py
global_thread_pool = ThreadPoolExecutor(
    max_workers=envs.VLLM_MEDIA_LOADING_THREAD_COUNT  # 可配置
)

async def load_from_url_async(self, url, media_io, ...):
    # HTTP 下载使用 aiohttp（纯异步，不占线程）
    data = await connection.async_get_bytes(url)
    # 解码在线程池中执行（OpenCV/NumPy 释放 GIL）
    return await loop.run_in_executor(global_thread_pool, media_io.load_bytes, data)

④ Renderer.render_chat_async() — tokenize + 组装

tokenize 通过 AsyncMicrobatchTokenizer 在独立线程中完成后，与多模态数据一起组装为 ProcessorInputs。

4.2 多模态缓存

mm_processor_cache（主进程缓存）:

• key = hash(mm_data) ← 同一图/视频跨请求复用
• 避免重复 CPU 解码 / resize

4.3 构造 EngineCoreRequest

prompt_token_ids: list[int]          ← 已含 mm placeholder token
mm_features: list[MultiModalFeature] ← 每个 mm 对象的 pixel_values / audio features
mm_hashes: list[str]                 ← 用于 encoder_cache 查找
mm_placeholders: list[PlaceholderRange] ← 在 token 序列中的位置信息

经 msgpack 序列化 → ZMQ input_socket → EngineCore 进程。

4.4 Encoder Cache (GPU 侧)

在 EngineCore 进程中，preprocess_request() 检查 mm_hash 是否在 encoder_cache 中：

• 命中 → 直接取已有 GPU embedding，跳过视觉编码
• 未命中 → 将 pixel_values 等打入 EngineCoreRequest，在 Worker 中由 Vision Encoder 处理

4.5 model.forward() 执行阶段

Vision Encoder:  pixel_values → vision embeddings (e.g. 4096 dims/image)
Projection Layer: vision_embeddings → LLM hidden_size
Merge: text token hidden states + vision hidden states → 统一 hidden sequence

合并后进入标准 Transformer decoder 流程。

五、Continuous Batching + Chunked Prefill

5.1 核心设计

每个 Engine Step 都有一个固定 Token Budget (max_num_batched_tokens, 默认 2048+)。Scheduler 在 budget 内打一个混合 batch：先塞满所有 decode 请求（每个占 1 token），剩余 budget 分给 prefill（可 chunk）。

5.2 序列 Flatten

Forward Pass 内部，所有序列被 flatten 拼接为一条超长序列（无 padding）。以 Step 2 为例：

input_ids: [R1_tok[512..1023], R2_tok[n], R3_tok[0..299]]

attention mask 保证每个序列只 attend 自身。slot_mapping 记录每个 token 对应的 KV cache 物理 block slot（PagedAttention 寻址）。

5.3 KV Cache 管理

• Step 1: R1 分配 2 blocks, R2 无新分配
• Step 2: R1 +2 blocks, R2 无新, R3 +2 blocks
• Step 3: R1 decode +1tok, R2 +1tok, R3 +2 blocks, R4 +1 block

5.4 抢占 (Preemption)

KV Cache 不足时，V1 默认 RECOMPUTE 模式（V0 支持 SWAP）：将低优先级 running 请求踢回 waiting 队列，释放其 KV blocks。下次调度时重新从头 prefill（开销低于 swap to CPU）。

5.5 Continuous Batching 核心特性

• 序列在 forward pass 中被 flatten 拼接（无 padding）
• 每个 step 结束后，已完成的请求立即被移出，新请求可在下个 step 加入
• V1 取消了 prefill/decode 的人为分界，统一用 {req_id: num_tokens} 表示调度决策
• KV Cache 用 PagedAttention 管理，物理块按需分配/回收

5.6 实例：Qwen3VL 64路视频理解的调度过程

完整的调度过程图可下载 vllm-scheduler-64videos.drawio 在 draw.io 中查看。

以 Qwen3VL 处理 64 个视频理解请求为例，展示 Continuous Batching 在真实多模态场景下的调度行为。

Step-by-Step 调度过程

Step 1 — 首批进入：

Budget: 8192
├── R1: prefill 7000 (一次完成)        → budget 剩余 1192
└── R2: prefill 1192 (分块! 仅部分)    → budget = 0
Batch size: 2, Waiting: 62

Step 2 — 混合调度开始：

Budget: 8192
├── R1: decode 1   (已完成 prefill)    → budget 8191
├── R2: prefill 5808 (完成剩余部分)    → budget 2383
└── R3: prefill 2383 (新请求, 分块)    → budget = 0
Batch size: 3, Waiting: 61

这就是 Continuous Batching 的核心：R1 在 decode，R2 在继续 prefill，R3 同时开始 prefill，三种状态共存于同一个 batch！

Step 4 — 阶梯模式形成：

Budget: 8192
├── R1-R3: decode × 3 = 3              → budget 8189
├── R4: prefill 3427 (完成剩余)        → budget 4762
└── R5: prefill 4762 (新请求, 分块)    → budget = 0
Batch size: 5, Waiting: 59

Step 6 — 加速效应：

Budget: 8192
├── R1-R5: decode × 5 = 5              → budget 8187
├── R6: prefill 1050 (完成)            → budget 7137
├── R7: prefill 7000 (一次完成!)       → budget 137
└── R8: prefill 137 (分块)             → budget = 0
Batch size: 8, Waiting: 56

关键洞察：decode 请求越多 → 每个只消耗 1 token → 剩余 budget 越大 → 新请求 prefill 越快！

Step 40+ — 全速运行（running 达到 max_num_seqs=32 上限）：

Budget: 8192
├── R1-R32: decode × 32 = 32 (仅占 0.4%!)  → budget 8160
├── R33: prefill 7000 (从 waiting 进入)     → budget 1160
└── R34: prefill 1160 (分块)                → budget = 0

Step 200+ — 请求轮替：

R1 生成完毕 → 释放 slot 和 KV blocks
→ R33-R64 中的下一个立即从 waiting 进入 running
→ 开始 prefill，循环直到所有 64 个请求完成

三阶段全景

对比：Static Batching vs Continuous Batching

Continuous Batching 在这个场景下可以带来约 30 倍的吞吐量提升。

六、GPUModelRunner V1 vs Model Runner V2 (MRV2)

MRV2 (VLLM_USE_V2_MODEL_RUNNER=1 开启) 目前仍为 experimental；V1 为当前默认。两者共用同一套 Scheduler / Executor / KV Cache 架构，仅 ModelRunner 层不同。

6.1 整体设计哲学

6.2 Persistent Batch 设计

V1 的问题：

persistent state tensor 直接用作 model input tensor（都在 CPU），block table 布局与请求顺序强耦合。这带来几个严重问题：

• 添加/删除请求 → 需要整张 tensor 重排（complex reorder）
• 需要 CachedRequestState 作冗余备份
• async scheduling 下容易引发 race condition
• 所有 CPU 操作必须在 async barrier 内，灵活性低

MRV2 的解法：

将持久状态与输入解耦为两部分：

• stable state table (CPU, fixed 1024 rows) — 每个 req 在生命周期内固定一行
• per-step input tensor (GPU) — Triton gather kernel 按当前顺序组装

# MRV2: 持久 state 与 input 解耦
self.states[req_idx] = new_req.data       # 写持久 state (非 pinned)
tmp_states = self.states.pin_memory()      # 每步拷贝到临时 pinned
states = tmp_states.to('cuda', non_blocking=True)  # H2D
input_tensors = triton_gather(states, schedule_indices)  # GPU gather

解耦好处：

• 添加/删除请求只改 state table 一行，O(1)
• 消除 CachedRequestState 冗余备份
• StagedWriteTensor: CPU 写 state, GPU 读 tmp copy → 无 race
• input_ids / positions / seq_lens / block_table 全在 GPU 组装
• 不再需要 async barrier

6.3 Input Preparation

6.4 Async Scheduling

6.5 Sampler

6.6 ModelState 抽象 (MRV2 核心创新)

class ModelState(ABC):
    def add_request(self, req: EngineCoreRequest): ...
    def remove_request(self, req_id: str): ...
    def get_mm_embeddings(self, req_id: str) -> Tensor: ...
    def prepare_inputs(self, scheduler_output) -> dict: ...

模型扩展方式对比：

• V1: override 方法，无统一抽象
• MRV2: ModelState ABC，职责清晰

Spec Decoding + Async 支持：

• V1: 难以同时支持，需要 hack
• MRV2: 天然支持，GPU prep 直接消费 rejection output

6.7 性能对比 (官方 benchmark)

Qwen3-0.6B · 1×GB200（小模型大 GPU → host overhead 占比大）

GLM-4.7-FP8 · 4×GB200 · MTP spec decoding

MRV2 实现 -6.3% TPOT（zero-sync spec decoding）。

6.8 当前功能覆盖 (v0.18.0)

6.9 核心差异一览表

七、torch.compile Backend：默认 Inductor vs VllmBackend

vLLM 对 torch.compile 进行了深度定制，实现了 VllmBackend 以替代默认的 Inductor backend。

7.1 触发入口

7.2 图切分策略 (Graph Splitting)

7.3 CUDA Graph 捕获方式

7.4 编译缓存机制

7.5 Post-grad 自定义 Pass

7.6 动态形状 (Dynamic Shape) 处理

7.7 多模型组件支持

7.8 返回值与序列化

八、进程拓扑与数据并行 (DP)

8.1 进程层次模型

vLLM V1 的进程架构是三层嵌套的：

API 进程 (AsyncLLM)
  │
  ├── EngineCore 进程 #0  (DP rank 0)
  │     └── MultiprocExecutor
  │           ├── Worker 进程 #0  (GPU 0) ← TP rank 0
  │           ├── Worker 进程 #1  (GPU 1) ← TP rank 1
  │           ├── Worker 进程 #2  (GPU 2) ← TP rank 2
  │           └── Worker 进程 #3  (GPU 3) ← TP rank 3
  │
  ├── EngineCore 进程 #1  (DP rank 1)
  │     └── MultiprocExecutor
  │           ├── Worker 进程 #4  (GPU 4) ← TP rank 0
  │           └── ...
  │
  └── (可选) DPCoordinator 进程
        └── 负载均衡统计、MoE wave 协调

关键的配置公式：

# config/parallel.py
self.world_size = (
    self.pipeline_parallel_size
    * self.tensor_parallel_size
    * self.prefill_context_parallel_size
)

8.2 典型部署拓扑举例

场景 A：单卡部署（最简）

TP=1, PP=1, DP=1 → 1 个 EngineCore + 1 个 Worker

API 进程 ──ZMQ──► EngineCore ──shm──► Worker (GPU 0)

场景 B：4 卡张量并行

TP=4, PP=1, DP=1 → 1 个 EngineCore + 4 个 Worker

API 进程 ──ZMQ──► EngineCore ──shm──► Worker 0 (GPU 0)
                                  ├──► Worker 1 (GPU 1)
                                  ├──► Worker 2 (GPU 2)
                                  └──► Worker 3 (GPU 3)
                                       ↕ NCCL AllReduce

场景 C：8 卡数据并行

TP=1, PP=1, DP=8 → 8 个 EngineCore，各带 1 个 Worker

API 进程 ──ZMQ──► EngineCore 0 ──► Worker (GPU 0)
           ├───► EngineCore 1 ──► Worker (GPU 1)
           ├───► ...
           └───► EngineCore 7 ──► Worker (GPU 7)
                 ↕
           DPCoordinator (负载均衡)

场景 D：DP + TP 混合（8 卡，DP=2, TP=4）

API 进程 ──ZMQ──► EngineCore 0 (DP rank 0) ──► 4 Workers (GPU 0-3, NCCL)
           └───► EngineCore 1 (DP rank 1) ──► 4 Workers (GPU 4-7, NCCL)

8.3 DP 下 EngineCore 与 Worker 的关系

每个 EngineCore 进程独立管理自己副本内的 Workers，不同 DP rank 之间的 Worker 完全隔离：

# multiproc_executor.py — 每个 EngineCore 内部
for local_rank in range(self.local_world_size):  # local_world_size = TP × PP
    WorkerProc.make_worker_process(
        local_rank=local_rank,
        rank=global_rank,
        ...
    )

跨 DP rank 的协调由 DPCoordinator 进程完成（仅在 data_parallel_size > 1 时启动）：

• 收集各 EngineCore 的队列长度统计，发布给前端做负载均衡
• MoE 模型的 wave coordination（确保所有 DP rank 同步执行 all-to-all）

8.4 Worker 启动流程

MultiprocExecutor._init_executor()
  ├─ 创建 MessageQueue (shm_broadcast 环形缓冲)
  ├─ for local_rank in range(local_world_size):
  │     WorkerProc.make_worker_process()
  │       └─ multiprocessing.Process(target=WorkerProc.__init__)
  │            ├─ WorkerWrapperBase → 实例化 Worker (gpu_worker.py)
  │            ├─ Worker.init_device()  → CUDA device 设置, NCCL 通信组
  │            └─ Worker.load_model()   → 加载权重, torch.compile
  └─ WorkerProc.wait_for_ready()  → 等待所有 Worker 就绪

Worker 就绪后进入 worker_busy_loop，通过 rpc_broadcast_mq.dequeue 等待 EngineCore 派发 RPC 调用（如 execute_model）。

8.5 通信机制总结