From 21ee597e4a56a3d2ab87b8bd008ccfe35c6e52d2 Mon Sep 17 00:00:00 2001 From: Hudson Xing <1277646412@qq.com> Date: Fri, 16 Jan 2026 12:15:33 +0800 Subject: [PATCH] =?UTF-8?q?ci:=20enable=20offline=20mode=20when=20local=20?= =?UTF-8?q?cache=20is=20complete=20to=20avoid=20HF=20Hub=20=E2=80=A6=20(#1?= =?UTF-8?q?6121)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- python/sglang/srt/configs/model_config.py | 69 +- .../srt/model_loader/ci_weight_validation.py | 1164 +++++++++++++++++ python/sglang/srt/model_loader/loader.py | 5 +- python/sglang/test/test_utils.py | 421 ++++-- scripts/ci/prepare_runner.sh | 5 + scripts/ci/prevalidate_cached_models.py | 407 ++++++ 6 files changed, 1959 insertions(+), 112 deletions(-) create mode 100755 scripts/ci/prevalidate_cached_models.py diff --git a/python/sglang/srt/configs/model_config.py b/python/sglang/srt/configs/model_config.py index 64b871485..cd62f3f76 100644 --- a/python/sglang/srt/configs/model_config.py +++ b/python/sglang/srt/configs/model_config.py @@ -590,39 +590,56 @@ class ModelConfig: hf_api = HfApi() try: - # Retry HF API call up to 3 times - file_exists = retry( - lambda: hf_api.file_exists( - self.model_path, "hf_quant_config.json" - ), - max_retry=2, - initial_delay=1.0, - max_delay=5.0, + # In offline mode, skip file_exists check to avoid OfflineModeIsEnabled error + # Instead, directly try to download/read from cache with local_files_only + file_exists = False # Initialize to avoid UnboundLocalError + if not huggingface_hub.constants.HF_HUB_OFFLINE: + # Online mode: check if file exists before attempting download (optimization) + file_exists = retry( + lambda: hf_api.file_exists( + self.model_path, "hf_quant_config.json" + ), + max_retry=2, + initial_delay=1.0, + max_delay=5.0, + ) + if not file_exists: + # File doesn't exist on hub, no need to try downloading + return quant_cfg # None + + # Download (online mode) or read from cache (offline mode) + if envs.SGLANG_USE_MODELSCOPE.get(): + quant_config_file = model_file_download( + model_id=self.model_path, + file_path="hf_quant_config.json", + revision=self.revision, + ) + else: + quant_config_file = hf_hub_download( + repo_id=self.model_path, + filename="hf_quant_config.json", + revision=self.revision, + local_files_only=huggingface_hub.constants.HF_HUB_OFFLINE, + ) + with open(quant_config_file) as f: + quant_config_dict = json.load(f) + quant_cfg = self._parse_modelopt_quant_config(quant_config_dict) + except huggingface_hub.errors.LocalEntryNotFoundError: + # Offline mode and file not in cache - this is normal for non-quantized models + logger.debug( + f"hf_quant_config.json not found in cache for {self.model_path} " + "(offline mode, normal for non-quantized models)" ) - if file_exists: - # Download and parse the quantization config for remote models - if envs.SGLANG_USE_MODELSCOPE.get(): - quant_config_file = model_file_download( - model_id=self.model_path, - file_path="hf_quant_config.json", - revision=self.revision, - ) - else: - quant_config_file = hf_hub_download( - repo_id=self.model_path, - filename="hf_quant_config.json", - revision=self.revision, - ) - with open(quant_config_file) as f: - quant_config_dict = json.load(f) - quant_cfg = self._parse_modelopt_quant_config(quant_config_dict) except huggingface_hub.errors.OfflineModeIsEnabled: + # Should not reach here after our changes, but keep for safety logger.warning( "Offline mode is enabled, skipping hf_quant_config.json check" ) except Exception as e: logger.warning( - f"Failed to check hf_quant_config.json: {self.model_path} {e}" + "Failed to load hf_quant_config.json for model %s: %s", + self.model_path, + e, ) elif os.path.exists(os.path.join(self.model_path, "hf_quant_config.json")): quant_config_file = os.path.join( diff --git a/python/sglang/srt/model_loader/ci_weight_validation.py b/python/sglang/srt/model_loader/ci_weight_validation.py index 48693f6dc..96054d5d3 100644 --- a/python/sglang/srt/model_loader/ci_weight_validation.py +++ b/python/sglang/srt/model_loader/ci_weight_validation.py @@ -7,6 +7,7 @@ These functions handle: - Checking for missing shards in sharded models - Cleaning up corrupted files (selective or full cache deletion) - Automatic retry logic for corrupted downloads +- Validating config/tokenizer files completeness to enable offline mode For regular users, weight_utils.py provides simple download functionality without the overhead of validation and automatic cleanup. The CI-specific behavior is @@ -14,11 +15,13 @@ gated by is_in_ci() checks in weight_utils.py. """ import glob as glob_module +import hashlib import json import logging import os import re import shutil +import tempfile from typing import List, Optional, Tuple import safetensors @@ -27,6 +30,1167 @@ from sglang.srt.utils import log_info_on_rank0 logger = logging.getLogger(__name__) +# Validation marker version - increment when validation logic changes +# v2: Added trust_remote_code module validation (modeling_*.py must exist in snapshot) +# v3: Added remote file existence checks for hf_quant_config.json +# v5: Invalidate all previous markers to force fresh validation +VALIDATION_MARKER_VERSION = "5" + + +def _remote_file_exists( + repo_id: str, filename: str, revision: Optional[str], allow_remote_check: bool +) -> Optional[bool]: + """ + Check if a file exists on Hugging Face Hub for a specific revision. + + Args: + repo_id: Repository ID (e.g., "meta-llama/Llama-2-7b-hf") + filename: File name to check (e.g., "hf_quant_config.json") + revision: Git revision (commit hash, branch, or tag). None means default branch. + allow_remote_check: Whether remote checks are allowed (e.g., CI validation phase) + + Returns: + True if file exists on hub, False if it doesn't exist, None if we cannot determine + (network error or remote check not allowed - be conservative and assume incomplete) + """ + if not allow_remote_check: + logger.debug( + "Remote check disabled for %s/%s, returning None (unknown)", + repo_id, + filename, + ) + return None + + try: + from huggingface_hub import HfApi + + api = HfApi() + exists = api.file_exists(repo_id=repo_id, filename=filename, revision=revision) + logger.debug( + "Remote file check: %s/%s (revision=%s) exists=%s", + repo_id, + filename, + revision or "default", + exists, + ) + return exists + except Exception as e: + # Network errors, auth issues, repo not found, etc. + # Return None (unknown) - caller will treat as optional + logger.debug( + "Failed to check remote file existence for %s/%s (revision=%s): %s. " + "Will treat as optional.", + repo_id, + filename, + revision or "default", + e, + ) + return None + + +def _get_validation_marker_path(snapshot_dir: str) -> Optional[str]: + """ + Get the path to validation marker file for a snapshot. + + Marker is stored in /tmp to avoid permission issues with HF cache directory. + Marker key is sha256(snapshot_dir) to avoid any collisions regardless of + model_name_or_path format. + + Args: + snapshot_dir: Path to snapshot directory + + Returns: + Path to marker file or None if snapshot_dir is invalid + """ + if not snapshot_dir or not os.path.isdir(snapshot_dir): + return None + + # Normalize path to avoid marker misses due to trailing slashes or symlinks + # realpath resolves symlinks, rstrip removes trailing slashes + normalized_dir = os.path.realpath(snapshot_dir).rstrip("/") + + # Use sha256 of normalized snapshot_dir path as unique key + # This avoids any collision issues with repo naming or snapshot hash reuse + dir_hash = hashlib.sha256(normalized_dir.encode("utf-8")).hexdigest()[:12] + + # Store in /tmp with directory hash + return f"/tmp/sglang_hf_validation_{dir_hash}.json" + + +def _get_per_run_marker_dir() -> str: + """ + Get the directory for per-run validation markers. + + These markers are specific to the current CI run and are not shared across + runners. They are stored in a temporary directory that is cleaned up after + the run completes. + + Returns: + Path to per-run marker directory + """ + # Prefer RUNNER_TEMP (GitHub Actions) or TMPDIR, fallback to /tmp + base_dir = os.environ.get("RUNNER_TEMP", os.environ.get("TMPDIR", "/tmp")) + marker_dir = os.path.join(base_dir, "sglang_ci_offline_markers") + os.makedirs(marker_dir, exist_ok=True) + return marker_dir + + +def _get_per_run_marker_path(snapshot_dir: str) -> Optional[str]: + """ + Get the path to per-run validation marker file for a snapshot. + + Per-run markers are specific to the current CI run and are not shared + across runners. This prevents cross-runner cache state pollution. + + Args: + snapshot_dir: Path to snapshot directory + + Returns: + Path to per-run marker file or None if snapshot_dir is invalid + """ + if not snapshot_dir or not os.path.isdir(snapshot_dir): + return None + + normalized_dir = os.path.realpath(snapshot_dir).rstrip("/") + dir_hash = hashlib.sha256(normalized_dir.encode("utf-8")).hexdigest()[:12] + + marker_dir = _get_per_run_marker_dir() + return os.path.join(marker_dir, f"{dir_hash}.json") + + +def _read_per_run_marker(snapshot_dir: str) -> Optional[dict]: + """ + Read per-run validation marker for a snapshot. + + Args: + snapshot_dir: Path to snapshot directory + + Returns: + Marker dict if exists and valid, None otherwise + """ + marker_path = _get_per_run_marker_path(snapshot_dir) + if not marker_path or not os.path.exists(marker_path): + return None + + try: + with open(marker_path, "r", encoding="utf-8") as f: + marker = json.load(f) + + # Validate marker structure + if not isinstance(marker, dict): + return None + + required_keys = ["timestamp", "model_id", "snapshot_hash", "validation_passed"] + if not all(k in marker for k in required_keys): + return None + + if marker.get("validation_passed") is not True: + return None + + return marker + + except Exception as e: + logger.debug("Failed to read per-run marker from %s: %s", marker_path, e) + return None + + +def _write_per_run_marker( + snapshot_dir: str, model_id: str, required_files: Optional[list] = None +) -> None: + """ + Write per-run validation marker for a snapshot. + + Args: + snapshot_dir: Path to snapshot directory + model_id: Model identifier + required_files: List of required files that were validated + """ + marker_path = _get_per_run_marker_path(snapshot_dir) + if not marker_path: + logger.debug("Cannot write per-run marker: invalid snapshot_dir") + return + + from datetime import datetime + + snapshot_hash = os.path.basename(snapshot_dir) + + marker = { + "timestamp": datetime.utcnow().isoformat() + "Z", + "model_id": model_id, + "snapshot_hash": snapshot_hash, + "validation_passed": True, + "required_files": required_files or [], + } + + try: + marker_dir = os.path.dirname(marker_path) + os.makedirs(marker_dir, exist_ok=True) + + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=marker_dir, + delete=False, + suffix=".tmp", + ) as f: + temp_path = f.name + json.dump(marker, f, indent=2) + + os.replace(temp_path, marker_path) + logger.debug("Wrote per-run marker to %s", marker_path) + except Exception as e: + logger.warning("Failed to write per-run marker to %s: %s", marker_path, e) + try: + if "temp_path" in locals() and os.path.exists(temp_path): + os.remove(temp_path) + except Exception: + pass + + +def _remove_per_run_marker(snapshot_dir: str) -> None: + """ + Remove per-run validation marker for a snapshot. + + Args: + snapshot_dir: Path to snapshot directory + """ + marker_path = _get_per_run_marker_path(snapshot_dir) + if marker_path and os.path.exists(marker_path): + try: + os.remove(marker_path) + logger.debug("Removed per-run marker: %s", marker_path) + except Exception as e: + logger.warning("Failed to remove per-run marker %s: %s", marker_path, e) + + +def _read_validation_marker(snapshot_dir: str) -> Optional[dict]: + """ + Read validation marker for a snapshot. + + Args: + snapshot_dir: Path to snapshot directory + + Returns: + Marker dict with keys: version, validated_at, validation_passed + None if marker doesn't exist or is invalid or validation_passed is not True + """ + marker_path = _get_validation_marker_path(snapshot_dir) + if not marker_path: + return None + + if not os.path.exists(marker_path): + return None + + try: + with open(marker_path, "r", encoding="utf-8") as f: + marker = json.load(f) + + # Validate marker structure + if not isinstance(marker, dict): + return None + + required_keys = ["version", "validated_at", "validation_passed"] + if not all(key in marker for key in required_keys): + return None + + # Check version match + if marker["version"] != VALIDATION_MARKER_VERSION: + logger.debug( + "Validation marker version mismatch: %s != %s, will re-validate", + marker["version"], + VALIDATION_MARKER_VERSION, + ) + return None + + # Explicitly check validation_passed is True (defensive check) + # Even though we only write markers on success, this guards against + # manual edits or future code changes + if marker.get("validation_passed") is not True: + logger.debug( + "Validation marker has validation_passed=%s, treating as invalid", + marker.get("validation_passed"), + ) + return None + + return marker + except (json.JSONDecodeError, OSError) as e: + logger.debug("Failed to read validation marker at %s: %s", marker_path, e) + return None + + +def _write_validation_marker(snapshot_dir: str, passed: bool) -> None: + """ + Write validation marker for a snapshot (atomic write). + + IMPORTANT: We only cache successful validations. Failed validations are NOT + cached to allow retry after files are downloaded. + + Args: + snapshot_dir: Path to snapshot directory + passed: Whether validation passed + """ + if not passed: + # Don't cache failures - allow retry on next launch + return + + marker_path = _get_validation_marker_path(snapshot_dir) + if not marker_path: + logger.debug("Cannot write marker: invalid snapshot_dir") + return + + from datetime import datetime + + marker = { + "version": VALIDATION_MARKER_VERSION, + "validated_at": datetime.utcnow().isoformat() + "Z", + "validation_passed": passed, + } + + try: + # Atomic write: write to temp file then os.replace + marker_dir = os.path.dirname(marker_path) + os.makedirs(marker_dir, exist_ok=True) + + with tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=marker_dir, + delete=False, + suffix=".tmp", + ) as f: + temp_path = f.name + json.dump(marker, f, indent=2) + + # Atomic replace (overwrites existing file if any) + os.replace(temp_path, marker_path) + logger.debug("Wrote validation marker to %s (passed=%s)", marker_path, passed) + except Exception as e: + logger.warning("Failed to write validation marker to %s: %s", marker_path, e) + # Clean up temp file if it exists + try: + if "temp_path" in locals() and os.path.exists(temp_path): + os.remove(temp_path) + except Exception: + pass + + +def _validate_json_file(file_path: str, file_name: str) -> bool: + """ + Validate that a JSON file exists, is non-empty, and can be parsed. + + Args: + file_path: Path to the JSON file + file_name: Name of the file (for logging) + + Returns: + True if the file is valid, False otherwise + """ + if not os.path.exists(file_path): + logger.debug("CI cache validation: %s not found at %s", file_name, file_path) + return False + + if not os.path.isfile(file_path): + logger.warning( + "CI cache validation: %s is not a file: %s", file_name, file_path + ) + return False + + # Check if file is non-empty + try: + file_size = os.path.getsize(file_path) + if file_size == 0: + logger.warning("CI cache validation: %s is empty: %s", file_name, file_path) + return False + except OSError as e: + logger.warning("CI cache validation: Cannot get size of %s: %s", file_name, e) + return False + + # Try to parse JSON + try: + with open(file_path, "r", encoding="utf-8") as f: + json.load(f) + return True + except json.JSONDecodeError as e: + logger.warning( + "CI cache validation: %s is not valid JSON: %s - %s", + file_name, + file_path, + e, + ) + return False + except Exception as e: + logger.warning( + "CI cache validation: Failed to read %s: %s - %s", + file_name, + file_path, + e, + ) + return False + + +def _validate_config_and_tokenizer_files( + snapshot_dir: str, + model_id: Optional[str] = None, + revision: Optional[str] = None, + allow_remote_check: bool = False, +) -> Tuple[bool, List[str]]: + """ + Validate that critical config and tokenizer files exist and are valid. + + This checks for: + - config.json (required) + - tokenizer_config.json (required) + - generation_config.json (optional but validated if present) + - hf_quant_config.json (conditionally required based on Hub) - for FP4/FP8/ModelOpt + - quantize_config.json / quant_config.json (optional but validated if present) - for AWQ/GPTQ + - params.json (optional but validated if present) - for Mistral native format + - preprocessor_config.json (optional but validated if present) - for vision models + - trust_remote_code dynamic modules (required if auto_map present in config.json) + - At least one tokenizer file: tokenizer.json, tokenizer.model, or tiktoken.model + + Args: + snapshot_dir: Path to the model snapshot directory + model_id: Model repository ID (e.g., "meta-llama/Llama-2-7b-hf"), used for remote checks + revision: Git revision (commit hash), used for remote checks + allow_remote_check: Whether to check Hub for file existence to determine requirements + + Returns: + Tuple of (is_valid, missing_files) + - is_valid: True if all required files are present and valid + - missing_files: List of missing or invalid file names + """ + missing_files = [] + + # Check required config files + required_files = [ + "config.json", + "tokenizer_config.json", + ] + + for file_name in required_files: + file_path = os.path.join(snapshot_dir, file_name) + if not _validate_json_file(file_path, file_name): + missing_files.append(file_name) + + # Check optional generation_config.json (validate if exists) + generation_config_path = os.path.join(snapshot_dir, "generation_config.json") + if os.path.exists(generation_config_path): + if not _validate_json_file(generation_config_path, "generation_config.json"): + missing_files.append("generation_config.json (exists but invalid)") + + # Check hf_quant_config.json with remote existence check + # This file is needed for quantized models (FP4/FP8/ModelOpt) + # Example: nvidia/Llama-3.1-8B-Instruct-FP8, nvidia/DeepSeek-V3-0324-FP4 + hf_quant_config_path = os.path.join(snapshot_dir, "hf_quant_config.json") + local_hf_quant_exists = os.path.exists(hf_quant_config_path) + + # Check if file exists on Hub for this revision + # Only do remote check if model_id looks like a HF repo_id (org/model format) + # Skip if it's a local path (absolute path or doesn't contain '/') + remote_hf_quant_exists = None + is_hf_repo = ( + model_id is not None + and "/" in model_id + and not os.path.isabs(model_id) + and not model_id.startswith("/") + ) + if is_hf_repo and allow_remote_check: + remote_hf_quant_exists = _remote_file_exists( + repo_id=model_id, + filename="hf_quant_config.json", + revision=revision, + allow_remote_check=allow_remote_check, + ) + + # Apply conditional requirement logic + if remote_hf_quant_exists is True: + # Hub has this file for this revision - it's REQUIRED + if not local_hf_quant_exists: + missing_files.append( + f"hf_quant_config.json (required: exists on Hub for revision {revision or 'default'} but missing locally)" + ) + log_info_on_rank0( + logger, + f"Hub has hf_quant_config.json for {model_id} revision {revision or 'default'} " + f"but local snapshot missing it. Cache incomplete, will not write marker.", + ) + elif not _validate_json_file(hf_quant_config_path, "hf_quant_config.json"): + missing_files.append("hf_quant_config.json (exists but invalid)") + elif remote_hf_quant_exists is False: + # Hub doesn't have this file - it's OPTIONAL + # Only validate if it happens to exist locally + if local_hf_quant_exists: + if not _validate_json_file(hf_quant_config_path, "hf_quant_config.json"): + missing_files.append("hf_quant_config.json (exists but invalid)") + else: + # remote_hf_quant_exists is None - unknown (network error or remote check disabled) + # Treat as OPTIONAL - only enforce when we can positively confirm Hub has it + if local_hf_quant_exists: + # Local file exists - validate it + if not _validate_json_file(hf_quant_config_path, "hf_quant_config.json"): + missing_files.append("hf_quant_config.json (exists but invalid)") + # If local file missing and remote unknown, just log it - don't block marker + logger.debug( + "Cannot verify hf_quant_config.json on Hub for %s (revision=%s), " + "treating as optional since remote status unknown", + model_id or "unknown", + revision or "default", + ) + + # Check optional quantize_config.json / quant_config.json (validate if exists) + # These files are needed for AWQ/GPTQ/AutoRound quantized models + # Example: TheBloke/Llama-2-7B-AWQ, casperhansen/vicuna-7b-v1.5-awq + for quant_config_name in ["quantize_config.json", "quant_config.json"]: + quant_config_path = os.path.join(snapshot_dir, quant_config_name) + if os.path.exists(quant_config_path): + if not _validate_json_file(quant_config_path, quant_config_name): + missing_files.append(f"{quant_config_name} (exists but invalid)") + break # Only need to check one of these + + # Check optional params.json (validate if exists) + # This file is needed for Mistral native format models + # Example: mistralai/Mistral-7B-v0.1 + params_json_path = os.path.join(snapshot_dir, "params.json") + if os.path.exists(params_json_path): + if not _validate_json_file(params_json_path, "params.json"): + missing_files.append("params.json (exists but invalid)") + + # Check optional preprocessor_config.json (validate if exists) + # This file is needed for vision/multimodal models + # Example: llava-hf/llava-1.5-7b-hf, Qwen/Qwen2-VL-7B-Instruct + preprocessor_config_path = os.path.join(snapshot_dir, "preprocessor_config.json") + if os.path.exists(preprocessor_config_path): + if not _validate_json_file( + preprocessor_config_path, "preprocessor_config.json" + ): + missing_files.append("preprocessor_config.json (exists but invalid)") + + # Check for trust_remote_code dynamic module files if needed + # When auto_map exists in config.json, the model requires custom Python files + # These files must be present for offline mode to work + config_path = os.path.join(snapshot_dir, "config.json") + if os.path.exists(config_path): + try: + with open(config_path, "r", encoding="utf-8") as f: + config = json.load(f) + + auto_map = config.get("auto_map", {}) + if auto_map and isinstance(auto_map, dict): + # Extract Python module files from auto_map + # auto_map format: {"AutoConfig": "configuration_xxx.ConfigClass", ...} + # We need to check if the .py files exist + custom_files = set() + for key, value in auto_map.items(): + if isinstance(value, str) and "." in value: + # Extract module name (e.g., "configuration_xxx" from "configuration_xxx.ConfigClass") + module_name = value.split(".")[0] + custom_files.add(f"{module_name}.py") + + # Check if all custom files exist in snapshot directory + # NOTE: Some models (like nvidia/DeepSeek-V3-0324-FP4) have auto_map + # but don't include modeling_*.py in their repo, relying on transformers + # to fetch it from the base model. We MUST mark these as missing to + # prevent offline mode, which would fail to load the dynamic modules. + for custom_file in custom_files: + custom_file_path = os.path.join(snapshot_dir, custom_file) + if not os.path.exists(custom_file_path): + missing_files.append( + f"{custom_file} (required for trust_remote_code)" + ) + logger.debug( + f"Custom module file not in snapshot: {custom_file} for {snapshot_dir}" + ) + elif not os.path.isfile(custom_file_path): + missing_files.append(f"{custom_file} (exists but not a file)") + except (json.JSONDecodeError, OSError, KeyError) as e: + # If we can't read config.json, it will be caught by earlier validation + logger.debug("Failed to check auto_map in config.json: %s", e) + + # Check for at least one tokenizer file + tokenizer_files = [ + "tokenizer.json", + "tokenizer.model", + "tiktoken.model", + ] + + tokenizer_found = False + for tokenizer_file in tokenizer_files: + tokenizer_path = os.path.join(snapshot_dir, tokenizer_file) + if os.path.exists(tokenizer_path) and os.path.isfile(tokenizer_path): + # For tokenizer.json, validate it's proper JSON + if tokenizer_file == "tokenizer.json": + if _validate_json_file(tokenizer_path, tokenizer_file): + tokenizer_found = True + break + else: + # For .model files, just check they're non-empty + try: + if os.path.getsize(tokenizer_path) > 0: + tokenizer_found = True + break + except OSError: + pass + + if not tokenizer_found: + missing_files.append("tokenizer file") + + is_valid = len(missing_files) == 0 + return is_valid, missing_files + + +def ci_validate_cache_and_enable_offline_if_complete( + snapshot_dir: str, + weight_files: List[str], + model_name_or_path: str, +) -> bool: + """ + Validate local cache completeness (config/tokenizer/weights) and determine + if offline mode can be safely enabled. + + This function uses a snapshot-level marker to cache validation results, + so the heavy validation is done at most once per snapshot per runner. + + This function checks: + 1. Validation marker (if exists and version matches, skip re-validation) + 2. Config and tokenizer files (config.json, tokenizer_config.json, etc.) + 3. Weight files (safetensors shards, index files, corruption check) + + If all are present and valid, it returns True to signal that offline + mode can be safely enabled. + + IMPORTANT: This should be called BEFORE any HF operations, and if it + returns True, the caller should set HF_HUB_OFFLINE=1 for the server + subprocess env ONLY (not global environment). + + Args: + snapshot_dir: Path to the model snapshot directory + weight_files: List of weight file paths to validate (must be non-empty) + model_name_or_path: Model identifier for logging + + Returns: + True if cache is complete and offline mode can be enabled, False otherwise + """ + # Guard: weight_files is required + if not weight_files: + log_info_on_rank0( + logger, + f"CI_OFFLINE: No weight files provided, skip offline, keep online allowed - {model_name_or_path}", + ) + return False + + # Fast-path: Check if validation marker exists and is valid + # We only cache successful validations, so if marker exists, it means cache is complete + marker = _read_validation_marker(snapshot_dir) + if marker is not None: + marker_path = _get_validation_marker_path(snapshot_dir) + marker_name = os.path.basename(marker_path) if marker_path else "unknown" + log_info_on_rank0( + logger, + f"CI_OFFLINE: Marker hit (marker={marker_name}), skip re-validation, offline mode will be enabled - {model_name_or_path}", + ) + return True + + # No marker - perform full validation + # (Failures are not cached, so we'll retry validation each time until success) + + # Extract revision (snapshot hash) from snapshot_dir path + # snapshot_dir format: /path/to/cache/models--org--model/snapshots/ + revision = os.path.basename(snapshot_dir) + + # Only allow remote checks if we're not in offline mode + # This avoids unnecessary API calls and warnings in offline CI environments + import huggingface_hub.constants + + allow_remote_check = not huggingface_hub.constants.HF_HUB_OFFLINE + + log_info_on_rank0( + logger, + f"CI_OFFLINE: No marker found, performing full validation " + f"(snapshot={revision}, allow_remote_check={allow_remote_check}) - {model_name_or_path}", + ) + + # Validate config and tokenizer files with remote existence checks + config_valid, missing_config_files = _validate_config_and_tokenizer_files( + snapshot_dir=snapshot_dir, + model_id=model_name_or_path, + revision=revision, + allow_remote_check=allow_remote_check, + ) + + if not config_valid: + log_info_on_rank0( + logger, + f"CI_OFFLINE: Missing config/tokenizer files {missing_config_files}, skip offline, keep online allowed - {model_name_or_path}", + ) + # Don't write marker for failures - allow retry after download + return False + + # Validate weight files using existing validation from PR #15216 + # This checks for missing shards, corrupted safetensors, etc. + weights_valid, error_msg, _ = _validate_sharded_model(snapshot_dir, weight_files) + if not weights_valid: + log_info_on_rank0( + logger, + f"CI_OFFLINE: Weight validation failed ({error_msg}), skip offline, keep online allowed - {model_name_or_path}", + ) + # Don't write marker for failures - allow retry after download + return False + + log_info_on_rank0( + logger, + f"CI_OFFLINE: Cache validation PASSED, offline mode will be enabled - {model_name_or_path}", + ) + + # Write marker with passed=True for future reuse + # (Failures are not cached, so this only happens on success) + _write_validation_marker(snapshot_dir, passed=True) + return True + + +def _infer_component_type(component_name: str, component_info: list) -> str: + """ + Infer component type from component name and info. + + Args: + component_name: Name of the component (e.g., "scheduler", "tokenizer") + component_info: Component info from model_index.json (e.g., ["diffusers", "SchedulerClass"]) + + Returns: + Component type string for validation rules + """ + # Normalize component name for type detection + name_lower = component_name.lower() + + # Infer type based on name + if "scheduler" in name_lower: + return "scheduler" + elif "tokenizer" in name_lower: + return "tokenizer" + elif "image_processor" in name_lower: + return "image_processor" + elif "feature_extractor" in name_lower: + return "feature_extractor" + elif "processor" in name_lower: + return "processor" + else: + # Default to model component (needs config.json + weights) + return "model" + + +def _check_component_config( + component_dir: str, component_type: str +) -> Tuple[bool, List[str]]: + """ + Check if component has required config files based on type. + + Args: + component_dir: Path to component directory + component_type: Type of component (scheduler, tokenizer, processor, model, etc.) + + Returns: + Tuple of (has_valid_config, list_of_candidates_tried) + """ + if component_type == "scheduler": + # Scheduler: scheduler_config.json or config.json + candidates = ["scheduler_config.json", "config.json"] + for candidate in candidates: + candidate_path = os.path.join(component_dir, candidate) + if _validate_json_file(candidate_path, candidate): + return True, candidates + return False, candidates + + elif component_type == "tokenizer": + # Tokenizer must have actual tokenizer files (not just tokenizer_config.json) + # Valid combinations: + # - tokenizer.json + # - tokenizer.model + # - vocab.json + merges.txt + candidates = [ + "tokenizer.json", + "tokenizer.model", + "vocab.json+merges.txt", + ] + + # Check tokenizer.json (validate as JSON) + tokenizer_json_path = os.path.join(component_dir, "tokenizer.json") + if _validate_json_file(tokenizer_json_path, "tokenizer.json"): + return True, candidates + + # Check tokenizer.model (non-empty file) + tokenizer_model_path = os.path.join(component_dir, "tokenizer.model") + if os.path.exists(tokenizer_model_path) and os.path.isfile( + tokenizer_model_path + ): + try: + if os.path.getsize(tokenizer_model_path) > 0: + return True, candidates + except OSError: + pass + + # Check vocab.json + merges.txt pair + vocab_path = os.path.join(component_dir, "vocab.json") + merges_path = os.path.join(component_dir, "merges.txt") + if _validate_json_file(vocab_path, "vocab.json") and os.path.exists( + merges_path + ): + return True, candidates + + return False, candidates + + elif component_type in ["processor", "feature_extractor", "image_processor"]: + # Processor/feature_extractor/image_processor: preprocessor_config.json or config.json + candidates = ["preprocessor_config.json", "config.json"] + for candidate in candidates: + candidate_path = os.path.join(component_dir, candidate) + if _validate_json_file(candidate_path, candidate): + return True, candidates + return False, candidates + + else: + # Default model components: config.json + candidates = ["config.json"] + config_path = os.path.join(component_dir, "config.json") + if _validate_json_file(config_path, "config.json"): + return True, candidates + return False, candidates + + +def _check_component_weights(component_dir: str) -> bool: + """ + Check if component directory has weight files. + + Args: + component_dir: Path to component directory + + Returns: + True if weight files found, False otherwise + """ + weight_patterns = ["*.safetensors", "*.bin", "*.pt", "*.pth"] + + for pattern in weight_patterns: + weight_files = glob_module.glob(os.path.join(component_dir, pattern)) + if weight_files: + return True + + return False + + +def _format_component_list(components: List[str], max_show: int = 5) -> str: + """ + Format component list with truncation. + + Args: + components: List of component names + max_show: Maximum number to show before truncating + + Returns: + Formatted string like "comp1, comp2, comp3" or "comp1, comp2, +3 more" + """ + if len(components) <= max_show: + return ", ".join(components) + else: + shown = components[:max_show] + remaining = len(components) - max_show + return f"{', '.join(shown)}, +{remaining} more" + + +def _validate_diffusion_model( + snapshot_dir: str, +) -> Tuple[bool, Optional[str]]: + """ + Validate diffusion model (diffusers pipeline) cache completeness. + + This validation is based on model_index.json as the single source of truth. + Error reporting uses coarse-grained error codes unless verbose mode is enabled. + + Error codes: + - DIFFUSERS_INVALID_INDEX: model_index.json missing or corrupted + - DIFFUSERS_INVALID_COMPONENTS: model_index.json has no valid components + - DIFFUSERS_MISSING_COMPONENT: component directory or config missing + - DIFFUSERS_MISSING_WEIGHTS: component weights missing + + Args: + snapshot_dir: Path to the model snapshot directory + + Returns: + Tuple of (is_valid, error_message) + - (True, None) if validation passed + - (False, error_code_with_components) if validation failed + """ + # Check verbose mode from environment + verbose = os.environ.get("SGLANG_CI_VALIDATE_VERBOSE") == "1" + + # 1. Check for model_index.json (required for diffusers models) + model_index_path = os.path.join(snapshot_dir, "model_index.json") + if not os.path.exists(model_index_path): + return False, "DIFFUSERS_INVALID_INDEX: model_index.json not found" + + # Parse model_index.json + try: + with open(model_index_path, "r", encoding="utf-8") as f: + model_index = json.load(f) + except (json.JSONDecodeError, OSError) as e: + if verbose: + return False, f"DIFFUSERS_INVALID_INDEX: model_index.json parse error - {e}" + return False, "DIFFUSERS_INVALID_INDEX: model_index.json corrupted" + + # 2. Extract components (non-underscore keys with list values) + components = { + k: v + for k, v in model_index.items() + if not k.startswith("_") and isinstance(v, list) + } + + if not components: + return False, "DIFFUSERS_INVALID_COMPONENTS: no valid components defined" + + # Categorize errors by type + missing_dirs = [] + missing_configs = [] + missing_configs_verbose = [] + missing_weights = [] + + # 3. Validate each component + for component_name, component_info in components.items(): + component_dir = os.path.join(snapshot_dir, component_name) + + # Component directory must exist + if not os.path.isdir(component_dir): + missing_dirs.append(component_name) + continue + + # Infer component type for validation rules + component_type = _infer_component_type(component_name, component_info) + + # Check for required config files based on component type + has_valid_config, config_candidates = _check_component_config( + component_dir, component_type + ) + + if not has_valid_config: + missing_configs.append(component_name) + if verbose: + candidates_str = ", ".join(config_candidates) + missing_configs_verbose.append( + f"{component_name} (tried: {candidates_str})" + ) + continue + + # 4. Check for weights if component needs them + # These components don't require weight files (config-only) + needs_weights = component_type not in [ + "scheduler", + "tokenizer", + "processor", + "feature_extractor", + "image_processor", + ] + + if needs_weights: + has_weights = _check_component_weights(component_dir) + if not has_weights: + missing_weights.append(component_name) + + # 5. Build error message based on categorized errors + if missing_dirs or missing_configs or missing_weights: + errors = [] + + if missing_dirs: + dir_str = _format_component_list(missing_dirs) + if verbose: + errors.append(f"DIFFUSERS_MISSING_COMPONENT (dirs): {dir_str}") + else: + errors.append(f"DIFFUSERS_MISSING_COMPONENT(dir): {dir_str}") + + if missing_configs: + if verbose: + config_str = "; ".join(missing_configs_verbose) + errors.append(f"DIFFUSERS_MISSING_COMPONENT (configs): {config_str}") + else: + config_str = _format_component_list(missing_configs) + errors.append(f"DIFFUSERS_MISSING_COMPONENT(cfg): {config_str}") + + if missing_weights: + weight_str = _format_component_list(missing_weights) + errors.append(f"DIFFUSERS_MISSING_WEIGHTS: {weight_str}") + + return False, " | ".join(errors) + + return True, None + + +def validate_cache_with_detailed_reason( + snapshot_dir: str, weight_files: List[str], model_name_or_path: str +) -> Tuple[bool, Optional[str]]: + """ + Validate cache and return detailed reason for failure. + + This function performs validation without relying on shared validation markers. + Used by prevalidate_cached_models.py to provide detailed feedback. + + Args: + snapshot_dir: Path to the model snapshot directory + weight_files: List of weight file paths to validate + model_name_or_path: Model identifier for logging + + Returns: + Tuple of (success, reason): + - (True, None) if validation passed + - (False, reason_str) if validation failed with specific reason + """ + # Guard: weight_files is required + if not weight_files: + return False, "No weight files provided" + + # Perform full validation and capture failure reasons + revision = os.path.basename(snapshot_dir) + + # Read from environment variable instead of huggingface_hub.constants + allow_remote_check = os.environ.get("HF_HUB_OFFLINE") != "1" + + # Validate config and tokenizer files + config_valid, missing_config_files = _validate_config_and_tokenizer_files( + snapshot_dir=snapshot_dir, + model_id=model_name_or_path, + revision=revision, + allow_remote_check=allow_remote_check, + ) + + if not config_valid: + missing_files_str = ", ".join(missing_config_files) + return False, f"Missing config/tokenizer files: {missing_files_str}" + + # Validate weight files + weights_valid, error_msg, _ = _validate_sharded_model(snapshot_dir, weight_files) + if not weights_valid: + return False, f"Weight validation failed: {error_msg}" + + # All validations passed + return True, None + + +def validate_cache_lightweight( + snapshot_dir: str, requires_hf_quant_config: bool = False +) -> bool: + """ + Lightweight runtime validation for cache completeness. + + This is used during test runs to ensure the current runner's cache + is complete before enabling offline mode. Much faster than full validation + as it only checks file existence, not corruption. + + Args: + snapshot_dir: Path to the model snapshot directory + requires_hf_quant_config: If True, hf_quant_config.json must exist + (required for modelopt quantization) + + Returns: + True if cache is complete, False otherwise + """ + # Check required config files + required_files = [ + "config.json", + "tokenizer_config.json", + ] + + for fname in required_files: + if not os.path.exists(os.path.join(snapshot_dir, fname)): + return False + + # Check tokenizer files (at least one must exist) + tokenizer_files = [ + "tokenizer.json", + "tokenizer.model", + "tiktoken.model", + ] + + has_tokenizer = any( + os.path.exists(os.path.join(snapshot_dir, fname)) for fname in tokenizer_files + ) + if not has_tokenizer: + return False + + # Check for trust_remote_code dynamic module files if needed + # When auto_map exists in config.json, the model requires custom Python files + # These files must be present for offline mode to work + config_path = os.path.join(snapshot_dir, "config.json") + if os.path.exists(config_path): + try: + with open(config_path, "r", encoding="utf-8") as f: + config = json.load(f) + + auto_map = config.get("auto_map", {}) + if auto_map and isinstance(auto_map, dict): + # Extract Python module files from auto_map + # auto_map format: {"AutoConfig": "configuration_xxx.ConfigClass", ...} + # We need to check if the .py files exist + custom_files = set() + for key, value in auto_map.items(): + if isinstance(value, str) and "." in value: + # Extract module name (e.g., "configuration_xxx" from "configuration_xxx.ConfigClass") + module_name = value.split(".")[0] + custom_files.add(f"{module_name}.py") + + # Check if all custom files exist in snapshot directory + for custom_file in custom_files: + custom_file_path = os.path.join(snapshot_dir, custom_file) + if not os.path.exists(custom_file_path): + logger.debug( + "Custom module file not in snapshot: %s for %s", + custom_file, + snapshot_dir, + ) + return False + elif not os.path.isfile(custom_file_path): + logger.debug( + "Custom module path exists but not a file: %s", + custom_file_path, + ) + return False + except (json.JSONDecodeError, OSError, KeyError) as e: + # If we can't read config.json, it will be caught by earlier validation + logger.debug("Failed to check auto_map in config.json: %s", e) + + # Check for weight files with index self-consistency + index_path = os.path.join(snapshot_dir, "model.safetensors.index.json") + has_index = os.path.exists(index_path) + + if has_index: + # If index exists, validate that all shards listed in it exist + try: + with open(index_path, "r", encoding="utf-8") as f: + index_data = json.load(f) + weight_map = index_data.get("weight_map", {}) + if weight_map: + # Check that all shard files referenced in index exist + required_shards = set(weight_map.values()) + for shard_name in required_shards: + shard_path = os.path.join(snapshot_dir, shard_name) + if not os.path.exists(shard_path): + logger.debug( + "Index validation failed: missing shard %s in %s", + shard_name, + snapshot_dir, + ) + return False + except (json.JSONDecodeError, OSError, KeyError) as e: + logger.debug("Failed to validate index file %s: %s", index_path, e) + return False + else: + # No index file, check for at least one shard + # *.safetensors already covers model-*.safetensors pattern + has_shards = bool(glob_module.glob(os.path.join(snapshot_dir, "*.safetensors"))) + if not has_shards: + return False + + # Check hf_quant_config.json if required (for modelopt quantization) + if requires_hf_quant_config: + hf_quant_path = os.path.join(snapshot_dir, "hf_quant_config.json") + if not os.path.exists(hf_quant_path): + return False + + return True + def _validate_safetensors_file(file_path: str) -> bool: """ diff --git a/python/sglang/srt/model_loader/loader.py b/python/sglang/srt/model_loader/loader.py index 2bebead89..959114b57 100644 --- a/python/sglang/srt/model_loader/loader.py +++ b/python/sglang/srt/model_loader/loader.py @@ -566,7 +566,9 @@ class DefaultModelLoader(BaseModelLoader): ) hf_config = AutoConfig.from_pretrained( - model_config.model_path, trust_remote_code=True + model_config.model_path, + trust_remote_code=True, + local_files_only=huggingface_hub.constants.HF_HUB_OFFLINE, ) with init_empty_weights(): torch_dtype = getattr(hf_config, "torch_dtype", torch.float16) @@ -599,6 +601,7 @@ class DefaultModelLoader(BaseModelLoader): device_map=device_map, **model_kwargs, trust_remote_code=True, + local_files_only=huggingface_hub.constants.HF_HUB_OFFLINE, ) # Handle both legacy modelopt_quant and unified quantization flags if hasattr(model_config, "modelopt_quant") and model_config.modelopt_quant: diff --git a/python/sglang/test/test_utils.py b/python/sglang/test/test_utils.py index ca0e7c735..8093faa8b 100644 --- a/python/sglang/test/test_utils.py +++ b/python/sglang/test/test_utils.py @@ -562,6 +562,249 @@ def popen_with_error_check(command: list[str], allow_exit: bool = False): return process +def _try_enable_offline_mode_if_cache_complete( + model_name_or_path: str, env: dict, other_args: Optional[list[str]] = None +) -> Optional[str]: + """ + CI helper: Check if model cache is complete and enable offline mode. + + Uses per-run validation markers that are NOT shared across runners. + Each runner independently validates its cache using lightweight checks + before enabling offline mode. + + IMPORTANT: Even if a per-run marker exists, this function ALWAYS validates + the current launch's requirements (e.g., hf_quant_config.json for modelopt). + The marker is only a hint that this snapshot was validated earlier in the run. + + Args: + model_name_or_path: Model identifier or path + env: Environment dict to modify (will add HF_HUB_OFFLINE=1 if validation passes) + other_args: Launch command arguments (used to detect quantization requirement) + + Returns: + Per-run marker path if offline mode was enabled, None otherwise + """ + from sglang.srt.model_loader.ci_weight_validation import ( + _get_per_run_marker_path, + _read_per_run_marker, + _write_per_run_marker, + validate_cache_lightweight, + ) + from sglang.srt.utils import find_local_repo_dir + + other_args = other_args or [] + + # Fast-path: If subprocess env already has HF_HUB_OFFLINE=1, skip + if env.get("HF_HUB_OFFLINE") == "1": + print( + f"CI_OFFLINE: Subprocess env already has HF_HUB_OFFLINE=1, skip - {model_name_or_path}" + ) + return None + + # Skip if already a local path + if os.path.isdir(model_name_or_path): + return None + + # Try to find local snapshot + try: + snapshot_dir = find_local_repo_dir(model_name_or_path, revision=None) + if not snapshot_dir or not os.path.isdir(snapshot_dir): + return None + except Exception: + return None + + # Detect if quantization requires hf_quant_config.json + # Do this BEFORE checking marker to ensure current launch requirements are known + requires_hf_quant_config = False + for i, arg in enumerate(other_args): + if arg == "--quantization" and i + 1 < len(other_args): + quant_value = other_args[i + 1].lower() + if quant_value in ["modelopt_fp4", "modelopt_fp8", "modelopt"]: + requires_hf_quant_config = True + break + + # Check per-run marker (fast hint - snapshot validated earlier in this run) + per_run_marker = _read_per_run_marker(snapshot_dir) + if per_run_marker is not None: + # Marker exists, but STILL validate for current launch requirements + # This prevents a test without --quantization from enabling offline + # for a later test with --quantization that needs hf_quant_config.json + is_valid = validate_cache_lightweight(snapshot_dir, requires_hf_quant_config) + + if not is_valid: + # Current launch requirements not met, ignore marker + print( + f"CI_OFFLINE: Per-run marker found but current validation failed " + f"(requires_hf_quant_config={requires_hf_quant_config}), " + f"will use online mode - {model_name_or_path}" + ) + return None + + # Marker exists and current validation passed + env["HF_HUB_OFFLINE"] = "1" + marker_path = _get_per_run_marker_path(snapshot_dir) + print( + f"CI_OFFLINE: Per-run marker found and current validation passed " + f"(requires_hf_quant_config={requires_hf_quant_config}), " + f"enabling offline mode - {model_name_or_path}" + ) + return marker_path + + # No per-run marker - perform lightweight validation + is_valid = validate_cache_lightweight(snapshot_dir, requires_hf_quant_config) + + if not is_valid: + # Validation failed - cache is incomplete on this runner + print( + f"CI_OFFLINE: Cache validation failed " + f"(requires_hf_quant_config={requires_hf_quant_config}), " + f"will use online mode - {model_name_or_path}" + ) + return None + + # Validation passed - enable offline mode and write per-run marker + env["HF_HUB_OFFLINE"] = "1" + + # Write per-run marker for subsequent tests in this run + _write_per_run_marker(snapshot_dir, model_name_or_path) + + # Return marker path for potential invalidation if offline launch fails + marker_path = _get_per_run_marker_path(snapshot_dir) + + snapshot_basename = os.path.basename(snapshot_dir) + print( + f"CI_OFFLINE: Enabled HF_HUB_OFFLINE=1 for subprocess - " + f"validation passed for {model_name_or_path} " + f"(snapshot={snapshot_basename}, requires_hf_quant_config={requires_hf_quant_config})" + ) + + return marker_path + + +def _create_clean_subprocess_env(env: dict) -> dict: + """Create a clean subprocess environment without internal CI keys. + + Removes all keys starting with '_CI_OFFLINE_' or 'CI_OFFLINE' to prevent + leaking implementation details to the server subprocess. + + Args: + env: Source environment dict + + Returns: + Clean copy of environment dict + """ + child_env = env.copy() + keys_to_remove = [ + k for k in child_env if k.startswith(("_CI_OFFLINE_", "CI_OFFLINE_")) + ] + for k in keys_to_remove: + del child_env[k] + return child_env + + +def _launch_server_process( + command: List[str], + env: dict, + return_stdout_stderr: Optional[tuple], + model: str, +) -> subprocess.Popen: + """Launch server subprocess with clean environment. + + Args: + command: Command list for subprocess + env: Environment dict (will be cleaned before use) + return_stdout_stderr: Optional tuple of (stdout_file, stderr_file) for output capture + model: Model name for logging + + Returns: + Started subprocess.Popen object + """ + child_env = _create_clean_subprocess_env(env) + + hf_hub_offline = child_env.get("HF_HUB_OFFLINE", "0") + print(f"CI_OFFLINE: Launching server HF_HUB_OFFLINE={hf_hub_offline} model={model}") + + if return_stdout_stderr: + proc = subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + env=child_env, + text=True, + bufsize=1, + ) + + def _dump(src, sinks): + for line in iter(src.readline, ""): + for sink in sinks: + sink.write(line) + sink.flush() + src.close() + + threading.Thread( + target=_dump, + args=(proc.stdout, [return_stdout_stderr[0], sys.stdout]), + daemon=True, + ).start() + threading.Thread( + target=_dump, + args=(proc.stderr, [return_stdout_stderr[1], sys.stderr]), + daemon=True, + ).start() + else: + proc = subprocess.Popen(command, stdout=None, stderr=None, env=child_env) + + return proc + + +def _wait_for_server_health( + proc: subprocess.Popen, + base_url: str, + api_key: Optional[str], + timeout_duration: float, +) -> Tuple[bool, Optional[str]]: + """Wait for server health check to pass. + + Args: + proc: Server subprocess + base_url: Base URL for health check + api_key: Optional API key for authorization + timeout_duration: Maximum wait time in seconds + + Returns: + Tuple of (success, error_message) + """ + start_time = time.perf_counter() + with requests.Session() as session: + while time.perf_counter() - start_time < timeout_duration: + return_code = proc.poll() + if return_code is not None: + return False, f"Server process exited with code {return_code}" + + try: + headers = { + "Content-Type": "application/json; charset=utf-8", + "Authorization": f"Bearer {api_key}", + } + response = session.get( + f"{base_url}/health_generate", + headers=headers, + timeout=5, + ) + if response.status_code == 200: + return True, None + except requests.RequestException: + pass + + return_code = proc.poll() + if return_code is not None: + return False, f"Server unexpectedly exited (return_code={return_code})" + + time.sleep(10) + + return False, "Server failed to start within the timeout period" + + def popen_launch_server( model: str, base_url: str, @@ -574,11 +817,22 @@ def popen_launch_server( pd_separated: bool = False, num_replicas: Optional[int] = None, ): - """Launch a server process with automatic device detection. + """Launch a server process with automatic device detection and offline/online retry. Args: - device: Device type ("auto", "cuda", "rocm" or "cpu"). - If "auto", will detect available platforms automatically. + model: Model path or identifier + base_url: Base URL for the server + timeout: Timeout for server startup + api_key: Optional API key for authentication + other_args: Additional command line arguments + env: Environment dict for subprocess + return_stdout_stderr: Optional tuple for output capture + device: Device type ("auto", "cuda", "rocm" or "cpu") + pd_separated: Whether to use PD separated mode + num_replicas: Number of replicas for mixed PD mode + + Returns: + Started subprocess.Popen object """ other_args = other_args or [] @@ -588,6 +842,25 @@ def popen_launch_server( other_args = list(other_args) other_args += ["--device", str(device)] + # CI-specific: Validate cache and enable offline mode if complete + if env is None: + env = os.environ.copy() + else: + env = env.copy() + + # Store per-run marker path for potential invalidation + per_run_marker_path = None + try: + from sglang.utils import is_in_ci + + if is_in_ci(): + per_run_marker_path = _try_enable_offline_mode_if_cache_complete( + model, env, other_args + ) + except Exception as e: + print(f"CI cache validation failed (non-fatal): {e}") + + # Build server command _, host, port = base_url.split(":") host = host[2:] @@ -607,104 +880,82 @@ def popen_launch_server( ] if pd_separated or use_mixed_pd_engine: - command.extend( - [ - "--lb-host", - host, - "--lb-port", - port, - ] - ) + command.extend(["--lb-host", host, "--lb-port", port]) else: - command.extend( - [ - "--host", - host, - "--port", - port, - ] - ) + command.extend(["--host", host, "--port", port]) if use_mixed_pd_engine: - command.extend( - [ - "--mixed", - "--num-replicas", - str(num_replicas), - ] - ) + command.extend(["--mixed", "--num-replicas", str(num_replicas)]) if api_key: command += ["--api-key", api_key] print(f"command={shlex.join(command)}") - if return_stdout_stderr: - process = subprocess.Popen( - command, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - env=env, - text=True, - bufsize=1, + # Track if offline mode was enabled for potential retry + offline_enabled = env.get("HF_HUB_OFFLINE") == "1" + + # First launch attempt + process = _launch_server_process(command, env, return_stdout_stderr, model) + success, error_msg = _wait_for_server_health(process, base_url, api_key, timeout) + + # If offline launch failed and offline was enabled, retry with online mode + if not success and offline_enabled: + print( + f"CI_OFFLINE: Offline launch failed ({error_msg}), retrying with online mode..." ) - def _dump(src, sinks): - for line in iter(src.readline, ""): - for sink in sinks: - sink.write(line) - sink.flush() - src.close() - - threading.Thread( - target=_dump, - args=(process.stdout, [return_stdout_stderr[0], sys.stdout]), - daemon=True, - ).start() - threading.Thread( - target=_dump, - args=(process.stderr, [return_stdout_stderr[1], sys.stderr]), - daemon=True, - ).start() - else: - process = subprocess.Popen(command, stdout=None, stderr=None, env=env) - - start_time = time.perf_counter() - with requests.Session() as session: - while time.perf_counter() - start_time < timeout: - return_code = process.poll() - if return_code is not None: - # Server failed to start (non-zero exit code) or crashed - raise Exception( - f"Server process exited with code {return_code}. " - "Check server logs for errors." - ) + # Kill failed process + try: + if process.poll() is None: + kill_process_tree(process.pid) + else: + process.wait(timeout=5) + except Exception as e: + print(f"CI_OFFLINE: Error cleaning up failed offline process: {e}") + # Invalidate per-run marker to prevent subsequent tests from using offline + if per_run_marker_path and os.path.exists(per_run_marker_path): try: - headers = { - "Content-Type": "application/json; charset=utf-8", - "Authorization": f"Bearer {api_key}", - } - response = session.get( - f"{base_url}/health_generate", - headers=headers, - timeout=5, - ) - if response.status_code == 200: - return process - except requests.RequestException: - pass + os.remove(per_run_marker_path) + print("CI_OFFLINE: Invalidated per-run marker due to offline failure") + except Exception as e: + print(f"CI_OFFLINE: Failed to remove per-run marker: {e}") - return_code = process.poll() - if return_code is not None: - raise Exception( - f"Server unexpectedly exits ({return_code=}). Usually there will be error logs describing the cause far above this line." - ) + # Retry with online mode + env["HF_HUB_OFFLINE"] = "0" + process = _launch_server_process(command, env, return_stdout_stderr, model) + success, error_msg = _wait_for_server_health( + process, base_url, api_key, timeout + ) - time.sleep(10) + if success: + print("CI_OFFLINE: Online retry succeeded") + return process - kill_process_tree(process.pid) - raise TimeoutError("Server failed to start within the timeout period.") + # Online retry also failed + try: + kill_process_tree(process.pid) + except Exception as e: + print(f"CI_OFFLINE: Error killing process after online retry failure: {e}") + + if "exited" in error_msg: + raise Exception(error_msg + ". Check server logs for errors.") + raise TimeoutError(error_msg) + + # First attempt succeeded or offline was not enabled + if success: + return process + + # First attempt failed and offline was not enabled + try: + kill_process_tree(process.pid) + except Exception as e: + print(f"CI_OFFLINE: Error killing process after first attempt failure: {e}") + + if "exited" in error_msg: + raise Exception(error_msg + ". Check server logs for errors.") + raise TimeoutError(error_msg) def popen_launch_pd_server( diff --git a/scripts/ci/prepare_runner.sh b/scripts/ci/prepare_runner.sh index fe0bf4400..3a6701bf2 100755 --- a/scripts/ci/prepare_runner.sh +++ b/scripts/ci/prepare_runner.sh @@ -11,4 +11,9 @@ echo "" python3 "${SCRIPT_DIR}/cleanup_hf_cache.py" echo "" +# Pre-validate cached models and write markers for offline mode +# This allows tests to run with HF_HUB_OFFLINE=1 for models that are fully cached +python3 "${SCRIPT_DIR}/prevalidate_cached_models.py" +echo "" + echo "CI runner preparation complete!" diff --git a/scripts/ci/prevalidate_cached_models.py b/scripts/ci/prevalidate_cached_models.py new file mode 100755 index 000000000..9d08f987b --- /dev/null +++ b/scripts/ci/prevalidate_cached_models.py @@ -0,0 +1,407 @@ +#!/usr/bin/env python3 +""" +Pre-validate all cached HuggingFace models to provide detailed feedback. + +This script runs once during CI initialization (in prepare_runner.sh) to: +1. Scan snapshots in ~/.cache/huggingface/hub/ (with time/quantity limits) +2. Validate completeness (config/tokenizer/weights) +3. Output detailed failure reasons for debugging + +NOTE: This script no longer writes shared validation markers. Each test run +independently validates its cache using per-run markers to avoid cross-runner +cache state pollution. +""" + +import glob +import json +import os +import sys +import time +from pathlib import Path + +# Add python directory to path to import sglang modules +REPO_ROOT = Path(__file__).parent.parent.parent +sys.path.insert(0, str(REPO_ROOT / "python")) + +from sglang.srt.model_loader.ci_weight_validation import ( # noqa: E402 + _validate_diffusion_model, + validate_cache_with_detailed_reason, +) + +# Limits to avoid spending too much time on validation +MAX_VALIDATION_TIME_SECONDS = 300 # Max 5 minutes total + + +def find_all_hf_snapshots(): + """ + Find all HuggingFace snapshots in cache. + + Returns: + List of (model_name, snapshot_dir) tuples, sorted by mtime (newest first) + """ + hf_home = os.environ.get("HF_HOME", os.path.expanduser("~/.cache/huggingface")) + hub_dir = os.path.join(hf_home, "hub") + + if not os.path.isdir(hub_dir): + print(f"HF hub directory not found: {hub_dir}") + return [] + + snapshots = [] + + # Pattern: models--org--model/snapshots/hash + for model_dir in glob.glob(os.path.join(hub_dir, "models--*")): + # Extract model name from directory (models--org--model -> org/model) + dir_name = os.path.basename(model_dir) + if not dir_name.startswith("models--"): + continue + + # models--meta-llama--Llama-2-7b-hf -> meta-llama/Llama-2-7b-hf + # Handle multi-part names: models--a--b--c -> a/b-c (join parts 1+ with /) + parts = dir_name.split("--") + if len(parts) < 3 or parts[0] != "models": + # Invalid format, skip + continue + # Standard format: models--org--repo -> org/repo + # Extended format: models--org--repo--extra -> org/repo-extra (join with -) + model_name = parts[1] + "/" + "-".join(parts[2:]) + + snapshots_dir = os.path.join(model_dir, "snapshots") + if not os.path.isdir(snapshots_dir): + continue + + # Find all snapshot hashes + for snapshot_hash_dir in os.listdir(snapshots_dir): + snapshot_path = os.path.join(snapshots_dir, snapshot_hash_dir) + if os.path.isdir(snapshot_path): + try: + mtime = os.path.getmtime(snapshot_path) + snapshots.append((model_name, snapshot_path, mtime)) + except OSError: + continue + + # Sort by mtime (newest first) - prioritize recently used models + snapshots.sort(key=lambda x: x[2], reverse=True) + + # Return without mtime + return [(name, path) for name, path, _ in snapshots] + + +def is_transformers_text_model(snapshot_dir): + """ + Check if a snapshot is a transformers text model. + + Only excludes (returns False) for models with STRONG evidence of being + diffusers/generation pipelines. Uses conservative heuristics to avoid + false negatives on multimodal LLMs with tokenizers. + + Args: + snapshot_dir: Path to snapshot directory + + Returns: + True if this looks like a transformers text model, False otherwise (N/A) + """ + # Check for diffusers pipeline markers (strong evidence) + diffusers_markers = [ + "model_index.json", # Diffusers pipeline config + "scheduler", # Scheduler directory (diffusers) + ] + if any( + os.path.exists(os.path.join(snapshot_dir, marker)) + for marker in diffusers_markers + ): + return False + + config_path = os.path.join(snapshot_dir, "config.json") + if not os.path.exists(config_path): + # No config.json - likely not a transformers model + return False + + try: + with open(config_path, "r", encoding="utf-8") as f: + config = json.load(f) + + # Check for explicit diffusers/generation model types (conservative keywords) + model_type = config.get("_class_name") or config.get("model_type") + if model_type: + model_type_lower = str(model_type).lower() + # Only exclude clear diffusion/generation models + if any( + keyword in model_type_lower + for keyword in [ + "diffusion", + "unet", + "vae", + "controlnet", + "stable-diffusion", + "latent-diffusion", + ] + ): + return False + + # Check architectures for explicit generation/diffusion classes + architectures = config.get("architectures", []) + if architectures: + arch_str = " ".join(architectures).lower() + # Conservative: only exclude obvious diffusion/generation architectures + # Use word boundaries to avoid false positives (e.g., "dit" in "conditional") + for keyword in [ + "diffusion", + "unet2d", + "unet3d", + "vaedecoder", # More specific than "vae" + "vaeencoder", + "controlnet", + "autoencoder", + "ditmodel", # Diffusion Transformer - use more specific pattern + "pixart", # PixArt diffusion model + ]: + if keyword in arch_str: + return False + + # Check for standalone vision encoder/image processor (no text component) + # Only if model name explicitly indicates non-text usage + model_name = config.get("_name_or_path", "").lower() + + if any( + keyword in model_name + for keyword in [ + "image-edit-", # Pure image editing (e.g., Qwen-Image-Edit) + "-image-editing", + "dit-", # DiT generation models + "pixart-", # PixArt generation models + ] + ): + # Additional check: does it have tokenizer? If yes, might be multimodal LLM + has_tokenizer = any( + os.path.exists(os.path.join(snapshot_dir, fname)) + for fname in ["tokenizer.json", "tokenizer.model", "tiktoken.model"] + ) + if not has_tokenizer: + # Image-edit model without tokenizer -> likely pure vision pipeline + return False + + # Default: assume it's a transformers text/multimodal model + # Even if it lacks tokenizer, let validation report the actual error + # (better false positive than false negative for text models) + return True + + except (json.JSONDecodeError, OSError, KeyError): + # Can't parse config - assume it's transformers and let validation report failure + return True + + +def scan_weight_files(snapshot_dir): + """ + Scan for weight files in a snapshot. + + Returns: + List of weight file paths, or empty list if scan fails + """ + weight_files = [] + + # First, look for index files + index_patterns = ["*.safetensors.index.json", "pytorch_model.bin.index.json"] + index_files = [] + for pattern in index_patterns: + index_files.extend(glob.glob(os.path.join(snapshot_dir, pattern))) + + # If we have safetensors index, collect shards from it + for index_file in index_files: + if index_file.endswith(".safetensors.index.json"): + try: + with open(index_file, "r", encoding="utf-8") as f: + index_data = json.load(f) + weight_map = index_data.get("weight_map", {}) + for weight_file in set(weight_map.values()): + weight_path = os.path.join(snapshot_dir, weight_file) + if os.path.exists(weight_path): + weight_files.append(weight_path) + except Exception as e: + print( + f" Warning: Failed to parse index {os.path.basename(index_file)}: {e}" + ) + + # If no index found or no shards from index, do recursive glob + if not weight_files: + matched = glob.glob( + os.path.join(snapshot_dir, "**/*.safetensors"), recursive=True + ) + MAX_WEIGHT_FILES = 1000 + if len(matched) > MAX_WEIGHT_FILES: + print( + f" Warning: Too many safetensors files ({len(matched)} > {MAX_WEIGHT_FILES})" + ) + return [] + + for f in matched: + if os.path.exists(f): # Filter out broken symlinks + weight_files.append(f) + + return weight_files + + +def validate_snapshot(model_name, snapshot_dir, weight_files, validated_cache): + """ + Validate a snapshot and return detailed status. + + Uses in-process cache to avoid duplicate validation within the same run. + + Args: + model_name: Model identifier + snapshot_dir: Path to snapshot directory + weight_files: List of weight files to validate + validated_cache: Dict to track already-validated snapshots in this run + + Returns: + Tuple of (result, reason): + - (True, None) if validation passed + - (False, reason_str) if validation failed + - (None, None) if skipped (already validated in this run) + """ + # Fast path: check in-process cache first + if snapshot_dir in validated_cache: + return None, None # Already validated in this run, skip + + try: + # Perform validation with detailed reason + is_complete, reason = validate_cache_with_detailed_reason( + snapshot_dir=snapshot_dir, + weight_files=weight_files, + model_name_or_path=model_name, + ) + + # Cache result to avoid re-validation in this run + validated_cache[snapshot_dir] = (is_complete, reason) + + return is_complete, reason + + except Exception as e: + error_msg = f"Validation raised exception: {e}" + return False, error_msg + + +def main(): + start_time = time.time() + + print("=" * 70) + print("CI_OFFLINE: Pre-validating cached HuggingFace models") + print("=" * 70) + print(f"Max time: {MAX_VALIDATION_TIME_SECONDS}s") + print() + + print("Scanning HuggingFace cache for models...") + snapshots = find_all_hf_snapshots() + + if not snapshots: + print("No cached models found, skipping validation") + print("=" * 70) + return + + print(f"Found {len(snapshots)} snapshot(s) in cache") + print() + + validated_count = 0 + failed_count = 0 + skipped_count = 0 + processed_count = 0 + + # In-process cache to avoid re-validating same snapshot in this run + validated_cache = {} + + for model_name, snapshot_dir in snapshots: + # Check time limit + elapsed = time.time() - start_time + if elapsed > MAX_VALIDATION_TIME_SECONDS: + print() + print( + f"Time limit reached ({elapsed:.1f}s > {MAX_VALIDATION_TIME_SECONDS}s)" + ) + print( + f"Stopping validation, {len(snapshots) - processed_count} snapshots remaining" + ) + break + + snapshot_hash = os.path.basename(snapshot_dir) + print( + f"[{processed_count + 1}/{len(snapshots)}] {model_name} ({snapshot_hash[:8]}...)" + ) + processed_count += 1 + + # Determine model type by checking for model_index.json (diffusers pipeline marker) + model_index_path = os.path.join(snapshot_dir, "model_index.json") + is_diffusion_model = os.path.exists(model_index_path) + + if is_diffusion_model: + # This is a diffusers pipeline - use diffusion validation + try: + is_valid, reason = _validate_diffusion_model(snapshot_dir) + + if is_valid: + print(" PASS (diffusion) - Cache complete & valid") + validated_count += 1 + else: + print(f" FAIL (diffusion) - {reason}") + failed_count += 1 + + except Exception as e: + print(f" FAIL (diffusion) - Validation raised exception: {e}") + failed_count += 1 + + continue + + # Transformers model - use standard validation + # First check if this looks like a transformers text model + if not is_transformers_text_model(snapshot_dir): + # Not a recognized model type, skip + print( + " SKIP (unknown type) - Not a diffusers pipeline or transformers model" + ) + skipped_count += 1 + continue + + # Scan weight files + weight_files = scan_weight_files(snapshot_dir) + + if not weight_files: + print(" SKIP (no weights) - empty or incomplete download") + skipped_count += 1 + continue + + # Validate + try: + result, reason = validate_snapshot( + model_name, snapshot_dir, weight_files, validated_cache + ) + + if result is True: + print(" PASS - Cache complete & valid") + validated_count += 1 + elif result is False: + # Print detailed failure reason + if reason: + print(f" FAIL (incomplete) - {reason}") + else: + print(" FAIL (incomplete) - cache validation failed") + failed_count += 1 + else: # None (skipped) + print(" SKIP (already validated in this run)") + skipped_count += 1 + + except Exception as e: + print(f" FAIL (error) - Validation raised exception: {e}") + failed_count += 1 + + elapsed_total = time.time() - start_time + + print() + print("=" * 70) + print(f"Validation summary (completed in {elapsed_total:.1f}s):") + print(f" PASS (complete & valid): {validated_count}") + print(f" FAIL (incomplete/corrupted): {failed_count}") + print(f" SKIP (no weights/duplicate): {skipped_count}") + print(f" Total processed: {processed_count}/{len(snapshots)}") + print("=" * 70) + + +if __name__ == "__main__": + main()