diff --git a/tests/test_config.py b/tests/test_config.py index 1fc00a8f8a9..8171950cdd0 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1558,20 +1558,31 @@ def test_fault_tolerance_requires_single_api_server(): def test_renderer_num_workers_with_mm_cache(): - """Disallow renderer_num_workers > 1 when mm processor cache is enabled, - since neither cache type is thread-safe.""" + """Disallow renderer_num_workers > 1 with the mm processor cache only for + pooling models, whose preprocessing runs on the renderer workers.""" mm_model = "Qwen/Qwen2-VL-2B-Instruct" - # Should raise: multi-worker + cache enabled (default cache_gb=4) + # Should raise: pooling + multi-worker + cache enabled (default cache_gb=4) with pytest.raises(ValueError, match="renderer-num-workers"): - ModelConfig(mm_model, renderer_num_workers=4) + ModelConfig(mm_model, runner="pooling", renderer_num_workers=4) - # Should raise: multi-worker + explicit cache size + # Should raise: pooling + multi-worker + explicit cache size with pytest.raises(ValueError, match="renderer-num-workers"): - ModelConfig(mm_model, renderer_num_workers=2, mm_processor_cache_gb=1.0) + ModelConfig( + mm_model, + runner="pooling", + renderer_num_workers=2, + mm_processor_cache_gb=1.0, + ) - # Should pass: multi-worker + cache disabled - config = ModelConfig(mm_model, renderer_num_workers=4, mm_processor_cache_gb=0) + # Should pass: pooling + multi-worker + cache disabled + config = ModelConfig( + mm_model, runner="pooling", renderer_num_workers=4, mm_processor_cache_gb=0 + ) + assert config.renderer_num_workers == 4 + + # Should pass: generate models preprocess on the dedicated mm executor + config = ModelConfig(mm_model, renderer_num_workers=4) assert config.renderer_num_workers == 4 # Should pass: single worker + cache enabled (default) diff --git a/vllm/config/model.py b/vllm/config/model.py index 20efeb03298..d64bd57e87d 100644 --- a/vllm/config/model.py +++ b/vllm/config/model.py @@ -748,12 +748,13 @@ class ModelConfig: if ( self.renderer_num_workers > 1 and self.multimodal_config.mm_processor_cache_gb > 0 + and self.runner_type == "pooling" ): raise ValueError( "Cannot use --renderer-num-workers > 1 with the " - "multimodal processor cache enabled. The cache is " - "not thread-safe and does not support concurrent " - "renderer workers. Please set " + "multimodal processor cache enabled for pooling models. " + "Pooling preprocessing runs on the renderer workers, and " + "the cache is not thread-safe. Please set " "--renderer-num-workers 1 (the default), or " "disable the cache with --mm-processor-cache-gb 0." ) diff --git a/vllm/renderers/base.py b/vllm/renderers/base.py index bb02c115cae..b278a8c13dc 100644 --- a/vllm/renderers/base.py +++ b/vllm/renderers/base.py @@ -79,16 +79,15 @@ class BaseRenderer(ABC, Generic[_T]): self.tokenizer = tokenizer - # Shared thread pool executor for blocking tokenizer and - # multimodal preprocessing operations. The multimodal processor - # receives a deep-copied tokenizer (see #36557) so it is safe to - # run tokenization and MM preprocessing concurrently. + # Thread pool executor for blocking tokenizer operations. The + # multimodal processor receives a deep-copied tokenizer (see #36557) + # so it is safe to run tokenization and MM preprocessing concurrently. pool_workers = config.model_config.renderer_num_workers self._executor = ThreadPoolExecutor(max_workers=pool_workers) - # Multimodal preprocessing is always offloaded to the thread pool - # to keep the asyncio event loop responsive under concurrent load. - self._mm_executor: Executor = self._executor + # Separate single-worker executor so tokenization never queues behind + # MM preprocessing; must stay single-worker per #38418 (P0/P1 order). + self._mm_executor: Executor = ThreadPoolExecutor(max_workers=1) # Offload tokenization to the thread pool. The sync # ``_tokenize_prompt`` already encapsulates the unified ``__call__`` @@ -103,7 +102,7 @@ class BaseRenderer(ABC, Generic[_T]): 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._executor + self.clear_mm_cache, executor=self._mm_executor ) self._process_multimodal_async = make_async( self._process_multimodal, executor=self._mm_executor @@ -282,7 +281,7 @@ class BaseRenderer(ABC, Generic[_T]): self._clear_processor_cache(self._readonly_mm_processor) async def clear_mm_cache_async(self) -> None: - """Serialize clear_mm_cache through the shared executor to avoid + """Serialize clear_mm_cache through the multimodal executor to avoid races with concurrent process_inputs on the mm_processor_cache.""" await self._clear_mm_cache_async()