[vlm] fix: recover multimodal decode and processor failures (#36983)

python/sglang/srt/multimodal/cache/identity.py

@@ -72,8 +72,11 @@ class MediaSnapshot:
 
 
 def _snapshot_pil(image: Image.Image) -> MediaSnapshot:
-    snapshot = image.copy()
-    snapshot.load()
+    try:
+        snapshot = image.copy()
+        snapshot.load()
+    except OSError as e:
+        raise ValueError(f"Could not decode image: {e}") from e
     payload = snapshot.tobytes()
     palette = snapshot.palette.tobytes() if snapshot.palette is not None else b""
     palette_mode = (

python/sglang/srt/multimodal/processors/base_processor.py

@@ -5,6 +5,7 @@ import dataclasses
 import multiprocessing as mp
 import os
 import re
+import threading
 from abc import ABC, abstractmethod
 from contextlib import contextmanager
 from typing import (
@@ -367,13 +368,8 @@ class BaseMultimodalProcessor(ABC):
                 self.mm_processor_worker_num,
                 "auto" if requested_mm_processor_worker_num == 0 else "explicit",
             )
-        cpu_worker_start_method = (
-            "spawn" if self.mm_feature_transport == "cuda_vmm" else "fork"
-        )
-        self.cpu_executor = concurrent.futures.ProcessPoolExecutor(
-            mp_context=mp.get_context(cpu_worker_start_method),
-            max_workers=int(os.environ.get("SGLANG_CPU_WORKERS", os.cpu_count())),
-        )
+        self._cpu_executor_lock = threading.Lock()
+        self.cpu_executor = self._create_cpu_executor()
 
         # Mapping from attribute names to modality types
         self.ATTR_NAME_TO_MODALITY = {
@@ -493,6 +489,41 @@ class BaseMultimodalProcessor(ABC):
         if self.mm_processor_executor is not None:
             self.mm_processor_executor.shutdown()
 
+    def _create_cpu_executor(self) -> concurrent.futures.ProcessPoolExecutor:
+        start_method = "spawn" if self.mm_feature_transport == "cuda_vmm" else "fork"
+        return concurrent.futures.ProcessPoolExecutor(
+            mp_context=mp.get_context(start_method),
+            max_workers=int(os.environ.get("SGLANG_CPU_WORKERS", os.cpu_count())),
+        )
+
+    def _replace_broken_cpu_executor(
+        self, failed_executor: concurrent.futures.ProcessPoolExecutor
+    ) -> None:
+        """Replace a failed preprocess pool once across concurrent requests."""
+        with self._cpu_executor_lock:
+            if self.cpu_executor is not failed_executor:
+                return
+            self.cpu_executor = self._create_cpu_executor()
+        logger.warning("Replaced a broken multimodal CPU preprocess pool")
+        threading.Thread(
+            target=self._shutdown_broken_cpu_executor,
+            args=(failed_executor,),
+            name="sglang-mm-cpu-pool-cleanup",
+            daemon=True,
+        ).start()
+
+    @staticmethod
+    def _shutdown_broken_cpu_executor(
+        failed_executor: concurrent.futures.ProcessPoolExecutor,
+    ) -> None:
+        try:
+            failed_executor.shutdown(wait=False, cancel_futures=True)
+        except Exception:
+            logger.warning(
+                "Failed to shut down a broken multimodal CPU preprocess pool",
+                exc_info=True,
+            )
+
     def compute_mrope_positions(self, input_ids, mm_items):
         """Compute M-RoPE positions from expanded input_ids and multimodal items.
 
@@ -863,11 +894,8 @@ class BaseMultimodalProcessor(ABC):
                 img, _ = load_image(data, cls.gpu_image_decode)
                 if isinstance(img, torch.Tensor):
                     return img  # JPEG already decoded on GPU by nvJPEG
-                # PIL decodes lazily; do it here in the io worker so the decode
-                # doesn't run later on the event-loop thread.
                 if discard_alpha_channel and img.mode != "RGB":
                     return img.convert("RGB")
-                img.load()
                 return img
             elif modality == Modality.VIDEO:
                 return load_video(data, frame_count_limit)

python/sglang/srt/multimodal/processors/llava.py

@@ -1,5 +1,6 @@
 import asyncio
 import os
+from concurrent.futures.process import BrokenProcessPool
 from typing import Dict, List, Optional, Union
 
 import numpy as np
@@ -28,7 +29,13 @@ from sglang.srt.multimodal.mm_utils import (
     process_anyres_image,
 )
 from sglang.srt.multimodal.processors.base_processor import BaseMultimodalProcessor
-from sglang.srt.utils import ImageData, get_image_bytes, load_image, logger
+from sglang.srt.utils import (
+    CLIENT_MEDIA_EXCEPTIONS,
+    ImageData,
+    get_image_bytes,
+    load_image,
+    logger,
+)
 from sglang.utils import get_exception_traceback
 
 
@@ -93,8 +100,11 @@ class LlavaImageProcessor(BaseMultimodalProcessor):
                     pixel_values = pixel_values.astype(np.float16)
 
                 return pixel_values, image_hash, image.size
+        except CLIENT_MEDIA_EXCEPTIONS as error:
+            raise ValueError(f"Error while processing image: {error}") from error
         except Exception:
             logger.error("Exception in TokenizerManager:\n" + get_exception_traceback())
+            raise
 
     async def _fetch_remote_image_bytes(self, url):
         # Fetch a remote image's compressed bytes in the io thread pool, retrying
@@ -137,17 +147,32 @@ class LlavaImageProcessor(BaseMultimodalProcessor):
 
         if self.cpu_executor is not None:
             loop = asyncio.get_running_loop()
-            fut = loop.run_in_executor(
-                self.cpu_executor,
-                LlavaImageProcessor._preprocess_image_task,
-                image_input,
-                image_hash,
-                aspect_ratio,
-                grid_pinpoints,
-                self._processor,
-            )
+            executor = self.cpu_executor
             timeout = int(os.environ.get("REQUEST_TIMEOUT", "10"))
-            return await asyncio.wait_for(fut, timeout=timeout)
+            deadline = loop.time() + timeout
+            try:
+                # ProcessPoolExecutor.submit() can itself block after a worker
+                # exits. Keep submission off the request event loop so the
+                # timeout can still replace the failed pool.
+                process_future = await asyncio.wait_for(
+                    asyncio.to_thread(
+                        executor.submit,
+                        LlavaImageProcessor._preprocess_image_task,
+                        image_input,
+                        image_hash,
+                        aspect_ratio,
+                        grid_pinpoints,
+                        self._processor,
+                    ),
+                    timeout=timeout,
+                )
+                remaining = max(0.0, deadline - loop.time())
+                return await asyncio.wait_for(
+                    asyncio.wrap_future(process_future), timeout=remaining
+                )
+            except (BrokenProcessPool, asyncio.TimeoutError):
+                self._replace_broken_cpu_executor(executor)
+                raise
         else:
             return LlavaImageProcessor._preprocess_image_task(
                 image_input,

python/sglang/srt/multimodal/processors/moss_vl.py

@@ -413,9 +413,16 @@ class MossVLImageProcessor(SGLangBaseProcessor):
     def _write_video_bytes_to_tempfile(
         self, video_bytes: bytes, suffix: str = ".mp4"
     ) -> str:
-        with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as f:
-            f.write(video_bytes)
-            return f.name
+        temp_path = None
+        try:
+            with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as f:
+                temp_path = f.name
+                f.write(video_bytes)
+            return temp_path
+        except BaseException:
+            if temp_path is not None:
+                self._remove_temp_video_paths([temp_path])
+            raise
 
     def _normalize_video_string(self, value: str) -> Tuple[str, Optional[str]]:
         if value.startswith("file://"):
@@ -428,9 +435,8 @@ class MossVLImageProcessor(SGLangBaseProcessor):
             timeout = int(os.getenv("REQUEST_TIMEOUT", "10"))
             content = download_remote_media(value, timeout=timeout)
             suffix = os.path.splitext(urlparse(value).path)[1] or ".mp4"
-            with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as f:
-                f.write(content)
-                return f.name, f.name
+            temp_path = self._write_video_bytes_to_tempfile(content, suffix=suffix)
+            return temp_path, temp_path
 
         if value.startswith("data:"):
             header, encoded = value.split(",", 1)
@@ -483,15 +489,46 @@ class MossVLImageProcessor(SGLangBaseProcessor):
             )
             for v in video_data
         ]
-        results = await asyncio.gather(*futures)
+        gather_task = asyncio.gather(*futures, return_exceptions=True)
+        cancelled_error = None
+        try:
+            results = await asyncio.shield(gather_task)
+        except asyncio.CancelledError as error:
+            cancelled_error = error
+            results = await gather_task
 
         normalized_inputs: List[Union[str, Dict]] = []
         temp_paths: List[str] = []
-        for normalized_input, created_paths in results:
+        errors = []
+        for result in results:
+            if isinstance(result, BaseException):
+                errors.append(result)
+                continue
+            normalized_input, created_paths = result
             normalized_inputs.append(normalized_input)
             temp_paths.extend(created_paths)
+
+        if cancelled_error is not None or errors:
+            self._remove_temp_video_paths(temp_paths)
+            if cancelled_error is not None:
+                raise cancelled_error
+            first_error = errors[0]
+            if len(errors) > 1:
+                first_error.add_note(
+                    f"{len(errors) - 1} additional video input(s) failed"
+                )
+            raise first_error
+
         return normalized_inputs, temp_paths
 
+    @staticmethod
+    def _remove_temp_video_paths(temp_paths: List[str]) -> None:
+        for temp_path in temp_paths:
+            try:
+                os.unlink(temp_path)
+            except FileNotFoundError:
+                pass
+
     async def process_mm_data_async(
         self,
         image_data: List[Union[str, bytes, Dict]],
@@ -569,8 +606,4 @@ class MossVLImageProcessor(SGLangBaseProcessor):
                 visible_frame_counts=visible_frame_counts,
             )
         finally:
-            for temp_path in temp_video_paths:
-                try:
-                    os.unlink(temp_path)
-                except FileNotFoundError:
-                    pass
+            self._remove_temp_video_paths(temp_video_paths)

python/sglang/srt/utils/common.py

@@ -1857,7 +1857,20 @@ def _load_image(
                     "Failed to decode JPEG on GPU, falling back to CPU. Error: %s",
                     e,
                 )
-    return Image.open(BytesIO(image_bytes))
+    try:
+        image = Image.open(BytesIO(image_bytes))
+    except OSError as e:
+        raise ValueError(f"Could not decode image: {e}") from e
+    return _fully_load_pil_image(image)
+
+
+def _fully_load_pil_image(image: Image.Image) -> Image.Image:
+    """Force PIL's lazy decode while malformed input is still request-local."""
+    try:
+        image.load()
+    except OSError as e:
+        raise ValueError(f"Could not decode image: {e}") from e
+    return image
 
 
 def load_image(
@@ -1874,7 +1887,7 @@ def load_image(
     image = None
     image_size: Optional[tuple[int, int]] = None
     if isinstance(image_file, Image.Image):
-        image = image_file
+        image = _fully_load_pil_image(image_file)
         image_size = (image.width, image.height)
     elif isinstance(image_file, bytes):
         image = _load_image(image_bytes=image_file, gpu_image_decode=gpu_image_decode)