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

[CI] Speed up dependency install: dual-ABI Rust ext cache and prevalidation pruning (#33619)

Browse Source
This commit is contained in:
Liangsheng Yin
2026-08-04 20:33:48 -07:00
committed by GitHub
parent 6c05aaae7e
commit 1033cae8d5
11 changed files with 155 additions and 1371 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/actions/download-rust-ext/action.yml
+1 -1
View File
@@ -12,7 +12,7 @@ inputs:
cache_key_prefix:
description: 'Must match what _pr-test-rust-ext-build.yml saves under.'
required: false
default: rust-ext-x86_64
default: rust-ext-x86_64-cp310-cp312
outputs:
hit:
.github/workflows/_pr-test-rust-ext-build.yml
+52 -23
View File
@@ -36,9 +36,9 @@ on:
type: string
default: rust-ext-x86_64
cache_key_prefix:
description: 'Cache key prefix. Callers share it on purpose to reuse each other''s build; it encodes the arch, since the modules are not portable across architectures.'
description: 'Cache key prefix. Callers share it on purpose to reuse each other''s build; it encodes the arch and the interpreter ABI set, since the modules are portable across neither.'
type: string
default: rust-ext-x86_64
default: rust-ext-x86_64-cp310-cp312
max_glibc:
description: 'Highest GLIBC symbol version the built .so files may require. Set by the oldest test runner image, jammy at glibc 2.35 - the pools are not all on one image.'
type: string
@@ -102,6 +102,26 @@ jobs:
path: python/sglang/srt/*/_core*.so
key: ${{ inputs.cache_key_prefix }}-${{ hashFiles('rust/**', 'python/setup.py') }}
# On a miss: different hash = rust/setup.py moved; no entries = evicted.
- name: Report cache lookup
env:
GH_TOKEN: ${{ github.token }}
PRIMARY_KEY: ${{ steps.cache.outputs.cache-primary-key }}
MATCHED_KEY: ${{ steps.cache.outputs.cache-matched-key }}
KEY_PREFIX: ${{ inputs.cache_key_prefix }}
run: |
if [ -n "${MATCHED_KEY}" ]; then
echo "hit: ${MATCHED_KEY}"
ls -l python/sglang/srt/*/_core*.so
else
echo "miss: ${PRIMARY_KEY}"
echo "entries under ${KEY_PREFIX}- (created / ref / size / key):"
gh cache list --repo "${GITHUB_REPOSITORY}" --key "${KEY_PREFIX}-" \
--limit 15 --json createdAt,ref,sizeInBytes,key \
--jq '.[] | [.createdAt, .ref, ((.sizeInBytes / 1048576 | floor | tostring) + " MiB"), .key] | @tsv' \
|| echo "(gh cache list unavailable: token lacks actions:read)"
fi
# No MAX_GLIBC: these are the bytes the compile job already checked before
# saving them under this key. The module count is still worth re-checking,
# so a truncated entry fails here rather than as a test import error.
@@ -139,15 +159,20 @@ jobs:
- uses: ./.github/actions/check-maintenance
# No crate sets abi3, so the ABI tag is minor-version specific and only the
# pools on this version can use the result - h20 ships 3.12 and falls back to
# compiling during install. Pinned rather than left to the image so the tag is
# at least predictable.
# No crate sets abi3, so build one module set per interpreter the pools
# run (h100 ships 3.10, h20 ships 3.12); EXT_SUFFIX keeps them apart.
- name: Set up Python 3.10
id: py310
uses: actions/setup-python@v5
with:
python-version: '3.10'
- name: Set up Python 3.12
id: py312
uses: actions/setup-python@v5
with:
python-version: '3.12'
- name: Install protoc and Rust toolchain
run: bash scripts/ci/utils/install_rust_protoc.sh
@@ -155,32 +180,36 @@ jobs:
run: |
set -euxo pipefail
export PATH="${CARGO_HOME:-$HOME/.cargo}/bin:${PATH}"
# Same path ci_install_dependency.sh uses, so a runner that also runs test
# stages keeps one warm cache. Its guard is repeated here because nothing
# prunes the tree on a runner that only ever builds.
export CARGO_TARGET_DIR="${HOME}/.cache/sglang-cargo-target"
mkdir -p "${CARGO_TARGET_DIR}"
used="$(df --output=pcent "${CARGO_TARGET_DIR}" 2>/dev/null | tr -dc '0-9')"
# Per-interpreter subdirs (set in the loop): PyO3's fingerprint tracks
# the interpreter, so a shared dir rebuilds on every ABI switch.
cargo_target_root="${HOME}/.cache/sglang-cargo-target"
mkdir -p "${cargo_target_root}"
used="$(df --output=pcent "${cargo_target_root}" 2>/dev/null | tr -dc '0-9')"
if [ "${used:-0}" -ge 85 ]; then
echo "cargo target dir filesystem at ${used}%; dropping ${CARGO_TARGET_DIR}"
rm -rf "${CARGO_TARGET_DIR}"
mkdir -p "${CARGO_TARGET_DIR}"
echo "cargo target dir filesystem at ${used}%; dropping ${cargo_target_root}"
rm -rf "${cargo_target_root}"
mkdir -p "${cargo_target_root}"
fi
python3 -m pip install --upgrade pip
command -v uv >/dev/null 2>&1 || pip install uv
# build_rust needs only the build backend, not sglang's ~294 runtime deps.
# Per-job path: these runners are persistent and shared, so a fixed one
# both inherits the previous job's venv and races a concurrent build.
venv="${RUNNER_TEMP:-/tmp}/sglang-ci-rust-ext-${GITHUB_RUN_ID:-norun}-$$"
venv_root="${RUNNER_TEMP:-/tmp}/sglang-ci-rust-ext-${GITHUB_RUN_ID:-norun}-$$"
# Best-effort, like ci_cleanup_venv.sh: under set -e a failing EXIT trap
# would fail the step, and nothing here is worth keeping for a postmortem.
trap 'rm -rf "${venv}" || true' EXIT
uv venv "${venv}" --python python3.10 --seed
# shellcheck disable=SC1091
source "${venv}/bin/activate"
uv pip install "setuptools>=61.0" "setuptools-rust>=1.10" "setuptools-scm>=8.0" wheel
cd python
SGLANG_BUILD_RUST_EXTS=all python setup.py build_rust --inplace
trap 'rm -rf "${venv_root}" || true' EXIT
for python_bin in "${{ steps.py310.outputs.python-path }}" "${{ steps.py312.outputs.python-path }}"; do
minor="$("${python_bin}" -c 'import sys; print(f"{sys.version_info.major}.{sys.version_info.minor}")')"
export CARGO_TARGET_DIR="${cargo_target_root}/py${minor}"
venv="${venv_root}/py${minor}"
uv venv "${venv}" --python "${python_bin}" --seed
# shellcheck disable=SC1091
source "${venv}/bin/activate"
uv pip install "setuptools>=61.0" "setuptools-rust>=1.10" "setuptools-scm>=8.0" wheel
(cd python && SGLANG_BUILD_RUST_EXTS=all python setup.py build_rust --inplace)
deactivate
done
- name: Verify modules and stage for upload
env:
.github/workflows/seed-rust-ext-cache.yml
+5 -2
View File
@@ -10,6 +10,8 @@ on:
paths:
- 'rust/**'
- 'python/setup.py'
# The key's prefix lives in this file, so a bump there also moves the key.
- '.github/workflows/_pr-test-rust-ext-build.yml'
workflow_dispatch:
# Only the newest merge needs to seed; earlier ones are already stale.
@@ -17,9 +19,10 @@ concurrency:
group: seed-rust-ext-cache
cancel-in-progress: true
# Declaring permissions at all drops everything not listed, and check-maintenance
# needs issues: read to reach the maintenance issue.
# Declaring permissions at all drops everything not listed. issues: read for
# check-maintenance; actions: read for the restore job's gh cache list.
permissions:
actions: read
contents: read
issues: read
.pre-commit-config.yaml
+6
View File
@@ -103,6 +103,12 @@ repos:
language: system
files: ^\.github/workflows/.*\.yml$
pass_filenames: false
- id: check-rust-ext-cache-prefix
name: check rust-ext cache key prefix defaults match
entry: python3 scripts/ci/check_rust_ext_cache_prefix.py
language: system
files: ^(\.github/actions/download-rust-ext/action\.yml|\.github/workflows/_pr-test-rust-ext-build\.yml)$
pass_filenames: false
- id: check-registered-tests
name: validate registered test CI registries
entry: python3 scripts/ci/check_registered_tests.py
python/sglang/srt/model_loader/ci_weight_validation.py
-908
View File
@@ -31,92 +31,6 @@ 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:
"""
@@ -248,828 +162,6 @@ def _write_per_run_marker(
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/<commit_hash>
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:
scripts/ci/check_rust_ext_cache_prefix.py
Executable
+39
View File
@@ -0,0 +1,39 @@
#!/usr/bin/env python3
"""Check that the rust-ext cache_key_prefix defaults stay in sync.
The build workflow saves cache entries under its default; the download action
restores with its own. Neither file can reference the other, and a mismatch
makes every pool silently fall back to source builds at install time.
"""
import sys
import yaml
BUILD_WORKFLOW = ".github/workflows/_pr-test-rust-ext-build.yml"
DOWNLOAD_ACTION = ".github/actions/download-rust-ext/action.yml"
def main() -> int:
with open(BUILD_WORKFLOW, encoding="utf-8") as f:
workflow = yaml.safe_load(f)
with open(DOWNLOAD_ACTION, encoding="utf-8") as f:
action = yaml.safe_load(f)
# yaml 1.1 parses the `on:` key as boolean True
triggers = workflow.get("on", workflow.get(True))
save_prefix = triggers["workflow_call"]["inputs"]["cache_key_prefix"]["default"]
restore_prefix = action["inputs"]["cache_key_prefix"]["default"]
if save_prefix != restore_prefix:
print("ERROR: rust-ext cache_key_prefix defaults do not match.")
print(f" {BUILD_WORKFLOW} saves under: {save_prefix}")
print(f" {DOWNLOAD_ACTION} restores with: {restore_prefix}")
print("Bump both together, or every pool falls back to source builds.")
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
scripts/ci/cuda/ci_install_dependency.sh
+18 -13
View File
@@ -223,7 +223,10 @@ setup_cargo_cache() {
}
setup_pip_toolchain() {
python3 -m pip install --upgrade pip
if [ "$USE_VENV" = "1" ]; then
# The bootstrap upgrade hit system pip; this upgrades the venv's own.
python3 -m pip install --upgrade pip
fi
if [ "$USE_VENV" != "1" ]; then
export UV_SYSTEM_PYTHON=1
@@ -468,7 +471,6 @@ install_sglang_kernel() {
install_sglang_router() {
$PIP_CMD install sglang-router $PIP_INSTALL_SUFFIX
$PIP_CMD list
mark_step_done "${FUNCNAME[0]}"
}
@@ -632,9 +634,9 @@ prepare_runner() {
setup_ld_library_path() {
# NVIDIA pip packages and torch ship .so files under site-packages that are
# not on the default LD_LIBRARY_PATH.
# not on the default LD_LIBRARY_PATH; lib/ always nests under nvidia/.
SITE_PACKAGES=$(python3 -c "import site, sys; print(site.getsitepackages()[0])")
NVIDIA_LIBS=$(find "$SITE_PACKAGES" -path "*/nvidia/*/lib" -type d 2>/dev/null | tr '\n' ':')
NVIDIA_LIBS=$( (find "$SITE_PACKAGES/nvidia" -type d -name lib 2>/dev/null || true) | tr '\n' ':')
TORCH_LIB="$SITE_PACKAGES/torch/lib"
VENV_LD="${NVIDIA_LIBS}${TORCH_LIB}"
export LD_LIBRARY_PATH="${VENV_LD}${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}"
@@ -652,13 +654,18 @@ setup_ld_library_path() {
verify_imports() {
$PIP_CMD list
python3 -c "import torch; print(torch.version.cuda)"
python3 -c "import cutlass; import cutlass.cute;"
# A shadowed sglang still imports, so without this the failure only surfaces
# as a missing submodule during the test step. find_spec, not import: the
# finders alone answer this and importing would pull in torch for nothing.
# One process; torch/cutlass do not import sglang, so the find_spec check
# still runs ahead of any sglang import.
SGLANG_EXPECTED_INIT="${REPO_ROOT}/python/sglang/__init__.py" python3 -c '
import torch
print(torch.version.cuda)
import cutlass
import cutlass.cute
# A shadowed sglang still imports, so without this the failure only surfaces
# as a missing submodule during the test step. find_spec, not import: the
# finders alone answer this without importing sglang.
import importlib.util, os
want = os.environ["SGLANG_EXPECTED_INIT"]
spec = importlib.util.find_spec("sglang")
@@ -671,11 +678,9 @@ if spec.origin != want:
"something in site-packages is shadowing the checkout"
)
print(f"sglang resolves to {spec.origin}")
'
# Import, not find_spec: the finders locate an extension without dlopening it,
# so a .so that cannot load passes find_spec and only fails inside some suite.
python3 -c '
# Import, not find_spec: the finders locate an extension without dlopening it,
# so a .so that cannot load passes find_spec and only fails inside some suite.
import importlib
for mod in ("server", "grpc", "multimodal"):
name = f"sglang.srt.{mod}._core"
scripts/ci/cuda/prepare_runner.sh
+3 -7
View File
@@ -1,5 +1,5 @@
#!/bin/bash
# Prepare the CI runner by cleaning up stale HuggingFace cache artifacts and validating models
# Prepare the CI runner by cleaning up stale HuggingFace cache artifacts
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
@@ -7,13 +7,9 @@ SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
echo "Preparing CI runner..."
echo ""
# Clean up stale HuggingFace cache artifacts from previous failed downloads
# Clean up stale HuggingFace cache artifacts from previous failed downloads.
# No prevalidation: launch/load-time validation covers and repairs each cache.
python3 "${SCRIPT_DIR}/../utils/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}/../utils/prevalidate_cached_models.py"
echo ""
echo "CI runner preparation complete!"
scripts/ci/utils/cleanup_hf_cache.py
+17 -6
View File
@@ -2,13 +2,14 @@
"""
Clean up stale HuggingFace cache artifacts from previous failed downloads.
This script removes incomplete marker files, temporary files, and lock files
from the HuggingFace cache directory. These artifacts can accumulate from
interrupted or failed downloads and may interfere with future downloads.
This script removes incomplete marker files and temporary files from the
HuggingFace cache directory. These artifacts can accumulate from interrupted
or failed downloads and may interfere with future downloads.
"""
import os
import sys
import time
from pathlib import Path
from typing import List
@@ -46,16 +47,26 @@ def find_stale_artifacts(cache_dir: str) -> List[Path]:
if not cache_path.exists():
return []
# Patterns for stale files to clean up
# Not "**/*.lock": another container may hold a download lock for 30+ min;
# unlinking it gives the next acquirer a fresh inode and both proceed.
patterns = [
"**/*.incomplete", # Incomplete download markers
"**/*.tmp", # Temporary files
"**/*.lock", # Lock files from interrupted downloads
]
# Another container may still be appending to its blobs/*.incomplete.
min_stale_age_seconds = 2 * 60 * 60
stale_files = []
now = time.time()
for pattern in patterns:
stale_files.extend(cache_path.glob(pattern))
for path in cache_path.glob(pattern):
try:
if now - path.stat().st_mtime < min_stale_age_seconds:
continue
except OSError:
continue # vanished mid-scan
stale_files.append(path)
return stale_files
scripts/ci/utils/prevalidate_cached_models.py
-407
View File
@@ -1,407 +0,0 @@
#!/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.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()
scripts/ci/utils/stage_rust_ext_modules.sh
+14 -4
View File
@@ -12,15 +12,25 @@ shopt -s nullglob
# module would silently shift the archive layout.
rm -rf rust-ext-staging
built=()
# Same suffix set across pkgs, or one ABI's Rust-server tests silently skip.
expected_suffixes=""
for pkg in server grpc multimodal; do
found=(python/sglang/srt/"${pkg}"/_core*.so)
if [ ${#found[@]} -ne 1 ]; then
echo "::error::expected exactly one extension module for ${pkg}, found ${#found[@]}"
if [ ${#found[@]} -eq 0 ]; then
echo "::error::no extension module found for ${pkg}"
exit 1
fi
suffixes=$(printf '%s\n' "${found[@]##*/_core}" | sort)
if [ -z "${expected_suffixes}" ]; then
expected_suffixes="${suffixes}"
elif [ "${suffixes}" != "${expected_suffixes}" ]; then
echo "::error::extension modules for ${pkg} do not match server's interpreter set"
printf 'have:\n%s\nwant:\n%s\n' "${suffixes}" "${expected_suffixes}"
exit 1
fi
mkdir -p "rust-ext-staging/${pkg}"
cp "${found[0]}" "rust-ext-staging/${pkg}/"
built+=("${found[0]}")
cp "${found[@]}" "rust-ext-staging/${pkg}/"
built+=("${found[@]}")
done
max_allowed="${MAX_GLIBC:-}"
[ -n "${max_allowed}" ] || exit 0