EngineCoreProc: ZMQ-Hintergrundprozess-Wrapper

源码版本v0.25.1

Verantwortung

EngineCoreProc ist eine Unterklasse von EngineCore (core.py:896-897); sie verlegt alles in einen separaten Hintergrundprozess (background process). Der Docstring ist direkt: ZMQ-wrapper for running EngineCore in background process.. Sie macht drei Dinge: beim Prozessstart mit dem Frontend handshaken und ZMQ-Socket-Adressen holen; zwei IO-Threads starten, die Socket und zwei queue.Queue überbrücken (core.py:915-916); danach betreibt run_busy_loop im Haupt-Thread dauerhaft das Muster: Anfrage aus der input_queue holen → step_fn() für Schedule und Forward → Outputs in die output_queue stecken (core.py:1259-1267).

Diese Schicht ist der Schlüssel der v1-Multiprozess-Architektur. Das Frontend-Prozess (SyncMPClient oder AsyncMPClient) serialisiert EngineCoreRequest über ZMQ; der Input-Thread des EngineCoreProc legt ihn nach Empfang per put_nowait in die input_queue; der Haupt-Loop holt ihn in einer Plan-Lücke aus der Queue und dispatcht ihn über _handle_client_request (core.py:1372-1405). Outputs laufen umgekehrt: EngineCoreOutputs aus step_fn() wird in die output_queue geworfen; der Output-Thread nimmt sie aus der Queue, serialisiert und schiebt sie über ZMQ ans Frontend. Der ZMQ-Socket-IO gibt beim Senden/Empfangen den GIL frei, sodass die beiden IO-Threads tatsächlich mit GPU-Forward überlappen (core.py:974-1001); Serialisierung/Deserialisierung verschwindet hinter dem Forward.

Entwurfsmotivation

• Prozessisolation: EngineCoreProc läuft in einem eigenen Prozess; wenn das vordere Python abstürzt, reißt es nicht die Worker mit, die bereits in der GPU laufen; ein Worker-OOM nimmt nicht auch noch den API-Prozess mit. Das ist der Kerngrund, warum v1 EngineCore auslagert — in v0 macht LLMEngine im selben Prozess alles; fällt eine Stelle um, stürzt alles ab.
• Zweistufiger Puffer aus ZMQ + queue.Queue: Der ZMQ-Socket wird nicht direkt vom Haupt-Loop verwendet; dazwischen liegt eine queue.Queue (core.py:915-916). Socket-IO läuft im IO-Thread; der Haupt-Loop macht nur queue.get(block=...), damit er nicht durch Socket-recv blockiert oder durch interne ZMQ-Locks festgehalten wird. Gleichzeitig kann der Haupt-Loop den GIL sofort übernehmen und Forward laufen, wenn Socket-IO den GIL freigibt.
• Zwei IO-Threads: process_input_sockets und process_output_sockets (core.py:980-1001) sind unabhängig; der Eingangs-Thread übernimmt Deserialisierung + put_nowait, der Ausgangs-Thread get + Serialisierung + send_multipart; beide blockieren einander nicht.
• Start-Handshake (handshake): Im Konstruktor holt _perform_handshakes (core.py:932-938) eine Gruppe EngineZmqAddresses; der Input-Thread sendet dann auf ROUTER/DEALER eine EngineCoreReadyResponse ans Frontend: "Ich bin oben, hier sind max_model_len, num_gpu_blocks, kv_cache_size_tokens" (core.py:1525-1546). Das Frontend wartet in MPClient.__init__ auf diese Ready-Nachricht; kommt sie nicht, gilt der Start als fehlgeschlagen.
• Prozesseinstieg run_engine_core: Diese Funktion ist das Target von multiprocessing.Process(target=run_engine_core, ...) (core.py:1154-1224). Sie instanziiert EngineCoreProc, registriert SIGTERM/SIGINT und ruft abschließend run_busy_loop auf. Jede Ausnahme läuft über except Exception → _send_engine_dead → raise; beim Prozessende wird in finally engine_core.shutdown() aufgerufen (core.py:1226-1242).

Schlüsseldateien

• EngineCoreProc class:896-910 — class EngineCoreProc(EngineCore): Definition + Klassenkonstante ENGINE_CORE_DEAD = b"ENGINE_CORE_DEAD".
• EngineCoreProc.__init__:903-1009 — startet input_queue/output_queue, Tensor-IPC-Receiver, _perform_handshakes, super().__init__ und die beiden IO-Threads.
• queues + executor_fail_callback:915-919 — input_queue = queue.Queue[...], output_queue = queue.Queue[...]; der Executor-Fehler-Callback legt EXECUTOR_FAILED in die input_queue.
• IO threads startup:974-1001 — Start der beiden Daemon-Threads process_input_sockets und process_output_sockets; ready_event wird erst gesetzt, wenn der Input-Thread bereit ist.
• run_engine_core:1154-1224 — Prozesseinstieg; wählt EngineCoreProc oder DPEngineCoreProc, registriert Signal-Handler, ruft run_busy_loop.
• signal handlers:1204-1222 — wakeup_engine legt WAKEUP in die input_queue; SIGTERM/SIGINT setzen shutdown_state auf REQUESTED.
• run_busy_loop:1259-1267 — while self._handle_shutdown(): _process_input_queue(); _process_engine_step(); nach Shutdown raise SystemExit.
• _process_input_queue:1269-1298 — blockiert bei Inaktivität auf input_queue.get(block=True); drain-and-non-blocking, wenn Arbeit ansteht.
• _process_engine_step:1300-1317 — ruft step_fn(), output_queue.put_nowait(output), post_step; gibt bei nicht ausgeführtem Forward, aber noch offenen Scheduler-Anfragen per time.sleep(0.001) den GIL frei.
• _handle_client_request:1372-1405 — Dispatch nach EngineCoreRequestType: WAKEUP/ADD/ABORT/UTILITY/EXECUTOR_FAILED.
• _send_engine_dead:1470-1482 — legt ENGINE_CORE_DEAD in die output_queue und wartet, bis der Output-Thread die Nachricht gesendet hat, bevor es beendet wird.
• process_input_sockets:1484-1587 — Input-Thread-Körper; verwendet zmq.Poller auf DEALER + optionalem coord XSUB, deserialisiert und legt Daten per put_nowait in die input_queue.
• process_output_sockets:1589-1654 — Output-Thread-Körper; holt sich Outputs per output_queue.get(), encoder.encode_into + send_multipart(copy=False, track=True) für Zero-Copy ans Frontend.

Datenfluss

Die Anfrage kommt über den ZMQ-Input-Socket; der Input-Thread deserialisiert sie und legt sie in die input_queue. Der Haupt-Loop holt sie in _process_input_queue und dispatcht sie über _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))

Beachten Sie, dass ABORT sowohl in aborts_queue als auch in input_queue gelegt wird, da Abort im Scheduler idempotent ist (core.py:1276-1279); der Haupt-Loop kann in Step-Lücken zusätzlich aus aborts_queue drainen und so verhindern, dass bereits abgebrochene Anfragen hängen bleiben.

Hat der Haupt-Loop die Anfrage, dispatcht _handle_client_request nach 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 ist ein generischer RPC-Kanal: Das Frontend sendet Methodenname + msgspec args; EngineCoreProc ruft per getattr reflexiv auf; das Ergebnis geht über die output_queue zurück. _invoke_utility_method behandelt auch den Fall, dass ein Future zurückgegeben wird (core.py:1440-1446); bei asynchronen Werkzeugen wird der Output erst nach Abschluss zurückgesendet.

Grenzen und Fehler

• Absturz-Benachrichtigung: run_engine_core ruft in except Exception engine_core._send_engine_dead() auf (core.py:1229-1235), das ENGINE_CORE_DEAD in die output_queue legt (core.py:1470-1474); der Output-Thread sendet diesen besonderen Wert an alle Output-Sockets (core.py:1625-1628); der vordere MPClient wirft beim Empfang einen Fehler, statt endlos zu warten.
• Executor-Fehler-Callback: __init__ registriert den executor_fail_callback-Closure bei model_executor (core.py:917-919); wenn im Worker ein Fehler auftritt, legt der Callback EXECUTOR_FAILED in die input_queue; der Haupt-Loop wirft beim Erhalt über _handle_client_request ein raise RuntimeError("Executor failed.") (core.py:1400-1401); der gesamte Prozess wird dann vom except in run_engine_core über _send_engine_dead behandelt.
• Shutdown-Modi: _handle_shutdown betrachtet shutdown_state (core.py:1324-1360); im Zustand REQUESTED geht shutdown_timeout==0 den Abort-Weg und beendet sofort alle in-flight-Anfragen, ungleich 0 den Drain-Weg und wartet, dass alle Anfragen fertig sind. Während des Shutdowns werden neu eintreffende ADD- und UTILITY-Aufrufe explizit abgelehnt (core.py:1407-1432); _reject_add_in_shutdown sendet dem Client einen Abort-Output zurück, _reject_utility_in_shutdown einen UtilityOutput mit failure_message="Server shutting down".
• Input-Thread-Absturz muss Fehler melden: In __init__ wartet ready_event.wait(timeout=10) zirkulär, bis der Input-Thread bereit ist (core.py:1003-1009); wenn nach 10 s immer noch nicht bereit ist und der Thread gestorben ist, raise RuntimeError("Input socket thread died during startup"). Ist nur noch nicht bereit, aber noch am Leben, wird weiter auf die READY-Nachricht des DP-Coordinators gewartet.
• Zero-Copy-Versand + MessageTracker: Der Output-Thread verwendet send_multipart(buffers, copy=False, track=True) (core.py:1646-1654), das einen MessageTracker zurückgibt; noch nicht gesendete Tracker und die zugehörigen Output-Referenzen und Buffer werden an einer pending-Deque aufgehängt, damit GC den noch nicht gesendeten Backing-Buffer nicht einsammelt. reuse_buffers begrenzt auf höchstens len(sockets) + 1 wiederverwendbare Buffer, um unbegrenztes Speicherwachstum zu verhindern.
• WAKEUP ist eine Null-Operation: Der Signal-Handler legt über wakeup_engine WAKEUP in die input_queue; das weckt nur den in input_queue.get(block=True) blockierten Haupt-Loop (core.py:1377-1378) und tut selbst nichts; die eigentliche Shutdown-Behandlung läuft in _handle_shutdown über shutdown_state.
• Signal-Restoration in finally: run_engine_core stellt in finally mit signal.signal(signal.SIGTERM, signal.SIG_DFL) (core.py:1236-1242) den Signal-Handler auf den Standard zurück und ruft dann engine_core.shutdown() auf, das model_executor / Scheduler / verteilten Zustand freigibt. shutdown ruft auch gc.unfreeze() (core.py:652-656) auf, um den beim Start per gc.freeze() eingefrorenen Startup-Heap wieder dem GC zu übergeben; sonst würde der In-Process-Modus (Unit-Test) GPU-Speicher leaken.

Zusammenfassung

EngineCoreProc ist die tragende Wand der v1-Multiprozess-Architektur: Sie erbt von EngineCore, betreibt in einem eigenen Prozess ZMQ + zwei queue.Queue + zwei IO-Threads und lässt Socket-IO und GPU-Forward tatsächlich überlappen; run_engine_core ist der Prozesseinstieg und verantwortet Instanziierung, Signal-Registrierung, Fehler-Fallback und shutdown. Die gesamte Kommunikation zwischen Frontend (SyncMPClient / AsyncMPClient) und Backend (EngineCore) läuft über diese Schicht. Weiter geht es mit /engine/engine-core für den Aufbau des Engine-Kerns oder /engine/llm-engine für die vordere Schale; die drei Client-Implementierungen stehen in /client/inproc-mp; das Scheduler-Innere in /scheduler/scheduler.

Siehe offizielle Dokumentation: vLLM 文档 · README.