EngineCoreProc:ZMQ 後臺行程包裝

源码版本v0.25.1

職責

EngineCoreProc 是 EngineCore 的子類(core.py:896-897),把它整個扔進一個獨立的後臺行程 (background process) 裡跑。docstring 寫得很直白:ZMQ-wrapper for running EngineCore in background process.。它幹三件事:在行程啟動時跟前端握手拿到 ZMQ socket 地址、起兩個 IO 執行緒把 socket 和兩個 queue.Queue 橋接起來(core.py:915-916)、然後 run_busy_loop 在主執行緒裡反覆從 input_queue 拿請求 → step_fn() 跑排程和前向 → 把 outputs 塞進 output_queue(core.py:1259-1267)。

這一層是 v1 多行程架構的關鍵。前端行程 (SyncMPClient 或 AsyncMPClient) 通過 ZMQ 把 EngineCoreRequest 序列化發過來,EngineCoreProc 的 input 執行緒收到後 put_nowait 進 input_queue,主循環在排程間隙從佇列裡取出來交給 _handle_client_request 分發(core.py:1372-1405)。outputs 反方向流:step_fn() 產出的 EngineCoreOutputs 被丟進 output_queue,output 執行緒從佇列拿出來序列化通過 ZMQ 推回前端。ZMQ socket IO 在收發時會釋放 GIL,所以這兩條 IO 執行緒可以真正和 GPU 前向重疊(core.py:974-1001),序列化/反序列化也被藏到前向背後。

設計動機

• 行程隔離:EngineCoreProc 跑在獨立行程裡,前端 Python 崩了不會拖垮已經跑進 GPU 的 worker,worker OOM 也不會把 API 行程一起帶走。這就是 v1 把 EngineCore 抽出來的核心理由——v0 的 LLMEngine 在同行程裡什麼都幹,崩一處全崩。
• ZMQ + queue.Queue 兩級緩衝:ZMQ socket 不直接被主循環用,中間夾一層 queue.Queue(core.py:915-916)。socket IO 在 IO 執行緒裡,主循環只 queue.get(block=...),免得主循環被 socket recv 阻塞或被 ZMQ 內部鎖拖住;同時 socket IO 釋放 GIL 時,主循環能立刻拿回 GIL 跑前向。
• 兩個 IO 執行緒:process_input_sockets 和 process_output_sockets(core.py:980-1001)各自獨立,輸入執行緒負責反序列化 + put_nowait,輸出執行緒負責 get + 序列化 + send_multipart,兩邊互不阻塞。
• 啟動握手 (handshake):構造函數里 _perform_handshakes(core.py:932-938)拿到一組 EngineZmqAddresses,然後 input 執行緒在 ROUTER/DEALER 上發一條 EngineCoreReadyResponse 告訴前端"我起來了,這是 max_model_len、num_gpu_blocks、kv_cache_size_tokens"(core.py:1525-1546)。前端在 MPClient.__init__ 裡等這條 ready 訊息,等不到就當啟動失敗。
• run_engine_core 行程入口:這函數是 multiprocessing.Process(target=run_engine_core, ...) 的 target(core.py:1154-1224)。它把 EngineCoreProc 實例化、註冊 SIGTERM/SIGINT、最後調 run_busy_loop。任何異常都走 except Exception → _send_engine_dead → raise,行程退出時 finally 裡調 engine_core.shutdown()(core.py:1226-1242)。

關鍵檔案

• EngineCoreProc class:896-910 — class EngineCoreProc(EngineCore): 定義 + 類常數 ENGINE_CORE_DEAD = b"ENGINE_CORE_DEAD"。
• EngineCoreProc.__init__:903-1009 — 起 input_queue/output_queue、tensor IPC receiver、_perform_handshakes、super().__init__、起兩個 IO 執行緒。
• queues + executor_fail_callback:915-919 — input_queue = queue.Queue[...]、output_queue = queue.Queue[...],executor 失敗回調往 input_queue 塞 EXECUTOR_FAILED。
• IO threads startup:974-1001 — process_input_sockets 和 process_output_sockets 兩個 daemon thread 啟動,ready_event 在 input 執行緒 ready 後才 set。
• run_engine_core:1154-1224 — 行程入口,選 EngineCoreProc 或 DPEngineCoreProc,註冊信號 handler,調 run_busy_loop。
• signal handlers:1204-1222 — wakeup_engine 往 input_queue 塞 WAKEUP,SIGTERM/SIGINT 把 shutdown_state 改成 REQUESTED。
• run_busy_loop:1259-1267 — while self._handle_shutdown(): _process_input_queue(); _process_engine_step() 跑到 shutdown 後 raise SystemExit。
• _process_input_queue:1269-1298 — 沒活時阻塞 input_queue.get(block=True),有活時非阻塞 drain。
• _process_engine_step:1300-1317 — 調 step_fn()、output_queue.put_nowait(output)、post_step,沒跑前向但 scheduler 還有活時 time.sleep(0.001) 讓 GIL。
• _handle_client_request:1372-1405 — EngineCoreRequestType 分發:WAKEUP/ADD/ABORT/UTILITY/EXECUTOR_FAILED。
• _send_engine_dead:1470-1482 — 往 output_queue 塞 ENGINE_CORE_DEAD,等 output 執行緒把訊息發出去再退出。
• process_input_sockets:1484-1587 — input 執行緒主體,用 zmq.Poller 監聽 DEALER + 可選 coord XSUB,反序列化後 put_nowait 進 input_queue。
• process_output_sockets:1589-1654 — output 執行緒主體,從 output_queue.get() 拿 outputs,encoder.encode_into + send_multipart(copy=False, track=True) 零拷貝發回前端。

資料流

請求從 ZMQ input socket 進來,先被 input 執行緒反序列化塞進 input_queue,主循環在 _process_input_queue 裡取出來交給 _handle_client_request 分發:

# vllm/v1/engine/core.py L1556-L1587
while True:
    for input_socket, _ in poller.poll():
        # (RequestType, RequestData)
        type_frame, *data_frames = input_socket.recv_multipart(copy=False)
        # NOTE(yongji): ignore READY message sent by DP coordinator
        # that is used to notify newly started engines
        if type_frame.buffer == b"READY":
            assert input_socket == coord_socket
            continue
        request_type = EngineCoreRequestType(bytes(type_frame.buffer))

        # Deserialize the request data.
        request: Any
        if request_type == EngineCoreRequestType.ADD:
            req: EngineCoreRequest = add_request_decoder.decode(data_frames)
            try:
                request = self.preprocess_add_request(req)
            except Exception:
                self._handle_request_preproc_error(req)
                continue
        else:
            request = generic_decoder.decode(data_frames)

            if request_type == EngineCoreRequestType.ABORT:
                # Aborts are added to *both* queues, allows us to eagerly
                # process aborts while also ensuring ordering in the input
                # queue to avoid leaking requests. This is ok because
                # aborting in the scheduler is idempotent.
                self.aborts_queue.put_nowait(request)

        # Push to input queue for core busy loop.
        self.input_queue.put_nowait((request_type, request))

注意 ABORT 被同時塞進 aborts_queue 和 input_queue,因為 abort 在排程器裡是冪等的 (core.py:1276-1279),讓主循環在 step 間隙也能從 aborts_queue 裡 drain 掉,避免 leak 已經 abort 的請求。

主循環拿到請求後,_handle_client_request 按 EngineCoreRequestType 分發(core.py:1372-1405):

# vllm/v1/engine/core.py L1377-L1401
if request_type == EngineCoreRequestType.WAKEUP:
    return
elif request_type == EngineCoreRequestType.ADD:
    req, request_wave = request
    if self._reject_add_in_shutdown(req):
        return
    self.add_request(req, request_wave)
elif request_type == EngineCoreRequestType.ABORT:
    self.abort_requests(request)
elif request_type == EngineCoreRequestType.UTILITY:
    client_idx, call_id, method_name, args = request
    if self._reject_utility_in_shutdown(client_idx, call_id, method_name):
        return
    output = UtilityOutput(call_id)
    # Lazily look-up utility method so that failure will be handled/returned.
    get_result = lambda: (
        (method := getattr(self, method_name))
        and method(*self._convert_msgspec_args(method, args))
    )
    enqueue_output = lambda out: self.output_queue.put_nowait(
        (client_idx, EngineCoreOutputs(utility_output=out))
    )
    self._invoke_utility_method(method_name, get_result, output, enqueue_output)
elif request_type == EngineCoreRequestType.EXECUTOR_FAILED:
    raise RuntimeError("Executor failed.")

UTILITY 是個通用 RPC 通道:前端把方法名 + msgspec args 發過來,EngineCoreProc 用 getattr 反射呼叫,結果走 output_queue 回送。_invoke_utility_method 還能處理返回 Future 的情況(core.py:1440-1446),非同步工具完成後才回送 output。

邊界與失敗

• 行程崩潰通知:run_engine_core 在 except Exception 裡調 engine_core._send_engine_dead()(core.py:1229-1235),往 output_queue 塞一個 ENGINE_CORE_DEAD 字節串(core.py:1470-1474),output 執行緒收到這個特殊值後向所有 output socket 廣播(core.py:1625-1628),前端 MPClient 收到後拋錯而不是無限等下去。
• Executor 失敗回調:__init__ 裡把 executor_fail_callback 閉包註冊到 model_executor(core.py:917-919),worker 內部報錯時回調往 input_queue 塞一條 EXECUTOR_FAILED,主循環 _handle_client_request 拿到就 raise RuntimeError("Executor failed.")(core.py:1400-1401),整個行程會被 run_engine_core 的 except 捕到走 _send_engine_dead。
• Shutdown 模式:_handle_shutdown 看 shutdown_state(core.py:1324-1360),REQUESTED 狀態下 shutdown_timeout==0 走 abort 立刻殺所有在飛請求,非 0 走 drain 等所有請求跑完。shutdown 期間新進來的 ADD 和 UTILITY 會被顯式拒絕(core.py:1407-1432),_reject_add_in_shutdown 給客戶端回一條 abort output,_reject_utility_in_shutdown 回一條帶 failure_message="Server shutting down" 的 UtilityOutput。
• input 執行緒死了要報錯:__init__ 裡 ready_event.wait(timeout=10) 循環等 input 執行緒 ready(core.py:1003-1009),10 秒還沒 ready 且執行緒死了就 raise RuntimeError("Input socket thread died during startup");如果只是沒 ready 還活著,就繼續等 DP Coordinator 的 READY 訊息。
• 零拷貝發送 + MessageTracker:output 執行緒用 send_multipart(buffers, copy=False, track=True)(core.py:1646-1654),返回 MessageTracker,沒發送完的 tracker + outputs 引用 + buffer 被掛到 pending deque 上,避免 GC 把還沒發出去的 backing buffer 回收。reuse_buffers 限制最多 len(sockets) + 1 個複用 buffer,防止記憶體無限增長。
• WAKEUP 是空操作:信號 handler 通過 wakeup_engine 往 input_queue 塞 WAKEUP,只是把阻塞在 input_queue.get(block=True) 的主循環喚醒(core.py:1377-1378),本身不做事,真正的 shutdown 處理在 _handle_shutdown 裡看 shutdown_state 完成。
• finally 裡恢復信號:run_engine_core 在 finally 裡 signal.signal(signal.SIGTERM, signal.SIG_DFL)(core.py:1236-1242),把信號 handler 恢復成預設,然後調 engine_core.shutdown() 釋放 model_executor / scheduler / 分佈式狀態。shutdown 也會 gc.unfreeze()(core.py:652-656),把啟動時 gc.freeze() 凍住的 startup heap 重新交給 GC,否則同行程模式(unit test)會洩漏顯存。

小結

EngineCoreProc 是 v1 多行程架構的承重牆:它繼承 EngineCore,在獨立行程裡跑 ZMQ + 兩個 queue.Queue + 兩個 IO 執行緒,讓 socket IO 和 GPU 前向真正重疊;run_engine_core 是行程入口,負責實例化、信號註冊、異常兜底和 shutdown。所有前端 (SyncMPClient / AsyncMPClient) 和後端 (EngineCore) 的通訊都經過這一層。繼續往裡看可以去 /engine/engine-core 看引擎核心本身怎麼裝配,去 /engine/llm-engine 看前端薄殼,客戶端三種實現見 /client/inproc-mp,排程器內部見 /scheduler/scheduler。

對照官方資料:vLLM 文件 · README