[VLM] feat: size the multimodal preprocessing pool by where preprocessing runs (#35349)

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

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", {})

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():

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

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()