From 2a336d8239d00565a99752cfa248f2326e2ac8aa Mon Sep 17 00:00:00 2001 From: Ning Xie Date: Fri, 4 Sep 2026 03:44:21 +0800 Subject: [PATCH] [warmup] overlap renderer warmup and engine core initialization (#54557) Signed-off-by: Andy Xie Co-authored-by: Cyrus Leung --- tests/renderers/test_warmup.py | 347 ++++++++++++++++++++++++++++++++- vllm/renderers/base.py | 141 +++++++++++--- vllm/v1/engine/async_llm.py | 8 + vllm/v1/engine/core_client.py | 90 ++++++++- vllm/v1/engine/llm_engine.py | 8 + 5 files changed, 561 insertions(+), 33 deletions(-) diff --git a/tests/renderers/test_warmup.py b/tests/renderers/test_warmup.py index ac951aae46..aff1f5ad05 100644 --- a/tests/renderers/test_warmup.py +++ b/tests/renderers/test_warmup.py @@ -6,13 +6,25 @@ These tests exercise: - Zero-limit modalities are filtered from mm_counts passed to get_dummy_processor_inputs (e.g. --limit-mm-per-prompt image=0 ...) - MM warmup is skipped entirely when mm_processor is None + - The multimodal warmup is launched as a task on the single-worker + _mm_executor to overlap engine-core init (future lifecycle, join by + warmup/reset/shutdown, no double-run). Routing it through the same + executor that serves _process_multimodal keeps the numba workqueue + parallel region single-threaded, avoiding the "Concurrent access has been + detected" fatal abort when a request arrives during warmup. + - That overlap only exists for the multiprocess clients: the in-process + EngineCore is built synchronously, so InprocClient rejects a renderer + (assert) and warmup stays inside renderer.warmup(). No model weights are required: warmup() is called directly on a MagicMock that acts as the renderer instance. """ +from concurrent.futures import Future, ThreadPoolExecutor from unittest.mock import MagicMock, patch +import pytest + from vllm.renderers.base import BaseRenderer from vllm.renderers.params import ChatParams @@ -39,6 +51,18 @@ def _make_renderer_mock(mm_limits: dict[str, int]) -> MagicMock: renderer, BaseRenderer ) renderer._clear_processor_cache = BaseRenderer._clear_processor_cache + renderer.warmup_mm = BaseRenderer.warmup_mm.__get__(renderer, BaseRenderer) + renderer.start_mm_warmup_in_background = ( + BaseRenderer.start_mm_warmup_in_background.__get__(renderer, BaseRenderer) + ) + renderer._join_mm_warmup = BaseRenderer._join_mm_warmup.__get__( + renderer, BaseRenderer + ) + renderer.shutdown = BaseRenderer.shutdown.__get__(renderer, BaseRenderer) + # No background warmup launched by default; warmup() takes the inline path. + renderer._mm_warmup_future = None + # MM warmup has not run yet; warmup_mm must actually execute on the mock. + renderer._mm_warmup_done = False renderer.clear_mm_cache = MagicMock() renderer.model_config.max_model_len = 128 renderer.model_config.get_multimodal_config.return_value.limit_per_prompt = {} @@ -88,7 +112,8 @@ class TestMmWarmupZeroLimitFiltering: class TestMmWarmupRunsNormally: - """MM warmup must run when mm_processor is set and limits > 0.""" + # MM warmup must run when mm_processor is set and limits > 0; the chat + # template warmup must run alongside it. def test_processor_apply_called(self): renderer = _make_renderer_mock({"image": 1}) @@ -106,6 +131,14 @@ class TestMmWarmupRunsNormally: renderer.clear_mm_cache.assert_called_once() + def test_render_chat_called_with_warmup_message(self): + renderer = _make_renderer_mock({"image": 1}) + + with patch("vllm.multimodal.processing.TimingContext", autospec=True): + BaseRenderer.warmup(renderer, ChatParams()) + + renderer.render_chat.assert_called_once() + class TestMmWarmupSkippedWhenNoProcessor: """MM warmup must be skipped when mm_processor is None (text-only model).""" @@ -133,3 +166,315 @@ class TestReadonlyMmWarmup: readonly_mm_processor.apply.assert_called_once() readonly_mm_processor.cache.clear_cache.assert_called_once() + + +class TestWarmupFaultIsolation: + # A failure during a multimodal processor warmup is caught so it does not + # abort the remaining warmup steps; warmup itself must not raise. + + def test_chat_failure_does_not_abort_mm_warmup(self): + renderer = _make_renderer_mock({"image": 1}) + renderer.render_chat.side_effect = RuntimeError("chat boom") + + with patch("vllm.multimodal.processing.TimingContext", autospec=True): + BaseRenderer.warmup(renderer, ChatParams()) # must not raise + + renderer.mm_processor.apply.assert_called_once() + + def test_mm_failure_does_not_abort_readonly_warmup(self): + renderer = _make_renderer_mock({"image": 1}) + readonly_mm_processor = MagicMock() + readonly_mm_processor.info.allowed_mm_limits = {"image": 1} + renderer._readonly_mm_processor = readonly_mm_processor + # main processor warmup blows up before apply() + renderer.mm_processor.dummy_inputs.get_dummy_processor_inputs.side_effect = ( + RuntimeError("mm boom") + ) + + with patch("vllm.multimodal.processing.TimingContext", autospec=True): + BaseRenderer.warmup(renderer, ChatParams()) # must not raise + + readonly_mm_processor.apply.assert_called_once() + # cache is still cleared in the failed task's finally + renderer.clear_mm_cache.assert_called_once() + + +class TestBackgroundMmWarmup: + # The multimodal warmup is launched as a task on the single-worker + # _mm_executor so it overlaps engine-core init (fork) while staying + # serialized with the serving path (_process_multimodal runs on the same + # executor). warmup()/reset_mm_cache/shutdown join it. This also fixes the + # numba workqueue "Concurrent access has been detected" abort: warmup and a + # concurrent serving request can no longer both enter the numba parallel + # region at once. + + def _make_bg_renderer(self, mm_limits: dict[str, int]): + # A renderer mock with a real single-worker executor so submit() + # returns a real Future and warmup_mm actually runs (in the worker). + renderer = _make_renderer_mock(mm_limits) + renderer._mm_executor = ThreadPoolExecutor(max_workers=1) + return renderer + + def test_start_submits_future_to_mm_executor(self): + renderer = self._make_bg_renderer({"image": 1}) + try: + with patch("vllm.multimodal.processing.TimingContext", autospec=True): + # Spy on the real executor's submit to confirm the warmup is + # dispatched through _mm_executor (not a separate Thread). + with patch.object( + renderer._mm_executor, "submit", wraps=renderer._mm_executor.submit + ) as spy: + renderer.start_mm_warmup_in_background() + assert isinstance(renderer._mm_warmup_future, Future) + assert spy.called + renderer._join_mm_warmup() + assert renderer._mm_warmup_future is None + renderer.mm_processor.apply.assert_called_once() + finally: + renderer._mm_executor.shutdown(wait=True) + + def test_start_noop_for_text_only_model(self): + renderer = _make_renderer_mock({}) + renderer.mm_processor = None + # _readonly_mm_processor is already None + + renderer.start_mm_warmup_in_background() + + assert renderer._mm_warmup_future is None + + def test_start_is_run_at_most_once(self): + renderer = self._make_bg_renderer({"image": 1}) + try: + with patch("vllm.multimodal.processing.TimingContext", autospec=True): + renderer.start_mm_warmup_in_background() + first = renderer._mm_warmup_future + renderer.start_mm_warmup_in_background() # must not spawn a second + assert renderer._mm_warmup_future is first + renderer._join_mm_warmup() + finally: + renderer._mm_executor.shutdown(wait=True) + + def test_warmup_joins_background_and_does_not_rerun_mm(self): + # When a background MM warmup is in flight, warmup() must join it and + # run only the chat warmup — the MM warmup must not run twice. + renderer = self._make_bg_renderer({"image": 1}) + try: + with patch("vllm.multimodal.processing.TimingContext", autospec=True): + renderer.start_mm_warmup_in_background() + BaseRenderer.warmup(renderer, ChatParams()) + + # MM apply called exactly once (by the background warmup task). + renderer.mm_processor.apply.assert_called_once() + # Chat warmup ran exactly once. + renderer.render_chat.assert_called_once() + # Background task has been joined and cleared. + assert renderer._mm_warmup_future is None + finally: + renderer._mm_executor.shutdown(wait=True) + + def test_warmup_does_not_rerun_mm_after_reset_joins_background(self): + # Regression: reset_mm_cache joins the background warmup before + # warmup() runs (it clears _mm_warmup_future). warmup() must still not + # re-run the MM warmup — the _mm_warmup_done flag survives the join. + renderer = self._make_bg_renderer({"image": 1}) + try: + with patch("vllm.multimodal.processing.TimingContext", autospec=True): + renderer.start_mm_warmup_in_background() + # Simulate reset_mm_cache joining the background warmup first + # (clears _mm_warmup_future, sets _mm_warmup_done via the join). + renderer._join_mm_warmup() + assert renderer._mm_warmup_future is None # joined & cleared + BaseRenderer.warmup(renderer, ChatParams()) + + # MM apply called exactly once (by the background warmup task); + # warmup() did not re-run warmup_mm despite the future being cleared. + renderer.mm_processor.apply.assert_called_once() + renderer.render_chat.assert_called_once() + finally: + renderer._mm_executor.shutdown(wait=True) + + def test_warmup_mm_runs_at_most_once(self): + # Direct repeated calls to warmup_mm run the MM warmup only once. + renderer = _make_renderer_mock({"image": 1}) + + with patch("vllm.multimodal.processing.TimingContext", autospec=True): + renderer.warmup_mm() + renderer.warmup_mm() # second call must be a no-op + + renderer.mm_processor.apply.assert_called_once() + + def test_shutdown_joins_background_warmup(self): + # shutdown() must join the background warmup before closing caches, + # so the mm_processor_cache is never touched concurrently. + renderer = self._make_bg_renderer({"image": 1}) + + with patch("vllm.multimodal.processing.TimingContext", autospec=True): + renderer.start_mm_warmup_in_background() + future = renderer._mm_warmup_future + renderer.shutdown() + + assert renderer._mm_warmup_future is None + # The background warmup was allowed to complete (apply ran) and the + # future is done after shutdown joined it. + assert future.done() + renderer.mm_processor.apply.assert_called_once() + + +class TestEngineStartWarmupHook: + # The renderer is handed to the client, which starts the MM warmup + # (renderer.start_mm_warmup_in_background) only after engine-core + # processes have been forked — a live warmup thread at fork() time would + # deadlock the child (it inherits locks whose owning thread vanishes). + # These tests pin the renderer plumbing contract on EngineCoreClient. + + def _mock_config(self): + from types import SimpleNamespace + + from vllm.v1.engine.core_client import EngineCoreClient # noqa: F401 + + cfg = SimpleNamespace( + parallel_config=SimpleNamespace( + data_parallel_size=1, + data_parallel_external_lb=False, + ), + model_config=SimpleNamespace(multimodal_config=None), + ) + return cfg + + def _mock_renderer(self): + return MagicMock() + + def test_inproc_client_rejects_renderer(self): + # InprocClient has no _start_mm_warmup (only MPClient does), so it + # takes no renderer at all: passing one is a TypeError at the call + # site rather than a silently-ignored kwarg. + from vllm.v1.engine import core_client as cc + + with pytest.raises(TypeError, match="renderer"): + cc.InprocClient( + MagicMock(), + MagicMock(), + MagicMock(), + renderer=self._mock_renderer(), + ) + + def test_inproc_client_forwards_fail_callback_but_not_renderer(self): + # EngineCore args (incl. executor_fail_callback) pass through; the + # renderer kwarg is intercepted by InprocClient.__init__ (keyword-only + # after *), so it is never forwarded to EngineCore — which has no + # renderer parameter at all. + from vllm.v1.engine import core_client as cc + + callback = MagicMock() + with patch.object(cc, "EngineCore") as mock_engine_core: + cc.InprocClient( + MagicMock(), + MagicMock(), + MagicMock(), + executor_fail_callback=callback, + ) + + mock_engine_core.assert_called_once() + args, kwargs = mock_engine_core.call_args + assert len(args) == 3 # vllm_config, executor_class, log_stats + assert "renderer" not in kwargs + assert kwargs.get("executor_fail_callback") is callback + + def test_make_client_does_not_pass_renderer_to_inproc_client(self): + # In-process: the renderer must never reach InprocClient, so MM + # warmup stays a plain inline call inside renderer.warmup(). + from vllm.v1.engine import core_client as cc + + cfg = self._mock_config() + renderer = self._mock_renderer() + + with ( + patch.object(cc, "InprocClient") as mock_inproc, + patch.object(cc, "SyncMPClient") as mock_sync, + ): + client = cc.EngineCoreClient.make_client( + multiprocess_mode=False, + asyncio_mode=False, + vllm_config=cfg, + executor_class=MagicMock(), + log_stats=False, + renderer=renderer, + ) + + assert client is mock_inproc.return_value + mock_inproc.assert_called_once() + assert "renderer" not in mock_inproc.call_args.kwargs + mock_sync.assert_not_called() + renderer.start_mm_warmup_in_background.assert_not_called() + + @pytest.mark.parametrize( + ("client_cls", "multiprocess_mode"), + [ + # SyncMPClient forks engine-core: the renderer must be handed to + # it so the warmup fires after proc.start(), not before fork(). + ("SyncMPClient", True), + ], + ) + def test_make_client_passes_renderer(self, client_cls, multiprocess_mode): + from vllm.v1.engine import core_client as cc + + cfg = self._mock_config() + renderer = self._mock_renderer() + + with patch.object(cc, client_cls) as mock_client: + cc.EngineCoreClient.make_client( + multiprocess_mode=multiprocess_mode, + asyncio_mode=False, + vllm_config=cfg, + executor_class=MagicMock(), + log_stats=False, + renderer=renderer, + ) + + mock_client.assert_called_once() + assert mock_client.call_args.kwargs["renderer"] is renderer + + @pytest.mark.parametrize( + ("client_cls", "dp_size", "external_lb"), + [ + ("AsyncMPClient", 1, False), + ("DPAsyncMPClient", 2, True), + ], + ) + def test_make_async_mp_client_passes_renderer( + self, client_cls, dp_size, external_lb + ): + from vllm.v1.engine import core_client as cc + + cfg = self._mock_config() + cfg.parallel_config.data_parallel_size = dp_size + cfg.parallel_config.data_parallel_external_lb = external_lb + renderer = self._mock_renderer() + + with patch.object(cc, client_cls) as mock_client: + cc.EngineCoreClient.make_async_mp_client( + cfg, + executor_class=MagicMock(), + log_stats=False, + renderer=renderer, + ) + + mock_client.assert_called_once() + assert mock_client.call_args.kwargs["renderer"] is renderer + + def test_mp_client_starts_mm_warmup_only_with_renderer(self): + # MPClient must call renderer.start_mm_warmup_in_background when a + # renderer was given, and skip it (no crash) when not. + from vllm.v1.engine import core_client as cc + + # No renderer: _start_mm_warmup is a no-op. + client = cc.MPClient.__new__(cc.MPClient) + client._renderer = None + client._start_mm_warmup() + + # With a renderer: it starts the background MM warmup. + renderer = self._mock_renderer() + client2 = cc.MPClient.__new__(cc.MPClient) + client2._renderer = renderer + client2._start_mm_warmup() + renderer.start_mm_warmup_in_background.assert_called_once() diff --git a/vllm/renderers/base.py b/vllm/renderers/base.py index 6f374147e7..264a811a9c 100644 --- a/vllm/renderers/base.py +++ b/vllm/renderers/base.py @@ -5,7 +5,7 @@ import time import weakref from abc import ABC, abstractmethod from collections.abc import Mapping, Sequence -from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import Future, ThreadPoolExecutor from contextlib import ExitStack from functools import cached_property from typing import TYPE_CHECKING, Any, Generic, overload @@ -85,6 +85,26 @@ class BaseRenderer(ABC, Generic[_T]): self.tokenizer = tokenizer + # Background multimodal warmup task (a future returned by + # _mm_executor) so it overlaps engine-core initialization while + # staying serialized with the serving path through the same + # single-worker executor. None when no background warmup has been + # launched (or it has already been joined). Routing the warmup through + # _mm_executor (max_workers=1) guarantees it can never run + # concurrently with a serving request inside the numba workqueue + # parallel region, which would otherwise hit numba's + # "Concurrent access has been detected" fatal abort (see + # vllm/utils/jit_monitor.py, which forces the workqueue layer). + # Joined by the first of reset_mm_cache / warmup / shutdown so the + # mm_processor_cache is also never touched concurrently by the warmup + # and the serving path. + self._mm_warmup_future: Future[None] | None = None + # Whether warmup_mm has already run. Survives _join_mm_warmup (which + # clears the thread pointer), so warmup() can tell that the background + # warmup already ran even after reset_mm_cache joined it first, and + # avoid running warmup_mm a second time. + self._mm_warmup_done: bool = False + # Thread pool executor for blocking tokenizer operations. The # multimodal processor receives a deep-copied tokenizer (see #36557) # so it is safe to run tokenization and MM preprocessing concurrently. @@ -243,12 +263,97 @@ class BaseRenderer(ABC, Generic[_T]): elapsed = time.perf_counter() - start_time logger.info("%s warmup completed in %.3fs", log_prefix, elapsed) + def warmup_mm(self) -> None: + # Run only the multimodal warmup (mm + readonly mm), serially. + # Idempotent: runs at most once. A background warmup + # (start_mm_warmup_in_background) may already have run it, and + # reset_mm_cache may have joined that background warmup task before + # warmup() reaches this point — the _mm_warmup_done flag survives the + # join (the future pointer does not), so warmup() can still tell it + # already ran. + if self._mm_warmup_done: + return + try: + # prevent MM processor hangs + with set_default_torch_num_threads(1): + if self.mm_processor: + try: + logger.debug("Warming up multi-modal processing...") + self._warmup_mm_processor( + self.mm_processor, + log_prefix="Multi-modal", + ) + except Exception: + logger.warning("Multi-modal warmup failed") + finally: + self.clear_mm_cache() + + if self._readonly_mm_processor is not None: + try: + logger.debug("Warming up readonly multi-modal processing...") + self._warmup_mm_processor( + self._readonly_mm_processor, + log_prefix="Readonly multi-modal", + ) + except Exception: + logger.warning("Readonly multi-modal warmup failed") + finally: + self._clear_processor_cache(self._readonly_mm_processor) + finally: + self._mm_warmup_done = True + + def start_mm_warmup_in_background(self) -> None: + # Launch the multimodal warmup as a task on the single-worker + # _mm_executor so it overlaps with engine-core initialization + # (model loading) while staying serialized with the serving path. + # + # Why the same executor as serving (not a separate Thread): + # _process_multimodal runs on _mm_executor (max_workers=1) and + # enters the numba workqueue parallel region (e.g. Kimi K2.5 vision + # preprocessing via kimi_k25_vision_fused). A separate warmup Thread + # would run concurrently with that single serving worker, so both + # threads would enter the numba workqueue parallel region at the same + # time and trip numba's "Concurrent access has been detected" -> + # "Fatal Python error: Aborted" (see vllm/utils/jit_monitor.py, which + # forces the non-threadsafe workqueue layer). Submitting the warmup to + # the same single-worker executor makes warmup and serving + # physically incapable of overlapping on the numba threading layer, + # while still overlapping at the process level with the (forked) + # engine-core init in the main thread. + # + # The warmup task is joined by the first of: reset_mm_cache, warmup, or + # shutdown. This also guarantees the frontend mm_processor_cache — + # which is not designed for concurrent access (its get_and_update must + # stay ordered with the engine-core cache) — is never touched + # concurrently by the warmup and the serving path. + if self.mm_processor is None and self._readonly_mm_processor is None: + return # text-only model: nothing to warm up + if self._mm_warmup_future is not None: + return # already launched + self._mm_warmup_future = self._mm_executor.submit(self.warmup_mm) + + def _join_mm_warmup(self) -> None: + # Wait for the background MM warmup task to finish, if one was started. + # Called from the main / event-loop thread (reset_mm_cache, warmup, + # shutdown) — never from inside the _mm_executor worker — so + # waiting on the future cannot deadlock. + future = getattr(self, "_mm_warmup_future", None) + if future is not None: + future.result() + self._mm_warmup_future = None + def warmup(self, chat_params: ChatParams) -> None: """ Warm up this renderer to avoid first-request latency. For chat requests: - Jinja2 template compilation + + If a background multimodal warmup was launched (to overlap + engine-core init) it is joined first so it is not still touching the + mm_processor_cache. warmup_mm is idempotent, so the call below is a + no-op if the background warmup — or a prior reset_mm_cache that joined + it — already ran it. """ from vllm.entrypoints.chat_utils import ChatTemplateResolutionError @@ -267,29 +372,12 @@ class BaseRenderer(ABC, Generic[_T]): except Exception: logger.warning("Chat template warmup failed", exc_info=True) - if self.mm_processor: - try: - logger.debug("Warming up multi-modal processing...") - self._warmup_mm_processor( - self.mm_processor, - log_prefix="Multi-modal", - ) - except Exception: - logger.warning("Multi-modal warmup failed") - finally: - self.clear_mm_cache() - - if self._readonly_mm_processor is not None: - try: - logger.debug("Warming up readonly multi-modal processing...") - self._warmup_mm_processor( - self._readonly_mm_processor, - log_prefix="Readonly multi-modal", - ) - except Exception: - logger.warning("Readonly multi-modal warmup failed") - finally: - self._clear_processor_cache(self._readonly_mm_processor) + if self._mm_warmup_future is not None: + # Join a background MM warmup launched to overlap engine-core init. + self._join_mm_warmup() + # Idempotent: no-op if the background warmup (or a prior reset_mm_cache) + # already ran the MM warmup. + self.warmup_mm() async def clear_mm_cache_async(self) -> None: """Serialize clear_mm_cache through the multimodal executor to avoid @@ -297,6 +385,11 @@ class BaseRenderer(ABC, Generic[_T]): await self._clear_mm_cache_async() def shutdown(self) -> None: + # Wait for any background MM warmup to finish first: it runs on the + # _mm_executor and touches the mm_processor_cache, so it must be + # quiescent before _resources.close() tears down the executors and + # the cache (both are registered as resources in __init__). + self._join_mm_warmup() self._resources.close() def get_bos_token_id(self) -> int | None: diff --git a/vllm/v1/engine/async_llm.py b/vllm/v1/engine/async_llm.py index 4329d9d076..fdeebdcceb 100644 --- a/vllm/v1/engine/async_llm.py +++ b/vllm/v1/engine/async_llm.py @@ -153,6 +153,10 @@ class AsyncLLM(EngineClient): ) # EngineCore (starts the engine in background process). + # Hand the renderer to the client so it can start the frontend MM + # warmup only after engine-core fork (the why is in + # BaseRenderer.start_mm_warmup_in_background). The warmup is joined + # by reset_mm_cache / warmup / shutdown. self.engine_core = EngineCoreClient.make_async_mp_client( vllm_config=vllm_config, executor_class=executor_class, @@ -160,6 +164,7 @@ class AsyncLLM(EngineClient): client_addresses=client_addresses, client_count=client_count, client_index=client_index, + renderer=renderer, ) # Loggers. @@ -1023,6 +1028,9 @@ class AsyncLLM(EngineClient): await asyncio.gather(*coros) async def reset_mm_cache(self) -> None: + # Join the background MM warmup first: the mm_processor_cache is not + # safe for concurrent access with its apply/clear. + self.renderer._join_mm_warmup() await self.renderer.clear_mm_cache_async() await self.engine_core.reset_mm_cache_async() diff --git a/vllm/v1/engine/core_client.py b/vllm/v1/engine/core_client.py index 2bf7cf03c1..fa85646132 100644 --- a/vllm/v1/engine/core_client.py +++ b/vllm/v1/engine/core_client.py @@ -26,6 +26,7 @@ from vllm.config import VllmConfig from vllm.envs import VLLM_ENGINE_READY_TIMEOUT_S from vllm.logger import init_logger from vllm.lora.request import LoRARequest +from vllm.renderers import BaseRenderer from vllm.tasks import SupportedTask from vllm.tracing import instrument from vllm.utils.async_utils import in_loop @@ -93,7 +94,15 @@ class EngineCoreClient(ABC): vllm_config: VllmConfig, executor_class: type[Executor], log_stats: bool, + renderer: BaseRenderer | None = None, ) -> "EngineCoreClient": + # renderer is passed through to the multiprocess clients, which start + # the frontend MM warmup (renderer.start_mm_warmup_in_background) once + # the engine-core processes have been forked, so warmup overlaps + # engine-core load without a live thread at fork() time. In-process + # clients do not take a renderer: the engine core is built in this + # process, so there is no other process to overlap with and the MM + # warmup runs synchronously inside renderer.warmup() instead. # TODO: support this for debugging purposes. if asyncio_mode and not multiprocess_mode: raise NotImplementedError( @@ -103,11 +112,19 @@ class EngineCoreClient(ABC): if multiprocess_mode and asyncio_mode: return EngineCoreClient.make_async_mp_client( - vllm_config, executor_class, log_stats + vllm_config, + executor_class, + log_stats, + renderer=renderer, ) if multiprocess_mode and not asyncio_mode: - return SyncMPClient(vllm_config, executor_class, log_stats) + return SyncMPClient( + vllm_config, + executor_class, + log_stats, + renderer=renderer, + ) return InprocClient(vllm_config, executor_class, log_stats) @@ -120,6 +137,7 @@ class EngineCoreClient(ABC): client_addresses: dict[str, Any] | None = None, client_count: int = 1, client_index: int = 0, + renderer: BaseRenderer | None = None, ) -> "AsyncMPClient": parallel_config = vllm_config.parallel_config client_args = ( @@ -133,10 +151,19 @@ class EngineCoreClient(ABC): if parallel_config.data_parallel_size > 1: if parallel_config.data_parallel_external_lb: # External load balancer - client per DP rank. - return DPAsyncMPClient(*client_args) + return DPAsyncMPClient( + *client_args, + renderer=renderer, + ) # Internal load balancer - client balances to all DP ranks. - return DPLBAsyncMPClient(*client_args) - return AsyncMPClient(*client_args) + return DPLBAsyncMPClient( + *client_args, + renderer=renderer, + ) + return AsyncMPClient( + *client_args, + renderer=renderer, + ) @abstractmethod def shutdown(self, timeout: float | None = None) -> None: ... @@ -313,8 +340,23 @@ class InprocClient(EngineCoreClient): * pulls EngineCoreOutputs by stepping the EngineCore """ - def __init__(self, *args, **kwargs): - self.engine_core = EngineCore(*args, **kwargs) + # Takes no renderer: this class has no _start_mm_warmup (only MPClient + # does), so a renderer would never be used here. EngineCore is built in + # this process and MM warmup runs synchronously inside renderer.warmup(). + def __init__( + self, + vllm_config: VllmConfig, + executor_class: type[Executor], + log_stats: bool, + *, + executor_fail_callback: Callable | None = None, + ): + self.engine_core = EngineCore( + vllm_config, + executor_class, + log_stats, + executor_fail_callback=executor_fail_callback, + ) def get_output(self) -> EngineCoreOutputs: outputs, model_executed = self.engine_core.step_fn() @@ -520,8 +562,10 @@ class MPClient(EngineCoreClient): executor_class: type[Executor], log_stats: bool, client_addresses: dict[str, Any] | None = None, + renderer: BaseRenderer | None = None, ): self.vllm_config = vllm_config + self._renderer: BaseRenderer | None = renderer # ZMQ setup. sync_ctx = zmq.Context(io_threads=2) @@ -583,6 +627,9 @@ class MPClient(EngineCoreClient): ) finally: actual_address_pipe.close() + # Engines are managed externally: this process does not fork + # them, so there is no fork race; start the MM warmup now. + self._start_mm_warmup() else: # Engines are managed by this client. addresses = get_engine_zmq_addresses(vllm_config) @@ -614,6 +661,14 @@ class MPClient(EngineCoreClient): coordinator = engine_launch.coordinator addresses = engine_launch.addresses tensor_queue = engine_launch.tensor_queue + # Engine-core processes have now all been forked/started + # (CoreEngineProcManager.proc.start()). It is now safe to + # launch the frontend background MM warmup: it must not + # run while fork() is in flight (a live thread holding a + # lock would deadlock the forked child), but the + # engine-core model load (minutes) that follows is exactly + # what the warmup should overlap with. + self._start_mm_warmup() self.stats_update_address = addresses.frontend_stats_publish_address if coordinator is not None: @@ -708,6 +763,14 @@ class MPClient(EngineCoreClient): def dp_engines_running(self) -> bool: return self.engines_running + def _start_mm_warmup(self) -> None: + # Called once the engine-core process(es) have been created (forked or + # externally managed). Overlap the frontend MM warmup with the + # engine-core model load. This is a no-op when no renderer was passed + # (e.g. text-only serving or tests). + if self._renderer is not None: + self._renderer.start_mm_warmup_in_background() + def start_engine_core_monitor(self): """Start a monitor thread for engine core processes.""" engine_manager = self.resources.engine_manager @@ -808,13 +871,18 @@ class SyncMPClient(MPClient): @instrument(span_name="SyncMPClient init") def __init__( - self, vllm_config: VllmConfig, executor_class: type[Executor], log_stats: bool + self, + vllm_config: VllmConfig, + executor_class: type[Executor], + log_stats: bool, + renderer: BaseRenderer | None = None, ): super().__init__( asyncio_mode=False, vllm_config=vllm_config, executor_class=executor_class, log_stats=log_stats, + renderer=renderer, ) self.is_dp = self.vllm_config.parallel_config.data_parallel_size > 1 @@ -987,6 +1055,7 @@ class AsyncMPClient(MPClient): client_addresses: dict[str, Any] | None = None, client_count: int = 1, client_index: int = 0, + renderer: BaseRenderer | None = None, ): super().__init__( asyncio_mode=True, @@ -994,6 +1063,7 @@ class AsyncMPClient(MPClient): executor_class=executor_class, log_stats=log_stats, client_addresses=client_addresses, + renderer=renderer, ) self.client_count = client_count @@ -1262,6 +1332,7 @@ class DPAsyncMPClient(AsyncMPClient): client_addresses: dict[str, Any] | None = None, client_count: int = 1, client_index: int = 0, + renderer: BaseRenderer | None = None, ): self.current_wave = 0 @@ -1272,6 +1343,7 @@ class DPAsyncMPClient(AsyncMPClient): client_addresses, client_count, client_index, + renderer, ) # List of [waiting, running, kv_cache_usage] per engine. @@ -1444,6 +1516,7 @@ class DPLBAsyncMPClient(DPAsyncMPClient): client_addresses: dict[str, Any] | None = None, client_count: int = 1, client_index: int = 0, + renderer: BaseRenderer | None = None, ): self.client_count = client_count @@ -1460,6 +1533,7 @@ class DPLBAsyncMPClient(DPAsyncMPClient): client_addresses, client_count, client_index, + renderer, ) assert len(self.core_engines) > 1 diff --git a/vllm/v1/engine/llm_engine.py b/vllm/v1/engine/llm_engine.py index 38a90a4b67..32096079e0 100644 --- a/vllm/v1/engine/llm_engine.py +++ b/vllm/v1/engine/llm_engine.py @@ -102,12 +102,17 @@ class LLMEngine: ) # EngineCore (gets EngineCoreRequests and gives EngineCoreOutputs) + # Hand the renderer to the client. In multiprocess mode the client + # starts the MM warmup only after engine-core fork (the why is in + # BaseRenderer.start_mm_warmup_in_background); InprocClient takes no + # renderer, so MM warmup stays inside renderer.warmup() there. self.engine_core = EngineCoreClient.make_client( multiprocess_mode=multiprocess_mode, asyncio_mode=False, vllm_config=vllm_config, executor_class=executor_class, log_stats=self.log_stats, + renderer=renderer, ) self.logger_manager: StatLoggerManager | None = None @@ -343,6 +348,9 @@ class LLMEngine: self.engine_core.profile(False) def reset_mm_cache(self): + # Join the background MM warmup first: the mm_processor_cache is not + # safe for concurrent access with its apply/clear. + self.renderer._join_mm_warmup() self.renderer.clear_mm_cache() self.engine_core.reset_mm_cache()