Record Frontier trace stalls without ranking them
This commit is contained in:
@@ -68,6 +68,7 @@ def parse_args() -> argparse.Namespace:
|
||||
parser.add_argument("--allreduce-csv", type=Path)
|
||||
parser.add_argument("--timeout-seconds", type=float, default=1800.0)
|
||||
parser.add_argument("--resume", action="store_true")
|
||||
parser.add_argument("--continue-on-failure", action="store_true")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
@@ -156,6 +157,12 @@ def trace_manifest_entry(trace: dict[str, Any]) -> dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
def classify_frontier_failure(stderr: str) -> str:
|
||||
if "Sequential simulation ended with non-empty scheduler state" in stderr:
|
||||
return "scheduler_stall"
|
||||
return "frontier_error"
|
||||
|
||||
|
||||
def score(path: Path, expected_shapes: list[tuple[int, int]]) -> dict[str, Any]:
|
||||
with path.open(newline="") as source:
|
||||
rows = list(csv.DictReader(source))
|
||||
@@ -312,9 +319,48 @@ def main() -> None:
|
||||
check=False,
|
||||
)
|
||||
if completed.returncode != 0:
|
||||
raise RuntimeError(
|
||||
f"Frontier failed for {config.name}/{trace['label']}: {completed.returncode}"
|
||||
stderr_path = run_dir / "stderr.log"
|
||||
result = {
|
||||
"status": "frontier_failed",
|
||||
"failure_kind": classify_frontier_failure(
|
||||
stderr_path.read_text(errors="replace")
|
||||
),
|
||||
"returncode": completed.returncode,
|
||||
"config": {
|
||||
"tp": config.tp,
|
||||
"mns": config.mns,
|
||||
"name": config.name,
|
||||
},
|
||||
"trace_label": trace["label"],
|
||||
"offered_request_rate": trace["offered_request_rate"],
|
||||
"offered_request_rate_per_gpu": (
|
||||
trace["offered_request_rate"] / config.tp
|
||||
),
|
||||
"request_count": trace["requests"],
|
||||
"elapsed_seconds": time.time() - started,
|
||||
"trace_sha256": trace["sha256"],
|
||||
"stderr_sha256": BASE.sha256(stderr_path),
|
||||
}
|
||||
BASE.write_json(result_path, result)
|
||||
loads.append(result)
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"config": config.name,
|
||||
"trace": trace["label"],
|
||||
"status": result["status"],
|
||||
"failure_kind": result["failure_kind"],
|
||||
},
|
||||
sort_keys=True,
|
||||
),
|
||||
flush=True,
|
||||
)
|
||||
if not args.continue_on_failure:
|
||||
raise RuntimeError(
|
||||
f"Frontier failed for {config.name}/{trace['label']}: "
|
||||
f"{completed.returncode}"
|
||||
)
|
||||
continue
|
||||
metrics = find_request_metrics(run_dir)
|
||||
result = {
|
||||
"status": "completed",
|
||||
@@ -353,21 +399,38 @@ def main() -> None:
|
||||
for slo in (f"tpot_{int(value)}ms" for value in TPOT_SLOS_MS):
|
||||
records = []
|
||||
for item in config_results:
|
||||
completed_loads = [
|
||||
load for load in item["loads"] if load["status"] == "completed"
|
||||
]
|
||||
invalid_loads = [
|
||||
load for load in item["loads"] if load["status"] != "completed"
|
||||
]
|
||||
feasible = [
|
||||
load["offered_request_rate"]
|
||||
for load in item["loads"]
|
||||
for load in completed_loads
|
||||
if load["score"]["slos"][slo]["feasible"]
|
||||
]
|
||||
capacity = max(feasible, default=None)
|
||||
records.append(
|
||||
{
|
||||
"config": item["config"],
|
||||
"ranking_valid": not invalid_loads,
|
||||
"invalid_loads": [
|
||||
{
|
||||
"trace_label": load["trace_label"],
|
||||
"failure_kind": load.get("failure_kind", "unknown"),
|
||||
}
|
||||
for load in invalid_loads
|
||||
],
|
||||
"maximum_tested_feasible_request_rate": capacity,
|
||||
"maximum_tested_feasible_request_rate_per_gpu": (
|
||||
capacity / item["config"]["tp"] if capacity is not None else None
|
||||
),
|
||||
"lower_censored": capacity is None,
|
||||
"upper_censored": capacity == traces[-1]["offered_request_rate"],
|
||||
"lower_censored": capacity is None and not invalid_loads,
|
||||
"upper_censored": (
|
||||
capacity == traces[-1]["offered_request_rate"]
|
||||
and not invalid_loads
|
||||
),
|
||||
}
|
||||
)
|
||||
records.sort(
|
||||
@@ -382,9 +445,20 @@ def main() -> None:
|
||||
)
|
||||
rankings[slo] = records
|
||||
|
||||
has_invalid_cells = any(
|
||||
load["status"] != "completed"
|
||||
for item in config_results
|
||||
for load in item["loads"]
|
||||
)
|
||||
if selected != list(BASE.GRID):
|
||||
manifest_status = "partial_not_decision_bearing"
|
||||
elif has_invalid_cells:
|
||||
manifest_status = "frozen_with_invalid_cells"
|
||||
else:
|
||||
manifest_status = "frozen_before_real"
|
||||
manifest = {
|
||||
"schema": "frontier-qwen30-exact-trace-surface-v1",
|
||||
"status": "frozen_before_real" if selected == list(BASE.GRID) else "partial_not_decision_bearing",
|
||||
"status": manifest_status,
|
||||
"contract": {
|
||||
"window_seconds": (
|
||||
WINDOW_SECONDS if args.rate_contract == "trace-window" else None
|
||||
|
||||
@@ -106,6 +106,16 @@ class FidelityEnvelopeTest(unittest.TestCase):
|
||||
|
||||
def test_exact_trace_parser_and_joint_slo_score(self) -> None:
|
||||
module = load("run_frontier_qwen30_exact_trace_surface.py")
|
||||
self.assertEqual(
|
||||
module.classify_frontier_failure(
|
||||
"Sequential simulation ended with non-empty scheduler state"
|
||||
),
|
||||
"scheduler_stall",
|
||||
)
|
||||
self.assertEqual(
|
||||
module.classify_frontier_failure("unexpected failure"),
|
||||
"frontier_error",
|
||||
)
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
root = Path(temporary)
|
||||
trace = root / "trace.csv"
|
||||
|
||||
Reference in New Issue
Block a user