From c8824a1e42bbd674751f808d657d58024c68fb0f Mon Sep 17 00:00:00 2001 From: Cyrus Leung Date: Mon, 21 Sep 2026 22:43:07 +0800 Subject: [PATCH] [MM] Move `supports_multimodal_inputs` and cache out of registry (#57913) Signed-off-by: DarkLight1337 --- tests/config/test_multimodal_config.py | 100 +++-- .../test_generative_scoring.py | 5 +- .../openai/chat_completion/test_chat_error.py | 3 +- .../chat_completion/test_serving_chat.py | 1 + .../completion/test_completion_error.py | 3 +- .../openai/completion/test_lora_resolvers.py | 1 + .../entrypoints/openai/test_render_parity.py | 3 +- .../test_generate_stream.py | 3 +- .../tokenize/test_serving_tokenization.py | 3 +- .../processing/test_audio_in_video.py | 4 +- tests/multimodal/test_cache.py | 49 +-- tests/multimodal/test_registry.py | 22 -- tests/renderers/test_cohere.py | 1 + tests/renderers/test_completions.py | 1 + tests/renderers/test_kimi_k3.py | 1 + tests/renderers/test_mistral.py | 1 + .../unit/test_handshake_pp_aggregation.py | 11 +- .../test_scheduler_streaming.py | 1 + .../test_gpu_autoregressive_speculator.py | 50 +-- vllm/config/model.py | 88 ++++- vllm/multimodal/cache/__init__.py | 38 ++ vllm/multimodal/{cache.py => cache/base.py} | 367 +----------------- vllm/multimodal/cache/factories.py | 107 +++++ vllm/multimodal/cache/lru.py | 142 +++++++ vllm/multimodal/cache/shm.py | 261 +++++++++++++ vllm/multimodal/encoder_budget.py | 9 +- vllm/multimodal/registry.py | 140 +------ vllm/renderers/base.py | 14 +- vllm/v1/core/sched/scheduler.py | 4 +- vllm/v1/engine/core.py | 11 +- vllm/v1/engine/input_processor.py | 2 +- vllm/v1/spec_decode/llm_base_proposer.py | 6 +- vllm/v1/worker/gpu/model_runner.py | 4 +- vllm/v1/worker/gpu/spec_decode/speculator.py | 5 +- vllm/v1/worker/gpu_model_runner.py | 4 +- vllm/v1/worker/worker_base.py | 26 +- 36 files changed, 784 insertions(+), 707 deletions(-) create mode 100644 vllm/multimodal/cache/__init__.py rename vllm/multimodal/{cache.py => cache/base.py} (54%) create mode 100644 vllm/multimodal/cache/factories.py create mode 100644 vllm/multimodal/cache/lru.py create mode 100644 vllm/multimodal/cache/shm.py diff --git a/tests/config/test_multimodal_config.py b/tests/config/test_multimodal_config.py index 84cf092c21..225f41bd01 100644 --- a/tests/config/test_multimodal_config.py +++ b/tests/config/test_multimodal_config.py @@ -1,12 +1,14 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project +import copy from unittest.mock import MagicMock, patch import pytest import torch from transformers import PretrainedConfig +from tests.models.utils import build_model_context from vllm.config.ec_transfer import ECRole, ECTransferConfig from vllm.config.model import ModelConfig from vllm.config.multimodal import MultiModalConfig @@ -17,6 +19,26 @@ from vllm.transformers_utils.model_arch_config_convertor import ( from vllm.v1.attention.backends.registry import AttentionBackendEnum +@pytest.mark.parametrize( + "model_id,limit_mm_per_prompt,expected", + [ + ("Qwen/Qwen2-0.5B-Instruct", {}, False), + ("Qwen/Qwen2.5-VL-3B-Instruct", {}, True), + ("Qwen/Qwen2.5-VL-3B-Instruct", {"image": 0, "video": 0}, False), + ("Qwen/Qwen2.5-VL-3B-Instruct", {"image": 0}, True), + ], +) +@pytest.mark.core_model +def test_supports_multimodal_inputs(model_id, limit_mm_per_prompt, expected): + """Test supports_multimodal_inputs returns correct boolean for various + configs.""" + ctx = build_model_context( + model_id, + limit_mm_per_prompt=limit_mm_per_prompt, + ) + assert ctx.model_config.supports_multimodal_inputs is expected + + def test_mm_encoder_attn_backend_str_conversion(): config = MultiModalConfig(mm_encoder_attn_backend="FLASH_ATTN") # type: ignore[arg-type] assert config.mm_encoder_attn_backend == AttentionBackendEnum.FLASH_ATTN @@ -84,59 +106,33 @@ def test_mm_encoder_attn_dtype_hash_updates(tmp_path): assert fp8_hash != fp8_static_hash +_MULTIMODAL_MODEL = "llava-hf/llava-1.5-7b-hf" +_TEXT_ONLY_MODEL = "TinyLlama/TinyLlama-1.1B-Chat-v1.0" + + def _make_mm_prefix_model_config( + model: str = _MULTIMODAL_MODEL, *, language_model_only: bool = False, ) -> ModelConfig: - model_config = MagicMock(spec=ModelConfig) - model_config.multimodal_config = MultiModalConfig( - language_model_only=language_model_only - ) - # Bind real helper methods onto the mock. - model_config._supports_multimodal_for_mm_prefix = ( - ModelConfig._supports_multimodal_for_mm_prefix.__get__( - model_config, ModelConfig - ) - ) - return model_config - - -@pytest.mark.parametrize("supports_mm", [True, False]) -def test_supports_multimodal_for_mm_prefix_uses_registry(supports_mm: bool): - model_config = _make_mm_prefix_model_config() - - with patch( - "vllm.multimodal.MULTIMODAL_REGISTRY.supports_multimodal_inputs", - return_value=supports_mm, - ) as mocked: - assert model_config._supports_multimodal_for_mm_prefix() is supports_mm - mocked.assert_called_once_with(model_config) - - # Sticky cache — registry must not be consulted again. - with patch( - "vllm.multimodal.MULTIMODAL_REGISTRY.supports_multimodal_inputs", - side_effect=AssertionError("should use cache"), - ): - assert model_config._supports_multimodal_for_mm_prefix() is supports_mm + return ModelConfig(model, language_model_only=language_model_only) def test_supports_multimodal_for_mm_prefix_before_multimodal_config(): - model_config = _make_mm_prefix_model_config() - model_config.multimodal_config = None + """A text-only model never builds a multimodal config, so the early + return must neither clear mm_prefix nor write the sticky cache.""" + model_config = _make_mm_prefix_model_config(_TEXT_ONLY_MODEL) + assert model_config.multimodal_config is None assert model_config._supports_multimodal_for_mm_prefix() is True - assert not hasattr(model_config, "_supports_multimodal_inputs_cached") + assert not getattr(model_config, "_supports_multimodal_inputs_cache", None) def test_language_model_only_disables_via_supports_multimodal_inputs(): """language_model_only zeros all limits, so registry reports text-only.""" model_config = _make_mm_prefix_model_config(language_model_only=True) - with patch( - "vllm.multimodal.MULTIMODAL_REGISTRY.supports_multimodal_inputs", - return_value=False, - ): - assert model_config._supports_multimodal_for_mm_prefix() is False + assert model_config._supports_multimodal_for_mm_prefix() is False def test_convertor_clears_mm_prefix_when_multimodal_disabled(): @@ -159,23 +155,19 @@ def test_convertor_clears_mm_prefix_when_multimodal_disabled(): def test_sticky_cache_survives_text_subconfig_regeneration(): """with_hf_config deepcopies the cached decision onto text submodules.""" model_config = _make_mm_prefix_model_config() - with patch( - "vllm.multimodal.MULTIMODAL_REGISTRY.supports_multimodal_inputs", - return_value=False, - ): - assert model_config._supports_multimodal_for_mm_prefix() is False + assert model_config._supports_multimodal_for_mm_prefix() is True - # Simulate deepcopy onto a Gemma4ForCausalLM-like config that would - # otherwise fail registry lookup / return False incorrectly. - text_config = _make_mm_prefix_model_config() - text_config._supports_multimodal_inputs_cached = ( - model_config._supports_multimodal_inputs_cached - ) - with patch( - "vllm.multimodal.MULTIMODAL_REGISTRY.supports_multimodal_inputs", - side_effect=AssertionError("must not re-query registry"), - ): - assert text_config._supports_multimodal_for_mm_prefix() is False + # `with_hf_config` deep-copies this config and swaps `hf_config` for a + # text-only submodule (e.g. Gemma4ForCausalLM). + text_config = copy.deepcopy(model_config) + text_config.hf_config = model_config.hf_text_config + assert text_config._supports_multimodal_for_mm_prefix() is True + + # Without the copied cache the submodule architecture has no registered + # multimodal processor, so re-querying the registry wrongly settles on + # text-only and would clear mm_prefix. + del text_config._supports_multimodal_inputs_cache + assert text_config._supports_multimodal_for_mm_prefix() is False @pytest.mark.parametrize( diff --git a/tests/entrypoints/generate/generative_scoring/test_generative_scoring.py b/tests/entrypoints/generate/generative_scoring/test_generative_scoring.py index b67efcc229..f413869c90 100644 --- a/tests/entrypoints/generate/generative_scoring/test_generative_scoring.py +++ b/tests/entrypoints/generate/generative_scoring/test_generative_scoring.py @@ -60,7 +60,10 @@ class MockModelConfig: encoder_config = None generation_config: str = "auto" media_io_kwargs: dict[str, dict[str, Any]] = field(default_factory=dict) - skip_tokenizer_init = False + skip_tokenizer_init: bool = False + is_encoder_decoder: bool = False + is_multimodal_model: bool = False + supports_multimodal_inputs: bool = False vocab_size = 151936 def get_diff_sampling_param(self): diff --git a/tests/entrypoints/openai/chat_completion/test_chat_error.py b/tests/entrypoints/openai/chat_completion/test_chat_error.py index b83341cc50..a1665a71de 100644 --- a/tests/entrypoints/openai/chat_completion/test_chat_error.py +++ b/tests/entrypoints/openai/chat_completion/test_chat_error.py @@ -56,9 +56,10 @@ class MockModelConfig: encoder_config = None generation_config: str = "auto" media_io_kwargs: dict[str, dict[str, Any]] = field(default_factory=dict) - skip_tokenizer_init = False + skip_tokenizer_init: bool = False is_encoder_decoder: bool = False is_multimodal_model: bool = False + supports_multimodal_inputs: bool = False renderer_num_workers: int = 1 def get_diff_sampling_param(self): diff --git a/tests/entrypoints/openai/chat_completion/test_serving_chat.py b/tests/entrypoints/openai/chat_completion/test_serving_chat.py index 2757320362..9f92e1ce35 100644 --- a/tests/entrypoints/openai/chat_completion/test_serving_chat.py +++ b/tests/entrypoints/openai/chat_completion/test_serving_chat.py @@ -556,6 +556,7 @@ class MockModelConfig: skip_tokenizer_init: bool = False is_encoder_decoder: bool = False is_multimodal_model: bool = False + supports_multimodal_inputs: bool = False renderer_num_workers: int = 1 enable_prompt_embeds: bool = False diff --git a/tests/entrypoints/openai/completion/test_completion_error.py b/tests/entrypoints/openai/completion/test_completion_error.py index 46f4fe452c..28a499e129 100644 --- a/tests/entrypoints/openai/completion/test_completion_error.py +++ b/tests/entrypoints/openai/completion/test_completion_error.py @@ -61,9 +61,10 @@ class MockModelConfig: encoder_config = None generation_config: str = "auto" media_io_kwargs: dict[str, dict[str, Any]] = field(default_factory=dict) - skip_tokenizer_init = False + skip_tokenizer_init: bool = False is_encoder_decoder: bool = False is_multimodal_model: bool = False + supports_multimodal_inputs: bool = False renderer_num_workers: int = 1 def get_diff_sampling_param(self): diff --git a/tests/entrypoints/openai/completion/test_lora_resolvers.py b/tests/entrypoints/openai/completion/test_lora_resolvers.py index 4d8c3a1fe2..8cd874996c 100644 --- a/tests/entrypoints/openai/completion/test_lora_resolvers.py +++ b/tests/entrypoints/openai/completion/test_lora_resolvers.py @@ -54,6 +54,7 @@ class MockModelConfig: skip_tokenizer_init: bool = False is_encoder_decoder: bool = False is_multimodal_model: bool = False + supports_multimodal_inputs: bool = False renderer_num_workers: int = 1 def get_diff_sampling_param(self): diff --git a/tests/entrypoints/openai/test_render_parity.py b/tests/entrypoints/openai/test_render_parity.py index 92b7fe71bd..4fc44c5a71 100644 --- a/tests/entrypoints/openai/test_render_parity.py +++ b/tests/entrypoints/openai/test_render_parity.py @@ -86,9 +86,10 @@ class MockModelConfig: generation_config: str = "auto" override_generation_config: dict[str, Any] = field(default_factory=dict) media_io_kwargs: dict[str, dict[str, Any]] = field(default_factory=dict) - skip_tokenizer_init = False + skip_tokenizer_init: bool = False is_encoder_decoder: bool = False is_multimodal_model: bool = False + supports_multimodal_inputs: bool = False renderer_num_workers: int = 1 enable_prompt_embeds: bool = False diff --git a/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py b/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py index ba5839990c..f7c6f05e7f 100644 --- a/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py +++ b/tests/entrypoints/scale_out/token_in_token_out/test_generate_stream.py @@ -61,9 +61,10 @@ class MockModelConfig: encoder_config = None generation_config: str = "auto" media_io_kwargs: dict[str, dict[str, Any]] = field(default_factory=dict) - skip_tokenizer_init = False + skip_tokenizer_init: bool = False is_encoder_decoder: bool = False is_multimodal_model: bool = False + supports_multimodal_inputs: bool = False renderer_num_workers: int = 1 def get_diff_sampling_param(self): diff --git a/tests/entrypoints/serve/tokenize/test_serving_tokenization.py b/tests/entrypoints/serve/tokenize/test_serving_tokenization.py index a74962ab0d..057d7d1e8a 100644 --- a/tests/entrypoints/serve/tokenize/test_serving_tokenization.py +++ b/tests/entrypoints/serve/tokenize/test_serving_tokenization.py @@ -56,9 +56,10 @@ class MockModelConfig: encoder_config = None generation_config: str = "auto" media_io_kwargs: dict[str, dict[str, Any]] = field(default_factory=dict) - skip_tokenizer_init = False + skip_tokenizer_init: bool = False is_encoder_decoder: bool = False is_multimodal_model: bool = False + supports_multimodal_inputs: bool = False renderer_num_workers: int = 1 def get_diff_sampling_param(self): diff --git a/tests/models/multimodal/processing/test_audio_in_video.py b/tests/models/multimodal/processing/test_audio_in_video.py index acb102f60f..bde7bd51ab 100644 --- a/tests/models/multimodal/processing/test_audio_in_video.py +++ b/tests/models/multimodal/processing/test_audio_in_video.py @@ -22,7 +22,7 @@ import numpy as np import pytest from vllm.multimodal import MULTIMODAL_REGISTRY -from vllm.multimodal.cache import MultiModalProcessorSenderCache +from vllm.multimodal.cache import LruKeyReplicatedSenderCache from ....multimodal.utils import random_audio, random_video from ...utils import build_model_context @@ -68,7 +68,7 @@ def test_audio_in_video_cache_correctness(model_id: str, num_videos: int) -> Non # 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) + sender_cache = LruKeyReplicatedSenderCache(ctx.model_config) video_token_id = processor.info.get_hf_config().video_token_id diff --git a/tests/multimodal/test_cache.py b/tests/multimodal/test_cache.py index 37ec5b0cb8..ffd7d61cf2 100644 --- a/tests/multimodal/test_cache.py +++ b/tests/multimodal/test_cache.py @@ -10,20 +10,23 @@ import torch from vllm.config import ModelConfig, ParallelConfig, VllmConfig from vllm.config.multimodal import MultiModalConfig -from vllm.multimodal import MULTIMODAL_REGISTRY from vllm.multimodal.cache import ( BaseMultiModalProcessorCache, BaseMultiModalReceiverCache, - MultiModalCache, + LruKeyReplicatedReceiverCache, + LruKeyReplicatedSenderCache, MultiModalCacheMissError, + MultiModalProcessorOnlyCache, + ShmObjectStoreReceiverCache, + ShmObjectStoreSenderCache, + engine_receiver_cache_from_config, + processor_cache_from_config, +) +from vllm.multimodal.cache.base import ( + MultiModalCache, MultiModalProcessorCacheInItem, MultiModalProcessorCacheItem, MultiModalProcessorCacheItemMetadata, - MultiModalProcessorOnlyCache, - MultiModalProcessorSenderCache, - MultiModalReceiverCache, - ShmObjectStoreReceiverCache, - ShmObjectStoreSenderCache, ) from vllm.multimodal.hasher import MultiModalHasher from vllm.multimodal.inputs import ( @@ -137,10 +140,10 @@ def _compare_caches( n_iter: int = 100, seed: int = 0, ): - cache_0_p0 = MULTIMODAL_REGISTRY.processor_cache_from_config(config_0) - cache_0_p1 = MULTIMODAL_REGISTRY.engine_receiver_cache_from_config(config_0) - cache_1_p0 = MULTIMODAL_REGISTRY.processor_cache_from_config(config_1) - cache_1_p1 = MULTIMODAL_REGISTRY.engine_receiver_cache_from_config(config_1) + cache_0_p0 = processor_cache_from_config(config_0) + cache_0_p1 = engine_receiver_cache_from_config(config_0) + cache_1_p0 = processor_cache_from_config(config_1) + cache_1_p1 = engine_receiver_cache_from_config(config_1) cache_size_gb = max( config_0.model_config.multimodal_config.mm_processor_cache_gb, @@ -275,8 +278,8 @@ def test_oversized_item_is_served_uncached(): assert p0_only.get_and_update_item((small, []), "small")[0] is small assert p0_only.is_cached_item("small") - p0 = MultiModalProcessorSenderCache(model_config) # type: ignore[arg-type] - p1 = MultiModalReceiverCache(model_config) # type: ignore[arg-type] + p0 = LruKeyReplicatedSenderCache(model_config) # type: ignore[arg-type] + p1 = LruKeyReplicatedReceiverCache(model_config) # type: ignore[arg-type] assert p0.get_and_update_item((item, []), "big")[0] is item assert not p0.is_cached_item("big") assert p1.get_and_update_item(item, "big") is item @@ -296,8 +299,8 @@ def test_mm_cache_miss_raises_and_recovers(): model or network. """ model_config = _StubModelConfig(mm_processor_cache_gb=1) - p0 = MultiModalProcessorSenderCache(model_config) # type: ignore[arg-type] - p1 = MultiModalReceiverCache(model_config) # type: ignore[arg-type] + p0 = LruKeyReplicatedSenderCache(model_config) # type: ignore[arg-type] + p1 = LruKeyReplicatedReceiverCache(model_config) # type: ignore[arg-type] mm_hash = "image_A" item = MultiModalKwargsItem.dummy(nbytes=64) @@ -338,7 +341,7 @@ def test_receiver_cache_replaces_stale_item_when_payload_resent(): and kills EngineCore. Prefer the fresh payload and keep it for later hits. """ model_config = _StubModelConfig(mm_processor_cache_gb=1) - p1 = MultiModalReceiverCache(model_config) # type: ignore[arg-type] + p1 = LruKeyReplicatedReceiverCache(model_config) # type: ignore[arg-type] small = MultiModalKwargsItem.dummy(nbytes=64) large = MultiModalKwargsItem.dummy(nbytes=256) mm_hash = "shared-id" @@ -351,7 +354,7 @@ def test_receiver_cache_replaces_stale_item_when_payload_resent(): def test_receiver_cache_features_keep_resent_payload(): """EngineCore updates features through get_and_update_features.""" model_config = _StubModelConfig(mm_processor_cache_gb=1) - p1 = MultiModalReceiverCache(model_config) # type: ignore[arg-type] + p1 = LruKeyReplicatedReceiverCache(model_config) # type: ignore[arg-type] small = MultiModalKwargsItem.dummy(nbytes=64) large = MultiModalKwargsItem.dummy(nbytes=256) mm_hash = "shared-id" @@ -384,7 +387,7 @@ def test_mm_cache_miss_batches_all_drifted_hashes(): non-drifted items in the same request are still ingested. """ model_config = _StubModelConfig(mm_processor_cache_gb=1) - p1 = MultiModalReceiverCache(model_config) # type: ignore[arg-type] + p1 = LruKeyReplicatedReceiverCache(model_config) # type: ignore[arg-type] item = MultiModalKwargsItem.dummy(nbytes=64) def _feature( @@ -582,8 +585,8 @@ def test_cache_eviction_lru_cache(): model="llava-hf/llava-onevision-qwen2-0.5b-ov-hf", mm_processor_cache_gb=6 / GiB_bytes, ) - sender_cache = MultiModalProcessorSenderCache(model_config) - receiver_cache = MultiModalReceiverCache(model_config) + sender_cache = LruKeyReplicatedSenderCache(model_config) + receiver_cache = LruKeyReplicatedReceiverCache(model_config) _run_test_cache_eviction_lru(sender_cache, receiver_cache, base_item_size=1) @@ -775,7 +778,7 @@ def test_processor_cache_shared_across_loras(): model="llava-hf/llava-onevision-qwen2-0.5b-ov-hf", mm_processor_cache_gb=1, ) - receiver_cache = MultiModalReceiverCache(model_config) + receiver_cache = LruKeyReplicatedReceiverCache(model_config) base_mm_hash = "image_hash_abc123" lora_a_identifier = f"12345:{base_mm_hash}" @@ -827,8 +830,8 @@ async def test_release_kv_cache_resends_mm_payload(use_async, release_error): ) model_config = ctx.model_config - sender = MultiModalProcessorSenderCache(model_config) - receiver = MultiModalReceiverCache(model_config) + sender = LruKeyReplicatedSenderCache(model_config) + receiver = LruKeyReplicatedReceiverCache(model_config) item = _dummy_item({"pixel_values": 16}) mm_hash = "image_A" payload, _ = sender.get_and_update_item((item, []), mm_hash) diff --git a/tests/multimodal/test_registry.py b/tests/multimodal/test_registry.py index d7947845af..f395d36933 100644 --- a/tests/multimodal/test_registry.py +++ b/tests/multimodal/test_registry.py @@ -10,31 +10,9 @@ import pytest from vllm.multimodal import MULTIMODAL_REGISTRY -from ..models.utils import build_model_context - pytestmark = pytest.mark.cpu_test -@pytest.mark.parametrize( - "model_id,limit_mm_per_prompt,expected", - [ - ("Qwen/Qwen2-0.5B-Instruct", {}, False), - ("Qwen/Qwen2.5-VL-3B-Instruct", {}, True), - ("Qwen/Qwen2.5-VL-3B-Instruct", {"image": 0, "video": 0}, False), - ("Qwen/Qwen2.5-VL-3B-Instruct", {"image": 0}, True), - ], -) -@pytest.mark.core_model -def test_supports_multimodal_inputs(model_id, limit_mm_per_prompt, expected): - """Test supports_multimodal_inputs returns correct boolean for various - configs.""" - ctx = build_model_context( - model_id, - limit_mm_per_prompt=limit_mm_per_prompt, - ) - assert MULTIMODAL_REGISTRY.supports_multimodal_inputs(ctx.model_config) is expected - - def test_create_processor_error_uses_served_model_name(): model_config = SimpleNamespace( is_multimodal_model=False, diff --git a/tests/renderers/test_cohere.py b/tests/renderers/test_cohere.py index 76b38734c2..ae2dfe6c7a 100644 --- a/tests/renderers/test_cohere.py +++ b/tests/renderers/test_cohere.py @@ -997,6 +997,7 @@ class _MockModelConfig: skip_tokenizer_init: bool = True is_encoder_decoder: bool = False is_multimodal_model: bool = False + supports_multimodal_inputs: bool = False renderer_num_workers: int = 1 diff --git a/tests/renderers/test_completions.py b/tests/renderers/test_completions.py index 3094eb6b11..99862219b9 100644 --- a/tests/renderers/test_completions.py +++ b/tests/renderers/test_completions.py @@ -39,6 +39,7 @@ class MockModelConfig: skip_tokenizer_init: bool = False is_encoder_decoder: bool = False is_multimodal_model: bool = False + supports_multimodal_inputs: bool = False renderer_num_workers: int = 1 hidden_size: int = 768 dtype: torch.dtype = torch.float32 diff --git a/tests/renderers/test_kimi_k3.py b/tests/renderers/test_kimi_k3.py index af8723f801..de43c0f6bb 100644 --- a/tests/renderers/test_kimi_k3.py +++ b/tests/renderers/test_kimi_k3.py @@ -40,6 +40,7 @@ class MockHFConfig: class MockModelConfig: runner_type: str = "generate" is_multimodal_model: bool = False + supports_multimodal_inputs: bool = False multimodal_config: Any = None hf_config: MockHFConfig = field(default_factory=MockHFConfig) allowed_local_media_path: str = "" diff --git a/tests/renderers/test_mistral.py b/tests/renderers/test_mistral.py index 3a8b49a4db..6ad55f2fb2 100644 --- a/tests/renderers/test_mistral.py +++ b/tests/renderers/test_mistral.py @@ -37,6 +37,7 @@ class MockModelConfig: skip_tokenizer_init: bool = False is_encoder_decoder: bool = False is_multimodal_model: bool = False + supports_multimodal_inputs: bool = False renderer_num_workers: int = 1 diff --git a/tests/v1/kv_connector/unit/test_handshake_pp_aggregation.py b/tests/v1/kv_connector/unit/test_handshake_pp_aggregation.py index e5cbd977b6..509d3d3234 100644 --- a/tests/v1/kv_connector/unit/test_handshake_pp_aggregation.py +++ b/tests/v1/kv_connector/unit/test_handshake_pp_aggregation.py @@ -79,11 +79,6 @@ def _run_engine_core_handshake( "resolve_kv_cache_block_sizes", lambda kv_cache_config, vllm_config: (16, 16), ) - monkeypatch.setattr( - engine_core_module, - "MULTIMODAL_REGISTRY", - SimpleNamespace(engine_receiver_cache_from_config=lambda vllm_config: None), - ) monkeypatch.setattr(engine_core_module, "freeze_gc_heap", lambda: None) monkeypatch.setattr( engine_core_module, "maybe_attach_gc_debug_callback", lambda: None @@ -105,7 +100,11 @@ def _run_engine_core_handshake( speculative_config=None, ec_transfer_config=None, max_concurrent_batches=1, - model_config=SimpleNamespace(runner_type="generate", is_diffusion=False), + model_config=SimpleNamespace( + runner_type="generate", + is_diffusion=False, + supports_multimodal_inputs=False, + ), cache_config=SimpleNamespace( enable_prefix_caching=False, prefix_caching_hash_algo="builtin", diff --git a/tests/v1/streaming_input/test_scheduler_streaming.py b/tests/v1/streaming_input/test_scheduler_streaming.py index af858eaa8c..0d259c1c69 100644 --- a/tests/v1/streaming_input/test_scheduler_streaming.py +++ b/tests/v1/streaming_input/test_scheduler_streaming.py @@ -53,6 +53,7 @@ def create_scheduler() -> Scheduler: vllm_config.model_config = MagicMock() vllm_config.model_config.skip_tokenizer_init = True vllm_config.model_config.is_multimodal_model = False + vllm_config.model_config.supports_multimodal_inputs = False vllm_config.model_config.is_encoder_decoder = False vllm_config.model_config.is_diffusion = False vllm_config.model_config.max_model_len = 1024 diff --git a/tests/v1/worker/test_gpu_autoregressive_speculator.py b/tests/v1/worker/test_gpu_autoregressive_speculator.py index b9b3570b8a..f73023cee3 100644 --- a/tests/v1/worker/test_gpu_autoregressive_speculator.py +++ b/tests/v1/worker/test_gpu_autoregressive_speculator.py @@ -148,8 +148,16 @@ def test_speculator_uses_draft_model_hidden_size(monkeypatch, hc_mult, expected) def test_mm_support_configured_after_model_load(monkeypatch): - target_model_config = object() - draft_model_config = object() + checked_configs = [] + + class _ModelConfig: + @property + def supports_multimodal_inputs(self): + checked_configs.append(self) + return True + + target_model_config = _ModelConfig() + draft_model_config = _ModelConfig() vllm_config = SimpleNamespace(model_config=target_model_config) draft_model = _MultimodalDraftModel() @@ -164,19 +172,8 @@ def test_mm_support_configured_after_model_load(monkeypatch): speculator.supports_mm_inputs = False speculator.use_acceptance_estimator = False - checked_configs = [] - - def supports_multimodal_inputs(model_config): - checked_configs.append(model_config) - return True - monkeypatch.setattr(DraftModelSpeculator, "__init__", init_base) _mock_base_model_load(monkeypatch) - monkeypatch.setattr( - base_spec_module.MULTIMODAL_REGISTRY, - "supports_multimodal_inputs", - supports_multimodal_inputs, - ) speculator = _TestSpeculator(vllm_config, torch.device("cpu")) @@ -198,7 +195,9 @@ def test_load_model_keeps_mm_support_for_capable_drafter(monkeypatch): speculator.supports_mm_inputs = False speculator.inputs_embeds = None speculator.use_acceptance_estimator = False - speculator.vllm_config = SimpleNamespace(model_config=object()) + speculator.vllm_config = SimpleNamespace( + model_config=SimpleNamespace(supports_multimodal_inputs=True) + ) speculator.max_num_tokens = 4 speculator.hidden_size = 3 speculator.dtype = torch.float32 @@ -206,11 +205,6 @@ def test_load_model_keeps_mm_support_for_capable_drafter(monkeypatch): draft_model = _MultimodalDraftModel() speculator.test_draft_model = draft_model _mock_base_model_load(monkeypatch) - monkeypatch.setattr( - base_spec_module.MULTIMODAL_REGISTRY, - "supports_multimodal_inputs", - lambda model_config: True, - ) speculator.load_model(torch.nn.Module()) @@ -223,16 +217,13 @@ def test_load_model_disables_mm_support_for_text_only_drafter(monkeypatch): speculator.supports_mm_inputs = False speculator.inputs_embeds = None speculator.use_acceptance_estimator = False - speculator.vllm_config = SimpleNamespace(model_config=object()) + speculator.vllm_config = SimpleNamespace( + model_config=SimpleNamespace(supports_multimodal_inputs=True) + ) draft_model = _TextOnlyDraftModel() speculator.test_draft_model = draft_model warning_messages = [] _mock_base_model_load(monkeypatch) - monkeypatch.setattr( - base_spec_module.MULTIMODAL_REGISTRY, - "supports_multimodal_inputs", - lambda model_config: True, - ) monkeypatch.setattr( base_spec_module.logger, "warning_once", @@ -257,7 +248,9 @@ def test_multi_module_mm_support_configured_after_model_load(monkeypatch): speculator.inputs_embeds = None speculator.use_acceptance_estimator = False speculator.cached_draft_input_embeds = None - speculator.vllm_config = SimpleNamespace(model_config=object()) + speculator.vllm_config = SimpleNamespace( + model_config=SimpleNamespace(supports_multimodal_inputs=True) + ) speculator.max_num_tokens = 4 speculator.max_num_reqs = 2 speculator.num_speculative_steps = 3 @@ -271,11 +264,6 @@ def test_multi_module_mm_support_configured_after_model_load(monkeypatch): "load_draft_model", lambda self, target_model, target_attn_layer_names: draft_model, ) - monkeypatch.setattr( - base_spec_module.MULTIMODAL_REGISTRY, - "supports_multimodal_inputs", - lambda model_config: True, - ) speculator.load_model(torch.nn.Module()) diff --git a/vllm/config/model.py b/vllm/config/model.py index 145814a268..6b36afd910 100644 --- a/vllm/config/model.py +++ b/vllm/config/model.py @@ -865,6 +865,80 @@ class ModelConfig: self._verify_quantization() self._verify_cuda_graph() + def _supports_multimodal_inputs(self) -> bool: + """Checks if the model supports multimodal inputs. + Returns True if the model is multimodal with any non-zero supported + modalities, otherwise returns False, effectively running in + text-only mode. + """ + if not self.is_multimodal_model: + return False + + from vllm.multimodal import MULTIMODAL_REGISTRY + + mm_config = self.get_multimodal_config() + try: + info = MULTIMODAL_REGISTRY.get_processing_info(self) + except ValueError: + # Speculative drafters for multimodal targets (e.g. Qwen3_5MTP, + # Exaone4_5_MTP, MiMoV2OmniMTP) declare `SupportsMultiModal` so + # that they can consume the embeddings merged by the target model, + # but they never run a multi-modal processor of their own. Running + # in text-only mode is the expected outcome for them, not a + # misconfiguration worth warning about. + if self.runner_type != "draft": + logger.warning_once( + "Model %s is treated as multimodal but has no registered " + "multimodal processor; running in text-only mode.", + self.model, + ) + return False + + # Check if all supported modalities have limit == 0 + if all( + mm_config.get_limit_per_prompt(modality) == 0 + for modality in info.supported_mm_limits + ): + # If enable_mm_embeds is True, we still need MM infrastructure + # to process pre-computed embeddings even though encoder won't run + if mm_config.enable_mm_embeds: + return True + + logger.info_once( + "All limits of multimodal modalities supported by the model " + "are set to 0, running in text-only mode." + ) + return False + + return True + + def _cached_supports_multimodal_inputs(self) -> bool: + cache = getattr(self, "_supports_multimodal_inputs_cache", None) + if cache is None: + self._supports_multimodal_inputs_cache = cache = {} + + mm_config = self.multimodal_config + if mm_config is None: + mm_cache_key = None + else: + limits_per_prompt = { + modality: mm_config.get_limit_per_prompt(modality) + for modality in mm_config.limit_per_prompt + } + mm_cache_key = ( + mm_config.language_model_only, + tuple(limits_per_prompt.items()), + mm_config.enable_mm_embeds, + ) + + cache_key = (self.is_multimodal_model, self.runner_type, mm_cache_key) + if cache_key in cache: + return cache[cache_key] + + supports_mm = self._supports_multimodal_inputs() + cache[cache_key] = supports_mm + return supports_mm + def _supports_multimodal_for_mm_prefix(self) -> bool: """Whether multimodal inputs can still appear for this deployment. @@ -881,24 +955,18 @@ class ModelConfig: vision modality is still enabled (e.g. ``image=0`` but video allowed). The deep-copied cache preserves the top-level decision instead. """ - cached = getattr(self, "_supports_multimodal_inputs_cached", None) - if cached is not None: - return cached - if self.multimodal_config is None: # Early call before multimodal init — do not clear mm_prefix yet. return True - from vllm.multimodal import MULTIMODAL_REGISTRY - - supports_mm = MULTIMODAL_REGISTRY.supports_multimodal_inputs(self) - self._supports_multimodal_inputs_cached = supports_mm + supports_mm = self._cached_supports_multimodal_inputs() if not supports_mm: logger.info_once( "Disabled mm_prefix attention mode because multimodal inputs " "are configuration-disabled. Attention backends without " "mm_prefix support may now be selected." ) + return supports_mm def get_model_arch_config(self) -> ModelArchitectureConfig: @@ -1859,6 +1927,10 @@ class ModelConfig: def is_multimodal_raw_input_only_model(self) -> bool: return self._model_info.supports_multimodal_raw_input_only + @property + def supports_multimodal_inputs(self) -> bool: + return self._cached_supports_multimodal_inputs() + @property def requires_raw_input_tokens(self) -> bool: return self._model_info.requires_raw_input_tokens diff --git a/vllm/multimodal/cache/__init__.py b/vllm/multimodal/cache/__init__.py new file mode 100644 index 0000000000..29411c1859 --- /dev/null +++ b/vllm/multimodal/cache/__init__.py @@ -0,0 +1,38 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project + +from .base import ( + BaseMultiModalProcessorCache, + BaseMultiModalReceiverCache, + MultiModalCacheMissError, + MultiModalProcessorOnlyCache, +) +from .factories import ( + engine_receiver_cache_from_config, + processor_cache_from_config, + processor_only_cache_from_config, + worker_receiver_cache_from_config, +) +from .lru import ( + LruKeyReplicatedReceiverCache, + LruKeyReplicatedSenderCache, +) +from .shm import ( + ShmObjectStoreReceiverCache, + ShmObjectStoreSenderCache, +) + +__all__ = [ + "MultiModalCacheMissError", + "BaseMultiModalProcessorCache", + "BaseMultiModalReceiverCache", + "MultiModalProcessorOnlyCache", + "LruKeyReplicatedSenderCache", + "LruKeyReplicatedReceiverCache", + "ShmObjectStoreReceiverCache", + "ShmObjectStoreSenderCache", + "processor_cache_from_config", + "processor_only_cache_from_config", + "engine_receiver_cache_from_config", + "worker_receiver_cache_from_config", +] diff --git a/vllm/multimodal/cache.py b/vllm/multimodal/cache/base.py similarity index 54% rename from vllm/multimodal/cache.py rename to vllm/multimodal/cache/base.py index 15d433112c..1b9b757917 100644 --- a/vllm/multimodal/cache.py +++ b/vllm/multimodal/cache/base.py @@ -4,26 +4,19 @@ import operator import sys from abc import ABC, abstractmethod from collections.abc import Mapping, Sequence -from multiprocessing.synchronize import Lock as LockType -from typing import TYPE_CHECKING, Generic, TypeAlias, TypeVar, cast +from typing import TYPE_CHECKING, Generic, TypeAlias, TypeVar import torch from typing_extensions import override -import vllm.envs as envs -from vllm.distributed.device_communicators.shm_object_storage import ( - MsgpackSerde, - SingleWriterShmObjectStorage, - SingleWriterShmRingBuffer, -) +from vllm.config import ModelConfig from vllm.logger import init_logger from vllm.utils.cache import CacheInfo, LRUCache from vllm.utils.jsontree import json_count_leaves, json_map_leaves, json_reduce_leaves -from vllm.utils.mem_constants import GiB_bytes, MiB_bytes +from vllm.utils.mem_constants import GiB_bytes from vllm.utils.mem_utils import format_gib -from .inputs import ( - MultiModalBatchedField, +from ..inputs import ( MultiModalFeatureSpec, MultiModalFieldElem, MultiModalKwargsItem, @@ -32,9 +25,7 @@ from .inputs import ( ) if TYPE_CHECKING: - from vllm.config import ModelConfig, VllmConfig - - from .processing.processor import ResolvedPromptUpdate + from ..processing.processor import ResolvedPromptUpdate logger = init_logger(__name__) @@ -399,7 +390,7 @@ class MultiModalProcessorOnlyCache(BaseMultiModalProcessorCache): tensor data and metadata) into the cache, and return the input. """ - def __init__(self, model_config: "ModelConfig") -> None: + def __init__(self, model_config: ModelConfig) -> None: super().__init__() mm_config = model_config.get_multimodal_config() @@ -440,216 +431,6 @@ class MultiModalProcessorOnlyCache(BaseMultiModalProcessorCache): return self._cache.stat(delta=delta) -class MultiModalProcessorSenderCache(BaseMultiModalProcessorCache): - """The cache which is used on P0 when IPC caching is enabled. - - How to update each item: - - - If the item is already in the cache, clear the input to avoid - unnecessary IPC. - - - If the item is not in the cache, store the metadata of that item so - that the eviction policy remains the same as the cache on P1, - and return the input. - By only storing the metadata, we avoid keeping the data itself in - memory inside P0. - """ - - def __init__(self, model_config: "ModelConfig") -> None: - super().__init__() - - mm_config = model_config.get_multimodal_config() - - self._cache = MultiModalCache.get_lru_cache( - mm_config.mm_processor_cache_gb, - MultiModalProcessorCacheItemMetadata, - ) - - @override - def is_cached_item(self, mm_hash: str) -> bool: - return mm_hash in self._cache - - @override - def get_and_update_item( - self, - mm_item: MultiModalProcessorCacheInItem, - mm_hash: str, - ) -> MultiModalProcessorCacheOutItem: - if (cached_item := self._cache.get(mm_hash)) is not None: - return None, cached_item.prompt_updates - - assert mm_item is not None, f"Expected a cached item for {mm_hash=}" - - self.cache_if_fits( - self._cache, mm_hash, MultiModalProcessorCacheItemMetadata(*mm_item) - ) - return mm_item - - @override - def touch_sender_cache_item(self, mm_hash: str) -> None: - self._cache.touch(mm_hash) - - @override - def clear_cache(self) -> None: - self._cache.clear() - - @override - def make_stats(self, *, delta: bool = False) -> CacheInfo: - return self._cache.stat(delta=delta) - - @override - def invalidate(self, mm_hash: str) -> None: - # Drop our stale shadow entry so the next request for this hash re-sends the - # data and repopulates P1 (see MultiModalCacheMissError). - self._cache.pop(mm_hash, None) - - -class ShmObjectStoreSenderCache(BaseMultiModalProcessorCache): - """The cache which is used on P0 when IPC caching is enabled. - - How to update each item: - - - If the item is already in the cache, clear the input to avoid - unnecessary IPC. - - - If the item is not in the cache, store the data in shared memory. - """ - - def __init__(self, vllm_config: "VllmConfig") -> None: - super().__init__() - - self.world_size = vllm_config.parallel_config.world_size - mm_config = vllm_config.model_config.get_multimodal_config() - - ring_buffer = SingleWriterShmRingBuffer( - data_buffer_size=int(mm_config.mm_processor_cache_gb * GiB_bytes), - name=envs.VLLM_OBJECT_STORAGE_SHM_BUFFER_NAME, - create=True, # sender is the writer - ) - self._shm_cache = SingleWriterShmObjectStorage( - max_object_size=mm_config.mm_shm_cache_max_object_size_mb * MiB_bytes, - n_readers=self.world_size, - ring_buffer=ring_buffer, - serde_class=MsgpackSerde, - ) - # cache prompt_updates for P0 only - self._p0_cache: dict[str, Sequence[ResolvedPromptUpdate]] = {} - - self._hits = 0 - self._total = 0 - self._last_info = CacheInfo(hits=0, total=0) - - def _stat(self, *, delta: bool = False) -> CacheInfo: - info = CacheInfo(hits=self._hits, total=self._total) - - if delta: - info_delta = info - self._last_info - self._last_info = info - info = info_delta - - return info - - @override - def is_cached_item(self, mm_hash: str) -> bool: - return self._shm_cache.is_cached(mm_hash) - - @override - def get_and_update_item( - self, - mm_item: MultiModalProcessorCacheInItem, - mm_hash: str, - ) -> MultiModalProcessorCacheOutItem: - if self._shm_cache.is_cached(mm_hash): - self._hits += 1 - self._total += 1 - - address, monotonic_id = self._shm_cache.get_cached(mm_hash) - prompt_updates = self._p0_cache[mm_hash] - return self.address_as_item(address, monotonic_id), prompt_updates - - assert mm_item is not None, f"Expected a cached item for {mm_hash=}" - item, prompt_updates = mm_item - - self._total += 1 - - try: - address, monotonic_id = self._shm_cache.put(mm_hash, item) - # Try to remove dangling items if p0 cache is too large. - if len(self._p0_cache) >= 2 * len(self._shm_cache.key_index): - self.remove_dangling_items() - - self._p0_cache[mm_hash] = prompt_updates - return self.address_as_item(address, monotonic_id), prompt_updates - except ValueError as e: - # `put` raises ValueError either for an oversize item or for a - # duplicate key (concurrent insert); the latter is benign so we - # only warn on the oversize case. Subsequent UUID-only requests - # for an oversize item will fail with a cache miss. - if "already exists" not in str(e): - logger.warning_once( - "mm_input %s too large to cache; " - "raise --mm-shm-cache-max-object-size-mb. (%s)", - mm_hash, - str(e), - ) - return mm_item - except MemoryError as e: - # Cache full and protected items prevent eviction. - logger.debug( - "mm_input %s not cached; shm cache full, " - "consider raising --mm-processor-cache-gb. (%s)", - mm_hash, - str(e), - ) - return mm_item - - @override - def touch_sender_cache_item(self, mm_hash: str) -> None: - """Touch the item in shared memory cache to prevent eviction. - Increments writer_flag on sender side.""" - self._shm_cache.touch(mm_hash) - - @override - def clear_cache(self) -> None: - self._shm_cache.clear() - self._p0_cache.clear() - - self._hits = 0 - self._total = 0 - self._last_info = CacheInfo(hits=0, total=0) - - @override - def make_stats(self, *, delta: bool = False) -> CacheInfo: - return self._stat(delta=delta) - - @override - def close(self) -> None: - self._shm_cache.close() - - def remove_dangling_items(self) -> None: - """Remove items that are no longer in the shared memory cache.""" - cached_hashes = self._shm_cache.key_index.keys() - dangling_hashes = set(self._p0_cache.keys()) - cached_hashes - for mm_hash in dangling_hashes: - del self._p0_cache[mm_hash] - - def address_as_item( - self, - address: int, - monotonic_id: int, - ) -> MultiModalKwargsItem: - addr_elem = MultiModalFieldElem( - data=address, - field=MultiModalBatchedField(), - ) - id_elem = MultiModalFieldElem( - data=monotonic_id, - field=MultiModalBatchedField(), - ) - - return MultiModalKwargsItem({"address": addr_elem, "monotonic_id": id_elem}) - - class BaseMultiModalReceiverCache( BaseMultiModalCache[MultiModalKwargsItem | None, MultiModalKwargsItem] ): @@ -703,139 +484,3 @@ class BaseMultiModalReceiverCache( """ raise NotImplementedError - - -class MultiModalReceiverCache(BaseMultiModalReceiverCache): - """The cache which is used on P1 when IPC caching is enabled. - - How to update each item: - - - If the caller sent tensor data, store it (replacing any cached item - under the same key) and return that data. P0 can miss after independent - LRU eviction and resend a different item for the same identity. - - If the caller sent no data and the item is cached, return the cached item. - - If the caller sent no data and the item is not cached, raise - ``MultiModalCacheMissError``. - """ - - def __init__(self, model_config: "ModelConfig") -> None: - super().__init__() - - mm_config = model_config.get_multimodal_config() - - self._cache = MultiModalCache.get_lru_cache( - mm_config.mm_processor_cache_gb, - MultiModalKwargsItem, - ) - - @override - def get_and_update_item( - self, - mm_item: MultiModalKwargsItem | None, - mm_hash: str, - ) -> MultiModalKwargsItem: - if mm_item is not None: - # P0 sent a payload. Never keep a stale cached tensor under this - # identity: a later P0 hit would pair new placeholders with the old - # item and crash the engine. Drop first so a too-large replacement - # cannot leave the previous entry in place. - self._cache.pop(mm_hash, None) - self.cache_if_fits(self._cache, mm_hash, mm_item) - return mm_item - - if (cached_item := self._cache.get(mm_hash)) is not None: - return cached_item - - # No data and not cached here: P0 sent data=None trusting its shadow, but - # the P0/P1 caches have drifted. Raise a retryable error (not assert) so the - # engine can have P0 drop the stale entry and the client resend the data. - raise MultiModalCacheMissError([mm_hash]) - - @override - def touch_receiver_cache_item( - self, - mm_hash: str, - mm_item: MultiModalKwargsItem | None = None, - ) -> None: - self._cache.touch(mm_hash) - - @override - def clear_cache(self) -> None: - self._cache.clear() - - -class ShmObjectStoreReceiverCache(BaseMultiModalReceiverCache): - """The cache which is used on P1 Worker Process when IPC caching is enabled. - - How to update each item: - - - If the item has an address, replace the input with the cached item. - - If not, return the input. - """ - - def __init__( - self, - vllm_config: "VllmConfig", - shared_worker_lock: LockType, - ) -> None: - super().__init__() - - self.world_size = vllm_config.parallel_config.world_size - mm_config = vllm_config.model_config.get_multimodal_config() - - ring_buffer = SingleWriterShmRingBuffer( - data_buffer_size=int(mm_config.mm_processor_cache_gb * GiB_bytes), - name=envs.VLLM_OBJECT_STORAGE_SHM_BUFFER_NAME, - create=False, # Server is a reader - ) - self._shm_cache = SingleWriterShmObjectStorage( - max_object_size=mm_config.mm_shm_cache_max_object_size_mb * MiB_bytes, - n_readers=self.world_size, - ring_buffer=ring_buffer, - serde_class=MsgpackSerde, - reader_lock=shared_worker_lock, - ) - - @override - def get_and_update_features( - self, mm_features: list["MultiModalFeatureSpec"] - ) -> list["MultiModalFeatureSpec"]: - # strip_covered_mm_data preserves address items, so None here represents - # a stripped uncached payload and has no SHM reference to acknowledge. - features_with_data = [ - feature for feature in mm_features if feature.data is not None - ] - super().get_and_update_features(features_with_data) - return mm_features - - @override - def get_and_update_item( - self, - mm_item: MultiModalKwargsItem | None, - mm_hash: str, - ) -> MultiModalKwargsItem: - assert mm_item is not None, f"Expected an address item for {mm_hash=}" - if "address" in mm_item: - address = cast(int, mm_item["address"].data) - monotonic_id = cast(int, mm_item["monotonic_id"].data) - return self._shm_cache.get(address, monotonic_id) - - return mm_item - - @override - def touch_receiver_cache_item( - self, - mm_hash: str, - mm_item: MultiModalKwargsItem | None = None, - ) -> None: - """Touch the item in shared memory cache to prevent eviction. - Increments reader_count on receiver side.""" - assert mm_item is not None - if "address" in mm_item: - address = cast(int, mm_item["address"].data) - monotonic_id = cast(int, mm_item["monotonic_id"].data) - self._shm_cache.touch(mm_hash, address=address, monotonic_id=monotonic_id) - - @override - def clear_cache(self) -> None: - self._shm_cache.clear() diff --git a/vllm/multimodal/cache/factories.py b/vllm/multimodal/cache/factories.py new file mode 100644 index 0000000000..aa4f5a4d2c --- /dev/null +++ b/vllm/multimodal/cache/factories.py @@ -0,0 +1,107 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +from multiprocessing.synchronize import Lock as LockType +from typing import Literal + +from vllm.config import VllmConfig + +from .base import ( + BaseMultiModalProcessorCache, + BaseMultiModalReceiverCache, + MultiModalProcessorOnlyCache, +) +from .lru import ( + LruKeyReplicatedReceiverCache, + LruKeyReplicatedSenderCache, +) +from .shm import ( + ShmObjectStoreReceiverCache, + ShmObjectStoreSenderCache, +) + + +def _get_cache_type( + vllm_config: VllmConfig, +) -> Literal[None, "processor_only", "lru", "shm"]: + model_config = vllm_config.model_config + if not model_config.supports_multimodal_inputs: + return None + + # Check if the cache is disabled. + mm_config = model_config.get_multimodal_config() + if mm_config.mm_processor_cache_gb <= 0: + return None + + # Check if IPC caching is supported. + parallel_config = vllm_config.parallel_config + is_ipc_supported = parallel_config._api_process_count == 1 and ( + parallel_config.data_parallel_size == 1 + or parallel_config.data_parallel_external_lb + ) + + if not is_ipc_supported: + return "processor_only" + + mm_config = model_config.get_multimodal_config() + return mm_config.mm_processor_cache_type + + +def processor_cache_from_config( + vllm_config: VllmConfig, +) -> BaseMultiModalProcessorCache | None: + """Return a `BaseMultiModalProcessorCache`, if enabled.""" + cache_type = _get_cache_type(vllm_config) + if cache_type is None: + return None + elif cache_type == "processor_only": + return MultiModalProcessorOnlyCache(vllm_config.model_config) + elif cache_type == "lru": + return LruKeyReplicatedSenderCache(vllm_config.model_config) + elif cache_type == "shm": + return ShmObjectStoreSenderCache(vllm_config) + else: + raise ValueError(f"Unknown cache type: {cache_type!r}") + + +def processor_only_cache_from_config( + vllm_config: VllmConfig, +) -> MultiModalProcessorOnlyCache | None: + """Return a `MultiModalProcessorOnlyCache`, if enabled.""" + cache_type = _get_cache_type(vllm_config) + if cache_type is None: + return None + + return MultiModalProcessorOnlyCache(vllm_config.model_config) + + +def engine_receiver_cache_from_config( + vllm_config: VllmConfig, +) -> BaseMultiModalReceiverCache | None: + """Return a `BaseMultiModalReceiverCache` for the engine process.""" + cache_type = _get_cache_type(vllm_config) + if cache_type in (None, "processor_only", "shm"): + return None + elif cache_type == "lru": + return LruKeyReplicatedReceiverCache(vllm_config.model_config) + else: + raise ValueError(f"Unknown cache type: {cache_type!r}") + + +def worker_receiver_cache_from_config( + vllm_config: VllmConfig, + shared_worker_lock: LockType | None, +) -> BaseMultiModalReceiverCache | None: + """Return a `BaseMultiModalReceiverCache` for the worker process.""" + cache_type = _get_cache_type(vllm_config) + if cache_type in (None, "processor_only", "lru"): + return None + elif cache_type == "shm": + if shared_worker_lock is None: + raise ValueError( + "Missing `shared_worker_lock` argument from executor. " + "This argument is needed for mm_processor_cache_type='shm'." + ) + + return ShmObjectStoreReceiverCache(vllm_config, shared_worker_lock) + else: + raise ValueError(f"Unknown cache type: {cache_type!r}") diff --git a/vllm/multimodal/cache/lru.py b/vllm/multimodal/cache/lru.py new file mode 100644 index 0000000000..6b5ec3516a --- /dev/null +++ b/vllm/multimodal/cache/lru.py @@ -0,0 +1,142 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Implementation of Key-Replicated Cache (see docs/configuration/optimization.md).""" + +from typing_extensions import override + +from vllm.config import ModelConfig +from vllm.utils.cache import CacheInfo + +from ..inputs import MultiModalKwargsItem +from .base import ( + BaseMultiModalProcessorCache, + BaseMultiModalReceiverCache, + MultiModalCache, + MultiModalCacheMissError, + MultiModalProcessorCacheInItem, + MultiModalProcessorCacheItemMetadata, + MultiModalProcessorCacheOutItem, +) + + +class LruKeyReplicatedSenderCache(BaseMultiModalProcessorCache): + """The cache which is used on P0 when LRU caching is enabled. + + How to update each item: + + - If the item is already in the cache, clear the input to avoid + unnecessary IPC. + + - If the item is not in the cache, store the metadata of that item so + that the eviction policy remains the same as the cache on P1, + and return the input. + By only storing the metadata, we avoid keeping the data itself in + memory inside P0. + """ + + def __init__(self, model_config: ModelConfig) -> None: + super().__init__() + + mm_config = model_config.get_multimodal_config() + + self._cache = MultiModalCache.get_lru_cache( + mm_config.mm_processor_cache_gb, + MultiModalProcessorCacheItemMetadata, + ) + + @override + def is_cached_item(self, mm_hash: str) -> bool: + return mm_hash in self._cache + + @override + def get_and_update_item( + self, + mm_item: MultiModalProcessorCacheInItem, + mm_hash: str, + ) -> MultiModalProcessorCacheOutItem: + if (cached_item := self._cache.get(mm_hash)) is not None: + return None, cached_item.prompt_updates + + assert mm_item is not None, f"Expected a cached item for {mm_hash=}" + + self.cache_if_fits( + self._cache, mm_hash, MultiModalProcessorCacheItemMetadata(*mm_item) + ) + return mm_item + + @override + def touch_sender_cache_item(self, mm_hash: str) -> None: + self._cache.touch(mm_hash) + + @override + def clear_cache(self) -> None: + self._cache.clear() + + @override + def make_stats(self, *, delta: bool = False) -> CacheInfo: + return self._cache.stat(delta=delta) + + @override + def invalidate(self, mm_hash: str) -> None: + # Drop our stale shadow entry so the next request for this hash re-sends the + # data and repopulates P1 (see MultiModalCacheMissError). + self._cache.pop(mm_hash, None) + + +class LruKeyReplicatedReceiverCache(BaseMultiModalReceiverCache): + """The cache which is used on P1 when LRU caching is enabled. + + How to update each item: + + - If the caller sent tensor data, store it (replacing any cached item + under the same key) and return that data. P0 can miss after independent + LRU eviction and resend a different item for the same identity. + - If the caller sent no data and the item is cached, return the cached item. + - If the caller sent no data and the item is not cached, raise + `MultiModalCacheMissError`. + """ + + def __init__(self, model_config: ModelConfig) -> None: + super().__init__() + + mm_config = model_config.get_multimodal_config() + + self._cache = MultiModalCache.get_lru_cache( + mm_config.mm_processor_cache_gb, + MultiModalKwargsItem, + ) + + @override + def get_and_update_item( + self, + mm_item: MultiModalKwargsItem | None, + mm_hash: str, + ) -> MultiModalKwargsItem: + if mm_item is not None: + # P0 sent a payload. Never keep a stale cached tensor under this + # identity: a later P0 hit would pair new placeholders with the old + # item and crash the engine. Drop first so a too-large replacement + # cannot leave the previous entry in place. + self._cache.pop(mm_hash, None) + self.cache_if_fits(self._cache, mm_hash, mm_item) + return mm_item + + if (cached_item := self._cache.get(mm_hash)) is not None: + return cached_item + + # No data and not cached here: P0 sent data=None trusting its shadow, but + # the P0/P1 caches have drifted. Raise a retryable error (not assert) so the + # engine can have P0 drop the stale entry and the client resend the data. + raise MultiModalCacheMissError([mm_hash]) + + @override + def touch_receiver_cache_item( + self, + mm_hash: str, + mm_item: MultiModalKwargsItem | None = None, + ) -> None: + self._cache.touch(mm_hash) + + @override + def clear_cache(self) -> None: + self._cache.clear() diff --git a/vllm/multimodal/cache/shm.py b/vllm/multimodal/cache/shm.py new file mode 100644 index 0000000000..dddf3494cf --- /dev/null +++ b/vllm/multimodal/cache/shm.py @@ -0,0 +1,261 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Implementation of Shared Memory Cache (see docs/configuration/optimization.md).""" + +from collections.abc import Sequence +from multiprocessing.synchronize import Lock as LockType +from typing import TYPE_CHECKING, cast + +from typing_extensions import override + +import vllm.envs as envs +from vllm.config import VllmConfig +from vllm.distributed.device_communicators.shm_object_storage import ( + MsgpackSerde, + SingleWriterShmObjectStorage, + SingleWriterShmRingBuffer, +) +from vllm.logger import init_logger +from vllm.utils.cache import CacheInfo +from vllm.utils.mem_constants import GiB_bytes, MiB_bytes + +from ..inputs import ( + MultiModalBatchedField, + MultiModalFeatureSpec, + MultiModalFieldElem, + MultiModalKwargsItem, +) +from .base import ( + BaseMultiModalProcessorCache, + BaseMultiModalReceiverCache, + MultiModalProcessorCacheInItem, + MultiModalProcessorCacheOutItem, +) + +if TYPE_CHECKING: + from ..processing.processor import ResolvedPromptUpdate + +logger = init_logger(__name__) + + +class ShmObjectStoreSenderCache(BaseMultiModalProcessorCache): + """The cache which is used on P0 when SHM caching is enabled. + + How to update each item: + + - If the item is already in the cache, clear the input to avoid + unnecessary IPC. + + - If the item is not in the cache, store the data in shared memory. + """ + + def __init__(self, vllm_config: VllmConfig) -> None: + super().__init__() + + self.world_size = vllm_config.parallel_config.world_size + mm_config = vllm_config.model_config.get_multimodal_config() + + ring_buffer = SingleWriterShmRingBuffer( + data_buffer_size=int(mm_config.mm_processor_cache_gb * GiB_bytes), + name=envs.VLLM_OBJECT_STORAGE_SHM_BUFFER_NAME, + create=True, # sender is the writer + ) + self._shm_cache = SingleWriterShmObjectStorage( + max_object_size=mm_config.mm_shm_cache_max_object_size_mb * MiB_bytes, + n_readers=self.world_size, + ring_buffer=ring_buffer, + serde_class=MsgpackSerde, + ) + # cache prompt_updates for P0 only + self._p0_cache: dict[str, Sequence[ResolvedPromptUpdate]] = {} + + self._hits = 0 + self._total = 0 + self._last_info = CacheInfo(hits=0, total=0) + + def _stat(self, *, delta: bool = False) -> CacheInfo: + info = CacheInfo(hits=self._hits, total=self._total) + + if delta: + info_delta = info - self._last_info + self._last_info = info + info = info_delta + + return info + + @override + def is_cached_item(self, mm_hash: str) -> bool: + return self._shm_cache.is_cached(mm_hash) + + @override + def get_and_update_item( + self, + mm_item: MultiModalProcessorCacheInItem, + mm_hash: str, + ) -> MultiModalProcessorCacheOutItem: + if self._shm_cache.is_cached(mm_hash): + self._hits += 1 + self._total += 1 + + address, monotonic_id = self._shm_cache.get_cached(mm_hash) + prompt_updates = self._p0_cache[mm_hash] + return self.address_as_item(address, monotonic_id), prompt_updates + + assert mm_item is not None, f"Expected a cached item for {mm_hash=}" + item, prompt_updates = mm_item + + self._total += 1 + + try: + address, monotonic_id = self._shm_cache.put(mm_hash, item) + # Try to remove dangling items if p0 cache is too large. + if len(self._p0_cache) >= 2 * len(self._shm_cache.key_index): + self.remove_dangling_items() + + self._p0_cache[mm_hash] = prompt_updates + return self.address_as_item(address, monotonic_id), prompt_updates + except ValueError as e: + # `put` raises ValueError either for an oversize item or for a + # duplicate key (concurrent insert); the latter is benign so we + # only warn on the oversize case. Subsequent UUID-only requests + # for an oversize item will fail with a cache miss. + if "already exists" not in str(e): + logger.warning_once( + "mm_input %s too large to cache; " + "raise --mm-shm-cache-max-object-size-mb. (%s)", + mm_hash, + str(e), + ) + return mm_item + except MemoryError as e: + # Cache full and protected items prevent eviction. + logger.debug( + "mm_input %s not cached; shm cache full, " + "consider raising --mm-processor-cache-gb. (%s)", + mm_hash, + str(e), + ) + return mm_item + + @override + def touch_sender_cache_item(self, mm_hash: str) -> None: + """Touch the item in shared memory cache to prevent eviction. + Increments writer_flag on sender side.""" + self._shm_cache.touch(mm_hash) + + @override + def clear_cache(self) -> None: + self._shm_cache.clear() + self._p0_cache.clear() + + self._hits = 0 + self._total = 0 + self._last_info = CacheInfo(hits=0, total=0) + + @override + def make_stats(self, *, delta: bool = False) -> CacheInfo: + return self._stat(delta=delta) + + @override + def close(self) -> None: + self._shm_cache.close() + + def remove_dangling_items(self) -> None: + """Remove items that are no longer in the shared memory cache.""" + cached_hashes = self._shm_cache.key_index.keys() + dangling_hashes = set(self._p0_cache.keys()) - cached_hashes + for mm_hash in dangling_hashes: + del self._p0_cache[mm_hash] + + def address_as_item( + self, + address: int, + monotonic_id: int, + ) -> MultiModalKwargsItem: + addr_elem = MultiModalFieldElem( + data=address, + field=MultiModalBatchedField(), + ) + id_elem = MultiModalFieldElem( + data=monotonic_id, + field=MultiModalBatchedField(), + ) + + return MultiModalKwargsItem({"address": addr_elem, "monotonic_id": id_elem}) + + +class ShmObjectStoreReceiverCache(BaseMultiModalReceiverCache): + """The cache which is used on P1 Worker Process when SHM caching is enabled. + + How to update each item: + + - If the item has an address, replace the input with the cached item. + - If not, return the input. + """ + + def __init__( + self, + vllm_config: VllmConfig, + shared_worker_lock: LockType, + ) -> None: + super().__init__() + + self.world_size = vllm_config.parallel_config.world_size + mm_config = vllm_config.model_config.get_multimodal_config() + + ring_buffer = SingleWriterShmRingBuffer( + data_buffer_size=int(mm_config.mm_processor_cache_gb * GiB_bytes), + name=envs.VLLM_OBJECT_STORAGE_SHM_BUFFER_NAME, + create=False, # Server is a reader + ) + self._shm_cache = SingleWriterShmObjectStorage( + max_object_size=mm_config.mm_shm_cache_max_object_size_mb * MiB_bytes, + n_readers=self.world_size, + ring_buffer=ring_buffer, + serde_class=MsgpackSerde, + reader_lock=shared_worker_lock, + ) + + @override + def get_and_update_features( + self, mm_features: list["MultiModalFeatureSpec"] + ) -> list["MultiModalFeatureSpec"]: + # strip_covered_mm_data preserves address items, so None here represents + # a stripped uncached payload and has no SHM reference to acknowledge. + features_with_data = [ + feature for feature in mm_features if feature.data is not None + ] + super().get_and_update_features(features_with_data) + return mm_features + + @override + def get_and_update_item( + self, + mm_item: MultiModalKwargsItem | None, + mm_hash: str, + ) -> MultiModalKwargsItem: + assert mm_item is not None, f"Expected an address item for {mm_hash=}" + if "address" in mm_item: + address = cast(int, mm_item["address"].data) + monotonic_id = cast(int, mm_item["monotonic_id"].data) + return self._shm_cache.get(address, monotonic_id) + + return mm_item + + @override + def touch_receiver_cache_item( + self, + mm_hash: str, + mm_item: MultiModalKwargsItem | None = None, + ) -> None: + """Touch the item in shared memory cache to prevent eviction. + Increments reader_count on receiver side.""" + assert mm_item is not None + if "address" in mm_item: + address = cast(int, mm_item["address"].data) + monotonic_id = cast(int, mm_item["monotonic_id"].data) + self._shm_cache.touch(mm_hash, address=address, monotonic_id=monotonic_id) + + @override + def clear_cache(self) -> None: + self._shm_cache.clear() diff --git a/vllm/multimodal/encoder_budget.py b/vllm/multimodal/encoder_budget.py index 13448b9096..32db904bfa 100644 --- a/vllm/multimodal/encoder_budget.py +++ b/vllm/multimodal/encoder_budget.py @@ -4,7 +4,10 @@ from collections.abc import Mapping from vllm.config import VllmConfig from vllm.logger import init_logger -from vllm.multimodal.cache import BaseMultiModalProcessorCache +from vllm.multimodal.cache import ( + BaseMultiModalProcessorCache, + processor_only_cache_from_config, +) from vllm.multimodal.inputs import MultiModalKwargsItem from vllm.multimodal.processing import BaseMultiModalProcessor from vllm.multimodal.registry import MultiModalRegistry @@ -57,9 +60,7 @@ class MultiModalBudget: with set_default_torch_num_threads(): # Avoid hang during startup cache = ( - mm_registry.processor_only_cache_from_config(vllm_config) - if enable_cache - else None + processor_only_cache_from_config(vllm_config) if enable_cache else None ) processor = mm_registry.create_processor(model_config) diff --git a/vllm/multimodal/registry.py b/vllm/multimodal/registry.py index 24a7a696b4..28383979f9 100644 --- a/vllm/multimodal/registry.py +++ b/vllm/multimodal/registry.py @@ -3,21 +3,11 @@ import threading from collections import defaultdict from dataclasses import dataclass -from multiprocessing.synchronize import Lock as LockType -from typing import TYPE_CHECKING, Generic, Literal, Protocol, TypeVar, cast +from typing import TYPE_CHECKING, Generic, Protocol, TypeVar, cast from vllm.logger import init_logger from vllm.tokenizers import TokenizerLike, cached_tokenizer_from_config -from .cache import ( - BaseMultiModalProcessorCache, - BaseMultiModalReceiverCache, - MultiModalProcessorOnlyCache, - MultiModalProcessorSenderCache, - MultiModalReceiverCache, - ShmObjectStoreReceiverCache, - ShmObjectStoreSenderCache, -) from .processing import ( BaseDummyInputsBuilder, BaseMultiModalProcessor, @@ -27,7 +17,7 @@ from .processing import ( ) if TYPE_CHECKING: - from vllm.config import ModelConfig, ObservabilityConfig, VllmConfig + from vllm.config import ModelConfig, ObservabilityConfig from vllm.model_executor.models.interfaces import SupportsMultiModal logger = init_logger(__name__) @@ -89,51 +79,6 @@ class _ProcessorFactories(Generic[_I]): class MultiModalRegistry: """A registry that dispatches data processing according to the model.""" - def supports_multimodal_inputs(self, model_config: "ModelConfig") -> bool: - """Checks if the model supports multimodal inputs. - Returns True if the model is multimodal with any non-zero supported - modalities, otherwise returns False, effectively running in - text-only mode. - """ - if not model_config.is_multimodal_model: - return False - - mm_config = model_config.get_multimodal_config() - try: - info = self._create_processing_info(model_config, tokenizer=None) - except ValueError: - # Speculative drafters for multimodal targets (e.g. Qwen3_5MTP, - # Exaone4_5_MTP, MiMoV2OmniMTP) declare `SupportsMultiModal` so - # that they can consume the embeddings merged by the target model, - # but they never run a multi-modal processor of their own. Running - # in text-only mode is the expected outcome for them, not a - # misconfiguration worth warning about. - if model_config.runner_type != "draft": - logger.warning_once( - "Model %s is treated as multimodal but has no registered " - "multimodal processor; running in text-only mode.", - model_config.model, - ) - return False - - # Check if all supported modalities have limit == 0 - if all( - mm_config.get_limit_per_prompt(modality) == 0 - for modality in info.supported_mm_limits - ): - # If enable_mm_embeds is True, we still need MM infrastructure - # to process pre-computed embeddings even though encoder won't run - if mm_config.enable_mm_embeds: - return True - - logger.info_once( - "All limits of multimodal modalities supported by the model " - "are set to 0, running in text-only mode." - ) - return False - - return True - def register_processor( self, processor: MultiModalProcessorFactory[_I], @@ -220,87 +165,6 @@ class MultiModalRegistry: return factories.build_processor(ctx) - def _get_cache_type( - self, - vllm_config: "VllmConfig", - ) -> Literal[None, "processor_only", "lru", "shm"]: - model_config = vllm_config.model_config - if not self.supports_multimodal_inputs(model_config): - return None - - # Check if the cache is disabled. - mm_config = model_config.get_multimodal_config() - if mm_config.mm_processor_cache_gb <= 0: - return None - - # Check if IPC caching is supported. - parallel_config = vllm_config.parallel_config - is_ipc_supported = parallel_config._api_process_count == 1 and ( - parallel_config.data_parallel_size == 1 - or parallel_config.data_parallel_external_lb - ) - - if not is_ipc_supported: - return "processor_only" - - mm_config = model_config.get_multimodal_config() - return mm_config.mm_processor_cache_type - - def processor_cache_from_config( - self, - vllm_config: "VllmConfig", - ) -> BaseMultiModalProcessorCache | None: - """Return a `BaseMultiModalProcessorCache`, if enabled.""" - cache_type = self._get_cache_type(vllm_config) - if cache_type is None: - return None - elif cache_type == "processor_only": - return MultiModalProcessorOnlyCache(vllm_config.model_config) - elif cache_type == "lru": - return MultiModalProcessorSenderCache(vllm_config.model_config) - elif cache_type == "shm": - return ShmObjectStoreSenderCache(vllm_config) - else: - raise ValueError(f"Unknown cache type: {cache_type!r}") - - def processor_only_cache_from_config( - self, - vllm_config: "VllmConfig", - ) -> MultiModalProcessorOnlyCache | None: - """Return a `MultiModalProcessorOnlyCache`, if enabled.""" - cache_type = self._get_cache_type(vllm_config) - if cache_type is None: - return None - - return MultiModalProcessorOnlyCache(vllm_config.model_config) - - def engine_receiver_cache_from_config( - self, - vllm_config: "VllmConfig", - ) -> BaseMultiModalReceiverCache | None: - """Return a `BaseMultiModalReceiverCache` for the engine process.""" - cache_type = self._get_cache_type(vllm_config) - if cache_type in (None, "processor_only", "shm"): - return None - elif cache_type == "lru": - return MultiModalReceiverCache(vllm_config.model_config) - else: - raise ValueError(f"Unknown cache type: {cache_type!r}") - - def worker_receiver_cache_from_config( - self, - vllm_config: "VllmConfig", - shared_worker_lock: LockType, - ) -> BaseMultiModalReceiverCache | None: - """Return a `BaseMultiModalReceiverCache` for the worker process.""" - cache_type = self._get_cache_type(vllm_config) - if cache_type in (None, "processor_only", "lru"): - return None - elif cache_type == "shm": - return ShmObjectStoreReceiverCache(vllm_config, shared_worker_lock) - else: - raise ValueError(f"Unknown cache type: {cache_type!r}") - class MultiModalTimingRegistry: def __init__(self, observability_config: "ObservabilityConfig | None") -> None: diff --git a/vllm/renderers/base.py b/vllm/renderers/base.py index 9b769686a7..53e3f50b4e 100644 --- a/vllm/renderers/base.py +++ b/vllm/renderers/base.py @@ -32,7 +32,11 @@ from vllm.inputs import ( ) from vllm.logger import init_logger from vllm.multimodal import MULTIMODAL_REGISTRY as mm_registry -from vllm.multimodal.cache import BaseMultiModalProcessorCache +from vllm.multimodal.cache import ( + BaseMultiModalProcessorCache, + processor_cache_from_config, + processor_only_cache_from_config, +) from vllm.multimodal.gpu_ipc_memory import maybe_init_mm_gpu_ipc_pool from vllm.multimodal.parse import ( MultiModalDataItems, @@ -149,7 +153,7 @@ class BaseRenderer(ABC, Generic[_T]): self._mm_cache_stats: MultiModalCacheStats | None = None - if mm_registry.supports_multimodal_inputs(config.model_config): + if config.model_config.supports_multimodal_inputs: # Install the process-global GPU memory pool used to gate # frontend GPU-side multimodal decoding (no-op when the budget # is 0). Lives in the API-server process only. @@ -166,10 +170,8 @@ class BaseRenderer(ABC, Generic[_T]): tokenizer=self.tokenizer, ) - self._mm_processor_cache = mm_registry.processor_cache_from_config(config) - self._mm_processor_only_cache = ( - mm_registry.processor_only_cache_from_config(config) - ) + self._mm_processor_cache = processor_cache_from_config(config) + self._mm_processor_only_cache = 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 diff --git a/vllm/v1/core/sched/scheduler.py b/vllm/v1/core/sched/scheduler.py index 7d8aee55dc..36a09f9adc 100644 --- a/vllm/v1/core/sched/scheduler.py +++ b/vllm/v1/core/sched/scheduler.py @@ -234,9 +234,7 @@ class Scheduler(SchedulerInterface): # Encoder-related. # Calculate encoder cache size if applicable - supports_mm_inputs = mm_registry.supports_multimodal_inputs( - vllm_config.model_config - ) + supports_mm_inputs = vllm_config.model_config.supports_multimodal_inputs mm_budget = ( MultiModalBudget(vllm_config, mm_registry) if supports_mm_inputs else None ) diff --git a/vllm/v1/engine/core.py b/vllm/v1/engine/core.py index a51278b761..9b27f50edc 100644 --- a/vllm/v1/engine/core.py +++ b/vllm/v1/engine/core.py @@ -31,8 +31,10 @@ from vllm.envs import enable_envs_cache from vllm.logger import init_logger from vllm.logging_utils.dump_input import dump_engine_exception from vllm.lora.request import LoRARequest -from vllm.multimodal import MULTIMODAL_REGISTRY -from vllm.multimodal.cache import MultiModalCacheMissError +from vllm.multimodal.cache import ( + MultiModalCacheMissError, + engine_receiver_cache_from_config, +) from vllm.tasks import POOLING_TASKS, SupportedTask from vllm.tracing import instrument, maybe_init_worker_tracer from vllm.transformers_utils.config import maybe_register_config_serialize_by_value @@ -182,10 +184,7 @@ class EngineCore: if self.scheduler.ec_connector is not None: # type: ignore self.model_executor.init_ec_output_aggregator() - mm_registry = MULTIMODAL_REGISTRY - self.mm_receiver_cache = mm_registry.engine_receiver_cache_from_config( - vllm_config - ) + self.mm_receiver_cache = engine_receiver_cache_from_config(vllm_config) # If a KV connector is initialized for scheduler, we want to collect # handshake metadata from all workers so the connector in the scheduler diff --git a/vllm/v1/engine/input_processor.py b/vllm/v1/engine/input_processor.py index 06b2b32efd..d0ccb1dacf 100644 --- a/vllm/v1/engine/input_processor.py +++ b/vllm/v1/engine/input_processor.py @@ -62,7 +62,7 @@ class InputProcessor: self.renderer = renderer or renderer_from_config(vllm_config) - self.supports_mm_inputs = mm_registry.supports_multimodal_inputs(model_config) + self.supports_mm_inputs = model_config.supports_multimodal_inputs self.mm_encoder_cache_size = 0 self.skip_prompt_length_check = False if self.supports_mm_inputs: diff --git a/vllm/v1/spec_decode/llm_base_proposer.py b/vllm/v1/spec_decode/llm_base_proposer.py index 9f7ad68a88..2affce2fdd 100644 --- a/vllm/v1/spec_decode/llm_base_proposer.py +++ b/vllm/v1/spec_decode/llm_base_proposer.py @@ -35,7 +35,6 @@ from vllm.model_executor.models.laguna_dflash import DFlashLagunaForCausalLM from vllm.model_executor.models.llama_eagle3 import Eagle3LlamaForCausalLM from vllm.model_executor.models.qwen3_dflash import DFlashQwen3ForCausalLM from vllm.model_executor.models.qwen3_eagle3 import Eagle3Qwen3ForCausalLM -from vllm.multimodal import MULTIMODAL_REGISTRY from vllm.platforms import current_platform from vllm.utils.torch_utils import PIN_MEMORY, async_tensor_h2d from vllm.v1.attention.backend import CommonAttentionMetadata @@ -165,10 +164,7 @@ class SpecDecodeBaseProposer: self.max_positions = self.max_num_tokens # Multi-modal data support - self.mm_registry = MULTIMODAL_REGISTRY - self.supports_mm_inputs = self.mm_registry.supports_multimodal_inputs( - vllm_config.model_config - ) + self.supports_mm_inputs = vllm_config.model_config.supports_multimodal_inputs self.draft_attn_groups: list[AttentionGroup] = [] self.kv_cache_gid: int = -1 diff --git a/vllm/v1/worker/gpu/model_runner.py b/vllm/v1/worker/gpu/model_runner.py index 1ba40c5fda..92a37008ce 100644 --- a/vllm/v1/worker/gpu/model_runner.py +++ b/vllm/v1/worker/gpu/model_runner.py @@ -257,9 +257,7 @@ class GPUModelRunner(LoRAModelRunnerMixin): # Multimodal self.mm_registry = MULTIMODAL_REGISTRY - self.supports_mm_inputs = self.mm_registry.supports_multimodal_inputs( - self.model_config - ) + self.supports_mm_inputs = self.model_config.supports_multimodal_inputs self.uses_inputs_embeds = ( self.supports_mm_inputs or self.model_config.enable_prompt_embeds ) diff --git a/vllm/v1/worker/gpu/spec_decode/speculator.py b/vllm/v1/worker/gpu/spec_decode/speculator.py index 534b6ef8eb..244fa8bc18 100644 --- a/vllm/v1/worker/gpu/spec_decode/speculator.py +++ b/vllm/v1/worker/gpu/spec_decode/speculator.py @@ -15,7 +15,6 @@ from vllm.distributed.eplb.eplb_state import EplbState from vllm.logger import init_logger from vllm.model_executor.layers.attention_layer_base import AttentionLayerBase from vllm.model_executor.models import supports_multimodal_embeddings -from vllm.multimodal import MULTIMODAL_REGISTRY from vllm.v1.kv_cache_interface import KVCacheConfig from vllm.v1.watermarking import create_watermarker from vllm.v1.watermarking.spec_decode import ( @@ -226,9 +225,7 @@ class DraftModelSpeculator(BaseSpeculator): ) self.draft_attn_layer_names = all_attn_layers - target_attn_layer_names - target_supports_mm = MULTIMODAL_REGISTRY.supports_multimodal_inputs( - self.vllm_config.model_config - ) + target_supports_mm = self.vllm_config.model_config.supports_multimodal_inputs draft_supports_mm = supports_multimodal_embeddings(self.model) self.supports_mm_inputs = target_supports_mm and draft_supports_mm if target_supports_mm and not draft_supports_mm: diff --git a/vllm/v1/worker/gpu_model_runner.py b/vllm/v1/worker/gpu_model_runner.py index bdabe398f6..00e33b47c6 100644 --- a/vllm/v1/worker/gpu_model_runner.py +++ b/vllm/v1/worker/gpu_model_runner.py @@ -550,9 +550,7 @@ class GPUModelRunner( self.mm_registry = MULTIMODAL_REGISTRY self.uses_mrope = model_config.uses_mrope self.mrope_num_dims = model_config.mrope_num_dims - self.supports_mm_inputs = self.mm_registry.supports_multimodal_inputs( - model_config - ) + self.supports_mm_inputs = model_config.supports_multimodal_inputs if self.model_config.is_encoder_decoder: # Maximum length of the encoder input, only for encoder-decoder diff --git a/vllm/v1/worker/worker_base.py b/vllm/v1/worker/worker_base.py index cf869edb26..b0839bc4ca 100644 --- a/vllm/v1/worker/worker_base.py +++ b/vllm/v1/worker/worker_base.py @@ -12,7 +12,7 @@ from vllm.config import VllmConfig, set_current_vllm_config from vllm.distributed.kv_transfer.kv_connector.utils import get_current_attn_backends from vllm.logger import init_logger from vllm.lora.request import LoRARequest -from vllm.multimodal import MULTIMODAL_REGISTRY +from vllm.multimodal.cache import worker_receiver_cache_from_config from vllm.tracing import instrument from vllm.utils.import_utils import resolve_obj_by_qualname from vllm.utils.system_utils import update_environment_variables @@ -315,26 +315,10 @@ class WorkerWrapperBase: ) shared_worker_lock = kwargs.pop("shared_worker_lock", None) - if shared_worker_lock is None: - msg = ( - "Missing `shared_worker_lock` argument from executor. " - "This argument is needed for mm_processor_cache_type='shm'." - ) - - mm_config = vllm_config.model_config.multimodal_config - if mm_config and mm_config.mm_processor_cache_type == "shm": - raise ValueError(msg) - else: - logger.warning_once(msg) - - self.mm_receiver_cache = None - else: - self.mm_receiver_cache = ( - MULTIMODAL_REGISTRY.worker_receiver_cache_from_config( - vllm_config, - shared_worker_lock, - ) - ) + self.mm_receiver_cache = worker_receiver_cache_from_config( + vllm_config, + shared_worker_lock, + ) with set_current_vllm_config(self.vllm_config): # To make vLLM config available during worker initialization