minke.yu/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

[VLM] remove AsyncMMDataProcessor wrapper (#21651)

Browse Source
This commit is contained in:
Yuhao Yang
2026-04-01 17:39:50 +08:00
committed by GitHub
parent 80b1bc5f56
commit 1aabe44b64
5 changed files with 8 additions and 512 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/sglang/srt/managers/async_mm_data_processor.py
-122
View File
@@ -1,122 +0,0 @@
import asyncio
import logging
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from typing import Any, Dict, List, Optional, Union
logger = logging.getLogger(__name__)
class AsyncMMDataProcessor:
"""
Async wrapper for a multimodal processor.
Behavior:
- If the underlying processor exposes `process_mm_data_async`, call/await it directly.
- Otherwise, fall back to running a synchronous `process_mm_data` in a thread pool.
- Optionally guard per-call concurrency via an asyncio.Semaphore.
- Optionally enforce per-call timeout via asyncio.wait_for.
"""
def __init__(
self,
mm_processor: Any,
*,
max_concurrent_calls: Optional[int] = None,
timeout_s: Optional[float] = None,
) -> None:
"""
Args:
mm_processor: An object exposing either
- async def process_mm_data_async(...): -> Dict[str, Any]
or
- def process_mm_data(...): -> Dict[str, Any]
max_concurrent_calls: Optional concurrency cap for per-call execution.
timeout_s: Optional timeout (seconds) for each `process()` call.
"""
self.mm_processor = mm_processor
self.timeout_s = timeout_s
# Concurrency guard (None -> unlimited)
self.semaphore = (
asyncio.Semaphore(max_concurrent_calls) if max_concurrent_calls else None
)
# Detect async path; if missing, prepare a fallback executor for sync path
self._proc_async = getattr(mm_processor, "process_mm_data_async", None)
self.is_async = asyncio.iscoroutinefunction(self._proc_async)
self.fallback_exec: Optional[ThreadPoolExecutor] = (
ThreadPoolExecutor(max_workers=max_concurrent_calls)
if not self.is_async
else None
)
async def process(
self,
*,
image_data: Optional[List[Union[str, bytes]]] = None,
audio_data: Optional[List[Union[str, bytes]]] = None,
input_text_or_ids: Union[str, List[int], None] = None,
request_obj: Any,
**kwargs: Any,
) -> Dict[str, Any]:
"""
Public entrypoint: process a single multimodal request without blocking the event loop.
"""
async def _invoke() -> Dict[str, Any]:
if self.is_async:
# Native async implementation
return await self._proc_async(
image_data=image_data,
audio_data=audio_data,
input_text=input_text_or_ids,
request_obj=request_obj,
**kwargs,
)
# Synchronous fallback
sync_fn = getattr(self.mm_processor, "process_mm_data", None)
if not callable(sync_fn):
raise RuntimeError(
"mm_processor has neither 'process_mm_data_async' nor 'process_mm_data'."
)
loop = asyncio.get_running_loop()
fn = partial(
sync_fn,
image_data=image_data,
audio_data=audio_data,
input_text=input_text_or_ids,
request_obj=request_obj,
**kwargs,
)
return await loop.run_in_executor(self.fallback_exec, fn)
# Apply optional concurrency guard
if self.semaphore is not None:
async with self.semaphore:
if self.timeout_s is not None:
return await asyncio.wait_for(_invoke(), timeout=self.timeout_s)
return await _invoke()
# No concurrency guard
if self.timeout_s is not None:
return await asyncio.wait_for(_invoke(), timeout=self.timeout_s)
return await _invoke()
def shutdown(self) -> None:
"""Gracefully shutdown resources owned by this wrapper."""
try:
if self.fallback_exec:
self.fallback_exec.shutdown(wait=False)
except Exception:
logger.exception(
"Error while shutting down fallback executor in AsyncMMDataProcessor"
)
def __del__(self):
# Best-effort shutdown
try:
self.shutdown()
except Exception:
pass
python/sglang/srt/managers/tokenizer_manager.py
+4 -10
View File
@@ -44,7 +44,6 @@ from sglang.srt.disaggregation.utils import DisaggregationMode
from sglang.srt.environ import envs
from sglang.srt.lora.lora_registry import LoRARef, LoRARegistry
from sglang.srt.managers.async_dynamic_batch_tokenizer import AsyncDynamicbatchTokenizer
from sglang.srt.managers.async_mm_data_processor import AsyncMMDataProcessor
from sglang.srt.managers.disagg_service import start_disagg_service
from sglang.srt.managers.io_struct import (
AbortReq,
@@ -269,11 +268,6 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi
self.mm_processor = get_mm_processor(
self.model_config.hf_config, server_args, _processor, transport_mode
)
self.mm_data_processor = AsyncMMDataProcessor(
self.mm_processor,
max_concurrent_calls=self.server_args.mm_max_concurrent_calls,
timeout_s=self.server_args.mm_per_request_timeout,
)
if server_args.skip_tokenizer_init:
self.tokenizer = self.processor = None
@@ -729,10 +723,10 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi
need_wait_for_mm_inputs=obj.need_wait_for_mm_inputs,
)
if mm_inputs is None:
mm_inputs: Dict = await self.mm_data_processor.process(
mm_inputs: Dict = await self.mm_processor.process_mm_data_async(
image_data=obj.image_data,
audio_data=obj.audio_data,
input_text_or_ids=(input_text or input_ids),
input_text=(input_text or input_ids),
request_obj=obj,
max_req_input_len=self.max_req_input_len,
)
@@ -743,10 +737,10 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi
):
# In language_only mode with zmq_to_scheduler, if we didn't dispatch
# to encoder (e.g., only one image), process locally like non-language_only mode
mm_inputs: Dict = await self.mm_data_processor.process(
mm_inputs: Dict = await self.mm_processor.process_mm_data_async(
image_data=obj.image_data,
audio_data=obj.audio_data,
input_text_or_ids=(input_text or input_ids),
input_text=(input_text or input_ids),
request_obj=obj,
max_req_input_len=self.max_req_input_len,
)
python/sglang/srt/multimodal/processors/llava.py
+4 -1
View File
@@ -1,4 +1,5 @@
import asyncio
import os
from typing import Dict, List, Optional, Union
import numpy as np
@@ -96,7 +97,7 @@ class LlavaImageProcessor(BaseMultimodalProcessor):
):
if self.cpu_executor is not None:
loop = asyncio.get_running_loop()
return await loop.run_in_executor(
fut = loop.run_in_executor(
self.cpu_executor,
LlavaImageProcessor._process_single_image_task,
image_data,
@@ -104,6 +105,8 @@ class LlavaImageProcessor(BaseMultimodalProcessor):
grid_pinpoints,
self._processor,
)
timeout = int(os.environ.get("REQUEST_TIMEOUT", "10"))
return await asyncio.wait_for(fut, timeout=timeout)
else:
return self._process_single_image_task(
image_data,
python/sglang/srt/server_args.py
-14
View File
@@ -726,8 +726,6 @@ class ServerArgs:
sm_group_num: int = 8
# For Multi-Modal
mm_max_concurrent_calls: int = 32
mm_per_request_timeout: float = 10.0
enable_broadcast_mm_inputs_process: bool = False
enable_prefix_mm_cache: bool = False
mm_enable_dp_encoder: bool = False
@@ -5892,18 +5890,6 @@ class ServerArgs:
)
# For Multi-Modal
parser.add_argument(
"--mm-max-concurrent-calls",
type=int,
default=ServerArgs.mm_max_concurrent_calls,
help="The max concurrent calls for async mm data processing.",
)
parser.add_argument(
"--mm-per-request-timeout",
type=int,
default=ServerArgs.mm_per_request_timeout,
help="The timeout for each multi-modal request in seconds.",
)
parser.add_argument(
"--enable-broadcast-mm-inputs-process",
action="store_true",