diff --git a/python/sglang/srt/arg_groups/fields/memory.py b/python/sglang/srt/arg_groups/fields/memory.py index 81e809967..c49ce09d3 100644 --- a/python/sglang/srt/arg_groups/fields/memory.py +++ b/python/sglang/srt/arg_groups/fields/memory.py @@ -41,8 +41,12 @@ class Memory(msgspec.Struct): "for what each policy optimizes for." ), choices=RADIX_EVICTION_POLICY_CHOICES, + resolvable=True, ), ] = "lru" + # The value alone cannot distinguish the default from an explicit LRU + # choice, which model-specific defaults must preserve. + _radix_eviction_policy_explicitly_set: A[bool, Arg(no_cli=True)] = False radix_eviction_policy_config: A[ Optional[Dict[str, Any]], Arg( diff --git a/python/sglang/srt/arg_groups/fields/mm.py b/python/sglang/srt/arg_groups/fields/mm.py index 7e6a98ba7..5313c09e0 100644 --- a/python/sglang/srt/arg_groups/fields/mm.py +++ b/python/sglang/srt/arg_groups/fields/mm.py @@ -99,10 +99,15 @@ class Mm(msgspec.Struct): ] = 64 mm_preprocess_cache_size_mb: A[ Optional[int], - "CPU memory budget for content-addressed multimodal preprocessing " - "artifacts. Unset selects a model-specific default (256 MiB for " - "Kimi-K3); 0 disables the cache. The budget is divided across " - "tokenizer workers and does not reserve GPU memory.", + Arg( + help=( + "CPU memory budget for content-addressed multimodal preprocessing " + "artifacts. Unset selects a model-specific default (256 MiB for " + "Kimi-K3); 0 disables the cache. The budget is divided across " + "tokenizer workers and does not reserve GPU memory." + ), + resolvable=True, + ), ] = None trust_mm_content_hashes: A[ bool, @@ -139,13 +144,18 @@ class Mm(msgspec.Struct): ] = False mm_feature_transport: A[ Optional[Literal["cpu", "cuda_ipc", "cuda_vmm"]], - "Transport multimodal features through CPU memory, a bounded CUDA IPC " - "pool, or a bounded CUDA VMM pool. " - "Unset uses cpu except for validated multi-node GB200/GB300 MNNVL models, " - "which use cuda_vmm when an IMEX channel is available. Select cuda_ipc " - "explicitly for single-node GPU transport. GPU transports reserve " - "SGLANG_MM_FEATURE_CACHE_MB (default 1024 MiB) on the base GPU and fall " - "back to CPU transport when the pool is full.", + Arg( + help=( + "Transport multimodal features through CPU memory, a bounded CUDA IPC " + "pool, or a bounded CUDA VMM pool. " + "Unset uses cpu except for validated multi-node GB200/GB300 MNNVL models, " + "which use cuda_vmm when an IMEX channel is available. Select cuda_ipc " + "explicitly for single-node GPU transport. GPU transports reserve " + "SGLANG_MM_FEATURE_CACHE_MB (default 1024 MiB) on the base GPU and fall " + "back to CPU transport when the pool is full." + ), + resolvable=True, + ), ] = None keep_mm_feature_on_device: A[ bool, diff --git a/python/sglang/srt/arg_groups/fields/schedule.py b/python/sglang/srt/arg_groups/fields/schedule.py index d067290a0..de4267d95 100644 --- a/python/sglang/srt/arg_groups/fields/schedule.py +++ b/python/sglang/srt/arg_groups/fields/schedule.py @@ -61,9 +61,12 @@ class Schedule(msgspec.Struct): "The maximum number of tokens in a chunk for the chunked prefill. Setting this to -1 means disabling chunked prefill.", ] = None prefill_decode_interval: A[ - int, - "The number of decode rounds to run after a prefill batch before scheduling the next prefill. In data-parallel attention mode, the interval is synchronized across all DP ranks. Set to 0 to disable.", - ] = 0 + Optional[int], + Arg( + help="The number of decode rounds to run after a prefill batch before scheduling the next prefill. By default, this is disabled except for profiled Qwen3-VL serving configurations on Hopper. In data-parallel attention mode, the interval is synchronized across all DP ranks. Set to 0 to disable.", + resolvable=True, + ), + ] = None enable_dynamic_chunking: A[ bool, "Enable dynamic chunk size adjustment for pipeline parallelism. When enabled, chunk sizes are dynamically calculated based on fitted function to maintain consistent execution time across chunks.", diff --git a/python/sglang/srt/arg_groups/memory_hook.py b/python/sglang/srt/arg_groups/memory_hook.py index 267738f41..ff326bbe4 100644 --- a/python/sglang/srt/arg_groups/memory_hook.py +++ b/python/sglang/srt/arg_groups/memory_hook.py @@ -170,10 +170,17 @@ def handle_gpu_memory_settings(server_args: Any, gpu_mem): if decode_cuda_graph_config.max_bs is None: decode_cuda_graph_config.max_bs = 160 + from sglang.srt.arg_groups.model_overrides.qwen3_vl import ( + expand_multimodal_decode_graph_to_running_limit, + ) + + expand_multimodal_decode_graph_to_running_limit( + server_args, decode_cuda_graph_config, gpu_mem + ) + # ------------------------------------------------------------------ # CUDA graph batch-size materialization # ------------------------------------------------------------------ - if cfg.device != "cpu": if decode_cuda_graph_config.bs is None: decode_cuda_graph_config.bs = generate_decode_cuda_graph_batch_sizes( diff --git a/python/sglang/srt/arg_groups/model_overrides/qwen3_vl.py b/python/sglang/srt/arg_groups/model_overrides/qwen3_vl.py index 0dcbede98..406048a39 100644 --- a/python/sglang/srt/arg_groups/model_overrides/qwen3_vl.py +++ b/python/sglang/srt/arg_groups/model_overrides/qwen3_vl.py @@ -1,20 +1,69 @@ """Config-time override declarations for qwen3_vl. -Architectures: Qwen3VLForConditionalGeneration. +Architectures: Qwen3VLForConditionalGeneration, Qwen3VLMoeForConditionalGeneration. """ import logging -from typing import Any +from typing import Any, Optional from sglang.srt.arg_groups.model_override_base import ( _register_for, + model_config_of, resolving_view, ) from sglang.srt.environ import envs from sglang.srt.runtime_context import get_platform +from sglang.srt.utils.common import get_device_memory_capacity, is_sm90_supported logger = logging.getLogger(__name__) +_HOPPER_QWEN3_VL_TYPES = {"qwen3_vl", "qwen3_vl_moe"} + + +def large_hopper_qwen3_vl_model_type(server_args: Any, gpu_mem=None) -> Optional[str]: + """Return the HF model_type on large Hopper Qwen3-VL, else None.""" + cfg = resolving_view(server_args) + if gpu_mem is None: + gpu_mem = get_device_memory_capacity(cfg.device) + if not is_sm90_supported() or gpu_mem is None or gpu_mem < 60 * 1024: + return None + model_config = model_config_of(server_args) + model_type = getattr(model_config.hf_config, "model_type", "") + if model_config.is_multimodal and model_type in _HOPPER_QWEN3_VL_TYPES: + return model_type + return None + + +def expand_multimodal_decode_graph_to_running_limit( + server_args: Any, decode_config: Any, gpu_mem +) -> None: + """Keep profiled high-concurrency Qwen3-VL decode inside CUDA graph.""" + from sglang.srt.model_executor.cuda_graph_config import Phase + + cfg = resolving_view(server_args) + locked = getattr(server_args, "_cuda_graph_config_locked", set()) + max_running_requests = cfg.max_running_requests + if not ( + gpu_mem is not None + and max_running_requests is not None + and max_running_requests <= 512 + and decode_config.max_bs is not None + and decode_config.max_bs < max_running_requests + and (Phase.DECODE, "max_bs") not in locked + and (Phase.DECODE, "bs") not in locked + ): + return + if large_hopper_qwen3_vl_model_type(server_args, gpu_mem) is None: + return + + logger.info( + "Expanding multimodal decode CUDA graph max_bs from %d to " + "max_running_requests=%d.", + decode_config.max_bs, + max_running_requests, + ) + decode_config.max_bs = max_running_requests + @_register_for("Qwen3VLForConditionalGeneration") def _qwen3vl_overrides(server_args: Any, hf_config: Any) -> dict: @@ -30,3 +79,63 @@ def _qwen3vl_overrides(server_args: Any, hf_config: Any) -> dict: ) return {"page_size": 16} return {} + + +@_register_for( + "Qwen3VLForConditionalGeneration", + "Qwen3VLMoeForConditionalGeneration", +) +def _qwen3vl_hopper_serving_overrides(server_args: Any, hf_config: Any) -> dict: + """Select the profiled Qwen3-VL serving path on large Hopper GPUs.""" + model_type = large_hopper_qwen3_vl_model_type(server_args) + if model_type is None: + return {} + + cfg = resolving_view(server_args) + if not envs.SGLANG_VLM_CACHE_SIZE_MB.is_set(): + # Repeated-image traffic can opt in to embedding retention. Disable + # it by default for streaming traffic, where every image is used once. + envs.SGLANG_VLM_CACHE_SIZE_MB.set(0) + + updates = {} + preprocess_cache_size_mb = cfg.mm_preprocess_cache_size_mb + if preprocess_cache_size_mb is None: + preprocess_cache_size_mb = 0 + updates["mm_preprocess_cache_size_mb"] = 0 + cache_retention_enabled = ( + preprocess_cache_size_mb > 0 or envs.SGLANG_VLM_CACHE_SIZE_MB.get() > 0 + ) + if ( + cfg.mm_feature_transport is None + and cfg.nnodes == 1 + and not cache_retention_enabled + ): + updates["mm_feature_transport"] = "cuda_ipc" + resolved_transport = updates.get("mm_feature_transport", cfg.mm_feature_transport) + if ( + not envs.SGLANG_MM_FEATURE_CACHE_MB.is_set() + and cfg.max_running_requests is not None + and cfg.max_running_requests >= 400 + and not cache_retention_enabled + and resolved_transport == "cuda_ipc" + ): + # Keep a full high-concurrency wave GPU-resident instead of + # falling back to CPU transport while the scheduler drains it. + envs.SGLANG_MM_FEATURE_CACHE_MB.set(3 * 1024) + if ( + cfg.radix_eviction_policy == "lru" + and not cfg._radix_eviction_policy_explicitly_set + ): + updates["radix_eviction_policy"] = "priority" + if cfg.prefill_decode_interval is None: + updates["prefill_decode_interval"] = 22 + if cfg.attention_backend is None and cfg.decode_attention_backend is None: + updates["decode_attention_backend"] = "flashinfer" + + if updates: + logger.info( + "Applying profiled %s serving defaults on a large Hopper GPU: %s", + model_type, + updates, + ) + return updates diff --git a/python/sglang/srt/arg_groups/pipeline.py b/python/sglang/srt/arg_groups/pipeline.py index 298675869..a0e5e82b1 100644 --- a/python/sglang/srt/arg_groups/pipeline.py +++ b/python/sglang/srt/arg_groups/pipeline.py @@ -93,6 +93,7 @@ def run_resolution_pipeline(server_args: Any) -> None: handle_offload_compatibility(server_args) from sglang.srt.arg_groups.validation_hook import ( + default_unset_prefill_decode_interval, validate_experimental_sgl_marlin, validate_prefill_decode_interval, validate_sampling_mask_max_tokens, @@ -234,6 +235,7 @@ def run_resolution_pipeline(server_args: Any) -> None: ) handle_model_specific_adjustments(server_args) + default_unset_prefill_decode_interval(server_args) # After the model overrides: Qwen4-Exp declares the PLE offload default there. handle_offload_compatibility(server_args) diff --git a/python/sglang/srt/arg_groups/validation_hook.py b/python/sglang/srt/arg_groups/validation_hook.py index 2a5e41dbd..38fab7c6a 100644 --- a/python/sglang/srt/arg_groups/validation_hook.py +++ b/python/sglang/srt/arg_groups/validation_hook.py @@ -435,10 +435,23 @@ def validate_experimental_sgl_marlin(server_args: Any): def validate_prefill_decode_interval(server_args: Any): cfg = resolving_view(server_args) - if cfg.prefill_decode_interval < 0: + if cfg.prefill_decode_interval is not None and cfg.prefill_decode_interval < 0: raise ValueError("--prefill-decode-interval must be non-negative.") +def default_unset_prefill_decode_interval(server_args: Any): + """Leave Qwen3-VL Hopper free to pick 22; everyone else stays disabled.""" + from sglang.srt.arg_groups.overrides import declare_resolution + + cfg = resolving_view(server_args) + if cfg.prefill_decode_interval is None: + declare_resolution( + server_args, + "prefill_decode_interval_default", + prefill_decode_interval=0, + ) + + def validate_sampling_mask_max_tokens(server_args: Any): if envs.SGLANG_DISAGGREGATION_SAMPLING_MASK_MAX_TOKENS.is_set(): raise ValueError( diff --git a/python/sglang/srt/entrypoints/openai/serving_chat.py b/python/sglang/srt/entrypoints/openai/serving_chat.py index 02d19391c..03e1444f9 100644 --- a/python/sglang/srt/entrypoints/openai/serving_chat.py +++ b/python/sglang/srt/entrypoints/openai/serving_chat.py @@ -6,6 +6,7 @@ import logging import math import time import uuid +from collections import OrderedDict from enum import Enum from http import HTTPStatus from typing import TYPE_CHECKING, Any, AsyncGenerator, Dict, List, Optional, Union @@ -112,6 +113,7 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) _MEDIA_CONTENT_PART_TYPES = frozenset({"image_url", "video_url", "audio_url"}) +_CHAT_TEMPLATE_CACHE_MAX_SIZE = 128 def normalize_tool_content(role: str, content): @@ -351,6 +353,9 @@ class OpenAIServingChat(OpenAIServingBase): ) except Exception: self._tokenizer_auto_adds_specials = True + self._chat_template_cache: OrderedDict[ + bytes, tuple[str, tuple[int, ...], str] + ] = OrderedDict() def _handle_last_assistant_message( self, @@ -1342,6 +1347,7 @@ class OpenAIServingChat(OpenAIServingBase): """Apply Jinja chat template""" prompt = "" prompt_ids = [] + decoded_prompt = None openai_compatible_messages = [] image_data = [] video_data = [] @@ -1515,16 +1521,14 @@ class OpenAIServingChat(OpenAIServingBase): else {} ) try: - rendered_prompt = self.tokenizer_manager.tokenizer.apply_chat_template( - openai_compatible_messages, - tokenize=False, - add_generation_prompt=True, - tools=tools, - return_dict=False, - **extra_template_kwargs, - ) - prompt_ids = self.tokenizer_manager.tokenizer.encode( - rendered_prompt, **encode_kwargs + rendered_prompt, prompt_ids, decoded_prompt = ( + self._render_and_encode_chat_template( + openai_compatible_messages, + tools=tools, + template_kwargs=extra_template_kwargs, + encode_kwargs=encode_kwargs, + use_cache=is_multimodal, + ) ) except Exception: # If the first attempt fails, try with flat function-only format. @@ -1535,19 +1539,15 @@ class OpenAIServingChat(OpenAIServingBase): else None ) try: - rendered_prompt = ( - self.tokenizer_manager.tokenizer.apply_chat_template( + rendered_prompt, prompt_ids, decoded_prompt = ( + self._render_and_encode_chat_template( openai_compatible_messages, - tokenize=False, - add_generation_prompt=True, tools=tools, - return_dict=False, - **extra_template_kwargs, + template_kwargs=extra_template_kwargs, + encode_kwargs=encode_kwargs, + use_cache=is_multimodal, ) ) - prompt_ids = self.tokenizer_manager.tokenizer.encode( - rendered_prompt, **encode_kwargs - ) except _CHAT_TEMPLATE_CLIENT_ERRORS as template_error: # Template errors (e.g., from raise_exception in Jinja templates) # and TypeError (e.g., tojson filter on Jinja2 Undefined variables) @@ -1559,9 +1559,15 @@ class OpenAIServingChat(OpenAIServingBase): prompt_ids = self._append_assistant_prefix_to_prompt_ids( prompt_ids, assistant_prefix ) + # The cached decode corresponds to prompt_ids before the prefix. + decoded_prompt = None if is_multimodal: - prompt = self.tokenizer_manager.tokenizer.decode(prompt_ids) + prompt = ( + decoded_prompt + if decoded_prompt is not None + else self.tokenizer_manager.tokenizer.decode(prompt_ids) + ) stop = request.stop image_data = image_data if image_data else None @@ -1578,6 +1584,70 @@ class OpenAIServingChat(OpenAIServingBase): stop=stop, ) + def _render_and_encode_chat_template( + self, + messages: List[Dict[str, Any]], + *, + tools: Optional[List[Dict]], + template_kwargs: Dict[str, Any], + encode_kwargs: Dict[str, Any], + use_cache: bool, + ) -> tuple[str, List[int], Optional[str]]: + cache_key = None + if use_cache: + try: + cache_key = orjson.dumps( + ( + getattr( + self.tokenizer_manager.tokenizer, + "chat_template", + None, + ), + messages, + tools, + template_kwargs, + encode_kwargs, + ), + option=orjson.OPT_SORT_KEYS, + ) + except TypeError: + pass + + if cache_key is not None: + cached = self._chat_template_cache.get(cache_key) + if cached is not None: + self._chat_template_cache.move_to_end(cache_key) + rendered_prompt, prompt_ids, decoded_prompt = cached + return rendered_prompt, list(prompt_ids), decoded_prompt + + rendered_prompt = self.tokenizer_manager.tokenizer.apply_chat_template( + messages, + tokenize=False, + add_generation_prompt=True, + tools=tools, + return_dict=False, + **template_kwargs, + ) + prompt_ids = self.tokenizer_manager.tokenizer.encode( + rendered_prompt, **encode_kwargs + ) + decoded_prompt = ( + self.tokenizer_manager.tokenizer.decode(prompt_ids) + if cache_key is not None + else None + ) + + if cache_key is not None: + self._chat_template_cache[cache_key] = ( + rendered_prompt, + tuple(prompt_ids), + decoded_prompt, + ) + if len(self._chat_template_cache) > _CHAT_TEMPLATE_CACHE_MAX_SIZE: + self._chat_template_cache.popitem(last=False) + + return rendered_prompt, prompt_ids, decoded_prompt + def _apply_conversation_template( self, request: ChatCompletionRequest, diff --git a/python/sglang/srt/managers/mm_schedule.py b/python/sglang/srt/managers/mm_schedule.py index dbf6d2f21..13dc18697 100644 --- a/python/sglang/srt/managers/mm_schedule.py +++ b/python/sglang/srt/managers/mm_schedule.py @@ -8,6 +8,7 @@ import torch from sglang.srt.managers.schedule_batch import MultimodalDataItem from sglang.srt.mem_cache.multimodal_cache import EmbeddingResult, MultiModalStaticCache from sglang.srt.multimodal.evs import EVSEmbeddingResult +from sglang.srt.multimodal.transport.cuda_ipc import BORROW_CUDA_IPC_FEATURE_KEY from sglang.srt.runtime_context import get_parallel, get_schedule from sglang.srt.utils import is_hip, is_npu, is_xpu from sglang.srt.utils.async_probe import maybe_assert_sum @@ -368,6 +369,12 @@ def _batch_encode_per_image_misses( ) unique_misses[cache_key] = (item, expected_token_count) elif cache_key not in unique_misses: + if ( + start >= chunk_start + and end < chunk_end + and item.can_defer_cuda_ipc_feature_reconstruction() + ): + item.model_specific_data[BORROW_CUDA_IPC_FEATURE_KEY] = True unique_misses[cache_key] = (item, expected_token_count) # Phase 1b: single ViT call for all unique cache misses diff --git a/python/sglang/srt/managers/schedule_batch.py b/python/sglang/srt/managers/schedule_batch.py index efde185a4..f699df9b9 100755 --- a/python/sglang/srt/managers/schedule_batch.py +++ b/python/sglang/srt/managers/schedule_batch.py @@ -126,7 +126,9 @@ from sglang.srt.model_executor.forward_batch_info import ( ForwardMode, ) from sglang.srt.multimodal.transport.cuda_ipc import ( + CUDA_IPC_FEATURE_COPY_EVENT_KEY, DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY, + RETAINED_CUDA_IPC_FEATURE_PROXY_KEY, CudaIpcTensorTransportProxy, ) from sglang.srt.observability.metrics_collector import ( @@ -494,6 +496,8 @@ class MultimodalDataItem(msgspec.Struct, kw_only=True, dict=True, array_like=Tru self.precomputed_embeddings.reconstruct_on_target_device(target_device) ) for extra_key in self.model_specific_data: + if extra_key == RETAINED_CUDA_IPC_FEATURE_PROXY_KEY: + continue if isinstance( self.model_specific_data[extra_key], CudaIpcTensorTransportProxy ): @@ -533,7 +537,11 @@ class MultimodalDataItem(msgspec.Struct, kw_only=True, dict=True, array_like=Tru def release_transport_proxies(self, consumer_count: int = 1) -> None: """Best-effort release of proxies left by an abandoned request.""" - values = [self.feature, self.precomputed_embeddings] + retained_proxy = self.model_specific_data.pop( + RETAINED_CUDA_IPC_FEATURE_PROXY_KEY, None + ) + self.model_specific_data.pop(CUDA_IPC_FEATURE_COPY_EVENT_KEY, None) + values = [self.feature, self.precomputed_embeddings, retained_proxy] values.extend(self.model_specific_data.values()) for value in values: if not isinstance(value, CudaIpcTensorTransportProxy): @@ -682,10 +690,9 @@ class MultimodalInputs: """Release feature tensors to free GPU memory.""" for item in self.mm_items: try: - # A request can be rejected before a deferred GPU feature is - # reconstructed. Acknowledge that transport lease before the - # proxy is dropped so the tokenizer pool can reuse its slice. - item.acknowledge_deferred_cuda_ipc_feature() + # Release both deferred features that were never used and + # borrowed features retained for possible re-prefill. + item.release_transport_proxies() except Exception: logger.warning( "Failed to release an unused multimodal feature transport", diff --git a/python/sglang/srt/managers/scheduler.py b/python/sglang/srt/managers/scheduler.py index f0914b18d..eca270231 100644 --- a/python/sglang/srt/managers/scheduler.py +++ b/python/sglang/srt/managers/scheduler.py @@ -1283,7 +1283,7 @@ class Scheduler( def init_chunked_prefill(self): self.chunked_prefill_size = get_schedule().chunked_prefill_size - self.prefill_decode_interval = get_schedule().prefill_decode_interval + self.prefill_decode_interval = get_schedule().prefill_decode_interval or 0 self._prefill_decode_interval_remaining = 0 uses_transformers_backend = ( get_resolved_model_impl(self.model_config) == ModelImpl.TRANSFORMERS diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py index 1069ced6a..91315321c 100644 --- a/python/sglang/srt/managers/tokenizer_manager.py +++ b/python/sglang/srt/managers/tokenizer_manager.py @@ -976,6 +976,7 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin): input_embeds = None input_text = obj.text token_type_ids = None + contains_mm_input = obj.contains_mm_input() is_cross_encoder_request = ( isinstance(obj, EmbeddingReqInput) and obj.is_cross_encoder_request ) @@ -998,9 +999,21 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin): "the engine with skip_tokenizer_init=False." ) + # Avoid tokenizing a raw multimodal prompt when the processor will + # expand its placeholders and replace input_ids below. + if ( + self.mm_processor + and contains_mm_input + and getattr( + self.mm_processor, + "generates_input_ids_from_raw_prompt", + False, + ) + ): + input_ids = None # For audio-only requests (e.g., Whisper), text may be empty. # The multimodal processor will provide input_ids later. - if not input_text and self.mm_processor and obj.contains_mm_input(): + elif not input_text and self.mm_processor and contains_mm_input: # Use empty placeholder - multimodal processor will override input_ids = [] else: @@ -1008,7 +1021,6 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin): input_text, is_cross_encoder_request ) - contains_mm_input = obj.contains_mm_input() if contains_mm_input and get_disagg().language_model_only: raise ValueError( "Multimodal inputs are not supported when --language-model-only " diff --git a/python/sglang/srt/mem_cache/radix_cache.py b/python/sglang/srt/mem_cache/radix_cache.py index f33812cc7..84f8bb422 100644 --- a/python/sglang/srt/mem_cache/radix_cache.py +++ b/python/sglang/srt/mem_cache/radix_cache.py @@ -517,6 +517,31 @@ class RadixCache(BasePrefixCache): result = self.insert( InsertParams(key=radix_key, value=values, priority=priority) ) + # A request that was never cached while unfinished can add its + # whole prompt and generated output as one leaf. Split that leaf at + # the prompt boundary so LRU eviction can discard output KV without + # also losing the reusable prompt KV. Reinserting a prefix only + # changes radix topology; it reuses the indices inserted above. + prompt_key = RadixKey( + token_ids[: len(req.origin_input_ids)], + req.extra_key, + is_bigram=self.is_eagle, + cache_salt=req.cache_salt, + ).page_aligned(self.page_size) + if 0 < len(prompt_key) < key_len: + self.insert( + InsertParams( + key=prompt_key, + value=values[: len(prompt_key)], + priority=priority + 1, + # Topology-only re-insert: this request created these + # nodes moments ago, so counting it as a hit is the + # same self-referencing inflation `chunked` exists to + # suppress. hit_count drives eviction order, so an + # extra bump would silently promote every prompt node. + chunked=True, + ) + ) freed_end = result.prefix_len else: freed_end = key_len diff --git a/python/sglang/srt/mem_cache/unified_radix_cache.py b/python/sglang/srt/mem_cache/unified_radix_cache.py index 19bed9e53..40d98a288 100644 --- a/python/sglang/srt/mem_cache/unified_radix_cache.py +++ b/python/sglang/srt/mem_cache/unified_radix_cache.py @@ -995,6 +995,44 @@ class UnifiedRadixCache(BasePrefixCache): insert_params.value = values result = self.insert(insert_params) + # Keep the prompt as an independent radix node. Finished requests + # append a short, request-specific output to a much longer prompt; + # without this split the prompt and output form one leaf and are + # evicted together. Re-inserting the prompt only changes topology: + # prev_prefix_len prevents the overlapping KV indices from being + # treated as duplicate allocations and freed. A declined rotation + # tail releases everything past the protected prefix below, so the + # split is skipped there rather than handing the tree rows that + # are about to be freed. + prompt_key = RadixKey( + req.origin_input_ids, + req.extra_key, + is_bigram=self.tree_core.is_eagle, + cache_salt=req.cache_salt, + ).page_aligned(self.page_size) + if ( + not result.rotation_tail_declined + and len(self._components_tuple) == 1 + and self._components_tuple[0].component_type == BASE_COMPONENT_TYPE + and 0 < len(prompt_key) < len(radix_key) + ): + self.insert( + replace( + insert_params, + key=prompt_key, + value=values[: len(prompt_key)], + prev_prefix_len=len(prompt_key), + priority=insert_params.priority + 1, + # Topology-only re-insert: the request itself created + # these nodes moments ago, so counting it as a hit is + # the same self-referencing inflation `chunked` exists + # to suppress. hit_count drives eviction order, so an + # extra bump here would silently promote every prompt + # node into the protected segment. + chunked=True, + ) + ) + # Free unaligned tail (+ deferred truncation tail). A rotation # decline inserted nothing, so the whole span past the protected # prefix stayed request-owned and is released here instead. diff --git a/python/sglang/srt/models/qwen3_vl.py b/python/sglang/srt/models/qwen3_vl.py index 92bd58687..db6ba8427 100644 --- a/python/sglang/srt/models/qwen3_vl.py +++ b/python/sglang/srt/models/qwen3_vl.py @@ -72,6 +72,12 @@ from sglang.srt.multimodal.mm_utils import ( materialize_multimodal_features, run_dp_sharded_mrope_vision_model, ) +from sglang.srt.multimodal.transport.cuda_ipc import ( + BORROW_CUDA_IPC_FEATURE_KEY, + CUDA_IPC_FEATURE_COPY_EVENT_KEY, + RETAINED_CUDA_IPC_FEATURE_PROXY_KEY, + CudaIpcTensorTransportProxy, +) from sglang.srt.multimodal.vit_cuda_graph_runner import ViTCudaGraphRunner from sglang.srt.runtime_context import get_exec, get_mm, get_parallel from sglang.srt.utils import ( @@ -1435,13 +1441,23 @@ class Qwen3VLForConditionalGeneration(nn.Module): pixel_values_device=self.visual.device, pixel_values_dtype=self.visual.dtype, ) - pixel_values = self._materialize_visual_items(items, range(len(items))) + pixel_values, borrowed_items, packed_ready = self._materialize_visual_items( + items, range(len(items)), preserve_for_reprefill=True + ) assert pixel_values.dim() == 2, pixel_values.dim() - return self.visual(pixel_values, grid_thw=grid_thw) + visual_features = self.visual(pixel_values, grid_thw=grid_thw) + if borrowed_items: + self._offload_packed_visual_inputs( + pixel_values, borrowed_items, packed_ready + ) + return visual_features def _materialize_visual_items( - self, items: List[MultimodalDataItem], indices: Iterable[int] - ) -> torch.Tensor: + self, + items: List[MultimodalDataItem], + indices: Iterable[int], + preserve_for_reprefill: bool = False, + ) -> Union[torch.Tensor, Tuple[torch.Tensor, list, Optional[torch.cuda.Event]]]: device = self.visual.device device_index = device.index if device.type == "cuda" and device_index is None: @@ -1451,14 +1467,114 @@ class Qwen3VLForConditionalGeneration(nn.Module): consumer_count = max(parallel.tp_size, 1) features = [] + borrowed_items = [] + feature_offset = 0 for index in indices: item = items[index] if device.type == "cuda": - item.reconstruct(device_index, ipc_consumer_count=consumer_count) + proxy = item.feature + model_specific_data = getattr(item, "model_specific_data", {}) + pending_copy = model_specific_data.pop( + CUDA_IPC_FEATURE_COPY_EVENT_KEY, None + ) + if pending_copy is not None: + torch.cuda.current_stream(device_index).wait_event(pending_copy) + borrow_requested = model_specific_data.pop( + BORROW_CUDA_IPC_FEATURE_KEY, False + ) + can_borrow = ( + preserve_for_reprefill + and consumer_count == 1 + and isinstance(proxy, CudaIpcTensorTransportProxy) + and borrow_requested + ) + borrowed = ( + proxy.borrow_on_target_device(device_index) if can_borrow else None + ) + if borrowed is not None: + item.feature = borrowed + model_specific_data[RETAINED_CUDA_IPC_FEATURE_PROXY_KEY] = proxy + borrowed_items.append( + (item, proxy, feature_offset, borrowed.shape[0]) + ) + elif RETAINED_CUDA_IPC_FEATURE_PROXY_KEY not in model_specific_data: + item.reconstruct(device_index, ipc_consumer_count=consumer_count) features.append(item.feature) - return materialize_multimodal_features( - features, device=device, dtype=self.visual.dtype - ) + feature_offset += item.feature.shape[0] + try: + materialized = materialize_multimodal_features( + features, device=device, dtype=self.visual.dtype + ) + except Exception: + for item, proxy, _, _ in borrowed_items: + try: + proxy.release_borrowed_on_current_stream() + item.model_specific_data.pop( + RETAINED_CUDA_IPC_FEATURE_PROXY_KEY, None + ) + except Exception: + logger.warning( + "Failed to release a borrowed CUDA IPC feature after " + "materialization failed", + exc_info=True, + ) + item.feature = None + raise + packed_ready = None + if borrowed_items: + source_stream = torch.cuda.current_stream(device_index) + packed_ready = torch.cuda.Event() + packed_ready.record(source_stream) + for item, proxy, offset, length in borrowed_items: + item.feature = materialized.narrow(0, offset, length) + item.model_specific_data[CUDA_IPC_FEATURE_COPY_EVENT_KEY] = packed_ready + try: + proxy.release_borrowed_on_current_stream() + item.model_specific_data.pop( + RETAINED_CUDA_IPC_FEATURE_PROXY_KEY, None + ) + except Exception: + logger.warning( + "Failed to release a copied CUDA IPC feature; retaining " + "its lease until request cleanup", + exc_info=True, + ) + if preserve_for_reprefill: + return materialized, borrowed_items, packed_ready + return materialized + + def _offload_packed_visual_inputs( + self, materialized: torch.Tensor, borrowed_items: list, packed_ready + ) -> None: + """Preserve inputs for re-prefill without delaying the ViT launch.""" + try: + host_features = torch.empty( + materialized.shape, + dtype=materialized.dtype, + device="cpu", + pin_memory=torch.cuda.is_available(), + ) + copy_stream = getattr(self, "_mm_feature_copy_stream", None) + if copy_stream is None: + copy_stream = torch.cuda.Stream(device=self.visual.device) + self._mm_feature_copy_stream = copy_stream + with torch.cuda.stream(copy_stream): + copy_stream.wait_event(packed_ready) + host_features.copy_(materialized, non_blocking=True) + host_ready = torch.cuda.Event() + host_ready.record(copy_stream) + if materialized.is_cuda: + materialized.record_stream(copy_stream) + for item, _, offset, length in borrowed_items: + item.feature = host_features.narrow(0, offset, length) + item.model_specific_data[CUDA_IPC_FEATURE_COPY_EVENT_KEY] = host_ready + except Exception: + # The generic multimodal path will offload the owned CUDA slices. + logger.warning( + "Failed to preserve CUDA IPC features on the copy stream; " + "falling back to the generic offload path", + exc_info=True, + ) def get_input_embeddings(self): return self.model.embed_tokens diff --git a/python/sglang/srt/multimodal/media_artifacts/base.py b/python/sglang/srt/multimodal/media_artifacts/base.py index 762c6343f..19dcea6ca 100644 --- a/python/sglang/srt/multimodal/media_artifacts/base.py +++ b/python/sglang/srt/multimodal/media_artifacts/base.py @@ -412,6 +412,66 @@ class MediaArtifactCacheMixin: raise RuntimeError("Artifact cache did not resolve every media item") return [artifact for artifact in artifacts if artifact is not None] + async def prepare_media_artifacts_without_cache( + self, + media_data: Sequence[Any], + *, + content_hashes: Optional[Sequence[Optional[str]]] = None, + modality: Optional[Modality] = None, + ) -> list[MediaArtifact]: + """Build request-local artifacts without cache lookup or retention.""" + modality = self._resolve_artifact_modality(modality) + media_count = len(media_data) + if content_hashes is None: + content_hashes = [None] * media_count + if len(content_hashes) != media_count: + raise ValueError( + f"mm_content_hashes has {len(content_hashes)} entries for " + f"{media_count} {modality.name.lower()} items" + ) + content_hashes = [parse_content_hash(value) for value in content_hashes] + + snapshot_futures = [ + self.io_executor.submit(self.snapshot_media_source, source, modality) + for source in media_data + ] + snapshots = [] + for index, future in enumerate(snapshot_futures): + snapshot = await asyncio.wrap_future(future) + caller_hash = content_hashes[index] + if caller_hash is not None and caller_hash != snapshot.content_digest: + raise ValueError( + f"content hash mismatch for media_data[{index}]: " + f"expected {caller_hash}, got {snapshot.content_digest}" + ) + snapshots.append(snapshot) + + decode_futures = [ + self.io_executor.submit(self.decode_media_snapshot, snapshot, modality) + for snapshot in snapshots + ] + entries = [] + for source, snapshot, future in zip(media_data, snapshots, decode_futures): + entries.append( + MediaArtifactInput( + content_digest=snapshot.content_digest, + artifact_key=self._artifact_key( + snapshot.content_digest, source, modality=modality + ), + modality=modality, + media=await asyncio.wrap_future(future), + ) + ) + + artifacts = await self._run_preprocess_and_build_artifact_batch(entries) + if len(artifacts) != len(entries): + raise ValueError( + "prepare_artifact_batch must return one artifact per media input" + ) + for artifact, entry in zip(artifacts, entries): + self.validate_artifact(artifact, entry) + return artifacts + async def _compute_cache_misses( self, misses_to_compute: Sequence[CacheMiss[str, MediaArtifact]], diff --git a/python/sglang/srt/multimodal/processors/base_processor.py b/python/sglang/srt/multimodal/processors/base_processor.py index a0d8926df..9c6d172b0 100644 --- a/python/sglang/srt/multimodal/processors/base_processor.py +++ b/python/sglang/srt/multimodal/processors/base_processor.py @@ -227,6 +227,9 @@ class BaseMultimodalProcessor(ABC): # 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 + # Artifact-based processors may keep their prompt/M-RoPE fast path even + # when artifact retention is disabled. + uses_media_artifacts_without_cache = 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 -- @@ -287,6 +290,7 @@ class BaseMultimodalProcessor(ABC): self.processor_fingerprint = ( build_processor_fingerprint(self, hf_config) if self.mm_preprocess_cache.enabled + or self.uses_media_artifacts_without_cache else None ) if self.mm_preprocess_cache.enabled: diff --git a/python/sglang/srt/multimodal/processors/qwen_vl.py b/python/sglang/srt/multimodal/processors/qwen_vl.py index e911611d5..cc49a1393 100644 --- a/python/sglang/srt/multimodal/processors/qwen_vl.py +++ b/python/sglang/srt/multimodal/processors/qwen_vl.py @@ -2,13 +2,17 @@ import math import os import re import time -from typing import List, Optional, Union +from copy import deepcopy +from dataclasses import dataclass, replace +from functools import lru_cache +from typing import Any, List, Optional, Union import numpy as np import torch import torchvision from PIL import Image from torchvision.transforms import InterpolationMode +from transformers import BaseImageProcessor from sglang.srt.environ import envs from sglang.srt.layers.rotary_embedding import MRotaryEmbedding @@ -33,6 +37,11 @@ from sglang.srt.models.qwen3_omni_moe import Qwen3OmniMoeForConditionalGeneratio from sglang.srt.models.qwen3_vl import Qwen3VLForConditionalGeneration from sglang.srt.models.qwen3_vl_moe import Qwen3VLMoeForConditionalGeneration from sglang.srt.models.qwen4_exp import Qwen4ExpForConditionalGeneration +from sglang.srt.multimodal.cache import resolve_multimodal_item_hash +from sglang.srt.multimodal.media_artifacts.base import ( + MediaArtifactCacheMixin, + MediaArtifactInput, +) from sglang.srt.multimodal.processors.base_processor import ( BaseMultimodalProcessor as SGLangBaseProcessor, ) @@ -42,7 +51,7 @@ from sglang.srt.multimodal.processors.base_processor import ( from sglang.srt.multimodal.transport.cuda_ipc import ( DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY, ) -from sglang.srt.runtime_context import get_mm +from sglang.srt.runtime_context import get_mm, get_parallel from sglang.srt.utils import cpu_has_amx_support, is_cpu from sglang.srt.utils.video_decoder import VideoDecoderWrapper from sglang.utils import logger @@ -68,6 +77,37 @@ FPS = 2.0 FPS_MIN_FRAMES = 4 FPS_MAX_FRAMES = 768 + +@dataclass(frozen=True) +class QwenVLImagePreprocessArtifact: + """Prompt-independent Qwen-VL processor output for one image.""" + + content_digest: str + artifact_key: str + feature_hash: int + feature: Optional[torch.Tensor] + model_specific_data: dict[str, Any] + + @property + def has_feature(self) -> bool: + return self.feature is not None + + def cache_value(self) -> "QwenVLImagePreprocessArtifact": + """Keep CPU processor outputs but never retain a CUDA tensor.""" + if self.feature is None or self.feature.device.type == "cpu": + return self + return replace(self, feature=None) + + def cache_size_items(self) -> tuple: + return ( + self.content_digest, + self.artifact_key, + self.feature_hash, + self.feature, + self.model_specific_data, + ) + + QWEN_VIDEO_PREPROCESS_CONFIG_KEYS = frozenset( { "fps", @@ -289,8 +329,10 @@ async def preprocess_video( # Compatible with Qwen-VL & Qwen-Omni Series -class QwenVLImageProcessor(SGLangBaseProcessor): +class QwenVLImageProcessor(MediaArtifactCacheMixin, SGLangBaseProcessor): supports_transformers_backend = True + generates_input_ids_from_raw_prompt = True + artifact_modality = Modality.IMAGE models = [ Qwen2VLForConditionalGeneration, Qwen2_5_VLForConditionalGeneration, @@ -308,6 +350,10 @@ class QwenVLImageProcessor(SGLangBaseProcessor): def __init__(self, hf_config, server_args, _processor, *args, **kwargs): self.model_type = hf_config.model_type + self.uses_media_artifacts_without_cache = self.model_type in ( + "qwen3_vl", + "qwen3_vl_moe", + ) if self.model_type in ( "qwen2_vl", "qwen2_5_vl", @@ -732,7 +778,194 @@ class QwenVLImageProcessor(SGLangBaseProcessor): mrope_position_delta=mrope_position_delta, ) - async def process_mm_data_async( + def prepare_artifact_batch( + self, + entries: list[MediaArtifactInput], + *, + processor=None, + ) -> list[QwenVLImagePreprocessArtifact]: + """Preprocess image cache misses without retaining prompt-specific state.""" + if not entries: + return [] + + processor, _ = self._resolve_processor(processor) + image_kwargs = dict(self.image_config or {}) + processor_device = None + if ( + isinstance(processor.image_processor, BaseImageProcessor) + and not self.disable_fast_image_processor + ): + processor_device = self._fast_image_processor_device(processor) + if processor_device is not None: + image_kwargs["device"] = processor_device + + with self._temporary_fast_processor_cuda_pool(processor_device): + result = processor.image_processor( + images=[entry.media for entry in entries], + return_tensors="pt", + **image_kwargs, + ) + features = self._get_processor_output_value(result, "pixel_values") + image_grid_thw = self._get_processor_output_value(result, "image_grid_thw") + if ( + isinstance(features, torch.Tensor) + and ( + self.mm_preprocess_cache.enabled + or not self.keep_mm_features_on_device + ) + and not self.precompute_hash_before_cpu_transfer + ): + features = features.cpu() + + if not isinstance(features, torch.Tensor): + raise TypeError("Qwen-VL image processor must return pixel_values") + if not isinstance(image_grid_thw, torch.Tensor): + image_grid_thw = torch.as_tensor(image_grid_thw, dtype=torch.long) + if image_grid_thw.ndim != 2 or image_grid_thw.shape != (len(entries), 3): + raise ValueError( + "Qwen-VL image processor returned an invalid image_grid_thw shape: " + f"{tuple(image_grid_thw.shape)}" + ) + feature_lengths = image_grid_thw.prod(dim=1).tolist() + if sum(feature_lengths) != features.shape[0]: + raise ValueError( + "Qwen-VL image processor feature count does not match image grids: " + f"{features.shape[0]} != {sum(feature_lengths)}" + ) + + artifacts = [] + feature_offset = 0 + for entry, grid, feature_length in zip( + entries, image_grid_thw, feature_lengths + ): + feature = features[ + feature_offset : feature_offset + feature_length + ].contiguous() + feature_offset += feature_length + # The artifact key already commits to the media content, processor + # fingerprint, and every preprocessing kwarg. Derive the downstream + # cache identity from it so independently preprocessed copies in + # different tokenizer workers share the same radix/VLM cache key. + feature_hash = resolve_multimodal_item_hash( + existing_hash=0, namespace=entry.artifact_key + ) + artifacts.append( + QwenVLImagePreprocessArtifact( + content_digest=entry.content_digest, + artifact_key=entry.artifact_key, + feature_hash=feature_hash, + feature=feature, + model_specific_data={"image_grid_thw": grid.unsqueeze(0)}, + ) + ) + return artifacts + + def compose_image_artifacts( + self, + input_text, + artifacts: list[QwenVLImagePreprocessArtifact], + ) -> MultimodalProcessorOutput: + """Compose prompt tokens and request-owned items from cached images.""" + image_grids = [] + for artifact in artifacts: + grid = self._as_grid_batch( + artifact.model_specific_data.get("image_grid_thw") + ) + if grid is None or grid.shape[0] != 1: + raise ValueError("Each Qwen-VL image artifact requires one image grid") + image_grids.append(grid) + image_grid_thw = torch.cat(image_grids, dim=0) + + grid_key = tuple( + tuple(int(value) for value in row.tolist()) for row in image_grid_thw + ) + if isinstance(input_text, str): + ( + input_ids_tuple, + offsets, + mrope_positions, + mrope_position_delta, + ) = self._cached_image_prompt_template(input_text, grid_key) + else: + ( + input_ids_tuple, + offsets, + mrope_positions, + mrope_position_delta, + ) = self._build_image_prompt_template(input_text, grid_key) + input_ids_list = list(input_ids_tuple) + + mm_items = [] + for artifact, offset in zip(artifacts, offsets): + item = MultimodalDataItem( + modality=Modality.IMAGE, + feature=artifact.feature, + offsets=[offset], + model_specific_data=deepcopy(artifact.model_specific_data), + ) + item.set_hash(artifact.feature_hash) + mm_items.append(item) + + padded_input_ids = MultimodalProcessorOutput.build_padded_input_ids( + input_ids_list, mm_items + ) + mm_items = self._prepare_mm_items_for_transport(mm_items) + self._mark_cuda_ipc_features_for_deferred_reconstruction(mm_items) + return MultimodalProcessorOutput( + input_ids=input_ids_list, + padded_input_ids=padded_input_ids, + mm_items=mm_items, + im_start_id=self.vision_start_token_id, + im_end_id=self.vision_end_token_id, + im_token_id=self.mm_tokens.image_token_id, + video_token_id=self.mm_tokens.video_token_id, + audio_token_id=self.mm_tokens.audio_token_id, + mrope_positions=mrope_positions.clone(), + mrope_position_delta=mrope_position_delta.clone(), + ) + + @lru_cache(maxsize=256) + def _cached_image_prompt_template(self, input_text: str, grid_key: tuple): + """Cache prompt expansion and M-RoPE by prompt and image-grid shape.""" + return self._build_image_prompt_template(input_text, grid_key) + + def _build_image_prompt_template(self, input_text, grid_key: tuple): + image_grid_thw = torch.tensor(grid_key, dtype=torch.long) + input_ids_list, offsets, modalities = self.build_input_ids( + input_text, img_grid_thw=image_grid_thw + ) + if modalities != [Modality.IMAGE] * len(grid_key): + raise ValueError("Qwen-VL image artifacts do not match prompt placeholders") + + template_items = [ + MultimodalDataItem( + modality=Modality.IMAGE, + feature=None, + offsets=[offset], + model_specific_data={"image_grid_thw": grid.unsqueeze(0)}, + ) + for grid, offset in zip(image_grid_thw, offsets) + ] + input_ids = torch.tensor(input_ids_list, dtype=torch.long) + mrope_result = self._compute_image_only_mrope_positions_from_offsets( + input_len=input_ids.numel(), + mm_items=template_items, + dtype=input_ids.dtype, + device=input_ids.device, + ) + if mrope_result is None: + mrope_result = self.compute_mrope_positions(input_ids_list, template_items) + mrope_positions, mrope_position_delta = mrope_result + if mrope_positions is not None and mrope_positions.ndim == 3: + mrope_positions = mrope_positions.squeeze(1) + return ( + tuple(input_ids_list), + tuple(offsets), + mrope_positions, + mrope_position_delta, + ) + + async def _process_mm_data_uncached( self, image_data: List[Union[str, bytes]], input_text, @@ -788,7 +1021,7 @@ class QwenVLImageProcessor(SGLangBaseProcessor): base_output, self.mm_tokens, **processor_kwargs ) - self._mark_dp_encoder_features_for_deferred_reconstruction(mm_items) + self._mark_cuda_ipc_features_for_deferred_reconstruction(mm_items) audio_feature_lengths = None @@ -902,10 +1135,46 @@ class QwenVLImageProcessor(SGLangBaseProcessor): mrope_position_delta=mrope_position_delta, ) - def _mark_dp_encoder_features_for_deferred_reconstruction(self, mm_items): + async def process_mm_data_async( + self, + image_data: List[Union[str, bytes]], + input_text, + request_obj, + *args, + **kwargs, + ): + if ( + not image_data + or request_obj.video_data + or request_obj.audio_data + or any(self._is_preprocessed_input(item) for item in image_data) + or ( + not self.mm_preprocess_cache.enabled + and not self.uses_media_artifacts_without_cache + ) + ): + return await self._process_mm_data_uncached( + image_data, input_text, request_obj, *args, **kwargs + ) + + prepare_artifacts = ( + self.prepare_media_artifacts + if self.mm_preprocess_cache.enabled + else self.prepare_media_artifacts_without_cache + ) + artifacts = await prepare_artifacts( + image_data, content_hashes=getattr(request_obj, "mm_content_hashes", None) + ) + return self.compose_image_artifacts(input_text, artifacts) + + def _mark_cuda_ipc_features_for_deferred_reconstruction(self, mm_items): + supports_deferred_reconstruction = get_mm().mm_enable_dp_encoder or ( + get_parallel().tp_size == 1 + and self.model_type in ("qwen3_vl", "qwen3_vl_moe") + ) if not ( self.keep_mm_features_on_device - and get_mm().mm_enable_dp_encoder + and supports_deferred_reconstruction and self.model_type in ("qwen3_vl", "qwen3_vl_moe", "qwen3_5", "qwen3_5_moe") ): diff --git a/python/sglang/srt/multimodal/transport/cuda_ipc.py b/python/sglang/srt/multimodal/transport/cuda_ipc.py index 1f8f14e2a..705031725 100644 --- a/python/sglang/srt/multimodal/transport/cuda_ipc.py +++ b/python/sglang/srt/multimodal/transport/cuda_ipc.py @@ -25,6 +25,9 @@ MM_ITEM_MEMORY_POOL_RECYCLE_INTERVAL = ( DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY = ( "_sglang_defer_cuda_ipc_feature_reconstruction" ) +BORROW_CUDA_IPC_FEATURE_KEY = "_sglang_borrow_cuda_ipc_feature" +CUDA_IPC_FEATURE_COPY_EVENT_KEY = "_sglang_cuda_ipc_feature_copy_event" +RETAINED_CUDA_IPC_FEATURE_PROXY_KEY = "_sglang_retained_cuda_ipc_feature_proxy" def get_mm_feature_pool_size_per_worker( @@ -233,6 +236,9 @@ class CudaIpcTensorTransportProxy(StreamOrderedPoolConsumerMixin): # Keep uncached mappings alive until the work enqueued on the consumer # stream has completed. self._pool_storage = None + self._borrowed_storage = None + self._borrowed_base_address = None + self._borrowed_device_id = None def _reconstruct_from_ipc_extra( self, ipc_extra, *, use_cache: bool, rebuild_device_idx: int @@ -321,9 +327,53 @@ class CudaIpcTensorTransportProxy(StreamOrderedPoolConsumerMixin): ) self._retain_storage_until_stream_completes(storage, device_id) + def borrow_on_target_device( + self, rebuild_device_idx: int + ) -> Optional[torch.Tensor]: + """Return a zero-copy view whose lease remains owned by this proxy.""" + ipc_extra = self.proxy_state["ipc_extra"] + if not ipc_extra["use_pool_handle_cache"] or self._consumer_acknowledged: + return None + + with torch.cuda.device(rebuild_device_idx): + slice_tensor, storage = self._open_pool_slice(rebuild_device_idx) + base_address = storage.data_ptr() + self._wait_until_ready(base_address, rebuild_device_idx) + borrowed = slice_tensor.view(ipc_extra["recons_dtype"]).reshape( + ipc_extra["recons_shape"] + ) + + self._borrowed_storage = storage + self._borrowed_base_address = base_address + self._borrowed_device_id = rebuild_device_idx + return borrowed + + def release_borrowed_on_current_stream( + self, consumer_count: int = 1, consumer_rank: Optional[int] = None + ) -> None: + """Release a borrowed view after all current-stream reads are enqueued.""" + storage = getattr(self, "_borrowed_storage", None) + if storage is None: + return + device_id = self._borrowed_device_id + with torch.cuda.device(device_id): + self._acknowledge_on_stream( + self._borrowed_base_address, + device_id, + consumer_count, + consumer_rank, + ) + self._retain_storage_until_stream_completes(storage, device_id) + self._borrowed_storage = None + self._borrowed_base_address = None + self._borrowed_device_id = None + def release_without_reconstruction(self, consumer_count: int = 1) -> None: """Release a pool slice when its request abandons this proxy.""" - self.acknowledge_consumption(consumer_count) + if getattr(self, "_borrowed_storage", None) is not None: + self.release_borrowed_on_current_stream(consumer_count) + else: + self.acknowledge_consumption(consumer_count) def reconstruct_on_target_device( self, diff --git a/python/sglang/srt/server_args.py b/python/sglang/srt/server_args.py index 0daef9243..d79bb75d7 100644 --- a/python/sglang/srt/server_args.py +++ b/python/sglang/srt/server_args.py @@ -710,7 +710,15 @@ def prepare_server_args(argv: list[str]) -> ServerArgs: config_merger = ConfigArgumentMerger(parser) argv = config_merger.merge_config_with_args(argv) + radix_eviction_policy_explicitly_set = any( + arg == "--radix-eviction-policy" or arg.startswith("--radix-eviction-policy=") + for arg in argv + ) + raw_args = parser.parse_args(argv) + raw_args._radix_eviction_policy_explicitly_set = ( + radix_eviction_policy_explicitly_set + ) # Set up basic logging before ServerArgs.__post_init__ so that # logger.info / logger.warning calls there are properly formatted. diff --git a/test/registered/chunked_prefill/test_mm_chunked_embedding_unit.py b/test/registered/chunked_prefill/test_mm_chunked_embedding_unit.py index 593b705be..ee53a3da0 100644 --- a/test/registered/chunked_prefill/test_mm_chunked_embedding_unit.py +++ b/test/registered/chunked_prefill/test_mm_chunked_embedding_unit.py @@ -17,6 +17,11 @@ import torch from sglang.srt.managers import mm_schedule from sglang.srt.managers.schedule_batch import Modality, MultimodalDataItem +from sglang.srt.multimodal.transport.cuda_ipc import ( + BORROW_CUDA_IPC_FEATURE_KEY, + DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY, + CudaIpcTensorTransportProxy, +) from sglang.srt.runtime_context import get_context, get_parallel from sglang.test.ci.ci_register import register_cpu_ci @@ -235,6 +240,30 @@ def test_batched_mismatched_cache_entry_is_reencoded(): encoder.assert_called_once() +def test_full_deferred_ipc_item_is_marked_for_borrow(): + mm_schedule.init_mm_embedding_cache(1 << 30) + proxy = object.__new__(CudaIpcTensorTransportProxy) + item = MultimodalDataItem( + modality=Modality.IMAGE, + hash=1000, + pad_value=1000, + feature=proxy, + offsets=[ITEM_OFFSETS[0]], + model_specific_data={DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY: True}, + ) + request = mm_schedule.PerImageRequestInfo( + req_idx=0, + items=[item], + items_offset=[ITEM_OFFSETS[0]], + extend_prefix_len=0, + extend_seq_len=TOTAL_LEN, + ) + + mm_schedule._batch_encode_per_image_misses(_encoder_list, [request], _CPU) + + assert item.model_specific_data[BORROW_CUDA_IPC_FEATURE_KEY] + + def test_batched_colliding_hashes_with_different_lengths_are_not_deduplicated(): mm_schedule.init_mm_embedding_cache(1 << 30) items = _make_items() diff --git a/test/registered/unit/entrypoints/openai/test_serving_chat.py b/test/registered/unit/entrypoints/openai/test_serving_chat.py index 1cac9012d..223314353 100644 --- a/test/registered/unit/entrypoints/openai/test_serving_chat.py +++ b/test/registered/unit/entrypoints/openai/test_serving_chat.py @@ -6,7 +6,7 @@ or python -m unittest discover -s tests -p "test_*unit.py" -v """ -from sglang.test.test_utils import enter_override, maybe_stub_sgl_kernel +from sglang.test.test_utils import CustomTestCase, enter_override, maybe_stub_sgl_kernel maybe_stub_sgl_kernel() # must precede any import that pulls in sgl_kernel @@ -195,6 +195,81 @@ class _MockTemplateManager: self.jinja_template_may_reorder_tool_results = False +class TestChatTemplateCache(CustomTestCase): + def setUp(self): + super().setUp() + reset_context() + self.addCleanup(reset_context) + publish( + ServerArgs(model_path="dummy", default_chat_template_kwargs=None), + role="tokenizer", + ) + self.tokenizer_manager = _MockTokenizerManager() + self.chat = OpenAIServingChat( + self.tokenizer_manager, + _MockTemplateManager(), + ) + self.tokenizer_manager.tokenizer.apply_chat_template.return_value = "rendered" + self.tokenizer_manager.tokenizer.encode.return_value = [11, 12] + self.tokenizer_manager.tokenizer.decode.return_value = "decoded" + self.tokenizer_manager.tokenizer.reset_mock() + + def _render(self, **overrides): + kwargs = { + "messages": [{"role": "user", "content": "same text prefix"}], + "tools": None, + "template_kwargs": {"enable_thinking": False}, + "encode_kwargs": {"add_special_tokens": False}, + "use_cache": True, + } + kwargs.update(overrides) + return self.chat._render_and_encode_chat_template(**kwargs) + + def test_cache_hit_reuses_render_encode_and_returns_an_owned_id_list(self): + first = self._render() + first[1].append(99) + second = self._render() + + self.assertEqual(second, ("rendered", [11, 12], "decoded")) + self.tokenizer_manager.tokenizer.apply_chat_template.assert_called_once() + self.tokenizer_manager.tokenizer.encode.assert_called_once() + self.tokenizer_manager.tokenizer.decode.assert_called_once() + + def test_cache_key_includes_template_and_encode_options(self): + self._render() + self._render(template_kwargs={"enable_thinking": True}) + self._render(encode_kwargs={"add_special_tokens": True}) + + self.assertEqual( + self.tokenizer_manager.tokenizer.apply_chat_template.call_count, + 3, + ) + self.assertEqual(self.tokenizer_manager.tokenizer.encode.call_count, 3) + + def test_cache_key_tracks_tokenizer_chat_template_updates(self): + self.tokenizer_manager.tokenizer.chat_template = "template-v1" + self._render() + self.tokenizer_manager.tokenizer.chat_template = "template-v2" + self._render() + + self.assertEqual( + self.tokenizer_manager.tokenizer.apply_chat_template.call_count, + 2, + ) + + def test_non_serializable_input_bypasses_cache(self): + messages = [{"role": "user", "content": object()}] + self._render(messages=messages) + self._render(messages=messages) + + self.assertEqual( + self.tokenizer_manager.tokenizer.apply_chat_template.call_count, + 2, + ) + self.assertEqual(self.tokenizer_manager.tokenizer.encode.call_count, 2) + self.tokenizer_manager.tokenizer.decode.assert_not_called() + + class ServingChatTestCase(unittest.TestCase): # ------------- common fixtures ------------- def setUp(self): @@ -386,9 +461,7 @@ class ServingChatTestCase(unittest.TestCase): ) self.tm.tokenizer.apply_chat_template.reset_mock() self.chat._apply_jinja_template(ordered_request, None, is_multimodal=True) - self.assertEqual( - rendered_messages, self.tm.tokenizer.apply_chat_template.call_args[0][0] - ) + self.tm.tokenizer.apply_chat_template.assert_not_called() self.template_manager.jinja_template_may_reorder_tool_results = False self.tm.tokenizer.apply_chat_template.reset_mock() diff --git a/test/registered/unit/mem_cache/test_radix_cache_unit.py b/test/registered/unit/mem_cache/test_radix_cache_unit.py index f635bf4f3..c72bb6bdf 100644 --- a/test/registered/unit/mem_cache/test_radix_cache_unit.py +++ b/test/registered/unit/mem_cache/test_radix_cache_unit.py @@ -50,6 +50,7 @@ from sglang.srt.mem_cache.events import KVCacheEventRecorder from sglang.srt.mem_cache.mamba_radix_cache import TreeNode as MambaTreeNode from sglang.srt.mem_cache.radix_cache import RadixCache, RadixKey, TreeNode from sglang.srt.utils import get_device +from sglang.test.test_utils import CustomTestCase # Test constants DEFAULT_PAGE_SIZE = 4 @@ -368,7 +369,7 @@ class TestTreeNode(unittest.TestCase): self.assertEqual(n4.get_prefix_hash_values(n3), ["h1", "h2", "h3"]) -class TestRadixCache(unittest.TestCase): +class TestRadixCache(CustomTestCase): """Test cases for RadixCache class.""" def setUp(self): @@ -540,6 +541,52 @@ class TestRadixCache(unittest.TestCase): cache.req_to_token_pool.req_to_token[0], tree_indices ) + def test_finished_request_splits_prompt_from_output_for_eviction(self): + class ReqToTokenPool: + def __init__(self, row): + self.req_to_token = row.unsqueeze(0) + + allocator = TokenToKVPoolAllocator( + size=16, + dtype=torch.float16, + device="cpu", + kvcache=None, + need_sort=False, + ) + cache = RadixCache.create_simulated(mock_allocator=allocator) + prompt_ids = array("q", [1, 2, 3]) + output_ids = array("q", [4, 5]) + kv_indices = allocator.alloc(len(prompt_ids) + len(output_ids)) + self.assertIsNotNone(kv_indices) + cache.req_to_token_pool = ReqToTokenPool(kv_indices) + req = unittest.mock.Mock( + origin_input_ids=prompt_ids, + output_ids=output_ids, + kv=ReqKvInfo(req_pool_idx=0, cache_protected_len=0), + extra_key=None, + cache_salt=None, + priority=0, + last_node=cache.root_node, + ) + + cache.cache_finished_req( + req, + is_insert=True, + kv_len_to_handle=len(prompt_ids) + len(output_ids), + ) + + (prompt_node,) = cache.root_node.children.values() + (output_node,) = prompt_node.children.values() + self.assertEqual(len(prompt_node.key), len(prompt_ids)) + self.assertEqual(len(output_node.key), len(output_ids)) + + result = cache.evict(EvictParams(num_tokens=len(output_ids))) + self.assertEqual(result.num_tokens_evicted, len(output_ids)) + match = cache.match_prefix( + MatchPrefixParams(key=RadixKey(prompt_ids + output_ids)) + ) + self.assertEqual(len(match.device_indices), len(prompt_ids)) + def test_kv_cache_events(self): """Test KV cache events functionality.""" test_cases = [ diff --git a/test/registered/unit/mem_cache/test_unified_radix_cache_unittest.py b/test/registered/unit/mem_cache/test_unified_radix_cache_unittest.py index 71c8d7e47..61a597f05 100644 --- a/test/registered/unit/mem_cache/test_unified_radix_cache_unittest.py +++ b/test/registered/unit/mem_cache/test_unified_radix_cache_unittest.py @@ -1558,6 +1558,28 @@ class UnifiedRadixCacheSuite: MatchPrefixParams(key=RadixKey(array("q", all_ids[:aligned_len]))) ) self.assertEqual(len(m.device_indices), aligned_len) + + prompt_aligned_len = (len(input_ids) // ps) * ps + if self.cfg.components == (ComponentType.FULL,): + (prompt_node,) = _node_children(cache, cache.root_node_handle()) + (output_node,) = _node_children(cache, prompt_node) + self.assertEqual(_node_key_length(cache, prompt_node), prompt_aligned_len) + self.assertEqual( + _node_key_length(cache, output_node), + aligned_len - prompt_aligned_len, + ) + + result = cache.evict( + EvictParams(num_tokens=aligned_len - prompt_aligned_len) + ) + self.assertEqual( + result.num_tokens_evicted, + aligned_len - prompt_aligned_len, + ) + prompt_only = cache.match_prefix( + MatchPrefixParams(key=RadixKey(array("q", all_ids[:aligned_len]))) + ) + self.assertEqual(len(prompt_only.device_indices), prompt_aligned_len) cache.sanity_check() def test_cache_finished_req_strips_thinking(self): diff --git a/test/registered/unit/models/test_qwen3_vl_feature_materialization.py b/test/registered/unit/models/test_qwen3_vl_feature_materialization.py index c62ff1983..34262c1ad 100644 --- a/test/registered/unit/models/test_qwen3_vl_feature_materialization.py +++ b/test/registered/unit/models/test_qwen3_vl_feature_materialization.py @@ -1,16 +1,24 @@ """Regression tests for Qwen3-VL multimodal feature materialization.""" import unittest +from contextlib import nullcontext from types import SimpleNamespace from unittest.mock import Mock, patch import torch -from sglang.srt.managers.schedule_batch import Modality, MultimodalDataItem +from sglang.srt.managers.schedule_batch import ( + Modality, + MultimodalDataItem, + MultimodalInputs, +) from sglang.srt.models.qwen3_vl import Qwen3VLForConditionalGeneration from sglang.srt.multimodal.processors.qwen_vl import QwenVLImageProcessor from sglang.srt.multimodal.transport.cuda_ipc import ( + BORROW_CUDA_IPC_FEATURE_KEY, + CUDA_IPC_FEATURE_COPY_EVENT_KEY, DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY, + CudaIpcTensorTransportProxy, ) from sglang.srt.runtime_context import get_context from sglang.test.ci.ci_register import register_cpu_ci @@ -69,7 +77,7 @@ class TestQwen3VLFeatureMaterialization(CustomTestCase): MultimodalDataItem(modality=Modality.AUDIO), ] - processor._mark_dp_encoder_features_for_deferred_reconstruction(items) + processor._mark_cuda_ipc_features_for_deferred_reconstruction(items) self.assertTrue( items[0].model_specific_data[ @@ -87,19 +95,104 @@ class TestQwen3VLFeatureMaterialization(CustomTestCase): ) def test_processor_does_not_defer_cpu_transport(self): - processor = QwenVLImageProcessor.__new__(QwenVLImageProcessor) - processor.mm_feature_transport = "cpu" - processor.server_args = SimpleNamespace(mm_enable_dp_encoder=True) - processor.model_type = "qwen3_vl" - item = MultimodalDataItem(modality=Modality.IMAGE) + with get_context().override_server_args(mm_enable_dp_encoder=True): + processor = QwenVLImageProcessor.__new__(QwenVLImageProcessor) + processor.mm_feature_transport = "cpu" + processor.model_type = "qwen3_vl" + item = MultimodalDataItem(modality=Modality.IMAGE) - processor._mark_dp_encoder_features_for_deferred_reconstruction([item]) + processor._mark_cuda_ipc_features_for_deferred_reconstruction([item]) self.assertNotIn( DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY, item.model_specific_data, ) + def test_processor_defers_cuda_ipc_for_single_tp_qwen3_vl(self): + processor = QwenVLImageProcessor.__new__(QwenVLImageProcessor) + processor.mm_feature_transport = "cuda_ipc" + processor.model_type = "qwen3_vl" + item = MultimodalDataItem(modality=Modality.IMAGE) + + processor._mark_cuda_ipc_features_for_deferred_reconstruction([item]) + + self.assertTrue( + item.model_specific_data[DEFER_CUDA_IPC_FEATURE_RECONSTRUCTION_KEY] + ) + + def test_retract_reprefill_waits_for_preserved_visual_input(self): + visual = Mock() + visual.device = torch.device("cuda:0") + visual.dtype = torch.bfloat16 + visual.side_effect = lambda pixel_values, *, grid_thw: pixel_values + model = self._model(visual, use_data_parallel=False) + + proxy = CudaIpcTensorTransportProxy.__new__(CudaIpcTensorTransportProxy) + proxy.total_consumer_count = 1 + borrowed_feature = torch.ones(2, 3) + packed_ready = Mock() + host_ready = Mock() + current_stream = Mock() + copy_stream = Mock() + proxy.reconstruct_on_target_device = Mock() + proxy.borrow_on_target_device = Mock(return_value=borrowed_feature) + proxy.release_borrowed_on_current_stream = Mock() + proxy.release_without_reconstruction = Mock() + item = MultimodalDataItem( + modality=Modality.IMAGE, + feature=proxy, + model_specific_data={BORROW_CUDA_IPC_FEATURE_KEY: True}, + ) + item.image_grid_thw = torch.tensor([[1, 1, 2]]) + + with ( + patch( + "sglang.srt.models.qwen3_vl.get_parallel", + return_value=SimpleNamespace(tp_size=1), + ), + patch( + "sglang.srt.models.qwen3_vl.materialize_multimodal_features", + side_effect=lambda features, **_kwargs: torch.cat(features), + ), + patch( + "sglang.srt.models.qwen3_vl.torch.cuda.current_stream", + return_value=current_stream, + ), + patch( + "sglang.srt.models.qwen3_vl.torch.cuda.Event", + side_effect=(packed_ready, host_ready), + ), + patch( + "sglang.srt.models.qwen3_vl.torch.cuda.Stream", + return_value=copy_stream, + ), + patch( + "sglang.srt.models.qwen3_vl.torch.cuda.stream", + return_value=nullcontext(), + ), + ): + first = model.get_image_feature([item]) + second = model.get_image_feature([item]) + + self.assertIsNot(item.feature, borrowed_feature) + self.assertTrue(torch.equal(item.feature, borrowed_feature)) + self.assertTrue(torch.equal(first, second)) + proxy.reconstruct_on_target_device.assert_not_called() + proxy.borrow_on_target_device.assert_called_once_with(0) + proxy.release_borrowed_on_current_stream.assert_called_once_with() + proxy.release_without_reconstruction.assert_not_called() + packed_ready.record.assert_called_once_with(current_stream) + copy_stream.wait_event.assert_called_once_with(packed_ready) + host_ready.record.assert_called_once_with(copy_stream) + current_stream.wait_event.assert_called_once_with(host_ready) + self.assertEqual(visual.call_count, 2) + + MultimodalInputs(mm_items=[item]).release_features() + + proxy.release_without_reconstruction.assert_not_called() + self.assertIsNone(item.feature) + self.assertNotIn(CUDA_IPC_FEATURE_COPY_EVENT_KEY, item.model_specific_data) + def test_image_features_are_packed_on_the_visual_device(self): visual = _RecordingVisual() model = self._model(visual, use_data_parallel=False) diff --git a/test/registered/unit/multimodal/rust/qwen/_fixtures.py b/test/registered/unit/multimodal/rust/qwen/_fixtures.py index 56d6a5cb8..aa6e4f101 100644 --- a/test/registered/unit/multimodal/rust/qwen/_fixtures.py +++ b/test/registered/unit/multimodal/rust/qwen/_fixtures.py @@ -71,6 +71,7 @@ def make_processor(case, config, image_processor_cls=None): model_impl="sglang", keep_mm_feature_on_device=False, mm_feature_transport="cpu", + mm_enable_dp_encoder=False, image_processor_backend="auto", disable_fast_image_processor=True, skip_tokenizer_init=False, diff --git a/test/registered/unit/multimodal/test_cuda_ipc_transport.py b/test/registered/unit/multimodal/test_cuda_ipc_transport.py index 80bf3572e..e9a9ccebf 100644 --- a/test/registered/unit/multimodal/test_cuda_ipc_transport.py +++ b/test/registered/unit/multimodal/test_cuda_ipc_transport.py @@ -128,6 +128,61 @@ class TestCudaIpcTransport(CustomTestCase): producer.join(timeout=10) self.assertEqual(producer.exitcode, 0) + def test_borrowed_tensor_keeps_lease_until_explicit_release(self): + ctx = mp.get_context("spawn") + proxy_queue = ctx.Queue() + producer_results = ctx.Queue() + consumer_done = ctx.Event() + producer = ctx.Process( + target=_produce_pooled_tensor, + args=(proxy_queue, consumer_done, producer_results), + ) + producer.start() + proxy = borrowed = consumed = None + producer_result = None + try: + try: + proxy, expected = proxy_queue.get(timeout=60) + except queue.Empty: + producer_result = producer_results.get(timeout=5) + _status, payload = producer_result + self.fail( + f"CUDA IPC producer failed before sending its proxy: {payload}" + ) + + borrowed = proxy.borrow_on_target_device(0) + self.assertIsNotNone(borrowed) + consumed = borrowed + 1 + torch.cuda.synchronize() + + self.assertFalse(proxy._consumer_acknowledged) + proxy.release_without_reconstruction() + torch.cuda.synchronize() + + self.assertEqual( + consumed.cpu().tolist(), + (torch.tensor(expected) + 1).tolist(), + ) + self.assertTrue(proxy._consumer_acknowledged) + self.assertIsNone(proxy._borrowed_storage) + finally: + del consumed, borrowed, proxy + _pool_handle_cache_clear() + gc.collect() + torch.cuda.ipc_collect() + consumer_done.set() + producer.join(timeout=60) + try: + if producer_result is None: + producer_result = producer_results.get(timeout=5) + status, payload = producer_result + self.assertEqual(status, "ok", payload) + finally: + if producer.is_alive(): + producer.terminate() + producer.join(timeout=10) + self.assertEqual(producer.exitcode, 0) + def test_failed_reconstruction_releases_pooled_tensor(self): ctx = mp.get_context("spawn") proxy_queue = ctx.Queue() diff --git a/test/registered/unit/multimodal/test_media_artifact_processor.py b/test/registered/unit/multimodal/test_media_artifact_processor.py index 39a6c8054..be4d8da6d 100644 --- a/test/registered/unit/multimodal/test_media_artifact_processor.py +++ b/test/registered/unit/multimodal/test_media_artifact_processor.py @@ -15,6 +15,7 @@ from sglang.srt.multimodal.media_artifacts import ( MediaArtifactInput, ) from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase register_cpu_ci(est_time=11, suite="base-a-test-cpu") @@ -88,7 +89,7 @@ class _Processor(MediaArtifactCacheMixin): self.io_executor.shutdown() -class TestMediaArtifactProcessor(unittest.TestCase): +class TestMediaArtifactProcessor(CustomTestCase): def test_default_image_decoder_rejects_lazy_pil_failure(self): malformed_png = base64.b64decode( "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVQIHWP4z8DwHwAFgAI/ScLJSwAAAABJRU5ErkJggg==" @@ -253,6 +254,39 @@ class TestMediaArtifactProcessor(unittest.TestCase): self.assertEqual(artifacts[0].feature, b"fresh") self.assertEqual(len(processor.batches), 1) + def test_without_cache_builds_request_local_artifacts_and_does_not_retain(self): + processor = _Processor() + digest = snapshot_media(b"image").content_digest + key = processor._artifact_key(digest, b"image") + cached = _Artifact(digest, key, 1, b"cached") + processor.mm_preprocess_cache.put(key, cached) + + try: + artifacts = asyncio.run( + processor.prepare_media_artifacts_without_cache([b"image", b"image"]) + ) + cached_after = asyncio.run(processor.prepare_media_artifacts([b"image"])) + finally: + processor.close() + + self.assertEqual([artifact.feature for artifact in artifacts], [b"image"] * 2) + self.assertEqual(len(processor.batches), 1) + self.assertEqual(len(processor.batches[0]), 2) + self.assertIs(cached_after[0], cached) + + def test_without_cache_still_validates_caller_content_hash(self): + processor = _Processor() + try: + with self.assertRaisesRegex(ValueError, "content hash mismatch"): + asyncio.run( + processor.prepare_media_artifacts_without_cache( + [b"image"], + content_hashes=[snapshot_media(b"different").content_digest], + ) + ) + finally: + processor.close() + if __name__ == "__main__": unittest.main() diff --git a/test/registered/unit/server_args/test_server_args.py b/test/registered/unit/server_args/test_server_args.py index cbaa1fc9a..ac88a21c1 100644 --- a/test/registered/unit/server_args/test_server_args.py +++ b/test/registered/unit/server_args/test_server_args.py @@ -111,6 +111,25 @@ _mock_device.start() class TestPrepareServerArgs(CustomTestCase): + def test_radix_eviction_policy_explicitness_is_preserved(self): + omitted = prepare_server_args(["--model-path", "dummy"]) + separated = prepare_server_args( + ["--model-path", "dummy", "--radix-eviction-policy", "lru"] + ) + joined = prepare_server_args( + ["--model-path", "dummy", "--radix-eviction-policy=lru"] + ) + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: + f.write("model-path: dummy\nradix-eviction-policy: lru\n") + config_path = f.name + self.addCleanup(os.unlink, config_path) + configured = prepare_server_args(["--config", config_path]) + + self.assertFalse(omitted._radix_eviction_policy_explicitly_set) + self.assertTrue(separated._radix_eviction_policy_explicitly_set) + self.assertTrue(joined._radix_eviction_policy_explicitly_set) + self.assertTrue(configured._radix_eviction_policy_explicitly_set) + def test_ple_embedding_offload_rejects_generic_weight_offload(self): for generic_offload in ( {"cpu_offload_gb": 1}, diff --git a/test/registered/unit/test_model_overrides.py b/test/registered/unit/test_model_overrides.py index 8e0933311..4dc5b5a3d 100644 --- a/test/registered/unit/test_model_overrides.py +++ b/test/registered/unit/test_model_overrides.py @@ -20,6 +20,7 @@ from sglang.srt.arg_groups import overrides as overrides_module from sglang.srt.arg_groups.arg_utils import A, Arg, resolvable_fields from sglang.srt.arg_groups.model_overrides import minicpm as minicpm_module from sglang.srt.arg_groups.model_overrides import qwen3_5 as qwen3_5_module +from sglang.srt.arg_groups.model_overrides import qwen3_vl as qwen3_vl_module from sglang.srt.arg_groups.overrides import ( collect_model_override_declarations, register_model_override, @@ -108,6 +109,10 @@ class TestModelOverridableWhitelist(CustomTestCase): "enable_symm_mem", "speculative_attention_mode", "speculative_draft_attention_backend", + "prefill_decode_interval", + "radix_eviction_policy", + "mm_preprocess_cache_size_mb", + "mm_feature_transport", } ), ) @@ -3107,5 +3112,107 @@ class TestDeclarationValidation(CustomTestCase): validate_declarations(args, [("src", {"nope": 1})]) +class TestQwen3VLHopperServingOverrides(CustomTestCase): + def setUp(self): + super().setUp() + self.addCleanup(envs.SGLANG_VLM_CACHE_SIZE_MB.clear) + self.addCleanup(envs.SGLANG_MM_FEATURE_CACHE_MB.clear) + envs.SGLANG_VLM_CACHE_SIZE_MB.clear() + envs.SGLANG_MM_FEATURE_CACHE_MB.clear() + + @staticmethod + def _args(**overrides): + from sglang.srt.server_args import ServerArgs + + values = { + "mm_preprocess_cache_size_mb": None, + "mm_feature_transport": None, + "max_running_requests": 400, + "radix_eviction_policy": "lru", + "prefill_decode_interval": None, + "attention_backend": None, + "decode_attention_backend": None, + } + values.update(overrides) + return ServerArgs(model_path="dummy", **values) + + @patch.object( + qwen3_vl_module, + "large_hopper_qwen3_vl_model_type", + return_value="qwen3_vl", + ) + def test_profiled_defaults_are_valid_model_overrides(self, _mock_model_type): + server_args = self._args() + updates = qwen3_vl_module._qwen3vl_hopper_serving_overrides(server_args, None) + + self.assertEqual( + updates, + { + "mm_preprocess_cache_size_mb": 0, + "mm_feature_transport": "cuda_ipc", + "radix_eviction_policy": "priority", + "prefill_decode_interval": 22, + "decode_attention_backend": "flashinfer", + }, + ) + validate_declarations( + server_args, + [("_qwen3vl_hopper_serving_overrides", updates)], + ) + self.assertEqual(envs.SGLANG_VLM_CACHE_SIZE_MB.get(), 0) + self.assertEqual(envs.SGLANG_MM_FEATURE_CACHE_MB.get(), 3 * 1024) + + @patch.object( + qwen3_vl_module, + "large_hopper_qwen3_vl_model_type", + return_value="qwen3_vl", + ) + def test_multinode_does_not_auto_select_cuda_ipc(self, _mock_model_type): + updates = qwen3_vl_module._qwen3vl_hopper_serving_overrides( + self._args(nnodes=2), None + ) + + self.assertNotIn("mm_feature_transport", updates) + self.assertFalse(envs.SGLANG_MM_FEATURE_CACHE_MB.is_set()) + + @patch.object( + qwen3_vl_module, + "large_hopper_qwen3_vl_model_type", + side_effect=AssertionError("must not load model config without GPU memory"), + ) + def test_decode_graph_expansion_skips_unknown_gpu_memory(self, _mock_model_type): + decode_config = SimpleNamespace(max_bs=256) + + qwen3_vl_module.expand_multimodal_decode_graph_to_running_limit( + self._args(), decode_config, gpu_mem=None + ) + + self.assertEqual(decode_config.max_bs, 256) + + @patch.object( + qwen3_vl_module, + "large_hopper_qwen3_vl_model_type", + return_value="qwen3_vl", + ) + def test_explicit_choices_are_not_replaced(self, _mock_model_type): + envs.SGLANG_VLM_CACHE_SIZE_MB.set(512) + envs.SGLANG_MM_FEATURE_CACHE_MB.set(2048) + updates = qwen3_vl_module._qwen3vl_hopper_serving_overrides( + self._args( + mm_preprocess_cache_size_mb=256, + mm_feature_transport="cpu", + radix_eviction_policy="lru", + _radix_eviction_policy_explicitly_set=True, + prefill_decode_interval=0, + decode_attention_backend="fa3", + ), + None, + ) + + self.assertEqual(updates, {}) + self.assertEqual(envs.SGLANG_VLM_CACHE_SIZE_MB.get(), 512) + self.assertEqual(envs.SGLANG_MM_FEATURE_CACHE_MB.get(), 2048) + + if __name__ == "__main__": unittest.main()