[warmup] overlap renderer warmup and engine core initialization (#54557)
Signed-off-by: Andy Xie <andy.xning@gmail.com> Co-authored-by: Cyrus Leung <tlleungac@connect.ust.hk>
This commit is contained in:
@@ -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()
|
||||
|
||||
+117
-24
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user