From c2c3320cf0a4f40e24ae1275e2a7af77e826b54e Mon Sep 17 00:00:00 2001 From: Mick Date: Thu, 27 Aug 2026 16:32:58 +0800 Subject: [PATCH] [VLM] feat: size the multimodal preprocessing pool by where preprocessing runs (#35349) Co-authored-by: Claude Opus 5 --- .../multimodal/processors/base_processor.py | 77 +++++++-- .../srt/multimodal/processors/ernie45_vl.py | 12 +- .../srt/multimodal/processors/executor.py | 19 ++- .../srt/multimodal/processors/midashenglm.py | 12 +- .../processors/sarashina2_vision.py | 65 +++++--- .../unit/managers/test_mm_process_config.py | 151 +++++++++++++++--- .../test_paddleocr_vl_serving_defaults.py | 19 ++- .../test_processor_async_call_sites.py | 108 +++++++++++++ .../test_processor_clone_isolation.py | 68 ++++++++ 9 files changed, 450 insertions(+), 81 deletions(-) create mode 100644 test/registered/unit/multimodal/test_processor_clone_isolation.py diff --git a/python/sglang/srt/multimodal/processors/base_processor.py b/python/sglang/srt/multimodal/processors/base_processor.py index ff3f60a8c..00283fede 100644 --- a/python/sglang/srt/multimodal/processors/base_processor.py +++ b/python/sglang/srt/multimodal/processors/base_processor.py @@ -191,6 +191,17 @@ class MultimodalSpecialTokens: return self.combined_regex +def _tokenizer_of(processor): + """The tokenizer reached from an HF processor. + + Some processors (e.g. InternVL) are handed a tokenizer directly as their + ``_processor`` rather than one that wraps a tokenizer. Every path that + resolves a tokenizer -- construction and per-worker processor clones alike -- + goes through here, so a clone cannot resolve differently from the original. + """ + return processor.tokenizer if hasattr(processor, "tokenizer") else processor + + class BaseMultimodalProcessor(ABC): models = [] gpu_image_decode = True # Enable GPU decoding by default @@ -199,12 +210,20 @@ class BaseMultimodalProcessor(ABC): # Set by processors that already build input_ids from the request's own # tokens, so the retokenize-avoidance rebuild below has nothing to add. preserve_processor_input_ids = False - auto_mm_processor_worker_num = 1 + # None lets the worker count follow where preprocessing actually runs; a + # model that measured its own optimum assigns a number instead. See + # `_resolve_auto_mm_processor_worker_num`. + auto_mm_processor_worker_num = None auto_mm_io_worker_num = 4 # Models opt in by assigning a non-zero default. A user-provided server # argument overrides this value; zero disables storage and cache-key work. auto_mm_preprocess_cache_size_mb = 0 - supports_mm_processor_concurrency = False + # Processors opt out only when their preprocessing is not thread-safe. The + # worker pool gives each thread its own `copy.deepcopy` of the HF processor + # and injects it, and the single function it runs -- + # `process_and_combine_mm_data` -- resolves that clone instead of + # `self._processor`, so isolation does not depend on the subclass. + supports_mm_processor_concurrency = True def __init__( self, hf_config, server_args, _processor, transport_mode, *args, **kwargs @@ -272,12 +291,7 @@ class BaseMultimodalProcessor(ABC): "trusted" if self.trust_mm_content_hashes else "verified", ) - # Resolve tokenizer: some processors (e.g. InternVL) pass a tokenizer - # directly as _processor rather than a processor that wraps a tokenizer. - if hasattr(self._processor, "tokenizer"): - self._tokenizer = self._processor.tokenizer - else: - self._tokenizer = self._processor + self._tokenizer = _tokenizer_of(self._processor) # Same guard as in serving_chat.py against double BOS. try: @@ -314,7 +328,8 @@ class BaseMultimodalProcessor(ABC): self.mm_processor_worker_num = ( 1 if skip_mm_pool - else requested_mm_processor_worker_num or self.auto_mm_processor_worker_num + else requested_mm_processor_worker_num + or self._resolve_auto_mm_processor_worker_num() ) if ( self.mm_processor_worker_num > 1 @@ -329,8 +344,11 @@ class BaseMultimodalProcessor(ABC): self.mm_processor_executor = None if self.mm_processor_worker_num > 1: try: + # A callable, not the object: subclasses finish customizing + # `_processor` after this returns, and the workers must clone it + # as the subclass left it. self.mm_processor_executor = MultimodalProcessorExecutor( - self._processor, self.mm_processor_worker_num + lambda: self._processor, self.mm_processor_worker_num ) except Exception: logger.warning( @@ -594,7 +612,44 @@ class BaseMultimodalProcessor(ABC): def _resolve_processor(self, processor=None): if processor is None: return self._processor, self._tokenizer - return processor, processor.tokenizer + return processor, _tokenizer_of(processor) + + def _preprocessing_competes_with_the_scheduler(self) -> bool: + """Whether image preprocessing submits its work to the serving GPU. + + The fast image processor runs inside the tokenizer process but on + ``cuda:{base_gpu_id}`` -- the device the scheduler serves from. A second + preprocessing worker there is one more competitor for that device rather + than added parallelism. + """ + if _is_cpu or self.server_args.rl_on_policy_target is not None: + return False + if self.disable_fast_image_processor: + return False + image_processor = getattr(self._processor, "image_processor", None) + return isinstance(image_processor, BaseImageProcessor) + + def _resolve_auto_mm_processor_worker_num(self) -> int: + """The worker count to use when the user did not ask for one. + + Two workers overlap preprocessing that runs on the CPU, where the second + thread is real parallelism: measured on Qwen2.5-VL with full-page images + at 32-way concurrency, 4.46 -> 6.08 req/s on H200 and 7.07 -> 8.76 on + GB300. + + The GPU path is capped at one worker even when a model declares more. + A declaration records what its author measured on one platform and one + image shape; contending for the device the scheduler is serving from is a + property of the path itself, and it does not go away because a subclass + asked for concurrency. Qwen-VL declares two and is the model that + measures 9.30 -> 4.02 req/s on GB300 full-page images, so honouring the + declaration here would exempt exactly the case that regresses. + `--mm-processor-worker-num` still overrides this. + """ + if self._preprocessing_competes_with_the_scheduler(): + return 1 + declared = self.auto_mm_processor_worker_num + return 2 if declared is None else declared def _fast_image_processor_device(self, processor) -> Optional[str]: """The device for the fast image processor, or None to leave it unset. diff --git a/python/sglang/srt/multimodal/processors/ernie45_vl.py b/python/sglang/srt/multimodal/processors/ernie45_vl.py index 86675c1fe..ca76deba7 100644 --- a/python/sglang/srt/multimodal/processors/ernie45_vl.py +++ b/python/sglang/srt/multimodal/processors/ernie45_vl.py @@ -282,7 +282,13 @@ class Ernie4_5_VLImageProcessor(SGLangBaseProcessor): return pixel_values def process_mm_data( - self, input_text, images=None, videos=None, audios=None, **kwargs + self, + input_text, + images=None, + videos=None, + audios=None, + processor=None, + **kwargs, ) -> dict: """ process multimodal data with transformers AutoProcessor @@ -296,7 +302,9 @@ class Ernie4_5_VLImageProcessor(SGLangBaseProcessor): if self.video_config: kwargs.setdefault("videos_kwargs", {}).update(self.video_config) - processor = self._processor + # Take the worker pool's per-thread clone when it hands one over; falling + # back to self._processor would put every worker on one shared object. + processor, _ = self._resolve_processor(processor) if ( hasattr(processor, "image_processor") and isinstance(processor.image_processor, BaseImageProcessor) diff --git a/python/sglang/srt/multimodal/processors/executor.py b/python/sglang/srt/multimodal/processors/executor.py index 43da2ea8a..a94f7dbb6 100644 --- a/python/sglang/srt/multimodal/processors/executor.py +++ b/python/sglang/srt/multimodal/processors/executor.py @@ -15,9 +15,14 @@ class _WorkerState(threading.local): class MultimodalProcessorExecutor: """Run processor calls on isolated, thread-local processor clones.""" - def __init__(self, processor: Any, max_workers: int): - self._processor = processor - self._processor_clones = [copy.deepcopy(processor) for _ in range(max_workers)] + def __init__(self, resolve_processor: Callable[[], Any], max_workers: int): + # Resolved per clone rather than captured here: a subclass keeps + # customizing `_processor` after `super().__init__()` has already built + # this pool, and a clone taken now would miss every one of those edits. + self._resolve_processor = resolve_processor + # Probe once, so a processor that cannot be cloned still falls back to + # synchronous processing at startup instead of failing inside a worker. + copy.deepcopy(resolve_processor()) self._executor = concurrent.futures.ThreadPoolExecutor( max_workers=max_workers, thread_name_prefix="sglang-mm-processor", @@ -39,12 +44,10 @@ class MultimodalProcessorExecutor: ) -> T: processor = self._worker_state.processor if processor is None: + # One clone at a time: cloning reads the shared processor, which the + # worker path also reads for the token-count helpers. with self._clone_lock: - processor = ( - self._processor_clones.pop() - if self._processor_clones - else copy.deepcopy(self._processor) - ) + processor = copy.deepcopy(self._resolve_processor()) self._worker_state.processor = processor return function(*args, processor=processor, **kwargs) diff --git a/python/sglang/srt/multimodal/processors/midashenglm.py b/python/sglang/srt/multimodal/processors/midashenglm.py index 858573b89..0a8f61c60 100644 --- a/python/sglang/srt/multimodal/processors/midashenglm.py +++ b/python/sglang/srt/multimodal/processors/midashenglm.py @@ -48,7 +48,13 @@ class MiDashengLMMultimodalProcessor(BaseMultimodalProcessor): self.FEATURE_NAMES.append("input_values") def process_mm_data( - self, input_text, images=None, videos=None, audios=None, **kwargs + self, + input_text, + images=None, + videos=None, + audios=None, + processor=None, + **kwargs, ): """Override to use correct audio parameter name for MiDashengLM processor.""" if images: @@ -62,7 +68,9 @@ class MiDashengLMMultimodalProcessor(BaseMultimodalProcessor): if self.audio_config: kwargs["audio_kwargs"].update(self.audio_config) - processor = self._processor + # Take the worker pool's per-thread clone when it hands one over; falling + # back to self._processor would put every worker on one shared object. + processor, _ = self._resolve_processor(processor) result = processor.__call__( text=[input_text], padding=True, diff --git a/python/sglang/srt/multimodal/processors/sarashina2_vision.py b/python/sglang/srt/multimodal/processors/sarashina2_vision.py index b8b284ecd..e0ec52d9c 100644 --- a/python/sglang/srt/multimodal/processors/sarashina2_vision.py +++ b/python/sglang/srt/multimodal/processors/sarashina2_vision.py @@ -1,3 +1,4 @@ +import types from typing import List, Union from sglang.srt.managers.schedule_batch import MultimodalProcessorOutput @@ -7,6 +8,43 @@ from sglang.srt.multimodal.processors.base_processor import ( MultimodalSpecialTokens, ) +# Sarashina2Vision's remote-code `_preprocess` takes a narrow kwarg set, while +# transformers' `preprocess` forwards its full one, so the extras have to be +# dropped. +_PREPROCESS_PARAMS = frozenset( + { + "do_resize", + "resample", + "do_rescale", + "rescale_factor", + "do_normalize", + "image_mean", + "image_std", + "do_convert_rgb", + "data_format", + "input_data_format", + } +) + + +def _install_preprocess_kwarg_filter(image_processor) -> None: + """Drop the kwargs Sarashina2Vision's `_preprocess` cannot accept. + + Bound with `types.MethodType` rather than closing over `image_processor`, so + that a preprocessing worker's `copy.deepcopy` rebinds `__self__` to its own + clone instead of routing every thread back into this one. + """ + unfiltered_preprocess = type(image_processor)._preprocess + + def _preprocess(self, *args, **kwargs): + return unfiltered_preprocess( + self, + *args, + **{k: v for k, v in kwargs.items() if k in _PREPROCESS_PARAMS}, + ) + + image_processor._preprocess = types.MethodType(_preprocess, image_processor) + class Sarashina2VisionProcessor(BaseMultimodalProcessor): models = [Sarashina2VisionForCausalLM] @@ -25,33 +63,10 @@ class Sarashina2VisionProcessor(BaseMultimodalProcessor): image_token_id=self.IM_TOKEN_ID, ).build(_processor) - # Patch the processor's image processor to handle parameter compatibility if hasattr(_processor, "image_processor") and hasattr( - _processor.image_processor, "_preprocess" + type(_processor.image_processor), "_preprocess" ): - original_preprocess = _processor.image_processor._preprocess - - def patched_preprocess(*args, **kwargs): - # Filter kwargs to only include parameters that the custom _preprocess method accepts - # Based on Sarashina2VisionImageProcessor._preprocess signature - allowed_params = { - "do_resize", - "resample", - "do_rescale", - "rescale_factor", - "do_normalize", - "image_mean", - "image_std", - "do_convert_rgb", - "data_format", - "input_data_format", - } - filtered_kwargs = { - k: v for k, v in kwargs.items() if k in allowed_params - } - return original_preprocess(*args, **filtered_kwargs) - - _processor.image_processor._preprocess = patched_preprocess + _install_preprocess_kwarg_filter(_processor.image_processor) async def process_mm_data_async( self, diff --git a/test/registered/unit/managers/test_mm_process_config.py b/test/registered/unit/managers/test_mm_process_config.py index e22ac834c..2cdc38a55 100644 --- a/test/registered/unit/managers/test_mm_process_config.py +++ b/test/registered/unit/managers/test_mm_process_config.py @@ -79,6 +79,7 @@ class TestBaseProcessorConfigExtraction(CustomTestCase): mm_process_config, mm_processor_worker_num=0, mm_io_worker_num=0, + image_processor=None, ): """Create a BaseMultimodalProcessor via the real __init__ with mocked deps.""" from sglang.srt.multimodal.processors.base_processor import ( @@ -99,9 +100,21 @@ class TestBaseProcessorConfigExtraction(CustomTestCase): self.addCleanup(override.restore) server_args = MagicMock() + server_args.mm_processor_worker_num = mm_processor_worker_num + server_args.mm_io_worker_num = mm_io_worker_num + server_args.mm_preprocess_cache_size_mb = None + server_args.tokenizer_worker_num = 1 + server_args.trust_mm_content_hashes = False + server_args.media_url_max_file_size_mb = 64 + # A bare MagicMock makes every attribute truthy, which silently sends + # the worker-count decision down the CPU branch. Pin what it reads. + server_args.disable_fast_image_processor = False + server_args.rl_on_policy_target = None hf_config = MagicMock() mock_hf_processor = MagicMock() + if image_processor is not None: + mock_hf_processor.image_processor = image_processor # Call real __init__ so we test actual config extraction with patch.object(BaseMultimodalProcessor, "__abstractmethods__", set()): @@ -111,6 +124,8 @@ class TestBaseProcessorConfigExtraction(CustomTestCase): _processor=mock_hf_processor, transport_mode=None, ) + if proc.mm_processor_executor is not None: + self.addCleanup(proc.mm_processor_executor.shutdown) return proc def test_configs_extracted(self): @@ -165,10 +180,86 @@ class TestBaseProcessorConfigExtraction(CustomTestCase): self.assertIsNone(proc.mm_processor_executor) def test_parallel_workers_require_processor_support(self): - proc = self._make_processor({}, mm_processor_worker_num=2) + from sglang.srt.multimodal.processors.base_processor import ( + BaseMultimodalProcessor, + ) + + with patch.object( + BaseMultimodalProcessor, "supports_mm_processor_concurrency", False + ): + proc = self._make_processor({}, mm_processor_worker_num=2) self.assertEqual(proc.mm_processor_worker_num, 1) self.assertIsNone(proc.mm_processor_executor) + def test_cpu_preprocessing_path_gets_two_workers(self): + """A processor whose preprocessing stays on the CPU: the second worker is + real parallelism there (H200 4.46 -> 6.08 req/s, GB300 7.07 -> 8.76).""" + proc = self._make_processor({}) + self.assertEqual(proc.mm_processor_worker_num, 2) + self.assertIsNotNone(proc.mm_processor_executor) + + def test_gpu_preprocessing_path_stays_at_one_worker(self): + """A fast image processor submits to the device the scheduler serves + from, so a second worker there only contends for it: flat on H200 and + 9.30 -> 4.02 req/s on GB300 for full-page images.""" + from transformers import BaseImageProcessor + + proc = self._make_processor( + {}, image_processor=MagicMock(spec=BaseImageProcessor) + ) + self.assertEqual(proc.mm_processor_worker_num, 1) + self.assertIsNone(proc.mm_processor_executor) + + def test_explicit_request_overrides_the_path_decision(self): + """The server argument wins: an operator who measured their own workload + can still ask for concurrency on the GPU path.""" + from transformers import BaseImageProcessor + + proc = self._make_processor( + {}, + mm_processor_worker_num=2, + image_processor=MagicMock(spec=BaseImageProcessor), + ) + self.assertEqual(proc.mm_processor_worker_num, 2) + self.assertIsNotNone(proc.mm_processor_executor) + + def test_gpu_path_caps_a_count_the_model_declared(self): + """Contending for the scheduler's device is a property of the path, so a + subclass asking for concurrency does not exempt it. Qwen-VL declares two + and is the model that measures 9.30 -> 4.02 req/s on GB300 full-page + images.""" + from transformers import BaseImageProcessor + + from sglang.srt.multimodal.processors.base_processor import ( + BaseMultimodalProcessor, + ) + + with patch.object(BaseMultimodalProcessor, "auto_mm_processor_worker_num", 3): + proc = self._make_processor( + {}, image_processor=MagicMock(spec=BaseImageProcessor) + ) + self.assertEqual(proc.mm_processor_worker_num, 1) + + def test_cpu_path_honours_a_count_the_model_declared(self): + """On the CPU path the extra threads are real parallelism, so a model's + own measured count stands.""" + from sglang.srt.multimodal.processors.base_processor import ( + BaseMultimodalProcessor, + ) + + with patch.object(BaseMultimodalProcessor, "auto_mm_processor_worker_num", 3): + proc = self._make_processor({}) + self.assertEqual(proc.mm_processor_worker_num, 3) + + def test_clone_resolves_tokenizer_like_init(self): + proc = self._make_processor({}) + + wrapping = MagicMock() + self.assertIs(proc._resolve_processor(wrapping)[1], wrapping.tokenizer) + + bare = MagicMock(spec=["encode"]) + self.assertIs(proc._resolve_processor(bare)[1], bare) + def test_explicit_io_worker_count_overrides_auto(self): from sglang.srt.multimodal.processors.base_processor import ( BaseMultimodalProcessor, @@ -452,8 +543,9 @@ class TestMultimodalProcessorConcurrency(unittest.IsolatedAsyncioTestCase): ): processor = BaseMultimodalProcessor() + hf_processor = SimpleNamespace(tokenizer=object()) processor.mm_processor_executor = MultimodalProcessorExecutor( - SimpleNamespace(tokenizer=object()), max_workers=2 + lambda: hf_processor, max_workers=2 ) processor.process_and_combine_mm_data = MagicMock( side_effect=lambda *_args, **_kwargs: threading.current_thread().name @@ -497,7 +589,8 @@ class TestMultimodalProcessorConcurrency(unittest.IsolatedAsyncioTestCase): MultimodalProcessorExecutor, ) - executor = MultimodalProcessorExecutor(object(), max_workers=2) + hf_processor = SimpleNamespace(tokenizer=object()) + executor = MultimodalProcessorExecutor(lambda: hf_processor, max_workers=2) return_processor = lambda *, processor: processor try: first = await executor.run(return_processor) @@ -507,30 +600,37 @@ class TestMultimodalProcessorConcurrency(unittest.IsolatedAsyncioTestCase): self.assertIs(first, second) - async def test_replacement_worker_lazily_clones_processor(self): - from sglang.srt.multimodal.processors import executor as executor_module + async def test_clone_carries_customization_applied_after_construction(self): + """A subclass keeps customizing `_processor` after `super().__init__()`. - source_processor = object() - replacement_clone = SimpleNamespace(tokenizer=object()) - with patch.object( - executor_module.copy, - "deepcopy", - return_value=replacement_clone, - ) as deepcopy: - executor = executor_module.MultimodalProcessorExecutor( - source_processor, max_workers=2 - ) - executor._processor_clones.clear() - return_processor = lambda *, processor: processor - try: - first = await executor.run(return_processor) - second = await executor.run(return_processor) - finally: - executor.shutdown() + Sarashina2Vision patches its image processor there and Pixtral sets + `patch_size` / `spatial_merge_size`; a clone snapshotted while the pool + was built would serve requests from a half-configured processor. + """ + from sglang.srt.multimodal.processors.executor import ( + MultimodalProcessorExecutor, + ) - self.assertIs(first, replacement_clone) - self.assertIs(first, second) - self.assertEqual(deepcopy.call_count, 3) + owner = SimpleNamespace(_processor=SimpleNamespace(patch_size=16)) + executor = MultimodalProcessorExecutor(lambda: owner._processor, max_workers=2) + self.addCleanup(executor.shutdown) + + owner._processor.patch_size = 14 + + seen = await executor.run(lambda *, processor: processor.patch_size) + self.assertEqual(seen, 14) + + async def test_worker_gets_a_clone_not_the_shared_processor(self): + from sglang.srt.multimodal.processors.executor import ( + MultimodalProcessorExecutor, + ) + + hf_processor = SimpleNamespace(tokenizer=object()) + executor = MultimodalProcessorExecutor(lambda: hf_processor, max_workers=2) + self.addCleanup(executor.shutdown) + + worker_processor = await executor.run(lambda *, processor: processor) + self.assertIsNot(worker_processor, hf_processor) class TestProcessMmDataKwargs(CustomTestCase): @@ -698,6 +798,7 @@ class TestOverrideProcessorsConfigInjection(CustomTestCase): proc.disable_fast_image_processor = server_args.disable_fast_image_processor proc.skip_tokenizer_init = server_args.skip_tokenizer_init proc._processor = mock_hf_processor + proc._tokenizer = mock_hf_processor.tokenizer proc.image_config = mm_process_config.get("image", {}) proc.video_config = mm_process_config.get("video", {}) proc.audio_config = mm_process_config.get("audio", {}) diff --git a/test/registered/unit/models/test_paddleocr_vl_serving_defaults.py b/test/registered/unit/models/test_paddleocr_vl_serving_defaults.py index 3d4c5b1ec..cccc6f3de 100644 --- a/test/registered/unit/models/test_paddleocr_vl_serving_defaults.py +++ b/test/registered/unit/models/test_paddleocr_vl_serving_defaults.py @@ -28,7 +28,7 @@ from sglang.test.ci.ci_register import register_cpu_ci register_cpu_ci(est_time=2, suite="base-a-test-cpu") -def test_processor_opts_into_concurrency(): +def test_processor_preprocesses_pages_concurrently(): assert PaddleOCRVLImageProcessor.supports_mm_processor_concurrency is True assert PaddleOCRVLImageProcessor.auto_mm_processor_worker_num > 1 assert PaddleOCRVLImageProcessor.auto_mm_io_worker_num > 1 @@ -42,13 +42,16 @@ def test_worker_count_stays_at_the_measured_optimum(): assert PaddleOCRVLImageProcessor.auto_mm_processor_worker_num == 2 -def test_concurrency_opt_in_is_not_inherited_by_accident(): - """The base class must stay conservative; this model opts in explicitly.""" - assert BaseMultimodalProcessor.supports_mm_processor_concurrency is False - assert BaseMultimodalProcessor.auto_mm_processor_worker_num == 1 - assert ( - PaddleOCRVLImageProcessor.__dict__["supports_mm_processor_concurrency"] is True - ), "the opt-in must be declared on PaddleOCRVLImageProcessor itself" +def test_io_worker_count_is_this_model_own(): + """Concurrency is the base default now, but the IO fan-out is not. + + Fetching a page is network-bound and cheap to overlap, so this model asks for + more IO workers than the conservative base default. That number has to be + declared here, not inherited. + """ + assert PaddleOCRVLImageProcessor.__dict__["auto_mm_io_worker_num"] > ( + BaseMultimodalProcessor.auto_mm_io_worker_num + ), "the IO fan-out must be declared on PaddleOCRVLImageProcessor itself" def test_prefill_breakable_cuda_graph_is_allowlisted(): diff --git a/test/registered/unit/multimodal/test_processor_async_call_sites.py b/test/registered/unit/multimodal/test_processor_async_call_sites.py index f9fabf99c..31b84d81e 100644 --- a/test/registered/unit/multimodal/test_processor_async_call_sites.py +++ b/test/registered/unit/multimodal/test_processor_async_call_sites.py @@ -88,6 +88,114 @@ def test_every_call_site_can_await(): ) +def test_default_worker_count_follows_the_preprocessing_path(): + """The count is resolved per path, not pinned to a number. + + Two workers overlap preprocessing that runs on the CPU, where the second + thread is real parallelism: 4.46 -> 6.08 req/s on H200 and 7.07 -> 8.76 on + GB300, full-page images at 32-way concurrency. On the GPU path the same + second worker only contends for the device the scheduler serves from -- + flat on H200, and 9.30 -> 4.02 req/s on GB300. + + Measuring one path gives the opposite answer from the other, so pinning a + single default here is what this asserts against. + """ + from sglang.srt.multimodal.processors.base_processor import ( + BaseMultimodalProcessor, + ) + + assert BaseMultimodalProcessor.supports_mm_processor_concurrency is True + assert BaseMultimodalProcessor.auto_mm_processor_worker_num is None + + +def _process_mm_data_overrides(): + """Yield (path, node) for every subclass override of `process_mm_data`.""" + for path in sorted(_MULTIMODAL_ROOT.rglob("*.py")): + if path.name in _EXEMPT: + continue + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.FunctionDef) and node.name == "process_mm_data": + yield path, node + + +def test_overrides_take_the_worker_pools_processor_clone(): + """An override that reaches for `self._processor` puts every worker thread on + one shared HF processor, which is exactly what the per-thread clone exists to + prevent. Either accept `processor=` and resolve it, or delegate to super. + """ + offenders = [] + for path, node in _process_mm_data_overrides(): + args = [a.arg for a in node.args.args] + [a.arg for a in node.args.kwonlyargs] + body = ast.dump(node) + reaches_for_shared = "attr='_processor'" in body + resolves_injected = "_resolve_processor" in body + if reaches_for_shared and not resolves_injected: + offenders.append(f"{path.relative_to(_MULTIMODAL_ROOT)}:{node.lineno}") + elif "processor" not in args and not ( + "'super'" in body or not reaches_for_shared + ): + offenders.append(f"{path.relative_to(_MULTIMODAL_ROOT)}:{node.lineno}") + assert not offenders, ( + "these `process_mm_data` overrides bypass the worker pool's processor " + "clone; accept `processor=None` and resolve it with " + "`self._resolve_processor(processor)`: " + ", ".join(offenders) + ) + + +# Processors that build their whole preprocessing chain themselves and never +# reach `process_and_combine_mm_data`, so the worker pool cannot help them. They +# are not broken by concurrency either -- they simply do not participate. Listed +# explicitly so that adding a processor forces a decision instead of silently +# leaving it at one-worker speed. +_NO_WORKER_POOL_ROUTE = { + "dots_note_omni.py", + "inkling.py", + "lightonocr.py", + "llava.py", + "mimo_v2.py", + "mimo_v2_asr.py", + "minicpmv4_6.py", + "moss_vl.py", + "nano_nemotron_vl.py", + "voxtral.py", + "whisper.py", +} + + +def test_processors_outside_the_worker_pool_are_declared(): + """A new processor must either route through the pool or be listed here. + + Without this, a processor added on the old call site keeps preprocessing on + the event loop and nobody notices: there is no error, just one-worker + throughput. Whichever way the list moves, the change should be deliberate. + """ + unrouted = set() + for path in sorted(_MULTIMODAL_ROOT.rglob("*.py")): + if path.name in _EXEMPT: + continue + source = path.read_text(encoding="utf-8") + entry_points = ( + "async def process_mm_data_async" in source + or "async def _process_special_format" in source + ) + if entry_points and "process_and_combine_mm_data_async" not in source: + unrouted.add(path.name) + + newly_unrouted = unrouted - _NO_WORKER_POOL_ROUTE + assert not newly_unrouted, ( + "these processors reach preprocessing without going through the worker " + "pool, so they will serve at one-worker speed; either route them through " + "`process_and_combine_mm_data_async` or add them to " + f"_NO_WORKER_POOL_ROUTE with a reason: {sorted(newly_unrouted)}" + ) + now_routed = _NO_WORKER_POOL_ROUTE - unrouted + assert not now_routed, ( + "these processors now reach the worker pool, so drop them from " + f"_NO_WORKER_POOL_ROUTE: {sorted(now_routed)}" + ) + + def test_the_scan_actually_finds_call_sites(): """Guard against the scan silently matching nothing after a rename.""" assert len(list(_call_sites())) > 20 diff --git a/test/registered/unit/multimodal/test_processor_clone_isolation.py b/test/registered/unit/multimodal/test_processor_clone_isolation.py new file mode 100644 index 000000000..6e1d2a8a0 --- /dev/null +++ b/test/registered/unit/multimodal/test_processor_clone_isolation.py @@ -0,0 +1,68 @@ +"""A preprocessing worker's processor clone must not alias the original. + +The worker pool hands each thread its own ``copy.deepcopy`` of the HF processor. +That isolation is only real if a processor's customizations both survive the +copy and rebind to the clone. An instance-level patch that closes over the +original object copies as itself, so every worker thread calls back into the one +shared object -- exactly what the per-thread clone exists to prevent -- and a +patch installed after ``super().__init__()`` is missing from the clone entirely. + +Sarashina2Vision is the processor that patches its image processor this way. +""" + +import copy +import unittest + +from sglang.srt.multimodal.processors.sarashina2_vision import ( + _install_preprocess_kwarg_filter, +) +from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase + +register_cpu_ci(est_time=2, suite="base-a-test-cpu") + + +class _NarrowImageProcessor: + """Stands in for Sarashina2Vision's remote-code image processor.""" + + def __init__(self): + self.owner = "original" + + def _preprocess(self, images, do_resize=None, do_rescale=None): + return self.owner + + +class TestSarashina2PreprocessFilterSurvivesCloning(CustomTestCase): + def test_unfiltered_preprocess_rejects_what_transformers_forwards(self): + """Why the filter exists: the raw method cannot take the full kwarg set.""" + with self.assertRaises(TypeError): + _NarrowImageProcessor()._preprocess(["img"], do_resize=True, do_pad=False) + + def test_filter_applies_to_the_patched_processor(self): + image_processor = _NarrowImageProcessor() + _install_preprocess_kwarg_filter(image_processor) + + self.assertEqual( + image_processor._preprocess(["img"], do_resize=True, do_pad=False), + "original", + ) + + def test_clone_runs_the_filter_against_itself(self): + image_processor = _NarrowImageProcessor() + _install_preprocess_kwarg_filter(image_processor) + + clone = copy.deepcopy(image_processor) + clone.owner = "clone" + + self.assertIs(clone._preprocess.__self__, clone) + self.assertEqual( + clone._preprocess(["img"], do_resize=True, do_pad=False), "clone" + ) + self.assertEqual( + image_processor._preprocess(["img"], do_resize=True, do_pad=False), + "original", + ) + + +if __name__ == "__main__": + unittest.main()