【vllm】 scheduler 过程

Scheduler 的进程位置与设计解析

一、Scheduler 所在位置

进程架构全景

┌─────────────────────────────────────────────────────────────────┐ │ API Server 进程 │ │ AsyncLLM / LLMEngine │ │ EngineCoreClient (ZMQ Client) │ └─────────────────────┬───────────────────────────────────────────┘ │ ZMQ (IPC/TCP) ▼ ┌─────────────────────────────────────────────────────────────────┐ │ EngineCore 进程 ← Scheduler 在这里 │ │ │ │ EngineCoreProc │ │ └── EngineCore │ │ └── Scheduler ← ★ │ │ └── ModelExecutor │ │ └── Worker进程组 │ └─────────────────────────────────────────────────────────────────┘

进程内的类层次

# 进程入口EngineCoreProc.run_engine_core()│ ├── EngineCoreProc.__init__()# ZMQ通信层│ │ │ └── EngineCore.__init__()# 核心逻辑层│ │ │ ├── Scheduler()# ★ 调度器实例化在这里│ │ │ └── ModelExecutor()│ └── EngineCoreProc.run_busy_loop()│ └── EngineCore.step()│ ├── Scheduler.schedule()# 每步调用└── Scheduler.update_from_output()

二、为什么 Scheduler 在 EngineCore 进程中

2.1 核心原因：与执行紧密耦合

Scheduler 与 ModelExecutor 必须在同一进程的根本原因: 每个调度步骤的数据流: Scheduler.schedule() │ │ SchedulerOutput (包含): │ - 哪些请求执行 │ - KV Cache块ID │ - token数量 │ - 编码器输入 ▼ ModelExecutor.execute_model(scheduler_output) │ │ ModelRunnerOutput (包含): │ - sampled_token_ids │ - logprobs │ - kv_connector_output ▼ Scheduler.update_from_output() │ ▼ 更新请求状态、释放Cache块、生成输出

如果分离到不同进程会发生什么：

❌ 错误设计: Scheduler 在 API Server 进程 API Server进程 EngineCore进程 ┌─────────────────┐ ┌─────────────────┐ │ Scheduler │──序列化→ │ ModelExecutor │ │ │←序列化── │ │ │ KVCacheManager │ (ZMQ) │ │ └─────────────────┘ └─────────────────┘ 问题1: KV Cache块ID是GPU内存地址，序列化无意义 问题2: 每步都需要跨进程传输大量调度数据，延迟极高 问题3: KVCacheManager 状态与实际GPU内存严重不同步 问题4: 抢占决策需要实时知道GPU内存状态，跨进程无法实现

2.2 KVCacheManager 绑定关系

# Scheduler 内部直接操作 KVCacheManagerclassScheduler:defschedule(self):# 直接分配GPU KV Cache块new_blocks=self.kv_cache_manager.allocate_slots(request,...)# 直接释放GPU KV Cache块self.kv_cache_manager.free(request)# 直接查询前缀缓存状态computed_blocks=self.kv_cache_manager.get_computed_blocks(request)

KVCacheManager 本质上是 GPU 内存的"账本": GPU 物理内存 ┌──────────────────────────────────┐ │ Block[0] Block[1] Block[2] ... │ ← 实际GPU显存 └──────────────────────────────────┘ ↕ 逻辑映射 KVCacheManager (在Scheduler进程中) ┌──────────────────────────────────┐ │ 空闲块列表: [2, 5, 7, ...] │ │ 请求块表: req1→[0,1], req2→[3] │ │ 前缀Hash表: hash→[block_id] │ └──────────────────────────────────┘ 这个"账本"必须和实际使用它的调度器放在同一进程 否则"账本"和实际分配会不一致

2.3 调度器状态的完整性要求

Scheduler 维护的关键状态: ┌─────────────────────────────────────────────────────┐ │ self.requests: dict[str, Request] │ │ └── 完整请求对象，包含: │ │ - token ids (可能几千个) │ │ - 采样参数 │ │ - 计算进度 num_computed_tokens │ │ - 投机解码状态 spec_token_ids │ │ - 结构化输出状态 │ │ │ │ self.running: list[Request] │ │ self.waiting: RequestQueue │ │ │ │ self.kv_cache_manager │ │ self.encoder_cache_manager │ │ self.finished_req_ids: set[str] │ └─────────────────────────────────────────────────────┘ 这些状态在 schedule() 和 update_from_output() 之间 必须保持严格一致，任何跨进程同步都会引入不一致窗口

2.4 抢占机制的实时性要求

抢占必须在调度时同步决策: while True: new_blocks = kv_cache_manager.allocate_slots(request, ...) if new_blocks is not None: break # 分配成功，继续 # 内存不足 → 立即抢占 preempted_req = self.running.pop() kv_cache_manager.free(preempted_req) # 立即释放 # 再次尝试分配... # 这个"分配-失败-抢占-重试"循环 # 必须是原子性的同步操作 # 任何异步/跨进程调用都会破坏这个循环

三、与其他组件的边界设计

3.1 Scheduler 与 Worker 的职责分离

明确的职责边界: Scheduler (EngineCore进程) Worker (独立进程/GPU) ┌────────────────────────┐ ┌────────────────────────┐ │ 决策层: │ │ 执行层: │ │ - 哪些请求执行 │ │ - 实际的GPU计算 │ │ - 分配多少token │─────────│ - 注意力机制 │ │ - 使用哪些Cache块 │ 调度输出 │ - 采样 │ │ - 何时抢占 │ │ - KV Cache读写 │ │ - 前缀缓存命中 │◄────────│ - 返回token ids │ │ - 结构化输出约束 │ 模型输出 │ │ └────────────────────────┘ └────────────────────────┘ Scheduler 只管"资源分配决策" Worker 只管"按指令执行计算" 两者通过 SchedulerOutput 和 ModelRunnerOutput 通信

3.2 多DP引擎的 Scheduler 隔离

数据并行时每个引擎有独立的 Scheduler: ┌─────────────────────┐ ┌─────────────────────┐ │ DP Engine 0 │ │ DP Engine 1 │ │ ┌───────────────┐ │ │ ┌───────────────┐ │ │ │ Scheduler │ │ │ │ Scheduler │ │ │ │ - requests │ │ │ │ - requests │ │ │ │ - running │ │ │ │ - running │ │ │ │ - KVManager │ │ │ │ - KVManager │ │ │ └───────────────┘ │ │ └───────────────┘ │ │ ┌───────────────┐ │ │ ┌───────────────┐ │ │ │ GPU 0,1 │ │ │ │ GPU 2,3 │ │ │ └───────────────┘ │ │ └───────────────┘ │ └─────────────────────┘ └─────────────────────┘ │ │ └──────── AllReduce ───────┘ (仅同步全局是否有未完成请求) 每个 Scheduler 完全独立管理自己的请求集合 DP间只需要极少量的同步(每32步一次AllReduce)

3.3 输入处理线程的角色

EngineCore进程内的线程分工: ┌─────────────────────────────────────────────────────┐ │ EngineCore 进程 │ │ │ │ ┌─────────────────┐ ┌─────────────────────────┐ │ │ │ 输入线程 │ │ 主线程 (busy loop) │ │ │ │ │ │ │ │ │ │ ZMQ接收请求 │ │ Scheduler.schedule() │ │ │ │ 预处理(mm特征) │───→│ ModelExecutor.execute()│ │ │ │ 放入input_queue │ │ Scheduler.update() │ │ │ └─────────────────┘ └─────────────────────────┘ │ │ │ │ ┌─────────────────┐ │ │ │ 输出线程 │←── output_queue │ │ │ ZMQ发送结果 │ │ │ └─────────────────┘ │ └─────────────────────────────────────────────────────┘ 关键点: Scheduler 只在主线程中运行 完全单线程访问，无需任何锁机制 输入/输出线程只操作队列，不接触Scheduler内部状态

四、设计总结

┌─────────────────────────────────────────────────────────────┐ │ 设计决策矩阵 │ ├──────────────────┬──────────────────┬───────────────────────┤ │ 设计选择 │ 原因 │ 收益 │ ├──────────────────┼──────────────────┼───────────────────────┤ │ Scheduler在 │ KVCacheManager │ 零序列化开销 │ │ EngineCore进程 │ 必须本地访问 │ 抢占决策实时 │ ├──────────────────┼──────────────────┼───────────────────────┤ │ 与API Server │ 请求量大, │ API Server可独立扩展 │ │ 进程分离 │ 避免GIL竞争 │ EngineCore专注GPU调度 │ ├──────────────────┼──────────────────┼───────────────────────┤ │ 主线程单线程 │ 调度状态复杂, │ 无锁设计 │ │ 访问Scheduler │ 避免并发错误 │ 逻辑简单可预测 │ ├──────────────────┼──────────────────┼───────────────────────┤ │ 每DP引擎 │ 请求集合完全 │ 无跨引擎调度同步开销 │ │ 独立Scheduler │ 独立 │ 线性扩展 │ └──────────────────┴──────────────────┴───────────────────────┘ 核心设计哲学: "调度决策" 和 "资源状态" 必须共存于同一进程 任何需要实时一致性的状态都不应该跨进程分离

vLLM V1 完整执行流程与线程调度

一、进程与线程全景图

┌─────────────────────────────────────────────────────────────────────────────┐ │ API Server 进程 │ │ │ │ 主线程: FastAPI/uvicorn │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ AsyncLLM │ │ │ │ └── EngineCoreClient │ │ │ │ ├── 输入线程: 发送请求到EngineCore (ZMQ PUSH) │ │ │ │ └── 输出线程: 接收结果从EngineCore (ZMQ PULL) │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ └──────────────────────────────┬──────────────────────────────────────────────┘ │ ZMQ (IPC/TCP) ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ EngineCore 进程 │ │ │ │ 主线程(busy loop): Scheduler + 调度决策 │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ run_busy_loop() │ │ │ │ ├── _process_input_queue() ← 消费 input_queue │ │ │ │ ├── Scheduler.schedule() │ │ │ │ ├── ModelExecutor.execute_model() → 提交给Worker │ │ │ │ └── Scheduler.update_from_output() │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ 输入线程: ZMQ接收 │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ process_input_sockets() │ │ │ │ └── ZMQ DEALER socket → 反序列化 → input_queue.put() │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ │ │ │ 输出线程: ZMQ发送 │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ process_output_sockets() │ │ │ │ └── output_queue.get() → 序列化 → ZMQ PUSH socket │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ └──────────────────────────────┬──────────────────────────────────────────────┘ │ 进程间通信(共享内存/pipe) ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ Worker 进程组 (每个GPU一个) │ │ │ │ Worker-0 (GPU 0) Worker-1 (GPU 1) Worker-N (GPU N) │ │ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ │ │ 主线程: │ │ 主线程: │ │ 主线程: │ │ │ │ execute_model() │ │ execute_model() │ │ execute_model() │ │ │ │ NCCL AllReduce │◄────►│ NCCL AllReduce │◄►│ NCCL AllReduce │ │ │ └──────────────────┘ └──────────────────┘ └──────────────────┘ │ └─────────────────────────────────────────────────────────────────────────────┘

二、请求的完整生命周期

2.1 阶段划分

请求生命周期的7个阶段: Phase 1: API接收 HTTP请求 → FastAPI → AsyncLLM.generate() Phase 2: 请求入队 EngineCoreClient → ZMQ → EngineCore输入线程 → input_queue Phase 3: 预处理 input_queue → 主线程 → Request对象构建 (多模态特征处理、结构化输出FSM初始化) Phase 4: 等待调度 waiting队列 → Scheduler管理 Phase 5: 执行 schedule() → execute_model() → update_from_output() (可能经历多个step: prefill + decode × N) Phase 6: 结果返回 output_queue → 输出线程 → ZMQ → EngineCoreClient → AsyncLLM Phase 7: 清理 KV Cache释放 → 请求对象删除

三、详细执行时序

3.1 启动握手阶段

时序图: 进程启动与握手 API Server进程 EngineCore进程 │ │ │ launch_core_engines() │ │ ┌─ 创建ZMQ ROUTER socket │ │ │ (handshake_address) │ │ │ │ │ │ 启动EngineCore进程 ─────────►│ run_engine_core() 开始 │ │ │ │ │ │ EngineCoreProc.__init__() │ │ │ │ │ │◄──── HELLO (ZMQ DEALER) ────│ │ startup_handshake() │ │ │ │ │ │─── EngineHandshakeMetadata ►│ │ 收到ZMQ地址配置 │ │ (ZMQ addresses, │ │ │ │ parallel_config) │ │ │ │ │ │ EngineCore.__init__() │ │ │ │ ├── 模型加载 │ │ │ │ ├── KV Cache初始化 │ │ │ │ └── Scheduler初始化 │ │ │ │ │ │◄──── READY (num_gpu_blocks)─│ │ 初始化完成 │ │ │ │ │ wait_for_engine_startup() │ │ │ 收集num_gpu_blocks │ │ └─ 握手完成，yield给调用者 │ │ │ │ │ 启动输入/输出线程 │ │ run_busy_loop() 开始

3.2 请求接收阶段

时序图: 请求从API到EngineCore 用户 API Server(主线程) EngineCoreClient EngineCore输入线程 │ │ 输入线程 输出线程 │ │─ HTTP POST ───────►│ │ │ │ │ │ │ │ │ │ │ generate() │ │ │ │ │ 创建Request │ │ │ │ │ 分配request_id│ │ │ │ │ │ │ │ │ │─ add_request ►│ │ │ │ │ (通过queue) │ │ │ │ │ │ │ │ │ │ │ 序列化 │ │ │ │ │ EngineCoreRequest │ │ │ │─ ZMQ DEALER send ──────────►│ │ │ │ │ │ │ │ │ │ ZMQ接收 │ │ │ │ │ 反序列化 │ │ │ │ │ │ │ │ │ │ preprocess_add_request() │ │ │ │ ├── mm特征处理 │ │ │ │ │ ├── FSM初始化 │ │ │ │ │ └── 构建Request对象│ │ │ │ │ │ │ │ │ │ input_queue.put((ADD, req))

3.3 主循环执行阶段

时序图: EngineCore主线程busy loop EngineCore主线程 Scheduler ModelExecutor Worker进程 │ │ │ │ │ run_busy_loop() │ │ │ │ ┌─────────────────────────────────────────────────────┐ │ │ │ 循环开始 │ │ │ │ │ │ │ │ _process_input_queue() │ │ │ │ └── input_queue.get(block=True) ← 无任务时阻塞 │ │ │ │ ├── ADD请求 → scheduler.add_request() │ │ │ │ ├── ABORT → scheduler.finish_requests() │ │ │ │ └── UTILITY→ 调用对应方法 │ │ │ │ │ │ │ │ _process_engine_step() │ │ │ │ │ │ │ │ │ ├──► schedule() ─────────────────────────────────►│ │ │ │ │ ├── 调度running请求 │ │ │ │ │ ├── 调度waiting请求 │ │ │ │ │ ├── 分配KV Cache块 │ │ │ │ │ └── 返回 SchedulerOutput │ │ │ │ │ │ │ │ │ │ ├──► execute_model(scheduler_output, non_block=True) │ │ │ │ │ │ │────────────►│ │ │ │ │ │ │ GPU计算 │ │ │ │ │ │ │ (异步) │ │ │ │ └── 返回 Future│ │ │ │ │ │ │ │ │ │ │ ├──► get_grammar_bitmask() │ │ │ │ │ └── 计算结构化输出掩码 │ │ │ │ │ │ │ │ │ │ ├──► future.result() ◄─────────────────── │◄──────────│ │ │ │ └── 阻塞等待GPU计算完成 │ 返回结果 │ │ │ │ │ │ │ │ │ ├──► _process_aborts_queue() │ │ │ │ │ └── 处理执行期间产生的abort │ │ │ │ │ │ │ │ │ │ └──► update_from_output() ───────────────►│ │ │ │ ├── 处理采样结果 │ │ │ │ ├── 检查停止条件 │ │ │ │ ├── 释放完成请求的KV Cache │ │ │ │ └── 生成EngineCoreOutputs │ │ │ │ │ │ │ │ │ output_queue.put(outputs) │ │ │ │ │ │ │ └─────────────────────────────────────────────────────┘ │

3.4 流水线并行的批处理队列

时序图: step_with_batch_queue() (PP场景) batch_queue_size = 2 的情况: 时间轴: T1 T2 T3 T4 T5 │ │ │ │ │ 主线程: schedule(B1) schedule(B2) schedule(B3) wait(B1) wait(B2) │ │ │ │ │ execute(B1) execute(B2) execute(B3) ↓ ↓ │ │ │ │ │ GPU: [B1计算中] [B2计算中] [B3计算中] [B1完成] [B2完成] │ │ 输出: update(B1) update(B2) batch_queue状态: T1后: [B1] T2后: [B2, B1] ← 队列满 T3: pop(B1) → 阻塞等待B1完成 schedule(B3), push(B3) T3后: [B3, B2] T4: pop(B2) → 阻塞等待B2完成 ... 关键优化: GPU计算和调度决策重叠执行 消除流水线气泡

3.5 结果返回阶段

时序图: 结果从EngineCore返回到用户 EngineCore主线程 输出线程 EngineCoreClient API Server主线程 │ │ 输出线程 │ │ │ │ │ │ │ │ output_queue │ │ │ │ │ .put((idx, │ │ │ │ │ EngineCoreOutputs)) │ │ │ │ │ │ │ │ │ │ output_queue │ │ │ │ │ .get() │ │ │ │ │ │ │ │ │ │ MsgpackEncoder │ │ │ │ │ .encode_into() │ │ │ │ │ (零拷贝序列化) │ │ │ │ │ │ │ │ │ │─ ZMQ PUSH send ►│ │ │ │ │ (copy=False, │ │ │ │ │ track=True) │ │ │ │ │ │ │ │ │ │ │ ZMQ PULL recv │ │ │ │ MsgpackDecoder │ │ │ │ .decode() │ │ │ │ │ │ │ │ │ per-request │ │ │ │ output_queues