vLLM 架构拆解

Deutsch

简体中文
English
日本語
Español
Français
繁體中文

Deutsch

简体中文
English
日本語
Español
Français
繁體中文

Appearance

EngineCoreClient: In-Process- und Multiprozess-Client-Pfade ​

源码版本v0.25.1

Verantwortung ​

Das vLLM-Frontend (LLM / AsyncLLM / Serving-Einstieg) greift nicht direkt auf EngineCore zu, sondern über eine Client-Schicht, die Anfragen hineinpusht und Outputs herauszieht. Diese Schicht ist EngineCoreClient; sie vereint die beiden orthogonalen Dimensionen "synchron vs. asynchron" und "In-Process vs. Unterprozess" auf denselben abstrakten Methoden. In vllm/v1/engine/core_client.py liegen alle Implementierungen.

Die zwei wichtigsten Implementierungen sind InprocClient und MPClient. Ersteres konstruiert EngineCore direkt im eigenen Prozess: ein Schritt pro Aufruf, kein Busy Loop, kein ZMQ — im Wesentlichen eine dünne Schale für das v0-artige LLMEngine.add_request()/step(). Letzteres verpackt EngineCore als EngineCoreProc-Unterprozess und sendet/empfängt Anfragen und Outputs über ein Paar ZMQ ROUTER/PULL-Sockets; der aufrufende Frontend-Thread interagiert nur mit den Sockets. MPClient ist die Basisklasse; konkret落地 gibt es SyncMPClient (synchron, mit Hintergrund-Thread, der Frames vom Output-Socket liest) und AsyncMPClient (asyncio, mit Hintergrund-Task, der denselben Socket betreibt); letzteres wird auf einer separaten Seite behandelt.

EngineCoreClient.make_client ist eine statische Factory, die anhand der beiden Booleans multiprocess_mode und asyncio_mode die Implementierung wählt. Die Kombination (asyncio=True, multiprocess=False) wirft direkt NotImplementedError — EngineCore selbst ist nicht asyncio-freundlich; Async muss immer zusammen mit Multiprozess verwendet werden.

Entwurfsmotivation ​

  • Lebenszyklus von EngineCore entkoppeln: Im In-Process ist der Aufruf von EngineCore einfach; will das Frontend aber mehrere EngineCore parallel betreiben (Datenparallel) oder elastisches EP (Rank dynamisch hinzufügen/entfernen), muss EngineCore in einen eigenen Prozess verlagert werden, und das Frontend hält nur noch den Socket. weakref.finalize(self, self.resources) in MPClient.__init__ ist der Kern: Selbst wenn während der Konstruktion eine Exception auftritt, werden der Hintergrund-Unterprozess und der ZMQ-Context durch BackgroundResources.__call__ als Fallback eingesammelt, ohne zu leaken.
  • Synchrone und asynchrone Schnittstellen teilen dasselbe ZMQ-Protokoll: SyncMPClient nutzt queue.Queue, um Frames vom Output-Socket in den Haupt-Thread zu tragen; AsyncMPClient nutzt asyncio.Queue. MPClient teilt sich jedoch Encoder/Decoder, Ready-Handshake, Engine-Monitor und die Future-Registry für Utility. Die Unterschiede stecken in den beiden Unterklassen.
  • sleep(mode="wait") in InprocClient direkt verboten (InprocClient.sleep L324-L328), weil es im In-Process-Modus keinen Scheduler gibt, der "warten kann, bis Leerlauf, dann sleep"; es geht nur abort. Diese Grenze wird im Unterprozess-Modus unterstützt — der Unterprozess hat seinen eigenen Busy Loop und einen Scheduler-Zustandsautomat.
  • Utility-Aufrufe über Future-Registry: RPCs wie collective_rpc oder save_sharded_state, die einen Rückgabewert brauchen, können nicht wie add_request als Fire-and-Forget arbeiten. call_utility erzeugt eine call_id, steckt das Future in self.utility_results und wartet, bis das zugehörige UtilityOutput vom Output-Socket zurückkommt; dann holt _process_utility_output per call_id das Future und ruft set_result. Dieser Mechanismus ist auf dem sync und async Client identisch; nur der Future-Typ unterscheidet sich (concurrent.futures.Future vs asyncio.Future).
  • Ausbreitung von Engine-Dead: Wenn der Unterprozess unerwartet stirbt, setzt der Monitor-Thread BackgroundResources.engine_dead auf True; jeder nachfolgende _send_input wird von ensure_alive blockiert und wirft EngineDeadError. Wenn auf der Output-Socket-Seite ein ENGINE_CORE_DEAD-Einzelframe ankommt, zieht validate_alive dasselbe Flag hoch — Anfrage- und Empfangsseite sehen beide den Tod.

Schlüsseldateien ​

Datenfluss ​

SyncMPClient holt Outputs über einen Daemon-Thread, der auf dem ZMQ-output_socket pollt, dekodiert und in eine queue.Queue steckt. Der Frontend-Thread ruft get_output() auf und führt einfach queue.get() aus — der nicht-blockierende ZMQ-Socket wird an eine Python-Synchron-Schnittstelle adaptiert. Dieser process_outputs_socket ist der Kern des ganzen Sync-Pfads:

python
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). Drei Details: Erstens läuft Shutdown über einen separaten zmq.PAIR-Inproc-Socket, der an shutdown_path gebunden ist; wenn BackgroundResources.__call__ schließt, sendet es ein leeres Byte an diesen Socket, um den Thread zum Beenden zu bringen, ohne dass er auf out_socket wartet. Zweitens wandelt validate_alive einen ENGINE_CORE_DEAD-Einzelframe in EngineDeadError um. Drittens laufen Utility-Antworten (z. B. Rückgabe von collective_rpc) und normale Outputs über dasselbe Frame; erstere fließen direkt in das Future von utility_results[call_id], letztere in die Queue.

Die Anfragerichtung ist einfacher: SyncMPClient._send_input sendet die Multi-Frames (identity, request_type, *encoder.encode(request)) direkt an ROUTER; bei vorhandenem Tensor-Buffer wird track=True verwendet, um einen MessageTracker zu erhalten, der in pending_messages eingetragen wird und verhindert, dass die Python-Seite den Backing-Tensor vorzeitig per GC einsammelt (_send_input:861-873).

Der Handshake-Phase läuft in MPClient.__init__ ab (wait ready:615-634): Für jeden engine_rank wird eine 2-Byte-Little-Endian-Identität erzeugt und auf dem input_socket per poll auf die EngineCoreReadyResponse des Unterprozesses gewartet; ein Timeout wird durch VLLM_ENGINE_READY_TIMEOUT_S gesichert. _apply_ready_response schreibt die im Unterprozess berechneten num_gpu_blocks, das ausgerichtete block_size und das automatisch angepasste max_model_len ins Frontend-vllm_config zurück — das ist die echte Kapazität, die das Frontend für Schedule und Throttling verwenden kann.

Grenzen und Fehler ​

  • asyncio_mode=True, multiprocess_mode=False direkt abgelehnt (make_client:91-95). EngineCore selbst ist nicht async-freundlich; diese Kombination ist nicht implementiert.
  • Konstruktionsfehler müssen den Unterprozess einsammeln: MPClient.__init__ verwendet try/finally + ein success-Flag; bei Fehler wird self._finalizer() aufgerufen, was BackgroundResources.__call__ auslöst und auch die bereits per launch_core_engines gestarteten Unterprozesse stoppt (__init__ finally:499-649).
  • InprocClient unterstützt wait-Pause nicht (InprocClient.sleep:324-328). Im In-Process-Modus gibt es niemanden, der auf Leerlauf wartet; es geht nur, die aktuelle Anfrage abzubrechen und dann zu schlafen.
  • Der Tod des Engine-Core-Unterprozesses wird in beide Richtungen sichtbar: Auf der Sendeseite über ensure_alive (ensure_alive:670-672); auf der Empfangsseite über validate_alive beim ENGINE_CORE_DEAD-Einzelframe (validate_alive:454-457). Beide Wege ziehen BackgroundResources.engine_dead hoch.
  • Toleranz von InvalidStateError bei Utility-Futures: _process_utility_output fängt asyncio.InvalidStateError ab (_process_utility_output:769-776) und deckt damit den Fall, dass die aufrufende Task bereits abgebrochen wurde und das Future bereits cancelled ist.
  • Output-Socket muss vor context.term geschlossen werden: BackgroundResources.__call__ macht zuerst explizit close_sockets (sync case:437-452); andernfalls hängt context.term fest.
  • output_socket von SyncMPClient wird vom Thread geschlossen: Am Ende der Konstruktion wird self.resources.output_socket = None gesetzt (socket ownership transfer:846-847), damit BackgroundResources und der Daemon-Thread nicht gleichzeitig schließen.
  • Ready-Handshake-Timeout mit lesbarem Fehler: Timed out waiting for engine core processes to start. This is often caused by slow weight loading for large models. (ready timeout:622-630) — weist den Nutzer direkt hin, VLLM_ENGINE_READY_TIMEOUT_S anzupassen.

Zusammenfassung ​

Die beiden Pfade InprocClient und MPClient wirken sehr unterschiedlich, sehen aber außen wie EngineCoreClient aus — ein add_request hier, ein get_output dort; der Frontend-Code muss nicht wissen, ob EngineCore ein In-Process-Objekt oder ein Unterprozess + ZMQ ist. Der In-Process-Pfad ist so einfach wie möglich: eine Objektreferenz. Der Unterprozess-Pfad kapselt ZMQ ROUTER/PULL, Handshake, Monitor, Future-Tabelle und Socket-Ownership vollständig. Die asynchrone Variante AsyncMPClient ist auf /client/async-mp einzeln aufgeführt; EngineCore selbst in /engine/engine-core; Busy Loop und Shutdown-Signal von EngineCoreProc in /engine/engine-core-proc.

Siehe offizielle Dokumentation: vLLM 文档 · README.