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
50 changes: 50 additions & 0 deletions docs/source/features/checkpoint-loading.md
Original file line number Diff line number Diff line change
Expand Up @@ -221,6 +221,56 @@ performance, check that both groups use comparable runtime images, hardware,
CI configurations, and cache conditions—otherwise, those differences could be
mistaken for a loader improvement.

The primary model's finalized policy is also published as
`startup_metrics.model_loader.checkpoint_io_policy` in `/server_info`.
This detached snapshot is captured after weight-session finalization, before a
separate draft checkpoint can reuse the loader. It preserves activation even
when a subsequent advisory read failure makes the effective policy native.
HF-derived formats such as Mistral and MX also report observed native disk
loads; this does not enable rank-striped read-ahead for those formats.
GMS writers using the built-in HF disk path capture the native policy after
target materialization, before draft loading; GMS readers remain unknown.
Unsupported, preloaded, or unobserved paths report an explicit unknown (`null`),
not an inferred native policy. The collector prefers this structured primary
status; only older runtimes without the field use `legacy_logs` fallback.
Draft statuses do not override the structured primary status.

Use the following fields to qualify a comparison (with `ctx_` or `gen_`
prefixes for disaggregated server roles):

- `s_checkpoint_io_policy_source`: `server_info`, `legacy_logs`, or `unknown`.
- `b_checkpoint_io_policy_complete`: every expected server provided one valid
structured primary policy. Legacy log attribution is not considered complete.
- `b_startup_metrics_timing_complete`: every expected server provided finite,
nonnegative preparation, population, finalization, and total model-loading
times. The existing collection-complete flag only checks successful fetches
and server counts; it does not imply these fields were present.

CI records include image provenance so startup comparisons can account for
runtime-environment differences instead of attributing them to the loader.
This describes the benchmark environment, not which images customers deploy.

The existing `s_runtime_image`, `s_runtime_image_source`,
`s_runtime_image_digest`, and `s_runtime_image_identity_strength` fields
describe the original OCI/base-image reference selected for the test
container, not the Jenkins dispatcher image. A valid digest-pinned reference
reports `digest_pinned` for that OCI reference only; a tagged reference reports
`reference_only` with an unknown digest. Missing provenance remains `unknown`.

Enroot may launch a derived fat `.sqsh` with dependencies and TRT-LLM already
installed. These fields do not identify or verify that derived artifact or
the environment after package installation. A tag hash or `.sqsh` cache
filename is not a content digest. Group comparisons by the recorded base
reference as well as model/configuration, GPU and parallelism, but do not
treat matching base references as proof of identical runtime environments.
Final launch-artifact provenance remains a separate follow-up.

These additions do not change loader policy or add startup collectives. Timing
remains rank-0-based per server, and failed launches are not comprehensively
persisted. Native counterfactual eligibility, checkpoint fingerprints,
slowest-rank timing, and process-launch-to-first-token measurement remain
separate follow-ups.

This policy remains separate from ModelStreamer, MX, GMS, or snapshot
integrations. Those systems may change the source or bypass raw loading without
requiring a new combined checkpoint format.
Expand Down
13 changes: 10 additions & 3 deletions jenkins/L0_Test.groovy
Original file line number Diff line number Diff line change
Expand Up @@ -2010,7 +2010,9 @@ def runLLMTestlistWithAgent(pipeline, platform, testList, config=VANILLA_CONFIG,
return err
}
try {
executeLLMTestOnSlurm(pipeline, platform, testList, config, stageName, splitId, splits, skipInstallWheel, slurmRunner, postTag, useClusterDurations, retryContext, classifySlurmFailure)
withEnv(["TRTLLM_CI_RUNTIME_IMAGE=${LLM_DOCKER_IMAGE}"]) {
executeLLMTestOnSlurm(pipeline, platform, testList, config, stageName, splitId, splits, skipInstallWheel, slurmRunner, postTag, useClusterDurations, retryContext, classifySlurmFailure)
}
} catch (InterruptedException e) {
throw e
} catch (Exception e) {
Expand Down Expand Up @@ -2949,7 +2951,8 @@ def runLLMTestlistWithSbatch(pipeline, platform, testList, config=VANILLA_CONFIG
'JOB_NAME',
'globalVars',
'gitlabCommit',
'TRTLLM_PERF_SANITY_CHECKPOINT_IO_POLICY'
'TRTLLM_PERF_SANITY_CHECKPOINT_IO_POLICY',
'TRTLLM_CI_RUNTIME_IMAGE'
]
def srunArgs = [
"--container-name=multi_node_test-\${SLURM_JOB_ID}",
Expand All @@ -2975,7 +2978,9 @@ def runLLMTestlistWithSbatch(pipeline, platform, testList, config=VANILLA_CONFIG
}

def envExportStatements = envVarNames.collect { varName ->
def escapedValue = env."${varName}"?.toString() ?: ''
// Original OCI/base reference; derived .sqsh identity is not captured.
def value = varName == 'TRTLLM_CI_RUNTIME_IMAGE' ? LLM_DOCKER_IMAGE : env."${varName}"
def escapedValue = value?.toString() ?: ''
escapedValue = escapedValue
.replace('\\', '\\\\') // Backslash
.replace('"', '\\"') // Double quote
Expand Down Expand Up @@ -4967,6 +4972,8 @@ ${scratchUnreachableNodeAffinity}
valueFrom:
fieldRef:
fieldPath: spec.nodeName
- name: TRTLLM_CI_RUNTIME_IMAGE
value: "${image}"
${extraDeviceEnv}
${serviceContainerConfig}
- name: jnlp
Expand Down
57 changes: 54 additions & 3 deletions tensorrt_llm/_torch/pyexecutor/model_loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -869,10 +869,14 @@ class ModelLoaderMetricNames(Enum):
WEIGHT_MANIFEST_SECONDS = "weight_manifest_seconds"


_CheckpointStartupMetadata = dict[str,
str | dict[str, str | bool | None] | None]
Comment thread
chienchunhung marked this conversation as resolved.
Resolved


def _checkpoint_startup_metadata(
checkpoint_loader: BaseCheckpointLoader,
load_format: LoadFormat,
) -> dict[str, str]:
) -> _CheckpointStartupMetadata:
"""Describe the concrete checkpoint source selected for this load."""
weight_loader = getattr(checkpoint_loader, "weight_loader", None)
source_kind = getattr(checkpoint_loader, "checkpoint_format", None)
Expand All @@ -886,6 +890,42 @@ def _checkpoint_startup_metadata(
str(source_kind or "unknown"),
"load_format":
load_format.name.lower(),
# Explicitly unknown until a supported weight load has finished.
"checkpoint_io_policy":
None,
Comment thread
chienchunhung marked this conversation as resolved.
Resolved
}


def _checkpoint_io_policy_metadata(
checkpoint_loader: BaseCheckpointLoader
) -> dict[str, str | bool | None] | None:
"""Snapshot observed policy for loads using the built-in HF weight loader.

Args:
checkpoint_loader: Loader whose completed weight load is observed.

Returns:
A detached policy snapshot, or None for unsupported or unobserved loads.
"""
from tensorrt_llm._torch.models.checkpoints.hf.checkpoint_loader import \
HfCheckpointLoader
from tensorrt_llm._torch.models.checkpoints.hf.weight_loader import \
HfWeightLoader

if (not isinstance(checkpoint_loader, HfCheckpointLoader)
or type(checkpoint_loader.weight_loader) is not HfWeightLoader
or checkpoint_loader.is_weights_preloaded()):
return None
status = checkpoint_loader.weight_loader.last_checkpoint_io_status
if status.effective not in (CheckpointIOPolicy.NATIVE,
CheckpointIOPolicy.RANK_STRIPED_READ_AHEAD):
return None
return {
"requested": status.requested,
"selected": status.selected,
"activated": status.activated,
"effective": status.effective,
"fallback_reason": status.fallback_reason,
}


Expand Down Expand Up @@ -1054,15 +1094,15 @@ def __init__(self,
self._checkpoint_loader: Optional[BaseCheckpointLoader] = None
# Mostly weight loading and processing time metrics, updated when load() is called.
self._metrics: dict[str, float] = {}
self._startup_metadata: dict[str, str] = {}
self._startup_metadata: _CheckpointStartupMetadata = {}

@property
def metrics(self) -> dict[str, float]:
"""Return weight loading and processing time metrics."""
return self._metrics

@property
def startup_metadata(self) -> dict[str, str]:
def startup_metadata(self) -> _CheckpointStartupMetadata:
"""Return checkpoint-source metadata captured by the latest load."""
return self._startup_metadata

Expand Down Expand Up @@ -1575,6 +1615,9 @@ def init_meta_tensor_in_pool(t: torch.Tensor):
"commit an unpopulated model to the GMS "
"pool.")

self._startup_metadata["checkpoint_io_policy"] = \
_checkpoint_io_policy_metadata(checkpoint_loader)

if loads_draft_weights:
self._load_separate_draft_weights(
model, checkpoint_loader)
Expand Down Expand Up @@ -2430,6 +2473,7 @@ def _materialize_checkpoint_weights(self, checkpoint_loader,
checkpoint_dir: str, model, config,
load_weights_kwargs: dict) -> bool:
"""Keep loader-specific work alive through weight materialization."""
self._startup_metadata["checkpoint_io_policy"] = None
with _timed_checkpoint_weight_session(
checkpoint_loader, checkpoint_dir, self._metrics,
ModelLoaderMetricNames.CHECKPOINT_PREPARATION_SECONDS.value,
Expand All @@ -2452,11 +2496,16 @@ def _materialize_checkpoint_weights(self, checkpoint_loader,
self.weight_mapper)
if torch.cuda.is_available(): # CPU guard
torch.cuda.synchronize()
# Session finalization can degrade activated read-ahead to native.
# Snapshot before a same-engine draft load resets the shared loader.
self._startup_metadata["checkpoint_io_policy"] = \
_checkpoint_io_policy_metadata(checkpoint_loader)
return weights_preloaded

def _materialize_draft_checkpoint_weights(self, checkpoint_loader,
model) -> None:
"""Keep loader-specific work alive through draft materialization."""
self._startup_metadata["draft_checkpoint_io_policy"] = None
with _timed_checkpoint_weight_session(
checkpoint_loader,
self.spec_config.speculative_model,
Expand Down Expand Up @@ -2489,6 +2538,8 @@ def _materialize_draft_checkpoint_weights(self, checkpoint_loader,
draft_weight_mapper)
if torch.cuda.is_available(): # CPU guard
torch.cuda.synchronize()
self._startup_metadata["draft_checkpoint_io_policy"] = \
_checkpoint_io_policy_metadata(checkpoint_loader)

def _call_load_weights(self,
load_method: Callable,
Expand Down
11 changes: 11 additions & 0 deletions tests/integration/defs/perf/perf_regression_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,13 @@ def get_job_info():
commit = os.getenv("gitlabCommit", "")
primary = global_vars.get("llm_artifacts_primary") or "aws"
artifact_base = _ARTIFACT_BASES.get(primary, _ARTIFACT_BASES["aws"])
# Original OCI/base reference, not the derived Enroot artifact or dispatcher.
runtime_image = os.getenv("TRTLLM_CI_RUNTIME_IMAGE", "").strip()
digest_match = re.fullmatch(r"[^@\s]+@(sha256:[0-9a-f]{64})", runtime_image)
runtime_image_digest = digest_match.group(1) if digest_match else "unknown"
runtime_image_identity_strength = (
"digest_pinned" if digest_match else "reference_only" if runtime_image else "unknown"
)

# Initialize PR-specific fields
trigger_mr_user = ""
Expand Down Expand Up @@ -123,6 +130,10 @@ def get_job_info():
"s_job_url": job_url,
"s_branch": branch,
"s_commit": commit,
"s_runtime_image": runtime_image or "unknown",
"s_runtime_image_source": "TRTLLM_CI_RUNTIME_IMAGE" if runtime_image else "unknown",
"s_runtime_image_digest": runtime_image_digest,
"s_runtime_image_identity_strength": runtime_image_identity_strength,
"b_is_post_merge": is_post_merge,
"b_is_pr_job": is_pr_job,
"s_trigger_mr_user": trigger_mr_user,
Expand Down
78 changes: 76 additions & 2 deletions tests/integration/defs/perf/test_perf_sanity.py
Original file line number Diff line number Diff line change
Expand Up @@ -1105,6 +1105,7 @@ def add_perf_metric_value(
"checkpoint_weight_loader_kind",
"checkpoint_source_kind",
"load_format",
"checkpoint_io_policy_source",
)


Expand Down Expand Up @@ -1276,6 +1277,42 @@ def parse_checkpoint_io_policies(log_paths: List[str]) -> List[dict]:
return statuses


def _normalize_checkpoint_io_policy(policy: object) -> Optional[dict]:
"""Validate a finalized primary-model status without inferring activation.
Args:
policy: Structured policy from the primary model's startup metrics.
Returns:
A normalized policy with a bounded fallback reason, or None if invalid.
"""
if not isinstance(policy, dict):
return None
policies = ("native", "rank_striped_read_ahead")
if (
policy.get("requested") not in (*policies, "auto")
or policy.get("selected") not in policies
or policy.get("effective") not in policies
or not isinstance(policy.get("activated"), bool)
or "fallback_reason" not in policy
or not isinstance(policy["fallback_reason"], (str, type(None)))
):
return None
if (policy["activated"] and policy["selected"] != "rank_striped_read_ahead") or (
policy["effective"] == "rank_striped_read_ahead" and not policy["activated"]
):
return None
reason = policy["fallback_reason"] or "none"
return {
"requested": policy["requested"],
"selected": policy["selected"],
"activated": policy["activated"],
"effective": policy["effective"],
"fallback_reason": _bounded_checkpoint_io_fallback_reason(reason),
"fallback_category": checkpoint_io_fallback_category(reason),
}


def make_startup_observation(server_info: dict, log_paths: List[str], role: str) -> dict:
"""Normalize startup data from ``/server_info`` and server logs."""
startup_metrics = server_info.get("startup_metrics", {})
Expand All @@ -1292,7 +1329,12 @@ def make_startup_observation(server_info: dict, log_paths: List[str], role: str)
continue
for metric_name in STARTUP_METRIC_NAMES:
value = loader_metrics.get(metric_name)
if isinstance(value, (int, float)) and not isinstance(value, bool):
if (
isinstance(value, (int, float))
and not isinstance(value, bool)
and math.isfinite(value)
and value >= 0
):
metrics[f"{metric_prefix}{metric_name}"] = float(value)
for metadata_name in STARTUP_METADATA_NAMES:
value = loader_metrics.get(metadata_name)
Expand All @@ -1308,11 +1350,23 @@ def make_startup_observation(server_info: dict, log_paths: List[str], role: str)
metrics[f"{metric_prefix}{name}"] for name in DRAFT_CHECKPOINT_PIPELINE_PHASES
)

primary_loader = startup_metrics.get("model_loader", {})
if isinstance(primary_loader, dict) and "checkpoint_io_policy" in primary_loader:
Comment thread
chienchunhung marked this conversation as resolved.
Resolved
# An explicit unknown status is authoritative. Do not substitute logs
# that could describe a draft load or a different initialization phase.
policy = _normalize_checkpoint_io_policy(primary_loader["checkpoint_io_policy"])
policies = [policy] if policy is not None else []
policy_source = "server_info"
else:
policies = parse_checkpoint_io_policies(log_paths)
policy_source = "legacy_logs" if policies else "unknown"
metadata["checkpoint_io_policy_source"] = policy_source

return {
"role": role,
"metrics": metrics,
"metadata": metadata,
"checkpoint_io_policies": parse_checkpoint_io_policies(log_paths),
"checkpoint_io_policies": policies,
}


Expand Down Expand Up @@ -1507,6 +1561,24 @@ def add_startup_metric_values(
)

successful = [entry for entry in selected if not entry.get("error")]
collection_complete = (
expected_server_count > 0 and len(successful) == expected_server_count and not failed
)
required_timings = (*CHECKPOINT_PIPELINE_PHASES, "total_model_loading_seconds")
new_data[f"b_{field_prefix}startup_metrics_timing_complete"] = collection_complete and all(
isinstance(value := entry.get("metrics", {}).get(name), (int, float))
and not isinstance(value, bool)
and math.isfinite(value)
and value >= 0
for entry in successful
for name in required_timings
)
new_data[f"b_{field_prefix}checkpoint_io_policy_complete"] = collection_complete and all(
entry.get("metadata", {}).get("checkpoint_io_policy_source") == "server_info"
and len(entry.get("checkpoint_io_policies", [])) == 1
and _normalize_checkpoint_io_policy(entry["checkpoint_io_policies"][0]) is not None
for entry in successful
)
metric_names = {name for entry in successful for name in entry.get("metrics", {})}
for metric_name in metric_names:
values = [entry.get("metrics", {}).get(metric_name) for entry in successful]
Expand Down Expand Up @@ -1575,6 +1647,8 @@ def add_startup_metric_values(
)
new_data[f"s_{field_prefix}checkpoint_io_experiment_classification"] = (
classify_checkpoint_io_experiment(assignment, policies)
if collection_complete and all(entry.get("checkpoint_io_policies") for entry in successful)
else "unknown"
)


Expand Down
1 change: 1 addition & 0 deletions tests/integration/test_lists/test-db/l0_cpu.yml
Original file line number Diff line number Diff line change
Expand Up @@ -340,6 +340,7 @@ l0_cpu:
- unittest/llmapi/test_kv_cache_dtype_override.py
- unittest/llmapi/test_kv_hints.py
- unittest/llmapi/test_llm_args.py
- unittest/llmapi/test_llm.py::test_startup_metrics_by_server_info
- unittest/llmapi/test_gms_args.py
- unittest/llmapi/test_mx_args.py
- unittest/llmapi/test_custom_tokenizer_aliases.py
Expand Down
Loading
Loading