diff --git a/runs/frontier-fidelity-envelope-v1/analyze_qwen235_fixed_pd_state.py b/runs/frontier-fidelity-envelope-v1/analyze_qwen235_fixed_pd_state.py index 339853f..934605b 100644 --- a/runs/frontier-fidelity-envelope-v1/analyze_qwen235_fixed_pd_state.py +++ b/runs/frontier-fidelity-envelope-v1/analyze_qwen235_fixed_pd_state.py @@ -126,7 +126,6 @@ OP_CATEGORIES = { "expert_parallel_alltoall_dispatch", "expert_parallel_alltoall_combine", "expert_parallel_allreduce", - "expert_parallel_allreduce_wait", }, "tp_dp_communication": { "attn_tensor_parallel_allreduce", @@ -153,6 +152,7 @@ OP_CATEGORIES = { "ray_comm_time", }, } +DIAGNOSTIC_OPS = {"expert_parallel_allreduce_wait"} def atomic_write(path: Path, text: str) -> None: @@ -328,7 +328,9 @@ def categorized_components(components: dict[str, Any]) -> dict[str, float]: } -def op_category(name: str) -> str: +def op_category(name: str) -> str | None: + if name in DIAGNOSTIC_OPS: + return None matches = [category for category, names in OP_CATEGORIES.items() if name in names] if len(matches) != 1: raise ValueError(f"unclassified or multiply classified op trace event: {name}") @@ -395,6 +397,7 @@ def load_op_trace_rows( { "tokens": tokens, "categories_ms": {key: 0.0 for key in CATEGORIES}, + "diagnostic_wait_ms": 0.0, "events": 0, }, ) @@ -402,36 +405,42 @@ def load_op_trace_rows( raise ValueError( f"op trace token metadata changed within batch {batch_id}" ) - entry["categories_ms"][category] += float(event["duration_ms"]) + if category is None: + entry["diagnostic_wait_ms"] += float(event["duration_ms"]) + else: + entry["categories_ms"][category] += float(event["duration_ms"]) entry["events"] += 1 if not grouped: raise ValueError(f"no op trace events: {trace_path}") - rows = [] - for batch_id, entry in sorted(grouped.items()): - tokens = entry["tokens"] - total = sum(entry["categories_ms"].values()) - rows.append( - { - "batch_id": batch_id, - "batch_size": len(tokens), - "decode_requests": sum(value == 1 for value in tokens), - "prefill_requests": sum(value > 1 for value in tokens), - "prefill_tokens": sum(value for value in tokens if value > 1), - "total_time_ms": total, - "categories_ms": entry["categories_ms"], - "per_expert_tokens": None, - } - ) - ledger_path = Path(result["state_artifacts"]["ledger"]["path"]) stage_span_by_batch = { int(row["batch_id"]): (float(row["stage_end_ts"]) - float(row["stage_start_ts"])) * 1000 for row in load_jsonl(ledger_path) } + rows = [] + for batch_id, entry in sorted(grouped.items()): + tokens = entry["tokens"] + if batch_id not in stage_span_by_batch: + raise ValueError(f"op trace batch missing from ledger: {batch_id}") + rows.append( + { + "batch_id": batch_id, + "batch_size": len(tokens), + "decode_requests": sum(value == 1 for value in tokens), + "prefill_requests": sum(value > 1 for value in tokens), + "prefill_tokens": sum(value for value in tokens if value > 1), + "total_time_ms": stage_span_by_batch[batch_id], + "categories_ms": entry["categories_ms"], + "diagnostic_wait_ms": entry["diagnostic_wait_ms"], + "per_expert_tokens": None, + } + ) + ratios = [ - row["total_time_ms"] / stage_span_by_batch[row["batch_id"]] + sum(row["categories_ms"].values()) + / stage_span_by_batch[row["batch_id"]] for row in rows if row["batch_id"] in stage_span_by_batch and stage_span_by_batch[row["batch_id"]] > 0 @@ -440,7 +449,12 @@ def load_op_trace_rows( "trace_path": str(trace_path), "trace_batches": len(rows), "ledger_batches": len(stage_span_by_batch), - "trace_total_over_stage_span": numeric(ratios), + "serialized_component_sum_over_stage_span": numeric(ratios), + "diagnostic_wait_ms": numeric( + row["diagnostic_wait_ms"] + for row in rows + if row["diagnostic_wait_ms"] > 0 + ), "event_names": dict(sorted(event_names.items())), } return rows, result, validation @@ -464,7 +478,10 @@ def weighted_component_mean( / denominator for category in CATEGORIES } - values["total"] = sum(values.values()) + values["total"] = sum( + row["total_time_ms"] * weight + for row, weight in zip(selected, weights, strict=True) + ) / denominator return values @@ -697,8 +714,13 @@ def analyze( "context lengths and graph buckets are unavailable in frozen logs" ), "graph_observability": ( - "Frontier ledger exposes additive execution components but no direct graph " - "bucket/padding/launch-overhead field; graph effects remain folded into predictors" + "Frontier state outputs have no direct graph bucket/padding/launch-overhead " + "field; graph effects remain folded into predictors" + ), + "op_trace_accounting": ( + "For TP8 shared-domain sync, total is ledger stage_end-start (critical path); " + "op categories are serialized work estimates and are not additive because " + "overlap and lane-summed wait diagnostics are represented separately" ), }, "real": real, @@ -798,6 +820,7 @@ def markdown(result: dict[str, Any]) -> str: f"- Real observed TPOT contrast: {result['reference']['observed_real_tp8_minus_tp4_ms']:+.4f} ms/token.", f"- {result['scope']['real_proxy_limitation']}.", f"- {result['scope']['graph_observability']}.", + f"- {result['scope']['op_trace_accounting']}.", "- Component deltas identify where Frontier creates its own TP8 advantage; without real per-stage measurements they are not yet root-cause proof.", "", ] diff --git a/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py b/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py index e13dba0..4221006 100644 --- a/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py +++ b/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py @@ -248,6 +248,7 @@ class FidelityEnvelopeTest(unittest.TestCase): for category, names in module.OP_CATEGORIES.items(): for name in names: self.assertEqual(module.op_category(name), category) + self.assertIsNone(module.op_category("expert_parallel_allreduce_wait")) with self.assertRaisesRegex(ValueError, "unclassified"): module.op_category("unknown_graph_overhead") component_row = {name: 1.0 for name in (*module.CATEGORIES, "total")}