【vllm】（二）vLLM v1 Engine — 模块超深度逐行分析之三

3.10 core.py - 引擎核心文件职责: 实现vLLM推理的内循环——调度→执行→更新这是GPU推理的真正驱动者。3.10.1 EngineCore.init() 初始化流程逐行解析加载插件:load_general_plugins()— 允许第三方插件注册创建ModelExecutor:executor_class(vllm_config)— 启动GPU workers注册失败回调:executor.register_failure_callback()KV Cache初始化:_initialize_kv_caches()— 最关键的初始化步骤创建StructuredOutputManager: 约束输出管理JSON grammar等创建Scheduler:vllm_config.scheduler_config.get_scheduler_cls()KV Connector握手: 收集所有worker的KV传输元数据Batch Queue: 如果PP1启用批处理队列消除pipeline气泡前缀缓存: 如果启用初始化request_block_hasherGC优化:freeze_gc_heap()enable_envs_cache()3.10.2 _initialize_kv_caches() 完整流程instrument(span_namePrepare model)def_initialize_kv_caches(self,vllm_config):获取KV Cache规格:model_executor.get_kv_cache_specs()GPU内存Profile:model_executor.determine_available_memory()运行dummy forward pass测量峰值内存可用GPU内存 总内存 - 模型权重 - 激活内存计算KV Cache配置:get_kv_cache_configs()根据可用内存计算num_gpu_blocks可能自动降低max_model_len以适配内存同步max_model_len: 如果auto-fit降低了max_model_len调用collective_rpc(update_max_model_len)同步到所有worker生成Scheduler KV Cache配置:generate_scheduler_kv_cache_config()初始化KV Cache:model_executor.initialize_from_config()分配GPU内存给KV Cache blocks预热模型执行3.10.3 step() 核心循环defstep(self)-tuple[dict[int,EngineCoreOutputs],bool]:完整执行流程逐行检查是否有待处理请求:if not self.scheduler.has_requests(): return {}, False调度:scheduler_output self.scheduler.schedule()决定哪些请求prefill、哪些decode分配KV Cache blocks生成SchedulerOutput异步执行模型:future model_executor.execute_model(scheduler_output, non_blockTrue)non_blockTrue返回Future不等待执行完成获取grammar bitmask:grammar_output scheduler.get_grammar_bitmask(scheduler_output)用于约束输出JSON grammar等等待模型执行:model_output future.result()阻塞直到GPU计算完成条件采样: 如果model_output为None调用model_executor.sample_tokens(grammar_output)pooling模型不需要采样处理中止请求:_process_aborts_queue()检查是否有用户在中止请求更新调度器:scheduler.update_from_output(scheduler_output, model_output)更新请求状态完成/继续/抢占释放已完成的KV Cache blocks返回EngineCoreOutputs返回:(engine_core_outputs, total_num_scheduled_tokens 0)3.10.4 step_with_batch_queue() Pipeline并行变体与step()的区别异步重叠调度和执行如果batch_queue未满立即调度新batch并放入队列同时之前的batch在GPU上执行延迟grammar采样当有pending_structured_output_tokens时延迟采样先处理上一轮输出再对当前轮进行grammar约束采样消除pipeline气泡多个batch流水线执行前一个batch的PP stage 2与下一个batch的PP stage 1重叠返回None如果本step只是填充队列而没有输出可返回返回None3.10.5 post_step() 推测解码处理defpost_step(self,model_executed:bool):ifnotself.async_schedulingandself.use_spec_decodeandmodel_executed:draft_token_idsself.model_executor.take_draft_token_ids()ifdraft_token_idsisnotNone:self.scheduler.update_draft_token_ids(draft_token_ids)仅在非异步调度 推测解码启用时执行获取draft model的token预测更新scheduler以验证draft tokens3.10.6 EngineCoreProc - 进程包装器classEngineCoreProc:staticmethoddefrun_engine_core(vllm_config,executor_class,handshake_address,...):ZMQ busy loop创建EngineCore实例完成就绪握手ZMQ DEALER → frontend ROUTER进入busy loop从input_socket接收请求ZMQ DEALER根据EngineCoreRequestType分派ADD →engine_core.add_request()ABORT →engine_core.abort_requests()UTILITY → 执行utility操作调用engine_core.step_fn()执行推理将结果编码发送到output_socketZMQ PUB/PUSH循环直到收到shutdown信号DEAD信号: 当EngineCore异常退出时发送预定义的ENGINE_CORE_DEAD字节序列frontend检测到此信号后设置engine_dead标志。3.10.7 EngineCoreActor - Ray Actor变体classEngineCoreActor:Ray actor version of EngineCoreProc for multi-node deployment.使用Ray remote actor替代multiprocessing.Process适用于多节点部署场景通过Ray的placement group管理GPU资源分配3.11 core_client.py - 引擎核心客户端文件职责: 提供前后端进程之间的通信桥梁封装ZMQ消息传递和零拷贝tensor传输。3.11.1 EngineCoreClient (ABC)classEngineCoreClient(ABC):staticmethoddefmake_client(multiprocess_mode,asyncio_mode,vllm_config,...):ifasyncio_modeandnotmultiprocess_mode:raiseNotImplementedError# 暂不支持async inprocifmultiprocess_modeandasyncio_mode:returnmake_async_mp_client(...)ifmultiprocess_modeandnotasyncio_mode:returnSyncMPClient(...)returnInprocClient(...)工厂模式根据运行模式自动选择客户端实现。3.11.2 InprocClient最简单的实现直接调用EngineCore方法无IPC开销classInprocClient(EngineCoreClient):def__init__(self,*args):self.engine_coreEngineCore(*args)defget_output(self):outputs,model_executedself.engine_core.step_fn()self.engine_core.post_step(model_executed)returnoutputs.get(0)orEngineCoreOutputs()defadd_request(self,request):req,waveself.engine_core.preprocess_add_request(request)self.engine_core.add_request(req,wave)3.11.3 MPClient 基类classMPClient(EngineCoreClient):def__init__(self,asyncio_mode,vllm_config,...):self.ctxzmq.asyncio.Context()ifasyncio_modeelsezmq.Context()self.resourcesBackgroundResources(ctxsync_ctx)self._finalizerweakref.finalize(self,self.resources)关键设计weakref.finalize: 确保进程退出时清理ZMQ资源BackgroundResources: 持有所有需要清理的资源引用ZMQ context使用2个IO线程io_threads2提升吞吐3.11.4 BackgroundResourcesdataclassclassBackgroundResources:ctx:zmq.Context engine_manager:CoreEngineProcManager|CoreEngineActorManager coordinator:DPCoordinator|Noneoutput_socket:zmq.Socket|zmq.asyncio.Socket input_socket:zmq.Socket|zmq.asyncio.Socket...engine_dead:boolFalsecall() 清理逻辑设置engine_dead True关闭engine_manager停止所有core进程关闭coordinator关闭所有ZMQ socket取消asyncio tasksasync模式发送shutdown信号到sync output线程sync模式3.11.5 AsyncMPClientclassAsyncMPClient(MPClient):asyncdefget_output_async(self):framesawaitself.output_socket.recv_multipart()self.resources.validate_alive(frames)outputsself.decoder.decode(frames[0])returnoutputsasyncdefadd_request_async(self,request):dataself.encoder.encode(EngineCoreRequestType.ADD,request)awaitself.input_socket.send_multipart(data)# 同时发送tensor数据self._send_tensors(request)核心异步操作get_output_async(): await ZMQ PULL接收add_request_async(): await ZMQ DEALER发送Tensor发送是同步的MPQueue.put不阻塞事件循环太久3.11.6 SyncMPClientclassSyncMPClient(MPClient):defget_output(self):# 后台线程从ZMQ PULL接收put到output_queuereturnself.output_queue.get()defadd_request(self,request):dataself.encoder.encode(EngineCoreRequestType.ADD,request)self.input_socket.send_multipart(data)使用后台线程_output_queue_loop从ZMQ拉取输出主线程通过标准queue.Queue.get()同步获取输出适配LLMEngine的同步step()模式3.11.7 DPLBAsyncMPClient - 内部DP负载均衡classDPLBAsyncMPClient(AsyncMPClient):Internal load balancer: client balances requests to all DP ranks.维护所有DP rank的连接使用round-robin或基于队列长度的策略分配请求聚合所有rank的输出处理wave_complete同步3.11.8 DPAsyncMPClient - 外部DP负载均衡每个client只连接一个特定的DP rank外部负载均衡器决定请求路由通过dp_rank参数指定目标rank3.12 llm_engine.py - LLM引擎同步文件职责: 提供同步的推理API适配V0风格的add_request()/step()使用模式。3.12.1 LLMEngine 类classLLMEngine:def__init__(self,vllm_config,executor_class,...):self.input_processorInputProcessor(...)self.output_processorOutputProcessor(...)self.engine_coreEngineCoreClient.make_client(multiprocess_modenotvllm_config.model_config.enable_lora,asyncio_modeFalse,...)注意: LoRA启用时使用InprocClient同进程因为LoRA管理需要直接访问engine_core。3.12.2 add_request()defadd_request(self,request_id,prompt,params,...):processed_inputsself.input_processor.process_inputs(...)self.output_processor.add_request(...)self.engine_core.add_request(processed_inputs)3.12.3 step()defstep(self)-list[RequestOutput]:outputsself.engine_core.get_output()request_outputsself.output_processor.process_outputs(outputs)returnrequest_outputs同步调用get_output()阻塞等待Core输出处理输出后返回给调用者典型用法while has_requests: outputs engine.step()3.13 async_llm.py - 异步LLM引擎文件职责: 提供基于asyncio的异步推理API支持高并发流式输出。3.13.1 AsyncLLM 类classAsyncLLM:def__init__(self,vllm_config,executor_class,...):self.input_processorInputProcessor(...)self.output_processorOutputProcessor(...)self.engine_coreEngineCoreClient.make_async_mp_client(...)self.output_handler:asyncio.Task|NoneNone3.13.2 generate() 异步生成器asyncdefgenerate(self,prompt,params,request_id,...):# 1. 处理输入processed_inputsself.input_processor.process_inputs(...)# 2. 注册输出追踪queueself.output_processor.add_request(...)# 3. 发送到Coreawaitself.engine_core.add_request_async(processed_inputs)# 4. 启动output_handler如果未启动self._ensure_output_handler()# 5. 返回异步生成器returnself._generate_stream(queue,request_id)_generate_stream() 逐行逻辑asyncdef_generate_stream(self,queue,request_id):whileTrue:outputawaitqueue.get()ifisinstance(output,EngineDeadError):raiseoutputyieldoutputifoutput.finished:break3.13.3 _output_handler() 后台任务asyncdef_output_handler(self):whileTrue:# 1. 从Core获取输出outputsawaitself.engine_core.get_output_async()# 2. 处理输出request_outputsself.output_processor.process_outputs(outputs)# 3. 分发到per-request Queueforoutputinrequest_outputs:stateself.output_processor.get_state(output.request_id)state.queue.put(output)3.13.4 _ensure_output_handler()def_ensure_output_handler(self):ifself.output_handlerisNoneorself.output_handler.done():self.output_handlerasyncio.create_task(self._output_handler())懒启动第一个请求到来时才创建后台任务如果任务异常退出自动重启3.13.5 scale_elastic_ep()asyncdefscale_elastic_ep(self,new_data_parallel_size):set_scaling_elastic_ep(True)try:awaitself.engine_core.scale_elastic_ep(new_data_parallel_size)self.vllm_config.parallel_config.data_parallel_sizenew_data_parallel_sizefinally:set_scaling_elastic_ep(False)动态调整DP大小弹性扩展使用set_scaling_elastic_ep保护配置更新期间的并发访问3.13.6 属性方法propertydefis_running(self)-bool:returnself.output_handlerisNoneornotself.output_handler.done()propertydefis_stopped(self)-bool:returnself.erroredpropertydeferrored(self)-bool:returnself.engine_core.resources.engine_deadornotself.is_running3.14 utils.py - 工具模块文件职责: 提供引擎进程管理、设备控制、ZMQ地址分配等基础设施。3.14.1 CoreEngineState 枚举classCoreEngineState(Enum):NEWauto()# 刚创建等待连接CONNECTEDauto()# 已连接到frontend等待就绪READYauto()# 初始化完成可以接收请求3.14.2 CoreEngineclassCoreEngine:def__init__(self,index0,localTrue):self.locallocal self.identityindex.to_bytes(2,little)# ZMQ ROUTER identityself.stateCoreEngineState.NEW每个DP rank对应一个CoreEngineidentity用于ZMQ ROUTER/DEALER模式的消息路由3.14.3 EngineZmqAddressesdataclassclassEngineZmqAddresses:inputs:list[str]# 前端→Core的输入地址outputs:list[str]# Core→前端的输出地址coordinator_input:str|None# Coordinator输入地址coordinator_output:str|None# Coordinator输出地址frontend_stats_publish_address:str|None# 统计发布地址3.14.4 CoreEngineProcManagerclassCoreEngineProcManager:def__init__(self,local_engine_count,start_index,...):# 1. 创建multiprocessing.Process列表# 2. 设置设备控制环境变量DP场景# 3. 配置NUMA亲和性# 4. 启动所有进程# 5. 如果任何进程启动失败停止所有进程设备控制set_device_control_env_var()在DP场景下每个engine进程只能看到自己的GPU临时修改CUDA_VISIBLE_DEVICES或等效环境变量使用unittest.mock.patch.dict安全地修改os.environNUMA亲和性numa_utils.configure_subprocess()将engine进程绑定到特定NUMA节点减少跨NUMA内存访问延迟3.14.5 CoreEngineActorManagerRay Actor版本的进程管理器支持多节点部署使用Ray placement group管理GPU资源支持MoE模型的DPMoEEngineCoreActor自动创建placement groups或使用传入的3.14.6 SignalCallbackclassSignalCallback:def__init__(self,callback):self._eventthreading.Event()self._threadthreading.Thread(targetself._run,daemonTrue)self._thread.start()deftrigger(self):self._event.set()# 安全地从信号处理器调用信号处理器中只能调用async-signal-safe函数Event.set()是async-signal-safe实际回调在专用线程中执行附录A. 关键配置参数参数默认值说明data_parallel_size1DP并行度data_parallel_external_lbFalse外部DP负载均衡enable_elastic_epFalse弹性EPenable_chunked_prefillTrue分块prefillenable_prefix_cachingFalse前缀缓存max_model_lenauto最大序列长度可能被auto-fit降低block_size16KV Cache block大小B. ZMQ Socket类型总结Socket类型用途input_socketDEALER (client) / ROUTER (core)请求输入output_socketPULL (client) / PUSH (core)输出拉取stats_socketSUB (client) / PUB (coordinator)统计订阅first_req_socketPAIRfirst-request-per-wavehandshake_socketDEALER/ROUTER就绪握手C. 模块依赖关系图D. 完整请求生命周期总结客户端调用generate(prompt, params)InputProcessor验证和转换请求OutputProcessor注册RequestStateEngineCoreClient通过ZMQ发送EngineCoreRequestTensorIpcSender传输多模态/LoRA张量EngineCore.step(): schedule → execute_model → update_from_outputEngineCoreOutput通过ZMQ返回FrontendOutputProcessor: detokenize → stop_check → logprobs → build_outputPer-request Queue分发到客户端生成器客户端yield RequestOutput直到finishedTrue