Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion python/sglang/srt/model_loader/ci_weight_validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -412,8 +412,11 @@ def _validate_sharded_model(
corrupted_files.append(f)

if group_info["suffix"] == "safetensors":
# Qwen3.5 names shards `model.safetensors-00001-of-00014.safetensors`,
# so the prefix may already end with the suffix.
index_stem = group_info["prefix"].removesuffix(".safetensors")
index_file = os.path.join(
snapshot_dir, f"{group_info['prefix']}.safetensors.index.json"
snapshot_dir, f"{index_stem}.safetensors.index.json"
)
if not os.path.exists(index_file):
return (
Expand Down
23 changes: 17 additions & 6 deletions python/sglang/srt/model_loader/weight_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import safetensors.torch
import torch
from huggingface_hub import HfFileSystem, hf_hub_download, snapshot_download
from huggingface_hub.errors import HfHubHTTPError
from pydantic import BaseModel, ConfigDict, ValidationInfo, model_validator
from tqdm.auto import tqdm

Expand Down Expand Up @@ -521,11 +522,10 @@ def _find_local_hf_snapshot_dir_unlocked(
),
)
rev_to_use = revision
if not rev_to_use:
ref_main = os.path.join(repo_folder, "refs", "main")
if os.path.isfile(ref_main):
with open(ref_main) as f:
rev_to_use = f.read().strip()
ref_path = os.path.join(repo_folder, "refs", revision or "main")
if os.path.isfile(ref_path):
with open(ref_path) as f:
rev_to_use = f.read().strip()
if rev_to_use:
rev_dir = os.path.join(repo_folder, "snapshots", rev_to_use)
if os.path.isdir(rev_dir):
Expand Down Expand Up @@ -664,7 +664,18 @@ def download_weights_from_hf(
if not huggingface_hub.constants.HF_HUB_OFFLINE:
# Before we download we look at what is available:
fs = HfFileSystem()
file_list = fs.ls(model_name_or_path, detail=False, revision=revision)
try:
file_list = fs.ls(model_name_or_path, detail=False, revision=revision)
except HfHubHTTPError as e:
# Fail open (e.g. a 429 rate limit): pick the format from the local
# snapshot; snapshot_download below re-raises errors it cannot recover.
logger.warning(
"Listing %s on the Hub failed, using the local snapshot: %s",
model_name_or_path,
e,
)
local_dir = find_local_repo_dir(model_name_or_path, revision)
file_list = os.listdir(local_dir) if local_dir else []

# depending on what is available we download different things
for pattern in allow_patterns:
Expand Down
12 changes: 6 additions & 6 deletions python/sglang/srt/utils/common.py
Original file line number Diff line number Diff line change
Expand Up @@ -4256,12 +4256,12 @@ def find_local_repo_dir(repo_id: str, revision: Optional[str] = None) -> Optiona
hf.constants.REPO_ID_SEPARATOR.join(["models", *repo_id.split("/")]),
)

# Get revision from main ref if not specified
if not revision:
ref_path = os.path.join(cache_path, "refs", "main")
if os.path.isfile(ref_path):
with open(ref_path) as f:
revision = f.read().strip()
# A branch or tag name (default "main") maps to a commit through refs/;
# snapshots/ is keyed by commit only.
ref_path = os.path.join(cache_path, "refs", revision or "main")
if os.path.isfile(ref_path):
with open(ref_path) as f:
revision = f.read().strip()

# List files from revision directory
if revision:
Expand Down
148 changes: 124 additions & 24 deletions python/sglang/test/test_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -401,18 +401,7 @@ def _try_enable_offline_mode_if_cache_complete(
) -> Optional[str]:
"""Set HF_HUB_OFFLINE=1 in `env` if the model cache validates; return the
per-run marker path, or None if offline mode was not enabled.

Markers are per-run and not shared across runners. A marker is only a hint;
the current launch's requirements (e.g. hf_quant_config.json) are revalidated.
"""
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 []

# Skip offline mode for LoRA scenarios (dynamic adapter loading may need online access)
Expand All @@ -428,6 +417,40 @@ def _try_enable_offline_mode_if_cache_complete(
)
return None

# Detect before the marker check so the current launch's 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

marker_path = _validate_cache_for_offline(
model_name_or_path, requires_hf_quant_config
)
if marker_path is not None:
env["HF_HUB_OFFLINE"] = "1"
return marker_path


def _validate_cache_for_offline(
model_name_or_path: str, requires_hf_quant_config: bool
) -> Optional[str]:
"""Return the per-run marker path if the model's local cache is complete
enough to load offline, else None.

Markers are per-run and not shared across runners. A marker is only a hint;
the current launch's requirements (e.g. hf_quant_config.json) are revalidated.
"""
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

# Skip if already a local path
if os.path.isdir(model_name_or_path):
return None
Expand All @@ -440,15 +463,6 @@ def _try_enable_offline_mode_if_cache_complete(
except Exception:
return None

# Detect before the marker check so the current launch's 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:
Expand All @@ -466,7 +480,6 @@ def _try_enable_offline_mode_if_cache_complete(
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 "
Expand All @@ -487,9 +500,6 @@ def _try_enable_offline_mode_if_cache_complete(
)
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)

Expand All @@ -506,6 +516,94 @@ def _try_enable_offline_mode_if_cache_complete(
return marker_path


@contextlib.contextmanager
def _hf_offline_if_cache_complete(engine_kwargs: dict):
"""Run an in-process Engine startup with HF_HUB_OFFLINE=1 when every Hub
repo it loads validates in the local cache.

A scheduler that fails at startup kills the test process, so there is no
online retry as in popen_launch_server; any doubt keeps the startup online.
"""
import huggingface_hub.constants

if "server_args" in engine_kwargs:
server_args = engine_kwargs["server_args"]
model_path = server_args.model_path
tokenizer_path = server_args.tokenizer_path
draft_path = server_args.speculative_draft_model_path
quantization = server_args.quantization
uses_lora = server_args.enable_lora or bool(server_args.lora_paths)
else:
model_path = engine_kwargs.get("model_path")
tokenizer_path = engine_kwargs.get("tokenizer_path")
draft_path = engine_kwargs.get("speculative_draft_model_path")
quantization = engine_kwargs.get("quantization")
uses_lora = bool(
engine_kwargs.get("enable_lora") or engine_kwargs.get("lora_paths")
)

# Adapters can be loaded by repo id after startup, from an offline scheduler.
if uses_lora or os.environ.get("HF_HUB_OFFLINE") == "1":
yield
return

requires_hf_quant_config = quantization in [
"modelopt_fp4",
"modelopt_fp8",
"modelopt",
]
repo_ids = {
path
for path in (model_path, tokenizer_path, draft_path)
if path is not None and not os.path.isdir(path)
}
if not repo_ids or not all(
_validate_cache_for_offline(
repo_id, requires_hf_quant_config and repo_id == model_path
)
for repo_id in repo_ids
):
yield
return

# Spawned children read the env at import; this process already imported
# huggingface_hub, which reads it once into a module constant.
prev_env = os.environ.get("HF_HUB_OFFLINE")
prev_constant = huggingface_hub.constants.HF_HUB_OFFLINE
os.environ["HF_HUB_OFFLINE"] = "1"
huggingface_hub.constants.HF_HUB_OFFLINE = True
try:
yield
finally:
if prev_env is None:
os.environ.pop("HF_HUB_OFFLINE", None)
else:
os.environ["HF_HUB_OFFLINE"] = prev_env
huggingface_hub.constants.HF_HUB_OFFLINE = prev_constant


def _install_ci_engine_offline_hook():
"""In CI, wrap Engine.__init__ once per process with _hf_offline_if_cache_complete."""
try:
from sglang.srt.entrypoints.engine import Engine
except ImportError as e:
# A test that never builds an Engine must not fail on this.
print(f"CI_OFFLINE: Engine import failed, skip the offline hook: {e}")
return

orig_init = Engine.__init__
if getattr(orig_init, "_ci_offline_wrapped", False):
return

@wraps(orig_init)
def init(self, **kwargs):
with _hf_offline_if_cache_complete(kwargs):
orig_init(self, **kwargs)

init._ci_offline_wrapped = True
Engine.__init__ = init


def _create_clean_subprocess_env(env: dict) -> dict:
"""Create a clean subprocess environment without internal CI keys."""
child_env = env.copy()
Expand Down Expand Up @@ -2301,6 +2399,8 @@ def __init_subclass__(cls, **kwargs):
def safe_setUpClass(klass):
try:
_wait_for_gpu_idle_in_ci()
if is_in_ci():
_install_ci_engine_offline_hook()
orig_func(klass)
except Exception:
# Best-effort cleanup; suppress teardown errors so the
Expand Down
12 changes: 0 additions & 12 deletions scripts/ci/cuda/ci_install_dependency.sh
Original file line number Diff line number Diff line change
Expand Up @@ -820,17 +820,6 @@ install_extra_deps() {
mark_step_done "${FUNCNAME[0]}"
}

install_test_tools() {
# Download kernels from kernels community
kernels download python || true
kernels lock python || true
[ -e "${HOME}/.cache/sglang" ] && [ ! -d "${HOME}/.cache/sglang" ] && rm -f "${HOME}/.cache/sglang"
mkdir -p "${HOME}/.cache/sglang/"
mv python/kernels.lock "${HOME}/.cache/sglang/" || true

mark_step_done "${FUNCNAME[0]}"
}

prepare_runner() {
bash "${SCRIPT_DIR}/prepare_runner.sh"

Expand Down Expand Up @@ -950,7 +939,6 @@ main() {
download_flashinfer_cache
stabilize_flashinfer_jit_paths
install_extra_deps
install_test_tools
install_nccl
prepare_runner
setup_ld_library_path
Expand Down
20 changes: 20 additions & 0 deletions test/manual/test_weight_validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,26 @@ def test_validate_sharded_model_all_present(self):
self.assertIsNone(error_msg)
self.assertEqual(corrupted_files, [])

def test_validate_sharded_model_suffixed_shard_prefix(self):
"""Shards named `model.safetensors-0000N-of-0000M.safetensors` (Qwen3.5)
pair with `model.safetensors.index.json`."""
with tempfile.TemporaryDirectory() as tmpdir:
names = [f"model.safetensors-0000{i}-of-00002.safetensors" for i in [1, 2]]
header = b'{"__metadata__":{}}'
for name in names:
with open(os.path.join(tmpdir, name), "wb") as f:
f.write(struct.pack("<Q", len(header)))
f.write(header)
with open(os.path.join(tmpdir, "model.safetensors.index.json"), "w") as f:
json.dump({"weight_map": {"a": names[0], "b": names[1]}}, f)

is_valid, error_msg, corrupted_files = _validate_sharded_model(
tmpdir, [os.path.join(tmpdir, name) for name in names]
)

self.assertTrue(is_valid, error_msg)
self.assertEqual(corrupted_files, [])

def test_validate_sharded_model_corrupted_shard(self):
"""
Test that corrupted shards are detected and returned in corrupted_files.
Expand Down
Loading