diff --git a/python/sglang/srt/layers/moe/fused_moe_triton/fused_moe.py b/python/sglang/srt/layers/moe/fused_moe_triton/fused_moe.py index aa38b8d74..fa809e982 100644 --- a/python/sglang/srt/layers/moe/fused_moe_triton/fused_moe.py +++ b/python/sglang/srt/layers/moe/fused_moe_triton/fused_moe.py @@ -555,7 +555,7 @@ def fused_experts_impl( gemm1_alpha, gemm1_limit, ) - elif _is_hip or _is_cuda: + elif _is_cuda or _is_hip: if not filter_expert: silu_and_mul(intermediate_cache1.view(-1, N), intermediate_cache2) else: @@ -575,7 +575,7 @@ def fused_experts_impl( elif activation == "gelu" and is_gated: assert gemm1_alpha is None, "gemm1_alpha is not supported for gelu" assert gemm1_limit is None, "gemm1_limit is not supported for gelu" - if _is_hip or _is_cuda: + if _is_cuda or _is_hip: if not filter_expert: gelu_and_mul(intermediate_cache1.view(-1, N), intermediate_cache2) else: diff --git a/python/sglang/srt/layers/moe/fused_moe_triton/fused_moe_triton_kernels.py b/python/sglang/srt/layers/moe/fused_moe_triton/fused_moe_triton_kernels.py index 8ce947d43..5afba6c01 100644 --- a/python/sglang/srt/layers/moe/fused_moe_triton/fused_moe_triton_kernels.py +++ b/python/sglang/srt/layers/moe/fused_moe_triton/fused_moe_triton_kernels.py @@ -810,9 +810,11 @@ def _apply_activation(x, ACTIVATION_TYPE: tl.constexpr): x = x.to(tl.float32) if ACTIVATION_TYPE == "silu": return x * tl.sigmoid(x) - else: + elif ACTIVATION_TYPE == "gelu": kAlpha = 0.7978845608028654 return 0.5 * x * (1 + tanh(kAlpha * (x + 0.044715 * x * x * x))) + else: + raise ValueError(f"Unsupported activation: {ACTIVATION_TYPE}") @triton.jit diff --git a/python/sglang/srt/managers/data_parallel_controller.py b/python/sglang/srt/managers/data_parallel_controller.py index de79ec056..d4967da1a 100644 --- a/python/sglang/srt/managers/data_parallel_controller.py +++ b/python/sglang/srt/managers/data_parallel_controller.py @@ -200,7 +200,7 @@ class DataParallelController: self.init_dispatcher() - self.watchdog = Watchdog.create( + self.soft_watchdog = Watchdog.create( debug_name="DataParallelController", watchdog_timeout=server_args.soft_watchdog_timeout, soft=True, @@ -564,7 +564,7 @@ class DataParallelController: def event_loop(self): while True: while True: - self.watchdog.feed() + self.soft_watchdog.feed() try: recv_req = self.recv_from_tokenizer.recv_pyobj(zmq.NOBLOCK) except zmq.ZMQError: diff --git a/python/sglang/srt/managers/detokenizer_manager.py b/python/sglang/srt/managers/detokenizer_manager.py index 6eb67435a..bd08fce53 100644 --- a/python/sglang/srt/managers/detokenizer_manager.py +++ b/python/sglang/srt/managers/detokenizer_manager.py @@ -116,7 +116,7 @@ class DetokenizerManager(MultiHttpWorkerDetokenizerMixin): self.is_tool_call_parser_gpt_oss = server_args.tool_call_parser == "gpt-oss" self.disable_tokenizer_batch_decode = server_args.disable_tokenizer_batch_decode - self.watchdog = Watchdog.create( + self.soft_watchdog = Watchdog.create( debug_name="DetokenizerManager", watchdog_timeout=server_args.soft_watchdog_timeout, soft=True, @@ -136,12 +136,12 @@ class DetokenizerManager(MultiHttpWorkerDetokenizerMixin): def event_loop(self): """The event loop that handles requests""" while True: - with self.watchdog.disable(): + with self.soft_watchdog.disable(): recv_obj = self.recv_from_scheduler.recv_pyobj() output = self._request_dispatcher(recv_obj) if output is not None: self.send_to_tokenizer.send_pyobj(output) - self.watchdog.feed() + self.soft_watchdog.feed() def trim_matched_stop( self, output: Union[str, List[int]], finished_reason: Dict, no_stop_trim: bool diff --git a/python/sglang/srt/managers/scheduler_metrics_mixin.py b/python/sglang/srt/managers/scheduler_metrics_mixin.py index 58f11885f..d44ff6027 100644 --- a/python/sglang/srt/managers/scheduler_metrics_mixin.py +++ b/python/sglang/srt/managers/scheduler_metrics_mixin.py @@ -78,7 +78,13 @@ class SchedulerMetricsMixin: ) if self.enable_metrics: - engine_type = "unified" + if self.server_args.disaggregation_mode == DisaggregationMode.PREFILL: + engine_type = "prefill" + elif self.server_args.disaggregation_mode == DisaggregationMode.DECODE: + engine_type = "decode" + else: + engine_type = "unified" + labels = { "model_name": self.server_args.served_model_name, "engine_type": engine_type, diff --git a/python/sglang/srt/managers/tokenizer_manager.py b/python/sglang/srt/managers/tokenizer_manager.py index 5ebd1c207..62cde8d7f 100644 --- a/python/sglang/srt/managers/tokenizer_manager.py +++ b/python/sglang/srt/managers/tokenizer_manager.py @@ -237,6 +237,7 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi if speculative_algorithm.is_none() else server_args.speculative_num_draft_tokens ) + self.validate_total_tokens = True def init_tokenizer_and_processor(self): server_args = self.server_args @@ -424,7 +425,7 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi if self.server_args.gc_warning_threshold_secs > 0.0: configure_gc_warning(self.server_args.gc_warning_threshold_secs) - self.watchdog = Watchdog.create( + self.soft_watchdog = Watchdog.create( debug_name="TokenizerManager", watchdog_timeout=self.server_args.soft_watchdog_timeout, soft=True, @@ -723,9 +724,10 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi """Validates that the input token count and the requested token count doesn't exceed the model's context length.""" # FIXME: unify the length validation logic with the one in the scheduler. _max_req_len = self.context_len - input_token_num = len(input_ids) if input_ids is not None else 0 input_token_num += self.reserve_input_token_num + + # Validate input length if input_token_num >= self.context_len: if self.server_args.allow_auto_truncate: logger.warning( @@ -741,16 +743,11 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi f"model's context length ({self.context_len} tokens)." ) - if isinstance(obj, EmbeddingReqInput) and self.is_generation: - raise ValueError( - "This model does not appear to be an embedding model by default. " - "Please add `--is-embedding` when launching the server or try another model." - ) - - # Check total tokens (input + max_new_tokens) + # Validate total tokens (input + max_new_tokens) max_new_tokens = obj.sampling_params.get("max_new_tokens") if ( - max_new_tokens is not None + self.validate_total_tokens + and max_new_tokens is not None and (max_new_tokens + input_token_num) >= _max_req_len ): if self.server_args.allow_auto_truncate: @@ -773,10 +770,18 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi ) raise ValueError(error_msg) - # Matryoshka embeddings validations + # Validate embedding requests + if isinstance(obj, EmbeddingReqInput) and self.is_generation: + raise ValueError( + "This model does not appear to be an embedding model by default. " + "Please add `--is-embedding` when launching the server or try another model." + ) + + # Validate Matryoshka embeddings if isinstance(obj, EmbeddingReqInput): self._validate_for_matryoshka_dim(obj) + # Validate custom logit processor if isinstance(obj, GenerateReqInput): if ( obj.return_hidden_states @@ -839,12 +844,22 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi ) def _validate_input_ids_in_vocab( - self, input_ids: List[int], vocab_size: int + self, input_ids: Union[List[int], List[List[int]]], vocab_size: int ) -> None: - if any(id >= vocab_size for id in input_ids): - raise ValueError( - f"The input_ids {input_ids} contains values greater than the vocab size ({vocab_size})." - ) + # Handle both single sequence and batch of sequences + if isinstance(input_ids[0], list): + # Batch of sequences + for seq in input_ids: + if any(id >= vocab_size for id in seq): + raise ValueError( + f"The input_ids {seq} contains values greater than the vocab size ({vocab_size})." + ) + else: + # Single sequence + if any(id >= vocab_size for id in input_ids): + raise ValueError( + f"The input_ids {input_ids} contains values greater than the vocab size ({vocab_size})." + ) def _get_sampling_params(self, sampling_kwargs: Dict) -> SamplingParams: return SamplingParams(**sampling_kwargs) @@ -1420,11 +1435,11 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi async def handle_loop(self): """The event loop that handles requests""" while True: - with self.watchdog.disable(): + with self.soft_watchdog.disable(): recv_obj = await self.recv_from_detokenizer.recv_pyobj() self._result_dispatcher(recv_obj) self.last_receive_tstamp = time.time() - self.watchdog.feed() + self.soft_watchdog.feed() def _handle_batch_output( self, @@ -1819,6 +1834,14 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi ): meta_info[attr_name] = getattr(recv_obj, attr_name)[index] + def _request_has_grammar(self, obj: GenerateReqInput) -> bool: + return ( + obj.sampling_params.get("json_schema", None) + or obj.sampling_params.get("regex", None) + or obj.sampling_params.get("ebnf", None) + or obj.sampling_params.get("structural_tag", None) + ) + def collect_metrics(self, state: ReqState, recv_obj: BatchStrOutput, i: int): completion_tokens = ( recv_obj.completion_tokens[i] @@ -1856,13 +1879,6 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi state.last_completion_tokens = completion_tokens if state.finished: - has_grammar = ( - state.obj.sampling_params.get("json_schema", None) - or state.obj.sampling_params.get("regex", None) - or state.obj.sampling_params.get("ebnf", None) - or state.obj.sampling_params.get("structural_tag", None) - ) - retraction_count = ( recv_obj.retraction_counts[i] if getattr(recv_obj, "retraction_counts", None) @@ -1876,7 +1892,7 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi completion_tokens, recv_obj.cached_tokens[i], state.finished_time - state.created_time, - has_grammar, + self._request_has_grammar(state.obj), retraction_count, ) diff --git a/python/sglang/srt/model_loader/ci_weight_validation.py b/python/sglang/srt/model_loader/ci_weight_validation.py index 2e4ddb5dd..48693f6dc 100644 --- a/python/sglang/srt/model_loader/ci_weight_validation.py +++ b/python/sglang/srt/model_loader/ci_weight_validation.py @@ -511,8 +511,6 @@ def ci_download_with_validation_and_retry( kwargs["disable"] = True super().__init__(*args, **kwargs) - log_info_on_rank0(logger, f"Using model weights format {allow_patterns}") - # Retry loop for handling corrupted downloads for attempt in range(max_retries): hf_folder = snapshot_download( diff --git a/python/sglang/srt/model_loader/weight_utils.py b/python/sglang/srt/model_loader/weight_utils.py index 596c58ec0..78972f01f 100644 --- a/python/sglang/srt/model_loader/weight_utils.py +++ b/python/sglang/srt/model_loader/weight_utils.py @@ -40,32 +40,20 @@ from sglang.srt.layers.quantization.modelopt_quant import ( ModelOptFp4Config, ModelOptFp8Config, ) +from sglang.srt.model_loader.ci_weight_validation import ( + ci_download_with_validation_and_retry, + ci_validate_and_cleanup_local_snapshot, +) from sglang.srt.utils import find_local_repo_dir, log_info_on_rank0, print_warning_once from sglang.utils import is_in_ci try: from fastsafetensors import SafeTensorsFileLoader, SingleGroup -except ImportError: - - class PlaceholderModule: - def __init__(self, name): - self.name = name - - def __getattr__(self, name): - raise ImportError(f"Please install {self.name}") - - fastsafetensors = PlaceholderModule("fastsafetensors") - SafeTensorsFileLoader = None - SingleGroup = None +except ImportError as e: + SafeTensorsFileLoader = SingleGroup = None logger = logging.getLogger(__name__) -# use system-level temp directory for file locks, so that multiple users -# can share the same lock without error. -# lock files in the temp directory will be automatically deleted when the -# system reboots, so users will not complain about annoying lock files -temp_dir = tempfile.gettempdir() - def enable_hf_transfer(): """automatically activates hf_transfer""" @@ -82,10 +70,11 @@ def enable_hf_transfer(): enable_hf_transfer() -class DisabledTqdm(tqdm): - def __init__(self, *args, **kwargs): - kwargs["disable"] = True - super().__init__(*args, **kwargs) +# use system-level temp directory for file locks, so that multiple users +# can share the same lock without error. +# lock files in the temp directory will be automatically deleted when the +# system reboots, so users will not complain about annoying lock files +temp_dir = tempfile.gettempdir() def get_lock( @@ -169,6 +158,12 @@ def replace_substrings(key: str, substring_mapping: dict[str, str]) -> str: return key +class DisabledTqdm(tqdm): + def __init__(self, *args, **kwargs): + kwargs["disable"] = True + super().__init__(*args, **kwargs) + + # TODO(woosuk): Move this to other place. def get_quant_config( model_config: ModelConfig, @@ -194,6 +189,7 @@ def get_quant_config( if hf_quant_config is not None: hf_quant_config["packed_modules_mapping"] = packed_modules_mapping return quant_cls.from_config(hf_quant_config) + # In case of bitsandbytes/QLoRA, get quant config from the adapter model. if model_config.quantization == "bitsandbytes": if ( @@ -204,9 +200,9 @@ def get_quant_config( model_name_or_path = load_config.model_loader_extra_config[ "qlora_adapter_name_or_path" ] - else: model_name_or_path = model_config.model_path + is_local = os.path.isdir(model_name_or_path) if not is_local: # Download the config files. @@ -357,10 +353,6 @@ def _find_local_hf_snapshot_dir_unlocked( # Only perform cache validation and cleanup in CI to avoid # unnecessary overhead for regular users if is_in_ci() and local_weight_files: - from sglang.srt.model_loader.ci_weight_validation import ( - ci_validate_and_cleanup_local_snapshot, - ) - is_valid = ci_validate_and_cleanup_local_snapshot( model_name_or_path, found_local_snapshot_dir, local_weight_files ) @@ -443,23 +435,10 @@ def download_weights_from_hf( allow_patterns = [pattern] break - # Only perform validation and retry in CI to avoid overhead for regular users - if is_in_ci(): - from sglang.srt.model_loader.ci_weight_validation import ( - ci_download_with_validation_and_retry, - ) + log_info_on_rank0(logger, f"Using model weights format {allow_patterns}") - return ci_download_with_validation_and_retry( - model_name_or_path=model_name_or_path, - allow_patterns=allow_patterns, - ignore_patterns=ignore_patterns, - cache_dir=cache_dir, - revision=revision, - max_retries=max_retries, - ) - else: + if not is_in_ci(): # Simple download without validation for non-CI environments - log_info_on_rank0(logger, f"Using model weights format {allow_patterns}") hf_folder = snapshot_download( model_name_or_path, allow_patterns=allow_patterns, @@ -470,6 +449,16 @@ def download_weights_from_hf( local_files_only=huggingface_hub.constants.HF_HUB_OFFLINE, ) return hf_folder + else: + # Only perform validation and retry in CI to avoid overhead for regular users + return ci_download_with_validation_and_retry( + model_name_or_path=model_name_or_path, + allow_patterns=allow_patterns, + ignore_patterns=ignore_patterns, + cache_dir=cache_dir, + revision=revision, + max_retries=max_retries, + ) def download_safetensors_index_file_from_hf(