From 751f6807d9cb3de50c27a5f27188c4fb04fe0e2b Mon Sep 17 00:00:00 2001 From: Cyrus Leung Date: Sat, 19 Sep 2026 23:38:00 +0800 Subject: [PATCH] [Refactor] Use dynamic MM cache in processor (#57674) Signed-off-by: DarkLight1337 --- .../processing/test_audio_in_video.py | 24 +-- .../multimodal/processing/test_common.py | 22 ++- .../multimodal/processing/test_gemma4.py | 15 +- .../multimodal/processing/test_mllama4.py | 6 +- .../multimodal/processing/test_moss_audio.py | 6 +- .../multimodal/processing/test_qwen2_vl.py | 2 +- .../processing/test_tensor_schema.py | 2 +- .../processing/test_transformers_image.py | 1 + .../processing/transformers_backend.py | 2 +- tests/multimodal/test_cache.py | 69 ++++--- tests/multimodal/test_processing.py | 37 +++- tests/multimodal/test_registry.py | 30 --- tests/renderers/test_warmup.py | 178 +++++++++--------- vllm/model_executor/models/llava.py | 5 - .../models/qwen3_asr_realtime.py | 35 ++-- .../model_executor/models/voxtral_realtime.py | 19 +- vllm/multimodal/encoder_budget.py | 65 +++---- vllm/multimodal/processing/inputs.py | 3 + vllm/multimodal/processing/processor.py | 43 ++++- vllm/multimodal/registry.py | 56 +----- vllm/renderers/base.py | 104 +++++----- vllm/v1/worker/gpu/model_runner.py | 6 +- vllm/v1/worker/gpu_model_runner.py | 14 +- 23 files changed, 343 insertions(+), 401 deletions(-) diff --git a/tests/models/multimodal/processing/test_audio_in_video.py b/tests/models/multimodal/processing/test_audio_in_video.py index f696d7a561..acb102f60f 100644 --- a/tests/models/multimodal/processing/test_audio_in_video.py +++ b/tests/models/multimodal/processing/test_audio_in_video.py @@ -63,36 +63,32 @@ def test_audio_in_video_cache_correctness(model_id: str, num_videos: int) -> Non mm_processor_cache_gb=1, ) - # Baseline: no cache, always processes from scratch. - baseline_processor = MULTIMODAL_REGISTRY.create_processor( - ctx.model_config, cache=None - ) + processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config) + # Sender cache: on a cache hit returns (None, prompt_updates) for each # item, setting mm_kwargs["video"] = [None] – the exact condition that # triggered the original bug. sender_cache = MultiModalProcessorSenderCache(ctx.model_config) - cached_processor = MULTIMODAL_REGISTRY.create_processor( - ctx.model_config, cache=sender_cache - ) - video_token_id = baseline_processor.info.get_hf_config().video_token_id + video_token_id = processor.info.get_hf_config().video_token_id mm_data = create_mm_data(num_videos) hf_processor_mm_kwargs = {"use_audio_in_video": True} - def run(processor): + def run(processor, cache): return processor( [video_token_id] * num_videos, - mm_items=baseline_processor.info.parse_mm_data(mm_data), + mm_items=processor.info.parse_mm_data(mm_data), hf_processor_mm_kwargs=hf_processor_mm_kwargs, + cache=cache, )["prompt_token_ids"] - baseline_ids = run(baseline_processor) + baseline_ids = run(processor, cache=None) # First call on the sender-cache processor: cache miss. # mm_kwargs["video"] items are real tensors; use_audio_in_video is # detected normally from the item data. - first_ids = run(cached_processor) + first_ids = run(processor, cache=sender_cache) assert first_ids == baseline_ids, ( "Cache-miss call produced different prompt_token_ids than baseline.\n" f" baseline : {baseline_ids}\n" @@ -103,7 +99,7 @@ def test_audio_in_video_cache_correctness(model_id: str, num_videos: int) -> Non # MultiModalProcessorSenderCache.get_and_update_item returns (None, …), # so mm_kwargs["video"] = [None]. Before the fix, use_audio_in_video was # not detected, yielding wrong token ids. - second_ids = run(cached_processor) + second_ids = run(processor, cache=sender_cache) assert second_ids == baseline_ids, ( "Cache-hit call produced different prompt_token_ids than baseline.\n" "This is the regression introduced when use_audio_in_video detection\n" @@ -123,7 +119,7 @@ def test_use_audio_in_video_without_audio_track(model_id: str) -> None: model_id, limit_mm_per_prompt={"audio": 1, "image": 0, "video": 1}, ) - processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config, cache=None) + processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config) video_token_id = processor.info.get_hf_config().video_token_id rng = np.random.RandomState(0) diff --git a/tests/models/multimodal/processing/test_common.py b/tests/models/multimodal/processing/test_common.py index ea4305976c..db6f37dc3d 100644 --- a/tests/models/multimodal/processing/test_common.py +++ b/tests/models/multimodal/processing/test_common.py @@ -72,6 +72,8 @@ def glmasr_patch_mm_data(mm_data: MultiModalDataDict) -> MultiModalDataDict: _IGNORE_MM_KEYS = { + # Dithering causes minor divergence + "cohere_asr": {"input_features"}, # In Ultravox, the audio_features can be different depending on padding # The slight difference should not be a problem though, since # attention_mask lets us ignore the difference. @@ -286,8 +288,7 @@ def _test_processing_correctness( } ) - baseline_processor = factories.build_processor(ctx, cache=None) - cached_processor = factories.build_processor(ctx, cache=cache) + processor = factories.build_processor(ctx) rng = np.random.RandomState(0) @@ -336,41 +337,42 @@ def _test_processing_correctness( _test_processing_correctness_one( model_config, mm_data, - baseline_processor, - cached_processor, + processor, batch_idx, hit_rate, num_batches, simplify_rate, + cache=cache, ) def _test_processing_correctness_one( model_config: ModelConfig, mm_data: MultiModalDataDict, - baseline_processor: BaseMultiModalProcessor, - cached_processor: BaseMultiModalProcessor, + processor: BaseMultiModalProcessor, batch_idx: int, hit_rate: float, num_batches: int, simplify_rate: float, + cache: MultiModalProcessorOnlyCache, ): model_type = model_config.hf_config.model_type - token_prompt = get_token_prompt(baseline_processor, mm_data) - mm_items = baseline_processor.info.parse_mm_data(mm_data) + token_prompt = get_token_prompt(processor, mm_data) + mm_items = processor.info.parse_mm_data(mm_data) ignore_mm_keys = _IGNORE_MM_KEYS.get(model_type, set[str]()) - baseline_tokenized_result = baseline_processor( + baseline_tokenized_result = processor( token_prompt, mm_items=mm_items, hf_processor_mm_kwargs={}, ) - cached_tokenized_result = cached_processor( + cached_tokenized_result = processor( token_prompt, mm_items=mm_items, hf_processor_mm_kwargs={}, + cache=cache, ) _assert_inputs_equal( diff --git a/tests/models/multimodal/processing/test_gemma4.py b/tests/models/multimodal/processing/test_gemma4.py index a28cc50be5..6430f40d8b 100644 --- a/tests/models/multimodal/processing/test_gemma4.py +++ b/tests/models/multimodal/processing/test_gemma4.py @@ -221,7 +221,7 @@ def test_video_cache_is_independent_of_image_kwargs( mm_processor_cache_gb=1, ) cache = MultiModalProcessorOnlyCache(ctx.model_config) - processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config, cache=cache) + processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config) hf_processor = processor.info.get_hf_processor() image = PILImage.new("RGB", (48, 48), color=(128, 128, 128)) frames = np.stack([np.asarray(image)] * 2) @@ -230,15 +230,16 @@ def test_video_cache_is_independent_of_image_kwargs( {"image": image, "video": [(frames, metadata)]} ) - def process(mm_kwargs): + def process(mm_kwargs, cache): return processor( hf_processor.image_token + hf_processor.video_token, mm_items, mm_uuid_items={"video": [video_uuid]}, hf_processor_mm_kwargs=mm_kwargs, + cache=cache, ) - baseline = process({}) + baseline = process({}, cache=None) if kwargs_on_init: ctx = build_model_context( model_id, @@ -247,12 +248,12 @@ def test_video_cache_is_independent_of_image_kwargs( mm_processor_cache_gb=1, ) cache = MultiModalProcessorOnlyCache(ctx.model_config) - processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config, cache=cache) + processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config) request_kwargs = {} if kwargs_on_init else kwargs - process(request_kwargs) - cached = process(request_kwargs) + process(request_kwargs, cache=cache) + cached = process(request_kwargs, cache=cache) cache.clear_cache() - fresh = process(request_kwargs) + fresh = process(request_kwargs, cache=cache) def pixels(result, modality, field): return result["mm_kwargs"][modality][0][field].data diff --git a/tests/models/multimodal/processing/test_mllama4.py b/tests/models/multimodal/processing/test_mllama4.py index e5ba6ae241..6cf946eff2 100644 --- a/tests/models/multimodal/processing/test_mllama4.py +++ b/tests/models/multimodal/processing/test_mllama4.py @@ -23,11 +23,9 @@ def test_profiling(model_id: str, max_model_len: int): model_config_kwargs=model_config_kwargs, limit_mm_per_prompt=mm_counts, ) + processor = MULTIMODAL_REGISTRY.create_processor(ctx.model_config) - mm_inputs = MULTIMODAL_REGISTRY.get_dummy_mm_inputs( - ctx.model_config, - mm_counts=mm_counts, - ) + mm_inputs = processor.get_dummy_mm_inputs(mm_counts) hf_config = ctx.get_hf_config(Llama4Config) image_size = hf_config.vision_config.image_size diff --git a/tests/models/multimodal/processing/test_moss_audio.py b/tests/models/multimodal/processing/test_moss_audio.py index 87e7079a81..17863b791c 100644 --- a/tests/models/multimodal/processing/test_moss_audio.py +++ b/tests/models/multimodal/processing/test_moss_audio.py @@ -168,7 +168,6 @@ def _build_moss_audio_processor(cache=None): MossAudioMultiModalProcessor( info, MossAudioDummyInputsBuilder(info), - cache=cache, ), ctx, ) @@ -259,21 +258,24 @@ def test_moss_audio_multimodal_processor_handles_token_and_cache_paths(): ) cache = MultiModalProcessorOnlyCache(ctx.model_config) - cached_processor, _ = _build_moss_audio_processor(cache=cache) + cached_processor, _ = _build_moss_audio_processor() cached_text_miss = cached_processor( prompt, mm_items=mm_items, hf_processor_mm_kwargs={}, + cache=cache, ) cached_text_hit = cached_processor( prompt, mm_items=mm_items, hf_processor_mm_kwargs={}, + cache=cache, ) cached_token_hit = cached_processor( token_prompt, mm_items=mm_items, hf_processor_mm_kwargs={}, + cache=cache, ) expected_audio_tokens = MossAudioEncoder.compute_num_audio_tokens(raw_mel_len) diff --git a/tests/models/multimodal/processing/test_qwen2_vl.py b/tests/models/multimodal/processing/test_qwen2_vl.py index 57b8e7fd39..963b5959ca 100644 --- a/tests/models/multimodal/processing/test_qwen2_vl.py +++ b/tests/models/multimodal/processing/test_qwen2_vl.py @@ -26,7 +26,6 @@ def test_jina_vl_processing_order() -> None: processor = MULTIMODAL_REGISTRY.create_processor( ctx.model_config, tokenizer=ctx.tokenizer, - cache=cache, ) placeholder = "<|vision_start|><|image_pad|><|vision_end|>" @@ -37,6 +36,7 @@ def test_jina_vl_processing_order() -> None: return processor( placeholder * len(images), mm_items=processor.info.parse_mm_data({"image": images}), + cache=cache, ) query = process([query_image]) diff --git a/tests/models/multimodal/processing/test_tensor_schema.py b/tests/models/multimodal/processing/test_tensor_schema.py index f249a25269..403a01a9e2 100644 --- a/tests/models/multimodal/processing/test_tensor_schema.py +++ b/tests/models/multimodal/processing/test_tensor_schema.py @@ -255,7 +255,7 @@ def test_model_tensor_schema(model_id: str): for modality, count in limit_mm_per_prompt.items() } ) - processor = factories.build_processor(ctx, cache=None) + processor = factories.build_processor(ctx) with initialize_dummy_model(model_cls, model_config) as model: for modality, _, mm_kwargs in create_batched_mm_kwargs(model_config, processor): diff --git a/tests/models/multimodal/processing/test_transformers_image.py b/tests/models/multimodal/processing/test_transformers_image.py index f5de7d66f4..8f5a47be7b 100644 --- a/tests/models/multimodal/processing/test_transformers_image.py +++ b/tests/models/multimodal/processing/test_transformers_image.py @@ -245,6 +245,7 @@ def test_repeated_image_hits_the_processor_cache(): prompt="\nWhat is this?", mm_items=mm_processor.info.parse_mm_data({"image": image}), hf_processor_mm_kwargs={}, + cache=cache, ) first, second = process(), process() diff --git a/tests/models/multimodal/processing/transformers_backend.py b/tests/models/multimodal/processing/transformers_backend.py index dc49036e83..869afe9456 100644 --- a/tests/models/multimodal/processing/transformers_backend.py +++ b/tests/models/multimodal/processing/transformers_backend.py @@ -48,4 +48,4 @@ def create_cached_processor(model_id: str, processor_cls): ) info = MultiModalProcessingInfo(ctx) cache = MultiModalProcessorOnlyCache(model_config) - return processor_cls(info, MultiModalDummyInputsBuilder(info), cache=cache), cache + return processor_cls(info, MultiModalDummyInputsBuilder(info)), cache diff --git a/tests/multimodal/test_cache.py b/tests/multimodal/test_cache.py index 0c68d2bac0..02bd511970 100644 --- a/tests/multimodal/test_cache.py +++ b/tests/multimodal/test_cache.py @@ -1,8 +1,6 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project import multiprocessing as mp -from concurrent.futures import ThreadPoolExecutor -from functools import partial from types import SimpleNamespace from unittest.mock import Mock @@ -37,10 +35,13 @@ from vllm.multimodal.inputs import ( PlaceholderRange, ) from vllm.multimodal.processing import PromptInsertion -from vllm.renderers.base import BaseRenderer +from vllm.renderers import renderer_from_config from vllm.utils.async_utils import make_async from vllm.utils.mem_constants import GiB_bytes, MiB_bytes +from vllm.v1.engine.async_llm import AsyncLLM +from vllm.v1.engine.llm_engine import LLMEngine +from ..models.utils import build_model_context from ..utils import create_new_process_for_each_test pytestmark = pytest.mark.cpu_test @@ -773,12 +774,14 @@ def test_processor_cache_shared_across_loras(): @pytest.mark.asyncio async def test_release_kv_cache_resends_mm_payload(use_async, release_error): """Release must not leave a sender hit pointing at a cleared receiver.""" - from vllm.v1.engine.async_llm import AsyncLLM - from vllm.v1.engine.llm_engine import LLMEngine - - model_config = SimpleNamespace( - get_multimodal_config=lambda: MultiModalConfig(mm_processor_cache_gb=1) + ctx = build_model_context( + "llava-hf/llava-v1.6-mistral-7b-hf", + mm_processor_kwargs=None, + limit_mm_per_prompt={"image": 1}, + mm_processor_cache_gb=1, ) + model_config = ctx.model_config + sender = MultiModalProcessorSenderCache(model_config) receiver = MultiModalReceiverCache(model_config) item = _dummy_item({"pixel_values": 16}) @@ -787,43 +790,39 @@ async def test_release_kv_cache_resends_mm_payload(use_async, release_error): assert receiver.get_and_update_item(payload, mm_hash) is item assert sender.get_and_update_item(None, mm_hash)[0] is None - renderer = SimpleNamespace(mm_processor_cache=sender, _mm_cache_stats=None) - renderer.clear_mm_cache = partial(BaseRenderer.clear_mm_cache, renderer) + renderer = renderer_from_config(VllmConfig(model_config=model_config)) + renderer._mm_processor_cache = sender def release(): if release_error: raise RuntimeError(release_error) receiver.clear_cache() - with ThreadPoolExecutor(max_workers=1) as executor: - renderer._clear_mm_cache_async = make_async( - renderer.clear_mm_cache, executor=executor - ) - renderer.clear_mm_cache_async = partial( - BaseRenderer.clear_mm_cache_async, renderer - ) - engine = SimpleNamespace( - renderer=renderer, - engine_core=SimpleNamespace( - release_kv_cache_memory=release, - release_kv_cache_memory_async=make_async(release, executor=executor), + engine = SimpleNamespace( + renderer=renderer, + engine_core=SimpleNamespace( + release_kv_cache_memory=release, + release_kv_cache_memory_async=make_async( + release, + executor=renderer._executor, ), - logger_manager=Mock(), - ) + ), + logger_manager=Mock(), + ) - async def call_release(): - if use_async: - await AsyncLLM.release_kv_cache_memory(engine) - else: - LLMEngine.release_kv_cache_memory(engine) - - if release_error: - with pytest.raises(RuntimeError, match=release_error): - await call_release() - engine.logger_manager.record_sleep_state.assert_not_called() + async def call_release(): + if use_async: + await AsyncLLM.release_kv_cache_memory(engine) else: + LLMEngine.release_kv_cache_memory(engine) + + if release_error: + with pytest.raises(RuntimeError, match=release_error): await call_release() - engine.logger_manager.record_sleep_state.assert_called_once_with(1, 0) + engine.logger_manager.record_sleep_state.assert_not_called() + else: + await call_release() + engine.logger_manager.record_sleep_state.assert_called_once_with(1, 0) payload, _ = sender.get_and_update_item((item, []), mm_hash) assert payload is item diff --git a/tests/multimodal/test_processing.py b/tests/multimodal/test_processing.py index de3e333886..0bd19a0ea0 100644 --- a/tests/multimodal/test_processing.py +++ b/tests/multimodal/test_processing.py @@ -7,7 +7,7 @@ from contextlib import nullcontext import numpy as np import pytest -from vllm.config import ModelConfig +from vllm.config import ModelConfig, SchedulerConfig from vllm.exceptions import VLLMValidationError from vllm.multimodal import MULTIMODAL_REGISTRY from vllm.multimodal.hasher import MultiModalHasher @@ -32,6 +32,7 @@ from vllm.multimodal.processing.processor import ( ) from vllm.utils.collection_utils import flatten_2d_lists +from ..models.utils import build_model_context from .utils import random_image pytestmark = pytest.mark.cpu_test @@ -1361,3 +1362,37 @@ def test_processor_inputs_hashes_distinguish_kwargs_shapes(left, right): ).get_mm_hashes("test-model", "blake3")["image"][0] assert hash_with(left) != hash_with(right) + + +@pytest.mark.parametrize( + ("chunked_prefill", "max_model_len", "expected_seq_len"), + [(None, 491520, 491520), (True, 491520, 8192), (True, 128, 128), (False, 128, 128)], +) +def test_dummy_inputs_scheduler_budget( + chunked_prefill, max_model_len, expected_seq_len +): + ctx = build_model_context( + "llava-hf/llava-v1.6-mistral-7b-hf", + mm_processor_kwargs=None, + limit_mm_per_prompt={"image": 1}, + ) + ctx.model_config.max_model_len = max_model_len + + processor = MULTIMODAL_REGISTRY.create_processor( + ctx.model_config, + tokenizer=ctx.tokenizer, + ) + processor.apply = lambda *args, **kwargs: {"prompt_token_ids": [7]} + + kwargs = {} + if chunked_prefill is not None: + kwargs["scheduler_config"] = SchedulerConfig( + max_model_len=max_model_len, + is_encoder_decoder=False, + max_num_batched_tokens=8192, + max_num_seqs=1, + enable_chunked_prefill=chunked_prefill, + ) + + result = processor.get_dummy_mm_inputs({"image": 1}, **kwargs) + assert len(result["prompt_token_ids"]) == expected_seq_len diff --git a/tests/multimodal/test_registry.py b/tests/multimodal/test_registry.py index 6ca8874014..d7947845af 100644 --- a/tests/multimodal/test_registry.py +++ b/tests/multimodal/test_registry.py @@ -5,11 +5,9 @@ Qwen2.5-VL visual component loading behavior. """ from types import SimpleNamespace -from unittest.mock import MagicMock import pytest -from vllm.config import SchedulerConfig from vllm.multimodal import MULTIMODAL_REGISTRY from ..models.utils import build_model_context @@ -17,34 +15,6 @@ from ..models.utils import build_model_context pytestmark = pytest.mark.cpu_test -@pytest.mark.parametrize( - ("chunked_prefill", "max_model_len", "expected_seq_len"), - [(None, 491520, 491520), (True, 491520, 8192), (True, 128, 128), (False, 128, 128)], -) -def test_dummy_inputs_scheduler_budget( - chunked_prefill, max_model_len, expected_seq_len -): - model_config = MagicMock() - model_config.max_model_len = max_model_len - processor = MagicMock() - processor.apply.return_value = {"prompt_token_ids": [7]} - kwargs = {} - if chunked_prefill is not None: - kwargs["scheduler_config"] = SchedulerConfig( - max_model_len=max_model_len, - is_encoder_decoder=False, - max_num_batched_tokens=8192, - max_num_seqs=1, - enable_chunked_prefill=chunked_prefill, - ) - result = MULTIMODAL_REGISTRY.get_dummy_mm_inputs( - model_config, {"image": 1}, processor=processor, **kwargs - ) - get_inputs = processor.dummy_inputs.get_dummy_processor_inputs - assert get_inputs.call_args.kwargs["seq_len"] == expected_seq_len - assert len(result["prompt_token_ids"]) == expected_seq_len - - @pytest.mark.parametrize( "model_id,limit_mm_per_prompt,expected", [ diff --git a/tests/renderers/test_warmup.py b/tests/renderers/test_warmup.py index f6f9f7a4db..2627cdd985 100644 --- a/tests/renderers/test_warmup.py +++ b/tests/renderers/test_warmup.py @@ -5,6 +5,8 @@ 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 ...) + - The warmup runs through the processor-only cache and clears it afterwards, + so its dummy inputs never land in the sender cache used by the serving path - 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 @@ -21,10 +23,13 @@ that acts as the renderer instance. """ from concurrent.futures import Future, ThreadPoolExecutor +from types import MethodType, SimpleNamespace from unittest.mock import MagicMock, patch import pytest +from vllm.entrypoints.chat_utils import ChatTemplateResolutionError +from vllm.multimodal.processing import BaseMultiModalProcessor from vllm.renderers.base import BaseRenderer from vllm.renderers.params import ChatParams @@ -34,9 +39,11 @@ def _make_renderer_mock(mm_limits: dict[str, int]) -> MagicMock: render_chat is mocked to raise ChatTemplateResolutionError so the chat warmup block is skipped cleanly, keeping the test focused on MM warmup. - """ - from vllm.entrypoints.chat_utils import ChatTemplateResolutionError + The processor is a mock at its I/O boundaries (``dummy_inputs`` and + ``apply``) but keeps the real ``get_dummy_mm_inputs``, since that is where + the warmup's seq_len capping and cache plumbing actually live. + """ renderer = MagicMock() # chat warmup: make render_chat raise so we skip past it cleanly @@ -45,13 +52,17 @@ def _make_renderer_mock(mm_limits: dict[str, int]) -> MagicMock: # MM processor with configurable limits mm_processor = MagicMock() mm_processor.info.allowed_mm_limits = mm_limits + mm_processor.info.ctx.model_config = renderer.model_config mm_processor.apply.return_value = {"prompt_token_ids": [1]} + mm_processor.get_dummy_mm_inputs = MethodType( + BaseMultiModalProcessor.get_dummy_mm_inputs, mm_processor + ) renderer.mm_processor = mm_processor - renderer._readonly_mm_processor = None + renderer._mm_processor_cache = MagicMock(name="sender_cache") + renderer._mm_processor_only_cache = MagicMock(name="processor_only_cache") renderer._warmup_mm_processor = BaseRenderer._warmup_mm_processor.__get__( 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) @@ -64,7 +75,6 @@ def _make_renderer_mock(mm_limits: dict[str, int]) -> MagicMock: 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.config.scheduler_config.max_num_batched_tokens = 8192 renderer.config.scheduler_config.enable_chunked_prefill = True @@ -80,8 +90,7 @@ class TestMmWarmupZeroLimitFiltering: """A modality with limit=0 must not appear in mm_counts.""" renderer = _make_renderer_mock({"image": 1, "video": 0}) - with patch("vllm.multimodal.processing.TimingContext", autospec=True): - BaseRenderer.warmup(renderer, ChatParams()) + BaseRenderer.warmup(renderer, ChatParams()) get_inputs = renderer.mm_processor.dummy_inputs.get_dummy_processor_inputs get_inputs.assert_called_once() @@ -93,8 +102,7 @@ class TestMmWarmupZeroLimitFiltering: """When all limits are 0, mm_counts must be empty.""" renderer = _make_renderer_mock({"image": 0, "video": 0}) - with patch("vllm.multimodal.processing.TimingContext", autospec=True): - BaseRenderer.warmup(renderer, ChatParams()) + BaseRenderer.warmup(renderer, ChatParams()) get_inputs = renderer.mm_processor.dummy_inputs.get_dummy_processor_inputs get_inputs.assert_called_once() @@ -105,8 +113,7 @@ class TestMmWarmupZeroLimitFiltering: """All modalities with limit > 0 must be present in mm_counts.""" renderer = _make_renderer_mock({"image": 2, "video": 1}) - with patch("vllm.multimodal.processing.TimingContext", autospec=True): - BaseRenderer.warmup(renderer, ChatParams()) + BaseRenderer.warmup(renderer, ChatParams()) get_inputs = renderer.mm_processor.dummy_inputs.get_dummy_processor_inputs get_inputs.assert_called_once() @@ -125,8 +132,7 @@ class TestMmWarmupRunsNormally: renderer = _make_renderer_mock({"image": 1, "video": 1}) renderer.model_config.max_model_len = max_model_len - with patch("vllm.multimodal.processing.TimingContext", autospec=True): - BaseRenderer.warmup(renderer, ChatParams()) + BaseRenderer.warmup(renderer, ChatParams()) get_inputs = renderer.mm_processor.dummy_inputs.get_dummy_processor_inputs assert get_inputs.call_args.kwargs["seq_len"] == expected_seq_len @@ -134,28 +140,46 @@ class TestMmWarmupRunsNormally: def test_processor_apply_called(self): renderer = _make_renderer_mock({"image": 1}) - with patch("vllm.multimodal.processing.TimingContext", autospec=True): - BaseRenderer.warmup(renderer, ChatParams()) + BaseRenderer.warmup(renderer, ChatParams()) renderer.mm_processor.apply.assert_called_once() - def test_mm_cache_cleared_after_warmup(self): - renderer = _make_renderer_mock({"image": 1}) - - with patch("vllm.multimodal.processing.TimingContext", autospec=True): - BaseRenderer.warmup(renderer, ChatParams()) - - 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()) + BaseRenderer.warmup(renderer, ChatParams()) renderer.render_chat.assert_called_once() +class TestMmWarmupCacheRouting: + # The warmup runs through the processor-only cache and drops it afterwards, + # so the dummy inputs never reach the sender cache used by the serving path. + + def test_warmup_passes_processor_only_cache(self): + renderer = _make_renderer_mock({"image": 1}) + + with patch.object( + renderer.mm_processor, + "get_dummy_mm_inputs", + wraps=renderer.mm_processor.get_dummy_mm_inputs, + ) as spy: + BaseRenderer.warmup(renderer, ChatParams()) + + assert spy.call_args.kwargs["cache"] is renderer._mm_processor_only_cache + assert ( + spy.call_args.kwargs["scheduler_config"] is renderer.config.scheduler_config + ) + + def test_only_cache_cleared_and_sender_cache_untouched(self): + renderer = _make_renderer_mock({"image": 1}) + + BaseRenderer.warmup(renderer, ChatParams()) + + renderer._mm_processor_only_cache.clear_cache.assert_called_once() + renderer._mm_processor_cache.clear_cache.assert_not_called() + + class TestMmWarmupSkippedWhenNoProcessor: """MM warmup must be skipped when mm_processor is None (text-only model).""" @@ -168,51 +192,30 @@ class TestMmWarmupSkippedWhenNoProcessor: renderer.model_config.get_multimodal_config.assert_not_called() -class TestReadonlyMmWarmup: - """Readonly MM processor warmup must mirror the render path behavior.""" - - def test_readonly_processor_apply_called_and_cache_cleared(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 - - with patch("vllm.multimodal.processing.TimingContext", autospec=True): - BaseRenderer.warmup(renderer, ChatParams()) - - 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. + # A failure during a warmup step is caught so it does not abort the other + # step; 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 + BaseRenderer.warmup(renderer, ChatParams()) # must not raise renderer.mm_processor.apply.assert_called_once() - def test_mm_failure_does_not_abort_readonly_warmup(self): + def test_mm_failure_is_swallowed_and_cache_cleared(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() + # the 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 + 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() + renderer.mm_processor.apply.assert_not_called() + # the cache is still cleared in the failed warmup's finally + renderer._mm_processor_only_cache.clear_cache.assert_called_once() class TestBackgroundMmWarmup: @@ -234,16 +237,15 @@ class TestBackgroundMmWarmup: 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() + # 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: @@ -252,7 +254,6 @@ class TestBackgroundMmWarmup: 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() @@ -261,12 +262,11 @@ class TestBackgroundMmWarmup: 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() + 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) @@ -275,9 +275,8 @@ class TestBackgroundMmWarmup: # 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()) + 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() @@ -294,13 +293,12 @@ class TestBackgroundMmWarmup: # 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()) + 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. @@ -313,9 +311,8 @@ class TestBackgroundMmWarmup: # 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.warmup_mm() + renderer.warmup_mm() # second call must be a no-op renderer.mm_processor.apply.assert_called_once() @@ -324,10 +321,9 @@ class TestBackgroundMmWarmup: # 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() + 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 @@ -344,10 +340,6 @@ class TestEngineStartWarmupHook: # 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, diff --git a/vllm/model_executor/models/llava.py b/vllm/model_executor/models/llava.py index 3ea0df9d16..22cd34fc85 100644 --- a/vllm/model_executor/models/llava.py +++ b/vllm/model_executor/models/llava.py @@ -25,7 +25,6 @@ from vllm.model_executor.layers.activation import get_act_fn from vllm.model_executor.layers.linear import ColumnParallelLinear, RowParallelLinear from vllm.model_executor.layers.quantization import QuantizationConfig from vllm.multimodal import MULTIMODAL_REGISTRY -from vllm.multimodal.cache import BaseMultiModalProcessorCache from vllm.multimodal.inputs import ( MultiModalFieldConfig, MultiModalKwargsItem, @@ -407,21 +406,17 @@ def _build_llava_or_pixtral_hf_info( def _build_llava_or_pixtral_hf_processor( info: _I, dummy_inputs: BaseDummyInputsBuilder[_I], - *, - cache: BaseMultiModalProcessorCache | None = None, ) -> BaseMultiModalProcessor: if isinstance(info, PixtralHFProcessingInfo): return PixtralHFMultiModalProcessor( info, dummy_inputs, # type: ignore - cache=cache, ) if isinstance(info, LlavaProcessingInfo): return LlavaMultiModalProcessor( info, dummy_inputs, # type: ignore - cache=cache, ) raise NotImplementedError(type(info)) diff --git a/vllm/model_executor/models/qwen3_asr_realtime.py b/vllm/model_executor/models/qwen3_asr_realtime.py index bab689c605..6eb762b628 100644 --- a/vllm/model_executor/models/qwen3_asr_realtime.py +++ b/vllm/model_executor/models/qwen3_asr_realtime.py @@ -25,21 +25,10 @@ import torch from vllm.config import ModelConfig, SpeechToTextConfig, VllmConfig from vllm.inputs import PromptType, TokensPrompt from vllm.logger import init_logger -from vllm.model_executor.models.interfaces import ( - SupportsRealtime, -) -from vllm.model_executor.models.qwen3_asr import ( - Qwen3ASRDummyInputsBuilder, - Qwen3ASRForConditionalGeneration, - Qwen3ASRMultiModalProcessor, - Qwen3ASRProcessingInfo, - _get_feat_extract_output_lengths, -) from vllm.multimodal import MULTIMODAL_REGISTRY -from vllm.multimodal.cache import _I, BaseMultiModalProcessorCache from vllm.multimodal.parse import MultiModalDataItems +from vllm.multimodal.processing import ProcessorInputs, TimingContext from vllm.multimodal.processing.processor import ( - BaseDummyInputsBuilder, MultiModalProcessingResult, PlaceholderFeaturesInfo, cached_encode, @@ -47,6 +36,15 @@ from vllm.multimodal.processing.processor import ( from vllm.tokenizers import cached_tokenizer_from_config from vllm.transformers_utils.processor import cached_processor_from_config +from .interfaces import SupportsRealtime +from .qwen3_asr import ( + Qwen3ASRDummyInputsBuilder, + Qwen3ASRForConditionalGeneration, + Qwen3ASRMultiModalProcessor, + Qwen3ASRProcessingInfo, + _get_feat_extract_output_lengths, +) + logger = init_logger(__name__) _PRE_ALLOCATE_BUFFER_SIZE_IN_S = 60 @@ -101,14 +99,13 @@ class Qwen3ASRRealtimeBuffer: class Qwen3ASRRealtimeMultiModalProcessor(Qwen3ASRMultiModalProcessor): - def __init__( + def _cached_apply_hf_processor( self, - info: _I, - dummy_inputs: BaseDummyInputsBuilder[_I], - *, - cache: BaseMultiModalProcessorCache | None = None, - ) -> None: - super().__init__(info, dummy_inputs, cache=None) + inputs: ProcessorInputs, + timing_ctx: TimingContext, + ) -> MultiModalProcessingResult: + # realtime can't make use of a cache yet + return self._apply_hf_processor(inputs, timing_ctx) def _maybe_apply_prompt_updates( self, diff --git a/vllm/model_executor/models/voxtral_realtime.py b/vllm/model_executor/models/voxtral_realtime.py index 162fecf038..9fdc946df9 100644 --- a/vllm/model_executor/models/voxtral_realtime.py +++ b/vllm/model_executor/models/voxtral_realtime.py @@ -27,9 +27,8 @@ from vllm.model_executor.models.voxtral import ( VoxtralProcessingInfo, ) from vllm.multimodal import MULTIMODAL_REGISTRY -from vllm.multimodal.cache import _I, BaseMultiModalProcessorCache from vllm.multimodal.parse import MultiModalDataItems -from vllm.multimodal.processing import BaseDummyInputsBuilder +from vllm.multimodal.processing import ProcessorInputs, TimingContext from vllm.multimodal.processing.processor import ( MultiModalProcessingResult, PlaceholderFeaturesInfo, @@ -38,23 +37,19 @@ from vllm.sequence import IntermediateTensors from vllm.tokenizers import cached_tokenizer_from_config from vllm.utils.torch_utils import is_torch_equal_or_newer -from .utils import ( - _flatten_embeddings, -) +from .utils import _flatten_embeddings logger = init_logger(__name__) class VoxtralRealtimeMultiModalProcessor(VoxtralMultiModalProcessor): - def __init__( + def _cached_apply_hf_processor( self, - info: _I, - dummy_inputs: BaseDummyInputsBuilder[_I], - *, - cache: BaseMultiModalProcessorCache | None = None, - ) -> None: + inputs: ProcessorInputs, + timing_ctx: TimingContext, + ) -> MultiModalProcessingResult: # realtime can't make use of a cache yet - super().__init__(info, dummy_inputs, cache=None) + return self._apply_hf_processor(inputs, timing_ctx) def _maybe_apply_prompt_updates( self, diff --git a/vllm/multimodal/encoder_budget.py b/vllm/multimodal/encoder_budget.py index ef3da192db..13448b9096 100644 --- a/vllm/multimodal/encoder_budget.py +++ b/vllm/multimodal/encoder_budget.py @@ -2,8 +2,9 @@ # SPDX-FileCopyrightText: Copyright contributors to the vLLM project from collections.abc import Mapping -from vllm.config import ModelConfig, VllmConfig +from vllm.config import VllmConfig from vllm.logger import init_logger +from vllm.multimodal.cache import BaseMultiModalProcessorCache from vllm.multimodal.inputs import MultiModalKwargsItem from vllm.multimodal.processing import BaseMultiModalProcessor from vllm.multimodal.registry import MultiModalRegistry @@ -14,26 +15,21 @@ logger = init_logger(__name__) def get_mm_max_toks_per_item( - model_config: ModelConfig, - mm_registry: MultiModalRegistry, processor: BaseMultiModalProcessor, mm_counts: Mapping[str, int], + cache: BaseMultiModalProcessorCache | None, ) -> Mapping[str, int]: """Get the maximum number of tokens per data item from each modality based on underlying model configuration. """ max_tokens_per_item = processor.info.get_mm_max_tokens_per_item( - seq_len=model_config.max_model_len, + seq_len=processor.info.ctx.model_config.max_model_len, mm_counts=mm_counts, ) if max_tokens_per_item is not None: return max_tokens_per_item - mm_inputs = mm_registry.get_dummy_mm_inputs( - model_config, - mm_counts=mm_counts, - processor=processor, - ) + mm_inputs = processor.get_dummy_mm_inputs(mm_counts, cache=cache) return { modality: sum(item.get_num_embeds() for item in placeholders) @@ -65,7 +61,7 @@ class MultiModalBudget: if enable_cache else None ) - processor = mm_registry.create_processor(model_config, cache=cache) + processor = mm_registry.create_processor(model_config) self.cache = cache self.processor = processor @@ -91,9 +87,8 @@ class MultiModalBudget: active_modalities = tower_modalities | embed_only_modalities all_mm_max_toks_per_item = get_mm_max_toks_per_item( - model_config, - mm_registry, processor, + cache=cache, mm_counts=dict.fromkeys(active_modalities, 1), ) @@ -194,26 +189,32 @@ class MultiModalBudget: def get_encoder_budget(self) -> int: return min(self.encoder_compute_budget, self.encoder_cache_size) + def get_dummy_encoder_profile_inputs( + self, + modality: str | None = None, + max_items_per_batch: int | None = None, + ) -> list[tuple[str, MultiModalKwargsItem]]: + if self.get_encoder_budget() <= 0 or not self.mm_max_toks_per_item: + return [] + + if modality is None: + modality = self.get_modality_with_max_tokens() + if max_items_per_batch is None: + max_items_per_batch = self.mm_max_items_per_batch[modality] + + # Don't use `max_items_per_batch` here to avoid redundant computation + dummy_mm_inputs = self.processor.get_dummy_mm_inputs( + mm_counts={modality: 1}, + cache=self.cache, + ) + dummy_mm_item = dummy_mm_inputs["mm_kwargs"][modality][0] + + # We use the cache so that the item is saved to the cache, + # but not read from the cache + assert dummy_mm_item is not None, "Item should not already be cached" + + return [(modality, dummy_mm_item)] * max_items_per_batch + def reset_cache(self) -> None: if self.cache is not None: self.cache.clear_cache() - - -def get_dummy_encoder_profile_inputs( - mm_registry: MultiModalRegistry, - budget: MultiModalBudget, -) -> list[tuple[str, MultiModalKwargsItem]]: - if budget.get_encoder_budget() <= 0 or not budget.mm_max_toks_per_item: - return [] - - modality = budget.get_modality_with_max_tokens() - max_items_per_batch = budget.mm_max_items_per_batch[modality] - dummy_mm_inputs = mm_registry.get_dummy_mm_inputs( - budget.model_config, - mm_counts={modality: 1}, - processor=budget.processor, - ) - dummy_mm_item = dummy_mm_inputs["mm_kwargs"][modality][0] - assert dummy_mm_item is not None, "Dummy item should be generated" - - return [(modality, dummy_mm_item)] * max_items_per_batch diff --git a/vllm/multimodal/processing/inputs.py b/vllm/multimodal/processing/inputs.py index 639045a408..6e547c0004 100644 --- a/vllm/multimodal/processing/inputs.py +++ b/vllm/multimodal/processing/inputs.py @@ -6,6 +6,7 @@ from dataclasses import dataclass, field from vllm.config.multimodal import MMHasherAlgorithm from vllm.inputs import MultiModalHashes +from ..cache import BaseMultiModalProcessorCache from ..hasher import MultiModalHasher from ..parse import MultiModalDataItems, MultiModalUUIDItems @@ -28,6 +29,8 @@ class ProcessorInputs: hf_processor_mm_kwargs: Mapping[str, object] = field(default_factory=dict) media_io_kwargs: Mapping[str, Mapping[str, object]] = field(default_factory=dict) + cache: BaseMultiModalProcessorCache | None = None + def get_mm_hashes( self, model_id: str, diff --git a/vllm/multimodal/processing/processor.py b/vllm/multimodal/processing/processor.py index 1efe84219e..b8e51b2053 100644 --- a/vllm/multimodal/processing/processor.py +++ b/vllm/multimodal/processing/processor.py @@ -18,6 +18,7 @@ from typing import ( import torch from typing_extensions import TypeVar, assert_never +from vllm.config import SchedulerConfig from vllm.inputs import ( MultiModalEncDecInput, MultiModalHashes, @@ -1039,8 +1040,6 @@ class BaseMultiModalProcessor(ABC, Generic[_I]): self, info: _I, dummy_inputs: "BaseDummyInputsBuilder[_I]", - *, - cache: BaseMultiModalProcessorCache | None = None, ) -> None: super().__init__() @@ -1053,16 +1052,50 @@ class BaseMultiModalProcessor(ABC, Generic[_I]): self.info = info self.dummy_inputs = dummy_inputs - self.cache = cache self.data_parser = self.info.get_data_parser() + def get_dummy_mm_inputs( + self, + mm_counts: Mapping[str, int], + *, + cache: BaseMultiModalProcessorCache | None = None, + scheduler_config: "SchedulerConfig | None" = None, + ) -> MultiModalInput: + """Create dummy data for profiling the memory usage of a model.""" + model_config = self.info.ctx.model_config + seq_len = model_config.max_model_len + if scheduler_config is not None and scheduler_config.enable_chunked_prefill: + seq_len = min(seq_len, scheduler_config.max_num_batched_tokens) + + mm_config = model_config.get_multimodal_config() + + processor_inputs = self.dummy_inputs.get_dummy_processor_inputs( + seq_len=seq_len, + mm_counts=mm_counts, + mm_options=mm_config.limit_per_prompt, + ) + processor_inputs.cache = cache + + mm_inputs = self.apply( + processor_inputs, + timing_ctx=TimingContext(enabled=False), + ) + + prompt_token_ids = mm_inputs["prompt_token_ids"] + total_len = len(prompt_token_ids) + if total_len < seq_len: + prompt_token_ids.extend([0] * (seq_len - total_len)) + + return mm_inputs + def __call__( self, prompt: str | list[int], mm_items: MultiModalDataItems, mm_uuid_items: MultiModalUUIDItems | None = None, hf_processor_mm_kwargs: Mapping[str, object] | None = None, + cache: BaseMultiModalProcessorCache | None = None, ) -> MultiModalInput: if isinstance(prompt, str): tokenizer = self.info.get_tokenizer() @@ -1076,6 +1109,7 @@ class BaseMultiModalProcessor(ABC, Generic[_I]): mm_items, mm_uuid_items, hf_processor_mm_kwargs=hf_processor_mm_kwargs or {}, + cache=cache, ) return self.apply(processor_inputs, TimingContext(enabled=False)) @@ -1420,7 +1454,7 @@ class BaseMultiModalProcessor(ABC, Generic[_I]): """Apply the HF processor on the full prompt text, caching the results and reusing cached results. """ - cache = self.cache + cache = inputs.cache has_passthrough_data = any( len(items.get_passthrough_data()) > 0 for items in inputs.mm_data_items.values() @@ -1827,6 +1861,7 @@ class EncDecMultiModalProcessor(BaseMultiModalProcessor[_I]): inputs.mm_uuid_items, hf_processor_mm_kwargs=inputs.hf_processor_mm_kwargs, media_io_kwargs=inputs.media_io_kwargs, + cache=inputs.cache, ) encoder_inputs = super().apply(encoder_processor_inputs, timing_ctx) diff --git a/vllm/multimodal/registry.py b/vllm/multimodal/registry.py index f5be2a0ebb..24a7a696b4 100644 --- a/vllm/multimodal/registry.py +++ b/vllm/multimodal/registry.py @@ -2,12 +2,10 @@ # SPDX-FileCopyrightText: Copyright contributors to the vLLM project import threading from collections import defaultdict -from collections.abc import Mapping from dataclasses import dataclass from multiprocessing.synchronize import Lock as LockType from typing import TYPE_CHECKING, Generic, Literal, Protocol, TypeVar, cast -from vllm.inputs import MultiModalInput from vllm.logger import init_logger from vllm.tokenizers import TokenizerLike, cached_tokenizer_from_config @@ -29,12 +27,7 @@ from .processing import ( ) if TYPE_CHECKING: - from vllm.config import ( - ModelConfig, - ObservabilityConfig, - SchedulerConfig, - VllmConfig, - ) + from vllm.config import ModelConfig, ObservabilityConfig, VllmConfig from vllm.model_executor.models.interfaces import SupportsMultiModal logger = init_logger(__name__) @@ -75,8 +68,6 @@ class MultiModalProcessorFactory(Protocol[_I]): # type: ignore[misc] self, info: _I, dummy_inputs: BaseDummyInputsBuilder[_I], - *, - cache: BaseMultiModalProcessorCache | None = None, ) -> BaseMultiModalProcessor[_I]: ... @@ -89,12 +80,10 @@ class _ProcessorFactories(Generic[_I]): def build_processor( self, ctx: InputProcessingContext, - *, - cache: BaseMultiModalProcessorCache | None = None, ): info = self.info(ctx) dummy_inputs_builder = self.dummy_inputs(info) - return self.processor(info, dummy_inputs_builder, cache=cache) + return self.processor(info, dummy_inputs_builder) class MultiModalRegistry: @@ -218,7 +207,6 @@ class MultiModalRegistry: model_config: "ModelConfig", *, tokenizer: TokenizerLike | None = None, - cache: BaseMultiModalProcessorCache | None = None, ) -> BaseMultiModalProcessor[BaseProcessingInfo]: """Create a multi-modal processor for a specific model and tokenizer.""" if not model_config.is_multimodal_model: @@ -230,45 +218,7 @@ class MultiModalRegistry: ctx = self._create_processing_ctx(model_config, tokenizer) - return factories.build_processor(ctx, cache=cache) - - def get_dummy_mm_inputs( - self, - model_config: "ModelConfig", - mm_counts: Mapping[str, int], - *, - cache: BaseMultiModalProcessorCache | None = None, - processor: BaseMultiModalProcessor | None = None, - scheduler_config: "SchedulerConfig | None" = None, - ) -> MultiModalInput: - """Create dummy data for profiling the memory usage of a model. - - The model is identified by `model_config`. - """ - seq_len = model_config.max_model_len - if scheduler_config is not None and scheduler_config.enable_chunked_prefill: - seq_len = min(seq_len, scheduler_config.max_num_batched_tokens) - - if processor is None: - processor = self.create_processor(model_config, cache=cache) - - mm_config = model_config.get_multimodal_config() - processor_inputs = processor.dummy_inputs.get_dummy_processor_inputs( - seq_len=seq_len, - mm_counts=mm_counts, - mm_options=mm_config.limit_per_prompt, - ) - mm_inputs = processor.apply( - processor_inputs, - timing_ctx=TimingContext(enabled=False), - ) - - prompt_token_ids = mm_inputs["prompt_token_ids"] - total_len = len(prompt_token_ids) - if total_len < seq_len: - prompt_token_ids.extend([0] * (seq_len - total_len)) - - return mm_inputs + return factories.build_processor(ctx) def _get_cache_type( self, diff --git a/vllm/renderers/base.py b/vllm/renderers/base.py index 2cb4dd646c..f4916faf6b 100644 --- a/vllm/renderers/base.py +++ b/vllm/renderers/base.py @@ -127,8 +127,7 @@ class BaseRenderer(ABC, Generic[_T]): self._async_tokenizer_decode = make_async(self._decode, executor=self._executor) self.mm_processor: BaseMultiModalProcessor | None = None - self._readonly_mm_processor: BaseMultiModalProcessor | None = None - self._mm_cache_stats: MultiModalCacheStats | None = None + self._clear_mm_cache_async = make_async( self.clear_mm_cache, executor=self._mm_executor ) @@ -138,6 +137,16 @@ class BaseRenderer(ABC, Generic[_T]): self._safe_load_prompt_embeds_async = make_async( safe_load_prompt_embeds, executor=self._executor ) + + self._mm_processor_cache: BaseMultiModalProcessorCache | None = None + + # A second processor with its own processor-only cache. + # Used by the tokenize endpoint so that tokenize-only + # requests don't pollute the sender cache. + self._mm_processor_only_cache: BaseMultiModalProcessorCache | None = None + + self._mm_cache_stats: MultiModalCacheStats | None = None + if mm_registry.supports_multimodal_inputs(config.model_config): # Install the process-global GPU memory pool used to gate # frontend GPU-side multimodal decoding (no-op when the budget @@ -149,30 +158,16 @@ class BaseRenderer(ABC, Generic[_T]): config.parallel_config._api_process_count, ) - mm_processor_cache = mm_registry.processor_cache_from_config(config) - with set_default_torch_num_threads(): self.mm_processor = mm_registry.create_processor( config.model_config, tokenizer=self.tokenizer, - cache=mm_processor_cache, ) - if mm_processor_cache: - self._mm_cache_stats = MultiModalCacheStats() - self._resources.callback(mm_processor_cache.close) - - # A second processor with its own processor-only cache. - # Used by the tokenize endpoint so that tokenize-only - # requests don't pollute the sender cache. - ro_cache = mm_registry.processor_only_cache_from_config(config) - if ro_cache is not None: - with set_default_torch_num_threads(): - self._readonly_mm_processor = mm_registry.create_processor( - config.model_config, - tokenizer=self.tokenizer, - cache=ro_cache, - ) + self._mm_processor_cache = mm_registry.processor_cache_from_config(config) + self._mm_processor_only_cache = ( + mm_registry.processor_only_cache_from_config(config) + ) # This is used to generate internal request ID for MM processing # It has no relation to the request ID for engine core @@ -181,6 +176,12 @@ class BaseRenderer(ABC, Generic[_T]): config.observability_config ) + if self._mm_processor_cache: + self._mm_cache_stats = MultiModalCacheStats() + self._resources.callback(self._mm_processor_cache.close) + if self._mm_processor_only_cache: + self._resources.callback(self._mm_processor_only_cache.close) + def get_tokenizer(self) -> _T: tokenizer = self.tokenizer if tokenizer is None: @@ -198,11 +199,8 @@ class BaseRenderer(ABC, Generic[_T]): return self.mm_processor @property - def mm_processor_cache(self) -> "BaseMultiModalProcessorCache | None": - if self.mm_processor is None: - return None - - return self.mm_processor.cache + def mm_processor_cache(self) -> BaseMultiModalProcessorCache | None: + return self._mm_processor_cache def stat_mm_cache(self) -> MultiModalCacheStats | None: mm_cache_stats = self._mm_cache_stats @@ -214,7 +212,7 @@ class BaseRenderer(ABC, Generic[_T]): return mm_cache_stats def update_mm_cache_stats(self) -> None: - mm_processor_cache = self.mm_processor_cache + mm_processor_cache = self._mm_processor_cache mm_cache_stats = self._mm_cache_stats if mm_processor_cache and mm_cache_stats: @@ -222,38 +220,26 @@ class BaseRenderer(ABC, Generic[_T]): mm_cache_stats.record(delta.total, delta.hits) def clear_mm_cache(self) -> None: - mm_processor_cache = self.mm_processor_cache + mm_processor_cache = self._mm_processor_cache if mm_processor_cache is not None: mm_processor_cache.clear_cache() if self._mm_cache_stats is not None: self._mm_cache_stats.reset = True - @staticmethod - def _clear_processor_cache( - processor: "BaseMultiModalProcessor | None", - ) -> None: - if processor is None: - return - - processor_cache = processor.cache - if processor_cache is not None: - processor_cache.clear_cache() - def _warmup_mm_processor( self, - processor: "BaseMultiModalProcessor", + processor: BaseMultiModalProcessor, + cache: BaseMultiModalProcessorCache | None, *, log_prefix: str, ) -> None: - model_config = self.model_config mm_limits = {k: v for k, v in processor.info.allowed_mm_limits.items() if v > 0} start_time = time.perf_counter() - _ = mm_registry.get_dummy_mm_inputs( - model_config, - mm_counts=dict.fromkeys(mm_limits, 1), - processor=processor, + _ = processor.get_dummy_mm_inputs( + dict.fromkeys(mm_limits, 1), + cache=cache, scheduler_config=self.config.scheduler_config, ) @@ -274,28 +260,20 @@ class BaseRenderer(ABC, Generic[_T]): # prevent MM processor hangs with set_default_torch_num_threads(1): if self.mm_processor: + mm_processor_only_cache = self._mm_processor_only_cache + try: logger.debug("Warming up multi-modal processing...") self._warmup_mm_processor( self.mm_processor, + mm_processor_only_cache, 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 mm_processor_only_cache: + mm_processor_only_cache.clear_cache() finally: self._mm_warmup_done = True @@ -323,7 +301,7 @@ class BaseRenderer(ABC, Generic[_T]): # 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: + if self.mm_processor is None: return # text-only model: nothing to warm up if self._mm_warmup_future is not None: return # already launched @@ -853,10 +831,7 @@ class BaseRenderer(ABC, Generic[_T]): *, skip_mm_cache: bool = False, ) -> "MultiModalInput": - if skip_mm_cache and self._readonly_mm_processor is not None: - mm_processor = self._readonly_mm_processor - else: - mm_processor = self.get_mm_processor() + mm_processor = self.get_mm_processor() mm_req_id = f"renderer{self.api_process_rank}-mm-{self._mm_req_counter.inc(1)}" @@ -873,6 +848,11 @@ class BaseRenderer(ABC, Generic[_T]): mm_uuid_items, hf_processor_mm_kwargs=mm_processor_kwargs or {}, media_io_kwargs=media_io_kwargs or {}, + cache=( + self._mm_processor_only_cache + if skip_mm_cache + else self._mm_processor_cache + ), ) mm_timing_ctx = self._mm_timing_registry.get(mm_req_id) diff --git a/vllm/v1/worker/gpu/model_runner.py b/vllm/v1/worker/gpu/model_runner.py index a05658e7ce..4dd8865cbd 100644 --- a/vllm/v1/worker/gpu/model_runner.py +++ b/vllm/v1/worker/gpu/model_runner.py @@ -57,7 +57,6 @@ from vllm.model_executor.warmup.jit_warmup import JitWarmupRegistry from vllm.multimodal import MULTIMODAL_REGISTRY from vllm.multimodal.encoder_budget import ( MultiModalBudget, - get_dummy_encoder_profile_inputs, ) from vllm.sequence import IntermediateTensors from vllm.tasks import SupportedTask @@ -935,10 +934,7 @@ class GPUModelRunner(LoRAModelRunnerMixin): self.mm_registry, enable_cache=False, ) - dummy_mm_inputs = get_dummy_encoder_profile_inputs( - self.mm_registry, - mm_budget, - ) + dummy_mm_inputs = mm_budget.get_dummy_encoder_profile_inputs() self.model_state.encoder_runner.profile_encoder_cache( dummy_mm_inputs, mm_budget ) diff --git a/vllm/v1/worker/gpu_model_runner.py b/vllm/v1/worker/gpu_model_runner.py index cedf95e319..7d933917bc 100644 --- a/vllm/v1/worker/gpu_model_runner.py +++ b/vllm/v1/worker/gpu_model_runner.py @@ -5837,21 +5837,15 @@ class GPUModelRunner( assert self.mm_budget is not None # Don't use `max_items_per_batch` here to avoid redundant computation - dummy_mm_inputs = self.mm_registry.get_dummy_mm_inputs( - self.model_config, - mm_counts={modality: 1}, - cache=self.mm_budget.cache, + dummy_mm_inputs = self.mm_budget.get_dummy_encoder_profile_inputs( + modality, + max_items_per_batch, ) - dummy_mm_item = dummy_mm_inputs["mm_kwargs"][modality][0] - - # We use the cache so that the item is saved to the cache, - # but not read from the cache - assert dummy_mm_item is not None, "Item should not already be cached" return next( mm_kwargs_batch for _, _, mm_kwargs_batch in group_and_batch_mm_kwargs( - [(modality, dummy_mm_item)] * max_items_per_batch, + dummy_mm_inputs, device=self.device, pin_memory=PIN_MEMORY, )