EngineCoreClient: in-process and multi-process client paths

源码版本v0.25.1

Responsibilities

The vLLM frontend (LLM / AsyncLLM / serving entry) does not touch EngineCore directly — it goes through a client layer to push requests in and pull outputs out. This layer is EngineCoreClient, which unifies two orthogonal dimensions — "sync vs async" and "same-process vs subprocess" — onto one set of abstract methods. The file vllm/v1/engine/core_client.py packs all implementations.

The two key implementations are InprocClient and MPClient. The former constructs EngineCore directly in-process and runs one step at a time — no busy loop, no ZMQ — mainly serving as a thin shell for the V0-style LLMEngine.add_request()/step(). The latter wraps EngineCore as an EngineCoreProc subprocess and exchanges requests and outputs via a pair of ZMQ ROUTER/PULL sockets; the calling thread only touches sockets. MPClient is the base class; concrete variants are SyncMPClient (sync, with a background thread receiving frames from the output socket) and AsyncMPClient (asyncio, with a background task running the same socket). The latter is covered on a separate page.

EngineCoreClient.make_client is a static factory picking an implementation based on two booleans, multiprocess_mode and asyncio_mode. The combination (asyncio=True, multiprocess=False) raises NotImplementedError — EngineCore itself is not asyncio-friendly, so async must be paired with multi-process.

Design motivation

• Decoupling frontend and EngineCore lifecycles: Calling EngineCore in-process is simple, but to run multiple EngineCores concurrently (data parallelism) or to support elastic EP (adding/removing ranks online), EngineCore must be pushed into a separate process and the frontend only holds a socket. The line weakref.finalize(self, self.resources) in MPClient.__init__ is the key: even if an exception is thrown mid-construction, the background subprocess and ZMQ context are reclaimed by BackgroundResources.__call__ without leaking.
• Sync and async interfaces share the same ZMQ protocol: SyncMPClient uses queue.Queue to move frames from the output socket to the main thread; AsyncMPClient uses asyncio.Queue. But MPClient shares the encoder/decoder, ready handshake, engine monitor, and the utility future registry. Differences are confined to the two subclasses.
• InprocClient's sleep(mode="wait") is directly disabled (InprocClient.sleep L324-L328), because in in-process mode there is no concurrent scheduler to "wait for idle then sleep" — it can only abort. This boundary is supported in subprocess mode — the subprocess has its own busy loop and scheduler state machine.
• Utility calls go through a future registry: RPCs like collective_rpc and save_sharded_state that need return values cannot be fire-and-forget like add_request. call_utility generates a call_id, puts a future into self.utility_results, and when the matching UtilityOutput comes back from the output socket, _process_utility_output uses the call_id to fetch the future and set_result it. This mechanism is identical across sync and async clients — only the future type differs (concurrent.futures.Future vs asyncio.Future).
• Engine death propagation: When the subprocess dies unexpectedly, the monitor thread sets BackgroundResources.engine_dead to True; any subsequent _send_input is blocked by ensure_alive and raises EngineDeadError. On the output side, if a single ENGINE_CORE_DEAD frame is received, validate_alive raises the same flag, ensuring both the sending and receiving sides observe the death.

Key files

• EngineCoreClient + make_client:71-105 — Abstract base class and static factory, choosing InprocClient/SyncMPClient/AsyncMPClient based on multiprocess_mode/asyncio_mode.
• make_async_mp_client:107-132 — Async factory, picking AsyncMPClient/DPAsyncMPClient/DPLBAsyncMPClient based on data_parallel_size and whether an external LB is used.
• InprocClient.__init__ / get_output:276-292 — Directly constructs EngineCore; get_output is just a sync call to step_fn() + post_step().
• InprocClient.add_request / abort / shutdown:297-306 — Directly forwards to self.engine_core; no serialization, no socket.
• InprocClient.sleep rejects wait:324-328 — In-process mode does not support wait pause; only abort.
• BackgroundResources:370-458 — Dataclass + __call__ as the weakref.finalize callback, uniformly closing sockets / stopping tasks / shutting down the engine manager.
• validate_alive:454-457 — On receiving a single ENGINE_CORE_DEAD frame, marks engine_dead and raises EngineDeadError.
• MPClient.__init__:480-649 — Creates ZMQ context/sockets, optionally launch_core_engines, handshakes ready, starts monitor; rolls back on failure.
• MPClient.shutdown:651-662 — detach()-es the finalizer, stops the engine manager, runs BackgroundResources cleanup.
• start_engine_core_monitor:685-712 — Daemon thread monitor_engine_liveness; on engine death sets engine_dead and triggers self.shutdown().
• _apply_ready_response:714-754 — Decodes EngineCoreReadyResponse, backfills max_model_len/num_gpu_blocks/block_size from the subprocess into the frontend's vllm_config.
• SyncMPClient.__init__:779-847 — Creates a queue.Queue, starts a process_outputs_socket daemon thread polling the output socket + shutdown socket.
• SyncMPClient.get_output / _send_input / call_utility:849-881 — Blocks on outputs_queue.get(); when sending requests, optionally tracks MessageTracker to prevent tensors from being GC'd early.

Data flow

SyncMPClient pulls outputs via a daemon thread that polls the ZMQ output_socket, decodes the frames, and pushes them into a queue.Queue. The frontend thread's get_output() is just queue.get() — adapting ZMQ's non-blocking socket into a synchronous blocking Python interface. This process_outputs_socket is the core of the sync path:

def process_outputs_socket():
    assert isinstance(out_socket, zmq.Socket)
    shutdown_socket = ctx.socket(zmq.PAIR)
    try:
        shutdown_socket.bind(shutdown_path)
        poller = zmq.Poller()
        poller.register(shutdown_socket, zmq.POLLIN)
        poller.register(out_socket, zmq.POLLIN)
        while True:
            socks = poller.poll()
            if not socks:
                continue
            if len(socks) == 2 or socks[0][0] == shutdown_socket:
                # shutdown signal, exit thread.
                break

            frames = out_socket.recv_multipart(copy=False)
            resources.validate_alive(frames)
            outputs: EngineCoreOutputs = decoder.decode(frames)
            if outputs.utility_output:
                _process_utility_output(outputs.utility_output, utility_results)
            else:
                outputs_queue.put_nowait(outputs)
    except Exception as e:
        outputs_queue.put_nowait(e)
    finally:
        # Close sockets.
        shutdown_socket.close(linger=0)
        out_socket.close(linger=0)

(SyncMPClient.process_outputs_socket:808-836). Three details to note: first, shutdown uses a separate zmq.PAIR inproc socket bound to shutdown_path; when BackgroundResources.__call__ runs, it sends an empty byte to this socket to notify the thread to exit, preventing the thread from blocking forever on out_socket. Second, validate_alive turns a single ENGINE_CORE_DEAD frame into EngineDeadError. Third, utility responses (e.g. the return of collective_rpc) and regular outputs travel in the same frame; the former is dispatched into the utility_results[call_id] future, the latter into the queue.

The send direction is simpler. SyncMPClient._send_input sends (identity, request_type, *encoder.encode(request)) as multipart frames directly to the ROUTER; when tensor buffers are involved, it uses track=True to obtain a MessageTracker and stores it in pending_messages to prevent the Python side from GC-ing the backing tensor early (_send_input:861-873).

The handshake phase is in MPClient.__init__ (waiting for ready:615-634): for each engine_rank it generates a 2-byte little-endian identity, then polls the input_socket waiting for the subprocess's EngineCoreReadyResponse, with the timeout gated by VLLM_ENGINE_READY_TIMEOUT_S. _apply_ready_response backfills the subprocess-computed num_gpu_blocks, the aligned block_size, and the auto-fitted max_model_len into the frontend's vllm_config — these are the real capacities the frontend can use for scheduling and rate limiting.

Boundaries and failure modes

• asyncio_mode=True, multiprocess_mode=False is rejected outright (make_client:91-95). EngineCore internals are not async-friendly; this combination is not implemented.
• Construction failure must reclaim the subprocess: MPClient.__init__ uses try/finally + a success flag; on failure it calls self._finalizer() to trigger BackgroundResources.__call__, which stops any subprocess already started by launch_core_engines (__init__ finally:499-649).
• InprocClient does not support wait pause (InprocClient.sleep:324-328). In in-process mode, nothing waits for idle; the only option is to abort current requests before sleeping.
• Engine core subprocess death is observed on both sides: The send side via ensure_alive (ensure_alive:670-672); the receive side via validate_alive receiving a single ENGINE_CORE_DEAD frame (validate_alive:454-457). Both paths raise BackgroundResources.engine_dead.
• utility future tolerates InvalidStateError: _process_utility_output catches asyncio.InvalidStateError (_process_utility_output:769-776), covering the case where the caller's task was cancelled and the future is already cancelled.
• output socket must be closed before context.term: BackgroundResources.__call__ explicitly close_sockets before continuing (sync case:437-452), otherwise the ZMQ context termination will hang.
• SyncMPClient's output_socket is owned by the thread: At the end of construction, self.resources.output_socket = None (socket ownership transferred:846-847), preventing both BackgroundResources and the daemon thread from closing it.
• ready handshake timeout gives a readable error: Timed out waiting for engine core processes to start. This is often caused by slow weight loading for large models. (ready timeout:622-630) — directly tells the user to tune VLLM_ENGINE_READY_TIMEOUT_S.

Summary

InprocClient and MPClient look very different on the inside, but from the outside both look like EngineCoreClient — call add_request, pull get_output, and the frontend code does not care whether EngineCore is an in-process object or a subprocess + ZMQ. The in-process path is dead simple, just an object reference; the subprocess path wraps up ZMQ ROUTER/PULL, handshake, monitor, future registry, and socket ownership. The async variant AsyncMPClient is covered separately at /client/async-mp; EngineCore itself is at /engine/engine-core; EngineCoreProc's busy loop and shutdown signal are at /engine/engine-core-proc.

See official docs: vLLM docs · README