EngineCoreProc: ZMQ background process wrapper

源码版本v0.25.1

Responsibilities

EngineCoreProc is a subclass of EngineCore (core.py:896-897) that runs the whole thing in a separate background process. The docstring is straightforward: ZMQ-wrapper for running EngineCore in background process.. It does three things: at process startup it handshakes with the frontend to obtain ZMQ socket addresses; it starts two IO threads that bridge sockets and two queue.Queue instances (core.py:915-916); and run_busy_loop repeatedly takes requests from input_queue → step_fn() runs scheduling and forward → pushes outputs into output_queue on the main thread (core.py:1259-1267).

This layer is the crux of the v1 multi-process architecture. The frontend process (SyncMPClient or AsyncMPClient) serializes an EngineCoreRequest and sends it over ZMQ. The EngineCoreProc input thread receives it and put_nowaits it into input_queue; the main loop pulls it out between scheduling rounds and hands it to _handle_client_request for dispatch (core.py:1372-1405). Outputs flow the other direction: the EngineCoreOutputs produced by step_fn() are pushed into output_queue; the output thread pulls them out, serializes them, and pushes them back to the frontend over ZMQ. ZMQ socket IO releases the GIL on send/receive, so these two IO threads genuinely overlap with GPU forward (core.py:974-1001), and serialization/deserialization is hidden behind the forward pass.

Design motivation

• Process isolation: EngineCoreProc runs in its own process. A frontend Python crash does not take down workers already running on the GPU, and a worker OOM does not take down the API process. This is the core reason v1 factored EngineCore out — v0's LLMEngine did everything in-process, so one crash anywhere took down everything.
• Two-stage buffering with ZMQ + queue.Queue: ZMQ sockets are not used directly by the main loop; a queue.Queue sits in between (core.py:915-916). Socket IO happens in IO threads; the main loop only does queue.get(block=...), so it is not blocked by socket recv or stuck behind ZMQ's internal locks. When socket IO releases the GIL, the main loop can immediately reclaim the GIL and run forward.
• Two IO threads: process_input_sockets and process_output_sockets (core.py:980-1001) are independent. The input thread handles deserialization + put_nowait; the output thread handles get + serialization + send_multipart; the two do not block each other.
• Startup handshake: the constructor calls _perform_handshakes (core.py:932-938) to obtain a set of EngineZmqAddresses. The input thread then sends an EngineCoreReadyResponse on the ROUTER/DEALER telling the frontend "I am up, here are max_model_len, num_gpu_blocks, kv_cache_size_tokens" (core.py:1525-1546). The frontend waits for this ready message in MPClient.__init__; if it does not arrive, startup is treated as failed.
• run_engine_core is the process entry point: this function is the target of multiprocessing.Process(target=run_engine_core, ...) (core.py:1154-1224). It instantiates EngineCoreProc, registers SIGTERM/SIGINT, and finally calls run_busy_loop. Any exception goes through except Exception → _send_engine_dead → raise; on exit the finally block calls engine_core.shutdown() (core.py:1226-1242).

Key files

• EngineCoreProc class:896-910 — class EngineCoreProc(EngineCore): definition + class constant ENGINE_CORE_DEAD = b"ENGINE_CORE_DEAD".
• EngineCoreProc.__init__:903-1009 — sets up input_queue/output_queue, tensor IPC receiver, _perform_handshakes, super().__init__, and starts the two IO threads.
• queues + executor_fail_callback:915-919 — input_queue = queue.Queue[...], output_queue = queue.Queue[...]; the executor failure callback pushes EXECUTOR_FAILED into input_queue.
• IO threads startup:974-1001 — starts the two daemon threads process_input_sockets and process_output_sockets; ready_event is only set after the input thread is ready.
• run_engine_core:1154-1224 — process entry; picks EngineCoreProc or DPEngineCoreProc, registers signal handlers, calls run_busy_loop.
• signal handlers:1204-1222 — wakeup_engine pushes WAKEUP into input_queue; SIGTERM/SIGINT switches shutdown_state to REQUESTED.
• run_busy_loop:1259-1267 — while self._handle_shutdown(): _process_input_queue(); _process_engine_step(); raises SystemExit once shutdown is complete.
• _process_input_queue:1269-1298 — blocking input_queue.get(block=True) when idle, non-blocking drain when there is work.
• _process_engine_step:1300-1317 — calls step_fn(), output_queue.put_nowait(output), post_step; when there is no forward to run but the scheduler still has work, does time.sleep(0.001) to yield the GIL.
• _handle_client_request:1372-1405 — dispatches by EngineCoreRequestType: WAKEUP/ADD/ABORT/UTILITY/EXECUTOR_FAILED.
• _send_engine_dead:1470-1482 — pushes ENGINE_CORE_DEAD into output_queue; waits for the output thread to send the message before exiting.
• process_input_sockets:1484-1587 — input thread body; uses a zmq.Poller to listen on DEALER + optional coord XSUB, deserializes, and put_nowaits into input_queue.
• process_output_sockets:1589-1654 — output thread body; pulls outputs from output_queue.get(), does encoder.encode_into + send_multipart(copy=False, track=True) to send them back to the frontend zero-copy.

Data flow

Requests come in from the ZMQ input socket, are deserialized by the input thread, and pushed into input_queue. The main loop pulls them out in _process_input_queue and hands them to _handle_client_request for dispatch:

# 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))

Note that ABORT is pushed to both aborts_queue and input_queue, because abort is idempotent in the scheduler (core.py:1276-1279). The main loop can also drain from aborts_queue between steps, avoiding leaks of already-aborted requests.

After the main loop picks up a request, _handle_client_request dispatches by 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 is a generic RPC channel: the frontend sends a method name + msgspec args; EngineCoreProc uses getattr to reflectively invoke the method and sends the result back through output_queue. _invoke_utility_method also handles the case where the return value is a Future (core.py:1440-1446); the output is only sent back after the async utility completes.

Boundaries and failure modes

• Process crash notification: in except Exception, run_engine_core calls engine_core._send_engine_dead() (core.py:1229-1235) which pushes an ENGINE_CORE_DEAD byte string into output_queue (core.py:1470-1474). When the output thread sees this special value, it broadcasts to all output sockets (core.py:1625-1628). The frontend MPClient then raises instead of waiting forever.
• Executor failure callback: in __init__, the executor_fail_callback closure is registered with model_executor (core.py:917-919). When a worker reports an error, the callback pushes an EXECUTOR_FAILED into input_queue. The main loop's _handle_client_request sees it and raise RuntimeError("Executor failed.") (core.py:1400-1401), and the whole process is caught by run_engine_core's except, which calls _send_engine_dead.
• Shutdown modes: _handle_shutdown looks at shutdown_state (core.py:1324-1360). In REQUESTED state, shutdown_timeout==0 takes the abort path and immediately kills all in-flight requests; a non-zero value takes the drain path and waits for all requests to complete. ADD and UTILITY arriving during shutdown are explicitly rejected (core.py:1407-1432). _reject_add_in_shutdown sends the client an abort output; _reject_utility_in_shutdown sends a UtilityOutput with failure_message="Server shutting down".
• Input thread death must be reported: in __init__, ready_event.wait(timeout=10) loops waiting for the input thread to be ready (core.py:1003-1009). If 10 seconds pass without ready and the thread is dead, it raise RuntimeError("Input socket thread died during startup"); if the thread is alive but not ready yet, it keeps waiting for the DP Coordinator's READY message.
• Zero-copy send + MessageTracker: the output thread uses send_multipart(buffers, copy=False, track=True) (core.py:1646-1654), which returns a MessageTracker. Trackers that have not finished sending, along with their outputs references and buffers, are attached to a pending deque, preventing GC from reclaiming backing buffers that have not been sent yet. reuse_buffers caps the number of reused buffers at len(sockets) + 1 to prevent unbounded memory growth.
• WAKEUP is a no-op: the signal handler uses wakeup_engine to push a WAKEUP into input_queue. This only wakes the main loop blocked on input_queue.get(block=True) (core.py:1377-1378); it does nothing itself. The real shutdown handling happens in _handle_shutdown by inspecting shutdown_state.
• finally restores signals: in finally, run_engine_core does signal.signal(signal.SIGTERM, signal.SIG_DFL) (core.py:1236-1242), restoring signal handlers to defaults, then calls engine_core.shutdown() to release model_executor / scheduler / distributed state. shutdown also calls gc.unfreeze() (core.py:652-656), returning the startup heap frozen by gc.freeze() at startup back to the GC. Otherwise the in-process mode (unit tests) would leak GPU memory.

Summary

EngineCoreProc is the load-bearing wall of the v1 multi-process architecture: it inherits from EngineCore and runs ZMQ + two queue.Queue instances + two IO threads in a separate process, so that socket IO and GPU forward truly overlap. run_engine_core is the process entry point, responsible for instantiation, signal registration, exception handling, and shutdown. All communication between the frontend (SyncMPClient / AsyncMPClient) and the backend (EngineCore) goes through this layer. To go deeper, read /engine/engine-core for how the engine core itself is assembled, and /engine/llm-engine for the frontend thin shell. The three client implementations are in /client/inproc-mp; the scheduler internals are in /scheduler/scheduler.

See official docs: vLLM docs · README