diff --git a/campaigns/musa_rlc_product/manifest/musa-rlc-32-product-v1.json b/campaigns/musa_rlc_product/manifest/musa-rlc-32-product-v1.json new file mode 100644 index 0000000..370ef63 --- /dev/null +++ b/campaigns/musa_rlc_product/manifest/musa-rlc-32-product-v1.json @@ -0,0 +1,551 @@ +{ + "schema_version": 1, + "campaign_id": "s5000-musa-rlc-32-product-v1-20260902-v3", + "experiment_kind": "musa-rlc-product-validation", + "permitted_claim": "S5000 MUSA product-dispatch correctness, layout-topology response, and same-contract native-event latency for this frozen corpus only", + "production_default_policy": "unchanged-off", + "source": { + "flagtree_commit": "3059917e2df16f8ea35c5efe1ed1af2d87117396", + "flaggems_commit": "46ef8080329190d8da8738340240de2d5a475950", + "zl_bench_commit": "da2288564584039366e7be07fe4e94539a560c81" + }, + "device": { + "ssh_alias": "S5000", + "container": "zixu_flagos_S5000", + "backend": "mthreads", + "model": "Moore Threads S5000", + "physical_device": "7", + "logical_device": "0", + "visible_device_count": 1 + }, + "protocol": { + "screen_cells": [ + "off", + "mask0", + "candidate" + ], + "timing_cells": [ + "off", + "candidate" + ], + "canary_order": [ + "dropout", + "softmax", + "instance_norm", + "fft", + "sort", + "batch_norm", + "topk", + "conv1d" + ], + "candidate_phase_mask": 13, + "candidate_profitability_policy": true, + "candidate_product_launch_count": 1, + "default_benchmark_dtypes": [ + "float16" + ], + "timer": "triton.testing.do_bench via MUSA device-interface Event", + "autotune_contract": "off-record then mask0/candidate replay from off", + "fresh_process_per_operator_cell": true, + "same_physical_device_for_all_cells": true, + "serial_measurement": true, + "full_campaign_requires_canary_terminal": true, + "correctness": "FlagGems public dispatch quick tests under candidate, narrowed to the variant each workload measures and run with the MThreads compatibility plugin, plus benchmark execution for every cell", + "classification": { + "G": "correct, convert count decreases, and stable speedup >= 5%", + "C": "correct and convert count decreases, but latency intervals overlap or speedup < 5%", + "B": "correct but candidate does not reduce the executed convert topology", + "R": "correct but stable regression <= -3%", + "K": "candidate correctness failure", + "X": "unsupported, timeout, harness failure, or same-contract replay violation" + }, + "timeout_seconds_by_cost": { + "light": 300, + "medium": 900, + "heavy": 1800 + }, + "measured_launch_floor": 1, + "supported_benchmark_dtypes": [ + "float16", + "float32", + "bfloat16", + "int16", + "int32", + "bool", + "complex64" + ] + }, + "workloads": [ + { + "id": "batch_norm", + "repo": "main", + "op": "batch_norm", + "op_file": "batch_norm", + "shape_key": "batch_norm", + "correctness_file": "tests/test_batch_norm.py", + "family": "normalization", + "nv_evidence": "strong-positive", + "nv_gain_percent": 43.1, + "prior_s5000": "B 0->0", + "cost": "medium", + "canary": true + }, + { + "id": "replication_pad3d", + "repo": "main", + "op": "replication_pad3d", + "op_file": "replication_pad3d", + "shape_key": "replication_pad3d", + "correctness_file": "tests/test_replication_pad3d.py", + "family": "padding", + "nv_evidence": "strong-positive", + "nv_gain_percent": 38.6, + "prior_s5000": "B 0->0 at author shape", + "cost": "medium", + "canary": false + }, + { + "id": "fft", + "repo": "main", + "op": "perf_fft", + "op_file": "fft", + "shape_key": "fft", + "correctness_file": "tests/test_fft.py", + "family": "fft", + "nv_evidence": "strong-positive", + "nv_gain_percent": 36.5, + "prior_s5000": "G 38->2", + "cost": "heavy", + "canary": true, + "benchmark_dtypes": [ + "complex64" + ] + }, + { + "id": "topk", + "repo": "main", + "op": "topk", + "op_file": "topk", + "shape_key": "topk", + "correctness_file": "tests/test_topk.py", + "family": "selection", + "nv_evidence": "strong-positive-shared-network-risk", + "nv_gain_percent": 20.8, + "prior_s5000": "B 0->0 at large author shape", + "cost": "medium", + "canary": true + }, + { + "id": "instance_norm", + "repo": "main", + "op": "instance_norm", + "op_file": "instance_norm", + "shape_key": "instance_norm", + "correctness_file": "tests/test_instance_norm.py", + "family": "normalization", + "nv_evidence": "strong-positive", + "nv_gain_percent": 17.8, + "prior_s5000": "G 1->0", + "cost": "medium", + "canary": true + }, + { + "id": "dropout", + "repo": "main", + "op": "dropout", + "op_file": "dropout", + "shape_key": "dropout", + "correctness_file": "tests/test_dropout.py", + "family": "rng", + "nv_evidence": "strong-positive-shared-network-risk", + "nv_gain_percent": 15.2, + "prior_s5000": "product G 8->0 +9.15%", + "cost": "light", + "canary": true + }, + { + "id": "router_gemm", + "repo": "main", + "op": "perf_router_gemm", + "op_file": "router_gemm", + "shape_key": "router_gemm", + "correctness_file": "tests/test_router_gemm.py", + "family": "mma", + "nv_evidence": "strong-positive-mma", + "nv_gain_percent": 14.5, + "prior_s5000": "public screen available; product v1 pending", + "cost": "heavy", + "canary": false, + "benchmark_dtypes": [ + "bfloat16" + ] + }, + { + "id": "fused_recurrent_gated_delta_rule", + "repo": "main", + "op": "perf_fused_recurrent_gated_delta_rule", + "op_file": "FLA/test_fused_recurrent_gated_delta_rule_perf", + "shape_key": null, + "correctness_file": "tests/test_FLA/test_fused_recurrent_gated_delta_rule.py", + "family": "recurrent", + "nv_evidence": "strong-positive-recurrent", + "nv_gain_percent": 14.2, + "prior_s5000": "not product-qualified", + "cost": "heavy", + "canary": false, + "benchmark_dtypes": [ + "bfloat16" + ] + }, + { + "id": "top_k_per_row_decode", + "repo": "main", + "op": "top_k_per_row_decode", + "op_file": "top_k_per_row_decode", + "shape_key": "top_k_per_row_decode", + "correctness_file": "tests/test_top_k_per_row_decode.py", + "family": "selection", + "nv_evidence": "strong-positive-shared-network-risk", + "nv_gain_percent": 13.0, + "prior_s5000": "not product-qualified", + "cost": "medium", + "canary": false + }, + { + "id": "sort", + "repo": "main", + "op": "perf_sort", + "op_file": "sort", + "shape_key": "sort", + "correctness_file": "tests/test_sort.py", + "family": "sort", + "nv_evidence": "strong-positive-shared-network-risk", + "nv_gain_percent": 12.3, + "prior_s5000": "G 2->1", + "cost": "medium", + "canary": true + }, + { + "id": "sort_stable", + "repo": "main", + "op": "perf_sort_stable", + "op_file": "sort", + "shape_key": "sort_stable", + "correctness_file": "tests/test_sort.py", + "family": "sort", + "nv_evidence": "strong-positive-shared-network-risk", + "nv_gain_percent": 12.3, + "prior_s5000": "G 2->1", + "cost": "medium", + "canary": false + }, + { + "id": "uniform", + "repo": "main", + "op": "uniform_", + "op_file": "uniform_", + "shape_key": "uniform_", + "correctness_file": "tests/test_uniform_.py", + "family": "rng", + "nv_evidence": "strong-positive-rng", + "nv_gain_percent": 10.3, + "prior_s5000": "public boundary risk; product v1 pending", + "cost": "light", + "canary": false + }, + { + "id": "rand", + "repo": "main", + "op": "rand", + "op_file": "rand", + "shape_key": "rand", + "correctness_file": "tests/test_rand.py", + "family": "rng", + "nv_evidence": "strong-positive-rng", + "nv_gain_percent": 8.7, + "prior_s5000": "product v1 pending", + "cost": "light", + "canary": false + }, + { + "id": "rand_like", + "repo": "main", + "op": "rand_like", + "op_file": "rand_like", + "shape_key": "rand_like", + "correctness_file": "tests/test_rand_like.py", + "family": "rng", + "nv_evidence": "strong-positive-rng", + "nv_gain_percent": 8.6, + "prior_s5000": "product v1 pending", + "cost": "light", + "canary": false + }, + { + "id": "max_pool2d_with_indices", + "repo": "main", + "op": "max_pool2d_with_indices", + "op_file": "max_pool2d", + "shape_key": "max_pool2d_with_indices", + "correctness_file": "tests/test_max_pool2d.py", + "family": "reduction-index", + "nv_evidence": "positive-reduction-index", + "nv_gain_percent": 6.1, + "prior_s5000": "B 0->0", + "cost": "medium", + "canary": false + }, + { + "id": "randperm", + "repo": "main", + "op": "randperm", + "op_file": "randperm", + "shape_key": "randperm", + "correctness_file": "tests/test_randperm.py", + "family": "rng", + "nv_evidence": "small-positive-rng", + "nv_gain_percent": 4.0, + "prior_s5000": "C 7->1 latency overlap", + "cost": "medium", + "canary": false, + "benchmark_dtypes": [ + "int32" + ] + }, + { + "id": "randn", + "repo": "main", + "op": "randn", + "op_file": "randn", + "shape_key": "randn", + "correctness_file": "tests/test_randn.py", + "family": "rng", + "nv_evidence": "diluted-positive-rng", + "nv_gain_percent": 4.0, + "prior_s5000": "product v1 pending", + "cost": "light", + "canary": false + }, + { + "id": "randn_like", + "repo": "main", + "op": "randn_like", + "op_file": "randn_like", + "shape_key": "randn_like", + "correctness_file": "tests/test_randn_like.py", + "family": "rng", + "nv_evidence": "diluted-positive-rng", + "nv_gain_percent": 4.0, + "prior_s5000": "product v1 pending", + "cost": "light", + "canary": false + }, + { + "id": "upsample_bicubic2d_aa", + "repo": "main", + "op": "upsample_bicubic2d_aa", + "op_file": "upsample_bicubic2d_aa", + "shape_key": "upsample_bicubic2d_aa", + "correctness_file": "tests/test_upsample_bicubic2d_aa.py", + "family": "interpolation", + "nv_evidence": "diluted-positive-cold-path", + "nv_gain_percent": 1.7, + "prior_s5000": "C 10->0 latency flat", + "cost": "medium", + "canary": false, + "correctness_selector": "not backward", + "correctness_scope_reason": "the benchmarked op is the forward path (benchmark/test_upsample_bicubic2d_aa.py::test_upsample_bicubic2d_aa); the backward baseline defect fails identically under off/mask0/candidate and is tracked as a separate MUSA numerics issue" + }, + { + "id": "grid_sample", + "repo": "main", + "op": "grid_sample", + "op_file": "grid_sample", + "shape_key": "grid_sample", + "correctness_file": "tests/test_grid_sample.py", + "family": "sampling", + "nv_evidence": "structure-change-no-win", + "nv_gain_percent": 0.0, + "prior_s5000": "B 0->0", + "cost": "medium", + "canary": false, + "benchmark_selector": "dtype0", + "benchmark_scope_reason": "benchmark/test_grid_sample.py parametrises over consts.FLOAT_DTYPES and declares dtypes=[dtype] per case, so a campaign-wide --dtypes float16 filter makes the float32 and bfloat16 cases raise; dtype0 is the float16 case this workload measures" + }, + { + "id": "max_pool2d_backward", + "repo": "main", + "op": "max_pool2d_backward", + "op_file": "max_pool2d", + "shape_key": "max_pool2d_backward", + "correctness_file": "tests/test_max_pool2d.py", + "family": "reduction-index", + "nv_evidence": "structure-change-no-win", + "nv_gain_percent": 0.0, + "prior_s5000": "B 0->0", + "cost": "medium", + "canary": false + }, + { + "id": "conv1d", + "repo": "main", + "op": "conv1d", + "op_file": "conv1d", + "shape_key": "conv1d", + "correctness_file": "tests/test_conv1d.py", + "family": "convolution", + "nv_evidence": "light-regression-review", + "nv_gain_percent": -4.2, + "prior_s5000": "C 1->0 via phase3, slight slowdown", + "cost": "heavy", + "canary": true + }, + { + "id": "layer_norm_backward", + "repo": "main", + "op": "layer_norm_backward", + "op_file": "layer_norm", + "shape_key": "layer_norm_backward", + "correctness_file": "tests/test_layer_norm.py", + "family": "normalization", + "nv_evidence": "light-regression-review", + "nv_gain_percent": -1.9, + "prior_s5000": "C 4->2 latency flat", + "cost": "medium", + "canary": false + }, + { + "id": "cutlass_scaled_mm", + "repo": "main", + "op": "cutlass_scaled_mm_benchmark", + "op_file": "cutlass_scaled_mm", + "shape_key": "cutlass_scaled_mm", + "correctness_file": "tests/test_cutlass_scaled_mm.py", + "family": "mma", + "nv_evidence": "light-regression-mma", + "nv_gain_percent": -1.2, + "prior_s5000": "likely unsupported CUDA-specific control", + "cost": "heavy", + "canary": false + }, + { + "id": "gelu_backward", + "repo": "main", + "op": "gelu_backward", + "op_file": "gelu", + "shape_key": "gelu_backward", + "correctness_file": "tests/test_gelu.py", + "family": "elementwise-backward", + "nv_evidence": "direct-no-cudagraph-noise", + "nv_gain_percent": null, + "prior_s5000": "B 0->0 with LLVM optimizations", + "cost": "medium", + "canary": false + }, + { + "id": "fused_marlin_moe_mxfp4", + "repo": "main", + "op": "fused_marlin_moe_w4a16_mxfp4", + "op_file": "fused_marlin_moe_w4a16_mxfp4", + "shape_key": null, + "correctness_file": "tests/test_fused_marlin_moe.py", + "family": "quantized-moe", + "nv_evidence": "autotune-config-change", + "nv_gain_percent": null, + "prior_s5000": "likely unsupported CUDA-specific control", + "cost": "heavy", + "canary": false, + "correctness_selector": "mxfp4", + "correctness_scope_reason": "tests/test_fused_marlin_moe.py also holds w4a16_int4 and w8a16_int8 variants; only the mxfp4 variant corresponds to this workload, and the int8 failures are a separate baseline defect", + "inventory_override_reason": "zl_bench da2288564584039366e7be07fe4e94539a560c81 lists benchmark/test_fused_marlin_moe.py, but FlagGems 46ef8080329190d8da8738340240de2d5a475950 split it into per-quantisation files; the mxfp4 target is benchmark/test_fused_marlin_moe_w4a16_mxfp4.py::test_fused_marlin_moe_w4a16_mxfp4" + }, + { + "id": "bincount", + "repo": "main", + "op": "bincount", + "op_file": "bincount", + "shape_key": "bincount", + "correctness_file": "tests/test_bincount.py", + "family": "histogram", + "nv_evidence": "autotune-config-change", + "nv_gain_percent": null, + "prior_s5000": "product v1 pending", + "cost": "medium", + "canary": false, + "benchmark_dtypes": [ + "float32" + ] + }, + { + "id": "masked_select", + "repo": "main", + "op": "masked_select", + "op_file": "masked_select", + "shape_key": "masked_select", + "correctness_file": "tests/test_masked_select.py", + "family": "selection", + "nv_evidence": "holdout-shared-network", + "nv_gain_percent": null, + "prior_s5000": "public qualify available", + "cost": "medium", + "canary": false + }, + { + "id": "mm", + "repo": "main", + "op": "mm", + "op_file": "mm", + "shape_key": "mm", + "correctness_file": "tests/test_mm.py", + "family": "mma-control", + "nv_evidence": "holdout-mma-control", + "nv_gain_percent": null, + "prior_s5000": "control only", + "cost": "heavy", + "canary": false + }, + { + "id": "scaled_dot_product_attention", + "repo": "main", + "op": "scaled_dot_product_attention", + "op_file": "scaled_dot_product_attention", + "shape_key": "scaled_dot_product_attention", + "correctness_file": "tests/test_scaled_dot_product_attention.py", + "family": "attention-control", + "nv_evidence": "holdout-attention-control", + "nv_gain_percent": null, + "prior_s5000": "control only", + "cost": "heavy", + "canary": false + }, + { + "id": "softmax", + "repo": "main", + "op": "softmax", + "op_file": "softmax", + "shape_key": "softmax", + "correctness_file": "tests/test_softmax.py", + "family": "reduction-control", + "nv_evidence": "holdout-reduction-control", + "nv_gain_percent": null, + "prior_s5000": "product B 0->0", + "cost": "light", + "canary": true + }, + { + "id": "layer_norm", + "repo": "main", + "op": "layer_norm", + "op_file": "layer_norm", + "shape_key": "layer_norm", + "correctness_file": "tests/test_layer_norm.py", + "family": "normalization-control", + "nv_evidence": "holdout-normalization-control", + "nv_gain_percent": null, + "prior_s5000": "control only", + "cost": "medium", + "canary": false + } + ], + "predecessor_campaign_id": "s5000-musa-rlc-32-product-v1-20260901-v2" +} diff --git a/campaigns/musa_rlc_product/prepare_musa_rlc_product.py b/campaigns/musa_rlc_product/prepare_musa_rlc_product.py new file mode 100644 index 0000000..e378c54 --- /dev/null +++ b/campaigns/musa_rlc_product/prepare_musa_rlc_product.py @@ -0,0 +1,478 @@ +#!/usr/bin/env python3 +"""Validate and freeze the single-device MUSA RLC product campaign.""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import json +import os +import shutil +from pathlib import Path +from typing import Any + + +COSTS = {"light", "medium", "heavy"} +CELLS = ["off", "mask0", "candidate"] +TIMING_CELLS = ["off", "candidate"] +ESTIMATED_SECONDS = {"light": 180, "medium": 600, "heavy": 1200} + + +class CampaignError(RuntimeError): + pass + + +def _canonical(value: object) -> bytes: + return (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8") + + +def _is_git_sha(value: object) -> bool: + return ( + isinstance(value, str) + and len(value) == 40 + and all(char in "0123456789abcdef" for char in value) + ) + + +def _correctness_scope_errors(row: dict[str, Any], index: int) -> list[str]: + """Check an optional narrowing of a workload's correctness gate. + + Running a whole test file lets an unrelated variant decide the verdict for + the operator actually being measured. A workload may therefore pin node ids + or a ``-k`` selector, but never both, and narrowing always requires a stated + reason so the exclusion stays visible in review. + """ + name = row.get("id", index) + selector = row.get("correctness_selector") + node_ids = row.get("correctness_node_ids") + errors: list[str] = [] + if selector is not None and node_ids is not None: + errors.append( + f"workload {name} must not set both correctness_selector and " + "correctness_node_ids" + ) + if selector is not None and (not isinstance(selector, str) or not selector.strip()): + errors.append(f"workload {name} correctness_selector must be a non-empty string") + if node_ids is not None and ( + not isinstance(node_ids, list) + or not node_ids + or not all(isinstance(value, str) and value.strip() for value in node_ids) + ): + errors.append( + f"workload {name} correctness_node_ids must be non-empty strings" + ) + if selector is None and node_ids is None: + return errors + reason = row.get("correctness_scope_reason") + if not isinstance(reason, str) or not reason.strip(): + errors.append( + f"workload {name} narrows correctness and must state " + "correctness_scope_reason" + ) + return errors + + +def _benchmark_scope_errors(row: dict[str, Any], index: int) -> list[str]: + """Check an optional narrowing of a workload's benchmark run. + + Some benchmarks parametrise one dtype per case, so the campaign-wide dtype + filter makes every non-matching case raise. A workload may pin the case it + measures, but narrowing what gets timed always requires a stated reason. + """ + name = row.get("id", index) + selector = row.get("benchmark_selector") + if selector is None: + return [] + errors: list[str] = [] + if not isinstance(selector, str) or not selector.strip(): + errors.append(f"workload {name} benchmark_selector must be a non-empty string") + reason = row.get("benchmark_scope_reason") + if not isinstance(reason, str) or not reason.strip(): + errors.append( + f"workload {name} narrows the benchmark and must state " + "benchmark_scope_reason" + ) + return errors + + +def _inventory_override_errors(row: dict[str, Any], index: int) -> list[str]: + """Check an explicit departure from the pinned zl_bench inventory. + + The inventory can lag the pinned FlagGems commit, for example when upstream + splits one benchmark file into several. A workload may then name the target + that actually exists, but it must say why, so a stale inventory is never + silently routed around. + """ + name = row.get("id", index) + reason = row.get("inventory_override_reason") + if reason is None: + return [] + if not isinstance(reason, str) or not reason.strip(): + return [ + f"workload {name} inventory_override_reason must be a non-empty string" + ] + return [] + + +def validate_manifest(manifest: dict[str, Any]) -> list[str]: + errors: list[str] = [] + if manifest.get("schema_version") != 1: + errors.append("schema_version must be 1") + if manifest.get("experiment_kind") != "musa-rlc-product-validation": + errors.append("experiment_kind must be musa-rlc-product-validation") + if manifest.get("production_default_policy") != "unchanged-off": + errors.append("production_default_policy must be unchanged-off") + if "S5000 MUSA product-dispatch" not in manifest.get("permitted_claim", ""): + errors.append("permitted_claim must bind S5000 MUSA product dispatch") + + source = manifest.get("source") + if not isinstance(source, dict): + errors.append("source must be an object") + else: + for key in ("flagtree_commit", "flaggems_commit", "zl_bench_commit"): + if not _is_git_sha(source.get(key)): + errors.append(f"source.{key} must be a lowercase 40-character Git SHA") + + device = manifest.get("device") + expected_device = { + "ssh_alias": "S5000", + "container": "zixu_flagos_S5000", + "backend": "mthreads", + "model": "Moore Threads S5000", + "physical_device": "7", + "logical_device": "0", + "visible_device_count": 1, + } + if not isinstance(device, dict): + errors.append("device must be an object") + else: + for key, expected in expected_device.items(): + if device.get(key) != expected: + errors.append(f"device.{key} must equal {expected!r}") + + protocol = manifest.get("protocol") + if not isinstance(protocol, dict): + errors.append("protocol must be an object") + else: + if protocol.get("screen_cells") != CELLS: + errors.append("protocol.screen_cells must be off/mask0/candidate") + if protocol.get("timing_cells") != TIMING_CELLS: + errors.append("protocol.timing_cells must be off/candidate") + if protocol.get("candidate_phase_mask") != 13: + errors.append("candidate_phase_mask must be 13") + canary_order = protocol.get("canary_order") + if not isinstance(canary_order, list) or len(canary_order) != 8: + errors.append("protocol.canary_order must contain eight workload ids") + for flag in ( + "candidate_profitability_policy", + "fresh_process_per_operator_cell", + "same_physical_device_for_all_cells", + "serial_measurement", + "full_campaign_requires_canary_terminal", + ): + if protocol.get(flag) is not True: + errors.append(f"protocol.{flag} must be true") + if protocol.get("timer") != "triton.testing.do_bench via MUSA device-interface Event": + errors.append("protocol.timer must bind the MUSA native-event path") + default_dtypes = protocol.get("default_benchmark_dtypes") + if not isinstance(default_dtypes, list) or not default_dtypes or not all( + isinstance(value, str) and value for value in default_dtypes + ): + errors.append("protocol.default_benchmark_dtypes must be non-empty strings") + supported = protocol.get("supported_benchmark_dtypes") + if not isinstance(supported, list) or not supported or not all( + isinstance(value, str) and value for value in supported + ): + errors.append( + "protocol.supported_benchmark_dtypes must be non-empty strings" + ) + timeouts = protocol.get("timeout_seconds_by_cost") + if not isinstance(timeouts, dict) or set(timeouts) != COSTS: + errors.append("timeout_seconds_by_cost must define light/medium/heavy") + elif not all(isinstance(value, int) and value > 0 for value in timeouts.values()): + errors.append("timeout_seconds_by_cost values must be positive integers") + floor = protocol.get("measured_launch_floor") + if not isinstance(floor, int) or isinstance(floor, bool) or floor < 0: + errors.append( + "protocol.measured_launch_floor must be a non-negative integer" + ) + + workloads = manifest.get("workloads") + if not isinstance(workloads, list) or len(workloads) != 32: + errors.append("workloads must contain exactly 32 entries") + return errors + + ids: list[str] = [] + nodes: list[tuple[str, str, str]] = [] + canaries = 0 + for index, row in enumerate(workloads): + if not isinstance(row, dict): + errors.append(f"workloads[{index}] must be an object") + continue + for key in ( + "id", "repo", "op", "op_file", "correctness_file", "family", + "nv_evidence", "prior_s5000", "cost", + ): + if not isinstance(row.get(key), str) or not row[key]: + errors.append(f"workloads[{index}] missing {key}") + if row.get("repo") != "main": + errors.append(f"workload {row.get('id', index)} must use the main FlagGems repo") + if row.get("cost") not in COSTS: + errors.append(f"workload {row.get('id', index)} has invalid cost") + if not isinstance(row.get("canary"), bool): + errors.append(f"workload {row.get('id', index)} canary must be boolean") + benchmark_dtypes = row.get( + "benchmark_dtypes", protocol.get("default_benchmark_dtypes", []) + ) + if not isinstance(benchmark_dtypes, list) or not benchmark_dtypes or not all( + isinstance(value, str) and value for value in benchmark_dtypes + ): + errors.append( + f"workload {row.get('id', index)} benchmark_dtypes must be non-empty strings" + ) + else: + # The benchmark CLI rejects a dtype it does not define, and that + # only surfaced as an argparse error inside a device run. + vocabulary = protocol.get("supported_benchmark_dtypes") + if isinstance(vocabulary, list): + unknown = [ + value for value in benchmark_dtypes if value not in vocabulary + ] + if unknown: + errors.append( + f"workload {row.get('id', index)} declares benchmark dtype(s) " + f"the benchmark CLI does not accept: {', '.join(unknown)}" + ) + errors.extend(_correctness_scope_errors(row, index)) + errors.extend(_benchmark_scope_errors(row, index)) + errors.extend(_inventory_override_errors(row, index)) + canaries += row.get("canary") is True + ids.append(row.get("id", "")) + nodes.append((row.get("repo", ""), row.get("op", ""), row.get("op_file", ""))) + if len(ids) != len(set(ids)): + errors.append("workload ids must be unique") + if len(nodes) != len(set(nodes)): + errors.append("workload repo/op/op_file triples must be unique") + if canaries != 8: + errors.append("exactly eight workloads must be canaries") + elif set(manifest["protocol"].get("canary_order", [])) != { + row["id"] for row in workloads if row["canary"] + }: + errors.append("protocol.canary_order must contain every canary exactly once") + return errors + + +def _load_inventory(inventory_dir: Path) -> dict[tuple[str, str, str], dict[str, str]]: + path = inventory_dir / "inventory_main.csv" + if not path.is_file(): + raise CampaignError(f"missing inventory: {path}") + rows: dict[tuple[str, str, str], dict[str, str]] = {} + with path.open(newline="", encoding="utf-8") as handle: + for row in csv.DictReader(handle): + key = ("main", row.get("op", ""), row.get("op_file", "")) + if key in rows: + raise CampaignError(f"duplicate inventory row: {key}") + rows[key] = row + return rows + + +def validate_inventory(workloads: list[dict[str, Any]], inventory_dir: Path) -> list[str]: + inventory = _load_inventory(inventory_dir) + errors: list[str] = [] + for row in workloads: + key = (row["repo"], row["op"], row["op_file"]) + found = inventory.get(key) + if found is None: + # A stated override admits a target the pinned inventory does not + # know about. The runner still resolves it against the real tree. + if row.get("inventory_override_reason"): + continue + errors.append(f"inventory missing {key}") + continue + if found.get("collected") == "no": + errors.append(f"inventory marks {key} collected=no") + if found.get("marked_skip") == "yes": + errors.append(f"inventory marks {key} skipped") + return errors + + +def build_plan(manifest: dict[str, Any], selection: str) -> tuple[dict[str, Any], dict[str, Any]]: + workloads = manifest["workloads"] + if selection == "canary": + by_id = {row["id"]: row for row in workloads if row["canary"]} + workloads = [by_id[task_id] for task_id in manifest["protocol"]["canary_order"]] + elif selection != "all": + raise CampaignError(f"unsupported selection: {selection}") + + timeouts = manifest["protocol"]["timeout_seconds_by_cost"] + tasks = [] + ordinal = 0 + for row in workloads: + ordinal += 1 + tasks.append( + { + "ordinal": ordinal, + "task_id": row["id"], + "repo": row["repo"], + "op": row["op"], + "op_file": row["op_file"], + "shape_key": row.get("shape_key"), + "correctness_file": row["correctness_file"], + "correctness_selector": row.get("correctness_selector"), + "correctness_node_ids": row.get("correctness_node_ids"), + "correctness_scope_reason": row.get("correctness_scope_reason"), + "benchmark_selector": row.get("benchmark_selector"), + "benchmark_scope_reason": row.get("benchmark_scope_reason"), + "inventory_override_reason": row.get("inventory_override_reason"), + "measured_launch_floor": manifest["protocol"]["measured_launch_floor"], + "family": row["family"], + "nv_evidence": row["nv_evidence"], + "nv_gain_percent": row.get("nv_gain_percent"), + "prior_s5000": row["prior_s5000"], + "cost": row["cost"], + "benchmark_dtypes": row.get( + "benchmark_dtypes", + manifest["protocol"]["default_benchmark_dtypes"], + ), + "timeout_seconds": timeouts[row["cost"]], + "estimated_seconds_per_cell": ESTIMATED_SECONDS[row["cost"]], + "cells": CELLS, + "status": "PENDING", + } + ) + estimate = sum(task["estimated_seconds_per_cell"] * len(CELLS) for task in tasks) + matrix = { + "schema_version": 1, + "campaign_id": manifest["campaign_id"], + "selection": selection, + "device": manifest["device"], + "screen_cells": CELLS, + "timing_cells": TIMING_CELLS, + "tasks": tasks, + } + budget = { + "schema_version": 1, + "campaign_id": manifest["campaign_id"], + "selection": selection, + "task_count": len(tasks), + "gpu_count": 1, + "serial_measurement": True, + "estimated_wall_seconds": estimate, + "estimated_gpu_hours": round(estimate / 3600.0, 4), + "note": "Scheduling estimate only; it is not performance evidence.", + } + return matrix, budget + + +def validate_canary_receipt( + manifest: dict[str, Any], selection: str, canary_receipt: Path | None +) -> list[str]: + if selection != "all": + return [] + if canary_receipt is None: + return ["full campaign requires --canary-receipt"] + try: + receipt = json.loads(canary_receipt.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + return [f"cannot read canary receipt: {exc}"] + errors = [] + if receipt.get("campaign_id") != manifest.get("campaign_id"): + errors.append("canary receipt campaign_id mismatch") + if receipt.get("selection") != "canary" or receipt.get("status") != "TERMINAL": + errors.append("canary receipt must be a terminal canary") + if receipt.get("canary_gate_pass") is not True: + errors.append("canary receipt does not authorize the full campaign") + return errors + + +def prepare( + manifest_path: Path, + inventory_dir: Path, + output_root: Path, + selection: str, + canary_receipt: Path | None = None, +) -> Path: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + if not isinstance(manifest, dict): + raise CampaignError("manifest root must be an object") + errors = validate_manifest(manifest) + if not errors: + errors.extend(validate_inventory(manifest["workloads"], inventory_dir)) + if not errors: + errors.extend(validate_canary_receipt(manifest, selection, canary_receipt)) + if errors: + raise CampaignError("campaign validation failed:\n- " + "\n- ".join(errors)) + + matrix, budget = build_plan(manifest, selection) + output = output_root.resolve() / f"{manifest['campaign_id']}--{selection}" + if output.exists(): + raise CampaignError(f"campaign output already exists and is immutable: {output}") + output.parent.mkdir(parents=True, exist_ok=True) + stage = output.parent / f".{output.name}.staging-{os.getpid()}" + if stage.exists(): + raise CampaignError(f"staging directory already exists: {stage}") + try: + stage.mkdir() + shutil.copy2(manifest_path, stage / "manifest.json") + manifest_sha = hashlib.sha256((stage / "manifest.json").read_bytes()).hexdigest() + matrix["manifest_sha256"] = manifest_sha + budget["manifest_sha256"] = manifest_sha + (stage / "manifest.sha256").write_text( + f"{manifest_sha} manifest.json\n", encoding="utf-8", newline="\n" + ) + (stage / "task-matrix.json").write_bytes(_canonical(matrix)) + (stage / "budget.json").write_bytes(_canonical(budget)) + (stage / "ops.txt").write_text( + "".join(f"{task['repo']}:{task['op']}@{task['op_file']}\n" for task in matrix["tasks"]), + encoding="utf-8", + newline="\n", + ) + stage.replace(output) + except Exception: + if stage.exists(): + shutil.rmtree(stage) + raise + return output + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("manifest", type=Path) + parser.add_argument("--inventory-dir", type=Path, required=True) + parser.add_argument("--selection", choices=("canary", "all"), default="canary") + parser.add_argument("--output-root", type=Path) + parser.add_argument("--canary-receipt", type=Path) + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args() + try: + manifest = json.loads(args.manifest.read_text(encoding="utf-8")) + errors = validate_manifest(manifest) if isinstance(manifest, dict) else ["manifest root must be an object"] + if not errors: + errors.extend(validate_inventory(manifest["workloads"], args.inventory_dir)) + if not errors and not args.dry_run: + errors.extend(validate_canary_receipt(manifest, args.selection, args.canary_receipt)) + if errors: + raise CampaignError("campaign validation failed:\n- " + "\n- ".join(errors)) + matrix, budget = build_plan(manifest, args.selection) + if args.dry_run: + print(json.dumps({"status": "DRY_RUN", "budget": budget, "tasks": matrix["tasks"]}, indent=2, sort_keys=True)) + return 0 + if args.output_root is None: + raise CampaignError("--output-root is required unless --dry-run is used") + output = prepare( + args.manifest, + args.inventory_dir, + args.output_root, + args.selection, + args.canary_receipt, + ) + except (OSError, json.JSONDecodeError, CampaignError) as exc: + print(str(exc)) + return 2 + print(json.dumps({"status": "PREPARED", "output": str(output)}, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/campaigns/musa_rlc_product/run_musa_rlc_product.py b/campaigns/musa_rlc_product/run_musa_rlc_product.py new file mode 100644 index 0000000..06470e0 --- /dev/null +++ b/campaigns/musa_rlc_product/run_musa_rlc_product.py @@ -0,0 +1,834 @@ +#!/usr/bin/env python3 +"""Run the frozen S5000 MUSA RLC product campaign serially.""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import importlib.util +import json +import math +import os +import signal +import statistics +import subprocess +import sys +import time +from pathlib import Path +from typing import Any + + +TERMINAL = {"G", "C", "B", "R", "K", "X"} +RLC_ENV_KEYS = { + "FLAGTREE_RLC_ENHANCE", + "FLAGTREE_RLC_PHASE_MASK", + "FLAGTREE_MUSA_RLC_ENHANCE", + "FLAGTREE_MUSA_RLC_PHASE_MASK", + "FLAGTREE_MUSA_RLC_PROFITABILITY_POLICY", + "FLAGTREE_MUSA_RLC_PRODUCT_LAUNCH_COUNT", + "FLAGTREE_MUSA_RLC_MIN_ADJUSTED_SAVED_COST_PER_TENSOR_OP", + "FLAGTREE_MUSA_RLC_PHASE3_SAVED_COST_MULTIPLIER", + "FLAGTREE_MUSA_RLC_MAX_EXTERNAL_USE_EDGES", + "FLAGTREE_MUSA_RLC_MIN_REMOVED_CONVERT_DENSITY_PER_1024_PROPOSAL_VALUES", + "FLAGTREE_MUSA_RLC_PRESERVE_INT_TO_FP_CONTIGUITY", + "FLAGTREE_MUSA_RLC_PRESERVE_FP_TO_FP_CONTIGUITY", +} + + +class CampaignError(RuntimeError): + pass + + +def _canonical(value: object) -> bytes: + return (json.dumps(value, indent=2, sort_keys=True) + "\n").encode("utf-8") + + +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _atomic_json(path: Path, value: object) -> None: + temporary = path.with_name(path.name + ".tmp") + temporary.write_bytes(_canonical(value)) + os.replace(temporary, path) + + +def _load_module(path: Path, name: str): + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise CampaignError(f"cannot load module: {path}") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _identity(path: Path) -> dict[str, str]: + rows: dict[str, str] = {} + for raw in path.read_text(encoding="utf-8").splitlines(): + key, separator, value = raw.partition("=") + if separator: + rows[key] = value + return rows + + +def _base_env(site: Path, flaggems: Path) -> dict[str, str]: + env = os.environ.copy() + for key in list(env): + if key.startswith("FLAGTREE_") and "RLC" in key: + env.pop(key) + env.pop("DISABLE_LLVM_OPT", None) + env.update( + { + "PYTHONNOUSERSITE": "1", + "PYTHONDONTWRITEBYTECODE": "1", + "PYTHONUNBUFFERED": "1", + "GEMS_VENDOR": "mthreads", + "FLAGGEMS_VENDOR": "mthreads", + "FLAGTREE_BACKEND": "mthreads", + "TRITON_BACKENDS_IN_TREE": "1", + "MUSA_VISIBLE_DEVICES": "7", + "MTHREADS_VISIBLE_DEVICES": "7", + "PYTHONPATH": f"{flaggems / 'src'}:{site}", + "FLAGTREE_RLC_TRACE_REJECTS": "1", + } + ) + return env + + +def _cell_env(base: dict[str, str], cell: str) -> dict[str, str]: + env = base.copy() + configs = { + "off": {"enhance": "0", "mask": "13", "policy": "0", "launches": "0"}, + "mask0": {"enhance": "1", "mask": "0", "policy": "1", "launches": "1"}, + "candidate": {"enhance": "1", "mask": "13", "policy": "1", "launches": "1"}, + } + try: + row = configs[cell] + except KeyError as exc: + raise CampaignError(f"unsupported cell: {cell}") from exc + env.update( + { + "FLAGTREE_MUSA_RLC_ENHANCE": row["enhance"], + "FLAGTREE_MUSA_RLC_PHASE_MASK": row["mask"], + "FLAGTREE_MUSA_RLC_PROFITABILITY_POLICY": row["policy"], + "FLAGTREE_MUSA_RLC_PRODUCT_LAUNCH_COUNT": row["launches"], + "FLAGTREE_MUSA_RLC_MIN_ADJUSTED_SAVED_COST_PER_TENSOR_OP": "1800", + "FLAGTREE_MUSA_RLC_PHASE3_SAVED_COST_MULTIPLIER": "2", + "FLAGTREE_MUSA_RLC_MAX_EXTERNAL_USE_EDGES": "8", + "FLAGTREE_MUSA_RLC_MIN_REMOVED_CONVERT_DENSITY_PER_1024_PROPOSAL_VALUES": "0", + "FLAGTREE_MUSA_RLC_PRESERVE_INT_TO_FP_CONTIGUITY": "1", + "FLAGTREE_MUSA_RLC_PRESERVE_FP_TO_FP_CONTIGUITY": "1", + } + ) + return env + + +def _run_process( + command: list[str], *, cwd: Path, env: dict[str, str], log: Path, timeout: int +) -> tuple[int, bool, float]: + started = time.time() + with log.open("w", encoding="utf-8", newline="\n") as handle: + process = subprocess.Popen( + command, + cwd=str(cwd), + env=env, + stdout=handle, + stderr=subprocess.STDOUT, + text=True, + start_new_session=True, + ) + timed_out = False + try: + return_code = process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + timed_out = True + try: + os.killpg(process.pid, signal.SIGTERM) + process.wait(timeout=10) + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGKILL) + process.wait() + return_code = 124 + return return_code, timed_out, time.time() - started + + +def _indexed_launches(ttgir_root: Path) -> dict[Path, int] | None: + """Map each dumped TTGIR file to its executed launch count. + + Returns None when ``index.tsv`` is missing or unreadable, which means the + measured-operator scope cannot be established for this cell. + """ + index = ttgir_root / "index.tsv" + if not index.is_file(): + return None + try: + lines = index.read_text(encoding="utf-8", errors="replace").splitlines() + except OSError: + return None + if not lines: + return None + header = lines[0].split("\t") + try: + shape_at = header.index("shape") + kernel_at = header.index("kernel") + file_at = header.index("file") + launch_at = header.index("launches") + except ValueError: + return None + width = max(shape_at, kernel_at, file_at, launch_at) + launches: dict[Path, int] = {} + for raw in lines[1:]: + fields = raw.split("\t") + if len(fields) <= width: + continue + name = fields[file_at].strip() + # Autotune losers are indexed for audit but never dumped; they carry a + # parenthesised placeholder instead of a file name. + if not name or name.startswith("("): + continue + try: + count = int(fields[launch_at]) + except ValueError: + continue + path = ttgir_root / fields[shape_at] / fields[kernel_at] / name + launches[path] = max(launches.get(path, 0), count) + return launches + + +def _convert_count(path: Path, *, measured_launch_floor: int = 1) -> dict[str, Any]: + """Count executed convert_layout ops, scoped to the measured operator. + + A benchmark call constructs its inputs once and then runs the operator many + times under the timer. Counting every executed kernel therefore credits the + candidate with converts removed from input construction, outside the timed + region. Only variants launched more than ``measured_launch_floor`` times + contribute to the measured count; the unscoped total is preserved too. + """ + ttgir_root = path / "ttgir" + files = sorted(ttgir_root.glob("**/*.ttgir")) + launches = _indexed_launches(ttgir_root) + total = 0 + measured = 0 + excluded: list[dict[str, Any]] = [] + unindexed: list[str] = [] + for file in files: + converts = file.read_text(encoding="utf-8", errors="replace").count( + "ttg.convert_layout" + ) + total += converts + relative = file.relative_to(ttgir_root).as_posix() + if launches is None: + measured += converts + continue + launch_count = launches.get(file) + if launch_count is None: + # Dumped but absent from the index: keep it and report the gap + # rather than silently dropping executed topology. + measured += converts + unindexed.append(relative) + continue + if launch_count > measured_launch_floor: + measured += converts + elif converts: + excluded.append( + { + "file": relative, + "launches": launch_count, + "convert_count": converts, + } + ) + return { + "measured_convert_count": measured, + "total_convert_count": total, + "ttgir_files": len(files), + "measured_launch_floor": measured_launch_floor, + "scope_complete": launches is not None and not unindexed, + "excluded_unmeasured": excluded, + "unindexed_files": unindexed, + } + + +def _median_gems_latency(rows: list[dict[str, str]]) -> float | None: + values = [] + for row in rows: + if row.get("row_status") != "SUCCESS": + continue + try: + value = float(row.get("gems_lat", "")) + except ValueError: + continue + if math.isfinite(value) and value > 0: + values.append(value) + return statistics.median(values) if values else None + + +def missing_targets(matrix: dict[str, Any], flaggems: Path) -> list[str]: + """Resolve every task's benchmark and correctness target against the tree. + + The inventory a matrix is built from can lag the pinned FlagGems commit, for + example when upstream splits one benchmark file into several. Before this + check, such a task only failed inside the worker's preflight, which exited + silently and left a zero-byte log and a cell receipt with nothing in it. + """ + missing: list[str] = [] + for task in matrix["tasks"]: + benchmark = flaggems / "benchmark" / f"test_{task['op_file']}.py" + if not benchmark.is_file(): + missing.append(f"{task['task_id']} benchmark -> {benchmark}") + correctness = flaggems / task["correctness_file"] + if not correctness.is_file(): + missing.append(f"{task['task_id']} correctness -> {correctness}") + return missing + + +def validate_contract( + manifest_path: Path, + matrix_path: Path, + build_root: Path, + flaggems: Path, + zl_bench: Path, + python: Path, +) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]: + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + matrix = json.loads(matrix_path.read_text(encoding="utf-8")) + if not isinstance(manifest, dict) or not isinstance(matrix, dict): + raise CampaignError("manifest and task matrix roots must be objects") + manifest_sha = _sha256(manifest_path) + if matrix.get("manifest_sha256") != manifest_sha: + raise CampaignError("task matrix manifest_sha256 does not match manifest") + if matrix.get("campaign_id") != manifest.get("campaign_id"): + raise CampaignError("campaign id mismatch") + if matrix.get("screen_cells") != ["off", "mask0", "candidate"]: + raise CampaignError("task matrix cell contract mismatch") + + missing = missing_targets(matrix, flaggems) + if missing: + raise CampaignError( + "targets missing from the frozen FlagGems tree: " + "; ".join(missing) + ) + + site = build_root / "site" + identity = _identity(build_root / "identity.txt") + expected = manifest["source"] + if identity.get("git_commit") != expected["flagtree_commit"]: + raise CampaignError("FlagTree build identity does not match manifest") + if not (build_root / "build.done").is_file(): + raise CampaignError("FlagTree product build is not terminal") + if (flaggems / ".source-commit").read_text(encoding="utf-8").strip() != expected["flaggems_commit"]: + raise CampaignError("FlagGems source identity does not match manifest") + if (zl_bench / ".source-commit").read_text(encoding="utf-8").strip() != expected["zl_bench_commit"]: + raise CampaignError("zl_bench source identity does not match manifest") + + probe_env = _base_env(site, flaggems) + probe_code = ( + "import json,torch,torch_musa,triton,flag_gems;" + "print(json.dumps({'count':torch.musa.device_count()," + "'device':str(flag_gems.device),'vendor':str(flag_gems.vendor_name)," + "'triton':triton.__file__,'flaggems':flag_gems.__file__},sort_keys=True))" + ) + probe = subprocess.run( + [str(python), "-B", "-c", probe_code], + cwd=str(flaggems), + env=probe_env, + text=True, + capture_output=True, + timeout=60, + ) + if probe.returncode: + raise CampaignError(f"product import probe failed: {probe.stderr.strip()}") + observed = json.loads(probe.stdout.strip().splitlines()[-1]) + if observed.get("count") != 1 or observed.get("vendor") != "mthreads": + raise CampaignError(f"MUSA device/vendor mismatch: {observed}") + if not Path(observed["triton"]).is_relative_to(site): + raise CampaignError("imported Triton does not come from the frozen product site") + if not Path(observed["flaggems"]).is_relative_to(flaggems / "src"): + raise CampaignError("imported FlagGems does not come from the frozen product source") + + observed.update( + { + "flagtree_commit": identity["git_commit"], + "libtriton_sha256": _sha256(site / "triton/_C/libtriton.so"), + "flaggems_commit": expected["flaggems_commit"], + "zl_bench_commit": expected["zl_bench_commit"], + "python": str(python), + } + ) + return manifest, matrix, observed + + +def run_cell( + task: dict[str, Any], + cell: str, + round_name: str, + replay_from: Path | None, + *, + output_root: Path, + worker: Path, + python: Path, + zl_bench: Path, + flaggems: Path, + site: Path, + shape_file: Path, + batch_summary: Any, + base_env: dict[str, str], +) -> dict[str, Any]: + out = output_root / "tasks" / task["task_id"] / round_name / cell + env = _cell_env(base_env, cell) + env.update( + { + "PYTHON_BIN": str(python), + "ZL_BENCH_DIR": str(zl_bench), + "FLAGGEMS_DIR": str(flaggems), + "FLAGTREE_SITE": str(site), + "OP": task["op"], + "OP_FILE": task["op_file"], + "SHAPE_FILE": str(shape_file), + "OUT_DIR": str(out), + "DTYPES": " ".join(task["benchmark_dtypes"]), + } + ) + if task.get("benchmark_selector"): + env["BENCH_SELECTOR"] = str(task["benchmark_selector"]) + if replay_from is not None: + env["REPLAY_FROM"] = str(replay_from) + command = ["bash", str(worker)] + # The shell worker owns run.log, so use a sibling controller log. + controller_log = out.parent / f"{cell}.controller.log" + controller_log.parent.mkdir(parents=True, exist_ok=True) + return_code, timed_out, elapsed = _run_process( + command, + cwd=worker.parent, + env=env, + log=controller_log, + timeout=int(task["timeout_seconds"]), + ) + run_log = out / "run.log" + text = run_log.read_text(encoding="utf-8", errors="replace") if run_log.is_file() else "" + rows = batch_summary.parse_run_log(text) + if timed_out: + status = "TIMEOUT" + else: + status = batch_summary.classify(return_code, rows, text) + scope = _convert_count( + out, measured_launch_floor=int(task.get("measured_launch_floor", 1)) + ) + # A worker can fail its preflight before creating OUT_DIR (for example, + # when a frozen inventory entry has no backend benchmark file). Preserve + # that failure as a cell receipt instead of aborting the whole campaign. + out.mkdir(parents=True, exist_ok=True) + receipt = { + "task_id": task["task_id"], + "round": round_name, + "cell": cell, + "status": status, + "return_code": return_code, + "timed_out": timed_out, + "elapsed_seconds": round(elapsed, 3), + "rows": rows, + "rows_success": sum(row.get("row_status") == "SUCCESS" for row in rows), + "median_gems_latency_ms": _median_gems_latency(rows), + "convert_count": scope["measured_convert_count"], + "convert_count_all": scope["total_convert_count"], + "convert_scope_complete": scope["scope_complete"], + "measured_launch_floor": scope["measured_launch_floor"], + "convert_excluded_unmeasured": scope["excluded_unmeasured"], + "convert_unindexed_files": scope["unindexed_files"], + "ttgir_files": scope["ttgir_files"], + "replay_fallback_count": text.count("AUTOTUNE_REPLAY_FALLBACK"), + "benchmark_selector": task.get("benchmark_selector"), + "benchmark_dtypes": list(task["benchmark_dtypes"]), + "effective_rlc": { + key: env[key] + for key in sorted(RLC_ENV_KEYS) + if key in env + }, + } + _atomic_json(out / "cell-receipt.json", receipt) + return receipt + + +def correctness_command(task: dict[str, Any], python: Path) -> list[str]: + """Build the correctness argv, narrowed to the variant this task measures. + + A task-level file runs every variant in that file, so an unrelated failing + variant can mask or fabricate a verdict for the operator under test. A + manifest may therefore pin explicit node ids or a ``-k`` selector; the + manifest schema requires a stated reason whenever it does. + """ + command = [str(python), "-B", "-m", "pytest", "-q"] + # The same vendor-scoped compatibility plugin the benchmark worker loads. + # Without it a stale CUDA-only guard skips the operator here too, and a + # SKIP would be read as "not measured" rather than "not gated". + command.extend(["-p", "_musa_benchmark_compat_plugin"]) + node_ids = task.get("correctness_node_ids") + if node_ids: + command.extend(str(node) for node in node_ids) + else: + command.append(task["correctness_file"]) + selector = task.get("correctness_selector") + if selector: + command.extend(["-k", str(selector)]) + # -rs records why a variant was skipped, so a SKIP verdict stays auditable + # without a second run. + command.extend(["-rs", "--quick"]) + return command + + +def run_correctness( + task: dict[str, Any], + cell: str, + *, + output_root: Path, + python: Path, + flaggems: Path, + site: Path, + base_env: dict[str, str], + plugin_dir: Path, +) -> dict[str, Any]: + out = output_root / "correctness" / task["task_id"] / cell + out.mkdir(parents=True, exist_ok=True) + env = _cell_env(base_env, cell) + env["PYTEST_DISABLE_PLUGIN_AUTOLOAD"] = "1" + env["PYTHONPATH"] = f"{plugin_dir}{os.pathsep}{env['PYTHONPATH']}" + env["TRITON_CACHE_DIR"] = str(out / "triton-cache") + env["FLAGGEMS_CACHE_DIR"] = str(out / "flaggems-cache") + (out / "triton-cache").mkdir() + (out / "flaggems-cache").mkdir() + command = correctness_command(task, python) + return_code, timed_out, elapsed = _run_process( + command, + cwd=flaggems, + env=env, + log=out / "correctness.log", + timeout=int(task["timeout_seconds"]), + ) + log_text = (out / "correctness.log").read_text(encoding="utf-8", errors="replace") + if timed_out: + status = "TIMEOUT" + elif return_code != 0: + status = "FAIL" + elif " passed" in log_text: + status = "PASS" + elif " skipped" in log_text: + status = "SKIP" + else: + status = "FAIL" + receipt = { + "task_id": task["task_id"], + "correctness_file": task["correctness_file"], + "correctness_node_ids": task.get("correctness_node_ids"), + "correctness_selector": task.get("correctness_selector"), + "correctness_scope_reason": task.get("correctness_scope_reason"), + "command": command, + "cell": cell, + "status": status, + "return_code": return_code, + "timed_out": timed_out, + "elapsed_seconds": round(elapsed, 3), + } + _atomic_json(out / "correctness-receipt.json", receipt) + return receipt + + +def _speedup(off: dict[str, Any], candidate: dict[str, Any]) -> float | None: + left = off.get("median_gems_latency_ms") + right = candidate.get("median_gems_latency_ms") + if not isinstance(left, (int, float)) or not isinstance(right, (int, float)) or right <= 0: + return None + return (left / right - 1.0) * 100.0 + + +def classify_task( + correctness: dict[str, Any], + cells: dict[str, dict[str, Any]], + correctness_isolation: dict[str, dict[str, Any]] | None = None, +) -> tuple[str, dict[str, Any]]: + detail: dict[str, Any] = {"reason": ""} + if correctness["status"] == "FAIL": + isolation = correctness_isolation or {} + off_status = isolation.get("off", {}).get("status") + mask0_status = isolation.get("mask0", {}).get("status") + detail.update( + { + "off_correctness_status": off_status, + "mask0_correctness_status": mask0_status, + "candidate_correctness_status": correctness["status"], + } + ) + if off_status == "FAIL" and mask0_status == "FAIL": + detail["reason"] = "baseline correctness fails under off/mask0/candidate" + return "X", detail + detail["reason"] = "candidate correctness failed" + return "K", detail + if correctness["status"] != "PASS": + detail["reason"] = f"candidate correctness is {correctness['status']}" + return "X", detail + required = ("a_off", "a_mask0", "a_candidate") + if any(cells.get(key, {}).get("status") != "PASS" for key in required): + detail["reason"] = "one or more screen cells did not pass" + return "X", detail + if any(cells[key].get("replay_fallback_count") for key in ("a_mask0", "a_candidate")): + detail["reason"] = "same-config replay fell back" + return "X", detail + # A topology claim is only admissible when every screen cell could be scoped + # to the measured operator. Older receipts predate scoping and are trusted. + if any(cells[key].get("convert_scope_complete", True) is False for key in required): + detail["reason"] = "measured convert topology scope is incomplete" + return "X", detail + off_count = cells["a_off"]["convert_count"] + mask0_count = cells["a_mask0"]["convert_count"] + candidate_count = cells["a_candidate"]["convert_count"] + detail.update( + { + "off_convert_count": off_count, + "mask0_convert_count": mask0_count, + "candidate_convert_count": candidate_count, + "off_convert_count_all": cells["a_off"].get("convert_count_all"), + "candidate_convert_count_all": cells["a_candidate"].get("convert_count_all"), + "round_a_speedup_percent": _speedup(cells["a_off"], cells["a_candidate"]), + } + ) + if off_count != mask0_count: + detail["reason"] = "mask0 topology differs from off" + return "X", detail + if candidate_count >= off_count: + speedup = detail["round_a_speedup_percent"] + if speedup is not None and speedup <= -3.0: + detail["reason"] = "no topology reduction and screening regression" + return "R", detail + detail["reason"] = "candidate did not reduce executed convert topology" + return "B", detail + + if cells.get("b_off", {}).get("status") != "PASS" or cells.get("b_candidate", {}).get("status") != "PASS": + detail["reason"] = "topology changed but reversed-order timing did not pass" + return "X", detail + if cells["b_off"].get("replay_fallback_count") or cells["b_candidate"].get("replay_fallback_count"): + detail["reason"] = "reversed-order timing replay fell back" + return "X", detail + round_b = _speedup(cells["b_off"], cells["b_candidate"]) + round_a = detail["round_a_speedup_percent"] + detail["round_b_speedup_percent"] = round_b + if round_a is None or round_b is None: + detail["reason"] = "missing comparable native-event latency" + return "X", detail + if round_a >= 5.0 and round_b >= 5.0: + detail["reason"] = "topology reduction and stable >=5% speedup" + return "G", detail + if round_a <= -3.0 and round_b <= -3.0: + detail["reason"] = "topology reduction but stable regression" + return "R", detail + detail["reason"] = "topology reduction without stable >=5% speedup" + return "C", detail + + +def run_campaign( + manifest: dict[str, Any], + matrix: dict[str, Any], + observed: dict[str, Any], + *, + output_root: Path, + worker: Path, + python: Path, + zl_bench: Path, + flaggems: Path, + build_root: Path, + resume: bool, +) -> dict[str, Any]: + if output_root.exists() and not resume: + raise CampaignError(f"output exists; use --resume: {output_root}") + observed = dict(observed) + observed.update( + { + "benchmark_runner_sha256": _sha256(Path(__file__)), + "benchmark_worker_sha256": _sha256(worker), + "musa_benchmark_compat_plugin_sha256": _sha256( + zl_bench / "_musa_benchmark_compat_plugin.py" + ), + } + ) + output_root.mkdir(parents=True, exist_ok=True) + site = build_root / "site" + shape_file = zl_bench / "ops/shapes_single.yaml" + batch_summary = _load_module(zl_bench / "batch_summary.py", "zl_batch_summary") + base_env = _base_env(site, flaggems) + status_csv = output_root / "status.csv" + rows: list[dict[str, Any]] = [] + + occupancy = subprocess.run( + ["mthreads-gmi"], text=True, capture_output=True, timeout=30 + ) + if occupancy.returncode: + raise CampaignError(f"mthreads-gmi failed: {occupancy.stderr.strip()}") + (output_root / "occupancy-before.txt").write_text(occupancy.stdout, encoding="utf-8", newline="\n") + if "No running processes found" not in occupancy.stdout: + raise CampaignError("S5000 has an active process; campaign not admitted") + + for task in matrix["tasks"]: + task_root = output_root / "tasks" / task["task_id"] + task_receipt_path = task_root / "task-receipt.json" + if resume and task_receipt_path.is_file(): + receipt = json.loads(task_receipt_path.read_text(encoding="utf-8")) + if receipt.get("classification") in TERMINAL: + rows.append(receipt) + continue + cells: dict[str, dict[str, Any]] = {} + off_root = task_root / "round-a" / "off" + cells["a_off"] = run_cell( + task, "off", "round-a", None, + output_root=output_root, worker=worker, python=python, + zl_bench=zl_bench, flaggems=flaggems, site=site, + shape_file=shape_file, batch_summary=batch_summary, base_env=base_env, + ) + cells["a_mask0"] = run_cell( + task, "mask0", "round-a", off_root, + output_root=output_root, worker=worker, python=python, + zl_bench=zl_bench, flaggems=flaggems, site=site, + shape_file=shape_file, batch_summary=batch_summary, base_env=base_env, + ) + cells["a_candidate"] = run_cell( + task, "candidate", "round-a", off_root, + output_root=output_root, worker=worker, python=python, + zl_bench=zl_bench, flaggems=flaggems, site=site, + shape_file=shape_file, batch_summary=batch_summary, base_env=base_env, + ) + correctness = run_correctness( + task, "candidate", output_root=output_root, python=python, flaggems=flaggems, + site=site, base_env=base_env, plugin_dir=zl_bench, + ) + correctness_isolation: dict[str, dict[str, Any]] = { + "candidate": correctness + } + if correctness["status"] != "PASS": + correctness_isolation["off"] = run_correctness( + task, "off", output_root=output_root, python=python, + flaggems=flaggems, site=site, base_env=base_env, + plugin_dir=zl_bench, + ) + correctness_isolation["mask0"] = run_correctness( + task, "mask0", output_root=output_root, python=python, + flaggems=flaggems, site=site, base_env=base_env, + plugin_dir=zl_bench, + ) + if ( + cells["a_off"]["status"] == "PASS" + and cells["a_candidate"]["status"] == "PASS" + and cells["a_candidate"]["convert_count"] < cells["a_off"]["convert_count"] + ): + cells["b_candidate"] = run_cell( + task, "candidate", "round-b", off_root, + output_root=output_root, worker=worker, python=python, + zl_bench=zl_bench, flaggems=flaggems, site=site, + shape_file=shape_file, batch_summary=batch_summary, base_env=base_env, + ) + cells["b_off"] = run_cell( + task, "off", "round-b", off_root, + output_root=output_root, worker=worker, python=python, + zl_bench=zl_bench, flaggems=flaggems, site=site, + shape_file=shape_file, batch_summary=batch_summary, base_env=base_env, + ) + classification, detail = classify_task( + correctness, cells, correctness_isolation + ) + receipt = { + "task_id": task["task_id"], + "op": task["op"], + "op_file": task["op_file"], + "family": task["family"], + "nv_evidence": task["nv_evidence"], + "nv_gain_percent": task.get("nv_gain_percent"), + "prior_s5000": task["prior_s5000"], + "classification": classification, + "classification_detail": detail, + "correctness": correctness, + "correctness_isolation": correctness_isolation, + "cells": cells, + } + task_root.mkdir(parents=True, exist_ok=True) + _atomic_json(task_receipt_path, receipt) + rows.append(receipt) + with status_csv.with_name(status_csv.name + ".tmp").open("w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter( + handle, + fieldnames=["task_id", "op", "family", "classification", "reason"], + ) + writer.writeheader() + for row in rows: + writer.writerow( + { + "task_id": row["task_id"], + "op": row["op"], + "family": row["family"], + "classification": row["classification"], + "reason": row["classification_detail"]["reason"], + } + ) + os.replace(status_csv.with_name(status_csv.name + ".tmp"), status_csv) + print( + f"MUSA_RLC_PRODUCT_TASK task={task['task_id']} class={classification} " + f"reason={detail['reason']}", + flush=True, + ) + + counts = {key: sum(row["classification"] == key for row in rows) for key in sorted(TERMINAL)} + key_rows = {row["task_id"]: row for row in rows} + canary_gate = None + if matrix.get("selection") == "canary": + canary_gate = ( + key_rows.get("dropout", {}).get("classification") in {"G", "C"} + and key_rows.get("softmax", {}).get("classification") == "B" + and counts["K"] == 0 + ) + final = { + "schema_version": 1, + "campaign_id": manifest["campaign_id"], + "selection": matrix["selection"], + "status": "TERMINAL", + "task_count": len(rows), + "counts": counts, + "canary_gate_pass": canary_gate, + "source_observed": observed, + "production_default_policy": "unchanged-off", + "claim_scope": manifest["permitted_claim"], + } + _atomic_json(output_root / "campaign-receipt.json", final) + return final + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--manifest", type=Path, required=True) + parser.add_argument("--task-matrix", type=Path, required=True) + parser.add_argument("--build-root", type=Path, required=True) + parser.add_argument("--flaggems", type=Path, required=True) + parser.add_argument("--zl-bench", type=Path, required=True) + parser.add_argument("--worker", type=Path, required=True) + parser.add_argument("--python", type=Path, default=Path("/usr/bin/python3")) + parser.add_argument("--output-root", type=Path, required=True) + parser.add_argument("--resume", action="store_true") + args = parser.parse_args() + try: + manifest, matrix, observed = validate_contract( + args.manifest.resolve(), args.task_matrix.resolve(), args.build_root.resolve(), + args.flaggems.resolve(), args.zl_bench.resolve(), args.python.resolve(), + ) + final = run_campaign( + manifest, matrix, observed, + output_root=args.output_root.resolve(), worker=args.worker.resolve(), + python=args.python.resolve(), zl_bench=args.zl_bench.resolve(), + flaggems=args.flaggems.resolve(), build_root=args.build_root.resolve(), + resume=args.resume, + ) + except (OSError, json.JSONDecodeError, CampaignError, subprocess.SubprocessError) as exc: + print(f"MUSA_RLC_PRODUCT_ERROR: {exc}", file=sys.stderr) + return 2 + print(json.dumps(final, indent=2, sort_keys=True)) + return 0 if final.get("canary_gate_pass") is not False else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/campaigns/musa_rlc_product/run_pytest_musa_rlc_product.sh b/campaigns/musa_rlc_product/run_pytest_musa_rlc_product.sh new file mode 100644 index 0000000..a12e530 --- /dev/null +++ b/campaigns/musa_rlc_product/run_pytest_musa_rlc_product.sh @@ -0,0 +1,111 @@ +#!/usr/bin/env bash +set -Eeuo pipefail + +: "${PYTHON_BIN:?PYTHON_BIN is required}" +: "${ZL_BENCH_DIR:?ZL_BENCH_DIR is required}" +: "${FLAGGEMS_DIR:?FLAGGEMS_DIR is required}" +: "${FLAGTREE_SITE:?FLAGTREE_SITE is required}" +: "${OP:?OP is required}" +: "${OP_FILE:?OP_FILE is required}" +: "${SHAPE_FILE:?SHAPE_FILE is required}" +: "${OUT_DIR:?OUT_DIR is required}" + +worker_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + +# A silent preflight exit leaves a zero-byte controller log and no run.log, so +# the campaign records a failed cell with nothing to diagnose it. Name the check +# that failed instead. +preflight() { + local description="$1" + shift + if ! "$@"; then + printf 'MUSA_RLC_PRODUCT_PREFLIGHT_FAILED op=%s check=%s\n' \ + "${OP:-?}" "$description" >&2 + exit 3 + fi +} + +preflight "python interpreter is executable: $PYTHON_BIN" test -x "$PYTHON_BIN" +preflight "zl_bench directory exists: $ZL_BENCH_DIR" test -d "$ZL_BENCH_DIR" +preflight "ir meta plugin exists: $ZL_BENCH_DIR/_ir_meta_plugin.py" \ + test -f "$ZL_BENCH_DIR/_ir_meta_plugin.py" +preflight "benchmark file exists: $FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py" \ + test -f "$FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py" +preflight "FlagGems source exists: $FLAGGEMS_DIR/src/flag_gems" \ + test -d "$FLAGGEMS_DIR/src/flag_gems" +preflight "libtriton is non-empty: $FLAGTREE_SITE/triton/_C/libtriton.so" \ + test -s "$FLAGTREE_SITE/triton/_C/libtriton.so" +preflight "shape file exists: $SHAPE_FILE" test -f "$SHAPE_FILE" +preflight "output directory is fresh: $OUT_DIR" test ! -e "$OUT_DIR" +mkdir -p "$OUT_DIR/autotune_records" "$OUT_DIR/triton-cache" "$OUT_DIR/flaggems-cache" + +export PYTHONNOUSERSITE=1 +export PYTHONDONTWRITEBYTECODE=1 +export PYTHONUNBUFFERED=1 +export GEMS_VENDOR=mthreads +export FLAGGEMS_VENDOR=mthreads +export FLAGTREE_BACKEND=mthreads +export TRITON_BACKENDS_IN_TREE=1 +export MUSA_VISIBLE_DEVICES=7 +export MTHREADS_VISIBLE_DEVICES=7 +export PYTHONPATH="$worker_dir:$ZL_BENCH_DIR:$FLAGGEMS_DIR/src:$FLAGTREE_SITE" +export FLAGGEMS_PERF_CURRENT_OP="$OP" +export FLAGGEMS_PERF_COLOR=never +export NO_COLOR=1 + +plugins=( + -p _musa_benchmark_compat_plugin + -p _device_guard_plugin + -p _seed_plugin + -p _shape_inject_plugin + -p _shape_iter_inject_plugin + -p _bespoke_shape_plugin + -p _autotune_record_plugin + -p _pretty_report_plugin + -p _ir_meta_plugin +) + +dtype_args=() +for dtype in ${DTYPES:-float16}; do + dtype_args+=(--dtypes "$dtype") +done + +# Some benchmarks parametrise one dtype per case, so a campaign-wide --dtypes +# filter makes every non-matching case raise. A workload may then pin the case +# it measures; the manifest requires a stated reason for that narrowing. +selector_args=() +if [[ -n "${BENCH_SELECTOR:-}" ]]; then + selector_args+=(-k "$BENCH_SELECTOR") +fi + +if [[ -n "${REPLAY_FROM:-}" ]]; then + autotune_env="FLAGGEMS_PERF_AUTOTUNE_REPLAY_DIR=$REPLAY_FROM/autotune_records" + mode_desc="replay($REPLAY_FROM)" +else + autotune_env="FLAGGEMS_PERF_AUTOTUNE_RECORD_DIR=$OUT_DIR/autotune_records" + mode_desc=record +fi + +cp "$SHAPE_FILE" "$OUT_DIR/shapes.yaml" +test_file="$FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py::test_${OP}" +status=0 +{ + printf 'MUSA_RLC_PRODUCT_RUN op=%s mode=%s out=%s\n' "$OP" "$mode_desc" "$OUT_DIR" + env \ + TRITON_CACHE_DIR="$OUT_DIR/triton-cache" \ + FLAGGEMS_CACHE_DIR="$OUT_DIR/flaggems-cache" \ + FLAGGEMS_PERF_TTGIR_DUMP_DIR="$OUT_DIR/ttgir" \ + "$autotune_env" \ + USE_FLAGTUNE=0 \ + "$PYTHON_BIN" -u -m pytest -s "$test_file" \ + "${plugins[@]}" \ + --color=no \ + --shape_file "$SHAPE_FILE" \ + "${dtype_args[@]}" \ + "${selector_args[@]}" \ + --level core --mode kernel || status=$? + printf 'MUSA_RLC_PRODUCT_RUN_RC=%s\n' "$status" + exit "$status" +} 2>&1 | tee "$OUT_DIR/run.log" || status=$? + +exit "$status" diff --git a/campaigns/musa_rlc_product/tests/test_musa_rlc_product.py b/campaigns/musa_rlc_product/tests/test_musa_rlc_product.py new file mode 100644 index 0000000..bb1189f --- /dev/null +++ b/campaigns/musa_rlc_product/tests/test_musa_rlc_product.py @@ -0,0 +1,567 @@ +from __future__ import annotations + +import copy +import importlib.util +import json +import sys +import tempfile +import unittest +from types import SimpleNamespace +from pathlib import Path + + +HERE = Path(__file__).resolve().parent +CAMPAIGN = HERE.parent +REPO_ROOT = CAMPAIGN.parents[1] +REFERENCE = CAMPAIGN / "manifest" / "musa-rlc-32-product-v1.json" +SPEC = importlib.util.spec_from_file_location( + "prepare_musa_rlc_product", CAMPAIGN / "prepare_musa_rlc_product.py" +) +assert SPEC is not None and SPEC.loader is not None +MODULE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(MODULE) +RUN_SPEC = importlib.util.spec_from_file_location( + "run_musa_rlc_product", CAMPAIGN / "run_musa_rlc_product.py" +) +assert RUN_SPEC is not None and RUN_SPEC.loader is not None +RUNNER = importlib.util.module_from_spec(RUN_SPEC) +RUN_SPEC.loader.exec_module(RUNNER) +COMPAT_SPEC = importlib.util.spec_from_file_location( + "_musa_benchmark_compat_plugin", REPO_ROOT / "_musa_benchmark_compat_plugin.py" +) +assert COMPAT_SPEC is not None and COMPAT_SPEC.loader is not None +COMPAT = importlib.util.module_from_spec(COMPAT_SPEC) +COMPAT_SPEC.loader.exec_module(COMPAT) + + +def _manifest(): + return json.loads(REFERENCE.read_text(encoding="utf-8")) + + +class MusaRlcProductTests(unittest.TestCase): + def test_musa_fft_compatibility_override_is_narrow(self): + stale = SimpleNamespace( + name="skipif", args=(True,), kwargs={"reason": "CUDA is not available"} + ) + unrelated = SimpleNamespace( + name="skipif", args=(True,), kwargs={"reason": "another reason"} + ) + fft = SimpleNamespace(path=Path("benchmark/test_fft.py"), name="test_perf_fft") + other = SimpleNamespace(path=Path("benchmark/test_fft.py"), name="test_other") + self.assertTrue(COMPAT._is_stale_fft_cuda_guard(fft, stale)) + self.assertFalse(COMPAT._is_stale_fft_cuda_guard(fft, unrelated)) + self.assertFalse(COMPAT._is_stale_fft_cuda_guard(other, stale)) + + def test_runner_preserves_preflight_failure_receipts(self): + runner = (CAMPAIGN / "run_musa_rlc_product.py").read_text(encoding="utf-8") + ensure_parent = "out.mkdir(parents=True, exist_ok=True)" + write_receipt = '_atomic_json(out / "cell-receipt.json", receipt)' + self.assertIn(ensure_parent, runner) + self.assertLess(runner.index(ensure_parent), runner.index(write_receipt)) + + def test_worker_dumps_only_executed_ttgir_variants(self): + worker = (CAMPAIGN / "run_pytest_musa_rlc_product.sh").read_text( + encoding="utf-8" + ) + self.assertIn( + 'FLAGGEMS_PERF_TTGIR_DUMP_DIR="$OUT_DIR/ttgir"', worker + ) + self.assertIn("-p _musa_benchmark_compat_plugin", worker) + self.assertIn( + 'PYTHONPATH="$worker_dir:$ZL_BENCH_DIR:$FLAGGEMS_DIR/src:$FLAGTREE_SITE"', + worker, + ) + + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + dumped = root / "ttgir" / "shape" / "kernel" / "used.ttgir" + cached = root / "triton-cache" / "hash" / "loser.ttgir" + dumped.parent.mkdir(parents=True) + cached.parent.mkdir(parents=True) + dumped.write_text("ttg.convert_layout\nttg.convert_layout\n", encoding="utf-8") + cached.write_text("ttg.convert_layout\n" * 99, encoding="utf-8") + scope = RUNNER._convert_count(root) + self.assertEqual(scope["total_convert_count"], 2) + self.assertEqual(scope["ttgir_files"], 1) + # Without an index the measured scope cannot be established, so the + # count stays unscoped and the cell is marked incomplete. + self.assertEqual(scope["measured_convert_count"], 2) + self.assertFalse(scope["scope_complete"]) + + @staticmethod + def _write_cell(root, rows): + """Materialise a cell's ttgir dump plus the index the dumper writes.""" + ttgir = root / "ttgir" + ttgir.mkdir(parents=True) + lines = [ + "\t".join( + [ + "shape", "kernel", "file", "launches", "sweep_launches", + "num_warps", "num_stages", "constexprs", "cache_hash", + ] + ) + ] + for shape, kernel, name, launches, converts in rows: + if name.startswith("("): + lines.append( + "\t".join([shape, kernel, name, str(launches), "0", "4", "3", "{}", "H"]) + ) + continue + path = ttgir / shape / kernel / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("ttg.convert_layout\n" * converts, encoding="utf-8") + lines.append( + "\t".join([shape, kernel, name, str(launches), "0", "4", "3", "{}", "H"]) + ) + (ttgir / "index.tsv").write_text("\n".join(lines) + "\n", encoding="utf-8") + + def test_convert_count_excludes_input_construction_kernels(self): + # Reproduces the sealed gelu_backward shape: every convert sits in the + # once-launched input builder, while the timed kernel has none. + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + self._write_cell( + root, + [ + ("1048576", "randn_kernel", "w16s3.ttgir", 1, 4), + ("1048576", "gelu_none_kernel_rank_1", "w4s3.ttgir", 1, 0), + ("1048576", "gelu_backward_none_kernel_rank_1", "w4s3.ttgir", 3795, 0), + ], + ) + scope = RUNNER._convert_count(root) + self.assertEqual(scope["total_convert_count"], 4) + self.assertEqual(scope["measured_convert_count"], 0) + self.assertTrue(scope["scope_complete"]) + self.assertEqual( + [row["file"] for row in scope["excluded_unmeasured"]], + ["1048576/randn_kernel/w16s3.ttgir"], + ) + + def test_convert_count_keeps_repeatedly_launched_kernels(self): + # Reproduces the sealed sort shape: converts in the dominant kernel and + # in an autotune loser that was indexed but never dumped. + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + self._write_cell( + root, + [ + ("1024x1024", "sweep", "w4s3.ttgir", 280, 3), + ("1024x1024", "broadcast_to_kernel", "w4s3.ttgir", 140, 1), + ("-", "arange_kernel", "(sweep loser, not dumped)", 0, 0), + ], + ) + scope = RUNNER._convert_count(root) + self.assertEqual(scope["measured_convert_count"], 4) + self.assertEqual(scope["total_convert_count"], 4) + self.assertTrue(scope["scope_complete"]) + self.assertEqual(scope["excluded_unmeasured"], []) + + def test_convert_count_reports_dumped_but_unindexed_files(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + self._write_cell(root, [("s", "measured", "w4s3.ttgir", 140, 2)]) + stray = root / "ttgir" / "s" / "stray" / "w4s3.ttgir" + stray.parent.mkdir(parents=True) + stray.write_text("ttg.convert_layout\n", encoding="utf-8") + scope = RUNNER._convert_count(root) + # An unindexed dump is counted rather than dropped, and flagged. + self.assertEqual(scope["measured_convert_count"], 3) + self.assertFalse(scope["scope_complete"]) + self.assertEqual(scope["unindexed_files"], ["s/stray/w4s3.ttgir"]) + + def test_incomplete_convert_scope_fails_closed(self): + def cell(convert_count, latency, complete=True): + return { + "status": "PASS", + "convert_count": convert_count, + "convert_scope_complete": complete, + "median_gems_latency_ms": latency, + "replay_fallback_count": 0, + } + + cells = { + "a_off": cell(8, 0.14), + "a_mask0": cell(8, 0.14), + "a_candidate": cell(0, 0.12, complete=False), + "b_candidate": cell(0, 0.121), + "b_off": cell(8, 0.141), + } + classification, detail = RUNNER.classify_task({"status": "PASS"}, cells) + self.assertEqual(classification, "X") + self.assertEqual( + detail["reason"], "measured convert topology scope is incomplete" + ) + + def test_correctness_command_narrows_to_the_measured_variant(self): + python = Path("/usr/bin/python3") + plain = RUNNER.correctness_command( + {"correctness_file": "tests/test_sort.py"}, python + ) + self.assertEqual(plain[-3:], ["tests/test_sort.py", "-rs", "--quick"]) + self.assertNotIn("-k", plain) + + selected = RUNNER.correctness_command( + { + "correctness_file": "tests/test_upsample_bicubic2d_aa.py", + "correctness_selector": "not backward", + }, + python, + ) + self.assertIn("tests/test_upsample_bicubic2d_aa.py", selected) + self.assertEqual(selected[selected.index("-k") + 1], "not backward") + + pinned = RUNNER.correctness_command( + { + "correctness_file": "tests/test_mm.py", + "correctness_node_ids": ["tests/test_mm.py::test_mm"], + }, + python, + ) + self.assertIn("tests/test_mm.py::test_mm", pinned) + self.assertNotIn("tests/test_mm.py", pinned) + + def test_narrowed_correctness_requires_a_stated_reason(self): + manifest = copy.deepcopy(_manifest()) + target = next( + row for row in manifest["workloads"] if row["id"] == "upsample_bicubic2d_aa" + ) + self.assertEqual(target["correctness_selector"], "not backward") + target.pop("correctness_scope_reason") + errors = MODULE.validate_manifest(manifest) + self.assertIn( + "workload upsample_bicubic2d_aa narrows correctness and must state " + "correctness_scope_reason", + errors, + ) + + both = copy.deepcopy(_manifest()) + row = next(item for item in both["workloads"] if item["id"] == "sort") + row["correctness_selector"] = "sort" + row["correctness_node_ids"] = ["tests/test_sort.py::test_sort"] + row["correctness_scope_reason"] = "reason" + self.assertIn( + "workload sort must not set both correctness_selector and " + "correctness_node_ids", + MODULE.validate_manifest(both), + ) + + def test_measured_launch_floor_is_a_frozen_protocol_field(self): + manifest = copy.deepcopy(_manifest()) + self.assertEqual(manifest["protocol"]["measured_launch_floor"], 1) + plan, _ = MODULE.build_plan(manifest, "all") + self.assertTrue( + all(task["measured_launch_floor"] == 1 for task in plan["tasks"]) + ) + scoped = next( + task for task in plan["tasks"] if task["task_id"] == "fused_marlin_moe_mxfp4" + ) + self.assertEqual(scoped["correctness_selector"], "mxfp4") + self.assertTrue(scoped["correctness_scope_reason"]) + + manifest["protocol"]["measured_launch_floor"] = -1 + self.assertIn( + "protocol.measured_launch_floor must be a non-negative integer", + MODULE.validate_manifest(manifest), + ) + + def test_compat_plugin_only_overrides_verified_stale_guards(self): + def mark(reason, name="skipif"): + return SimpleNamespace(name=name, args=(True,), kwargs={"reason": reason}) + + def item(file_name, test_name): + return SimpleNamespace(path=Path("tests") / file_name, name=test_name) + + # Verified on S5000: unskipped, these pass. + self.assertTrue( + COMPAT._is_stale_cuda_guard( + item("test_top_k_per_row_decode.py", "test_accuracy_top_k"), + mark("CUDA device required"), + ) + ) + self.assertTrue( + COMPAT._is_stale_fft_cuda_guard( + item("test_fft.py", "test_perf_fft"), mark("CUDA is not available") + ) + ) + # The fft guard stays pinned to the one test it was verified against. + self.assertFalse( + COMPAT._is_stale_cuda_guard( + item("test_fft.py", "test_other"), mark("CUDA is not available") + ) + ) + # Probed and rejected: grid_sample hardcodes device="cuda" in its own + # helper, and cutlass_scaled_mm gates on vLLM as well as CUDA. + self.assertFalse( + COMPAT._is_stale_cuda_guard( + item("test_grid_sample.py", "test_nearest_zeros_4d_small"), + mark("CUDA is required."), + ) + ) + self.assertFalse( + COMPAT._is_stale_cuda_guard( + item("test_cutlass_scaled_mm.py", "test_cutlass_scaled_mm"), + mark("Feature requires vLLM and NVIDIA Hopper architecture"), + ) + ) + # Only skipif marks, and only the exact reason string. + self.assertFalse( + COMPAT._is_stale_cuda_guard( + item("test_top_k_per_row_decode.py", "test_accuracy_top_k"), + mark("CUDA device required", name="xfail"), + ) + ) + self.assertFalse( + COMPAT._is_stale_cuda_guard( + item("test_top_k_per_row_decode.py", "test_accuracy_top_k"), + mark("some other reason"), + ) + ) + + def test_compat_plugin_strips_module_level_guards(self): + # top_k_per_row_decode carries its guard as a module-level pytestmark, + # which resolves through the node chain rather than the item's own marks. + stale = SimpleNamespace( + name="skipif", args=(True,), kwargs={"reason": "CUDA device required"} + ) + keep = SimpleNamespace( + name="skipif", args=(True,), kwargs={"reason": "vLLM not installed"} + ) + module = SimpleNamespace(own_markers=[stale, keep]) + function = SimpleNamespace( + path=Path("tests/test_top_k_per_row_decode.py"), + name="test_accuracy_top_k", + own_markers=[], + ) + function.listchain = lambda: [module, function] + + sys.modules.setdefault( + "flag_gems", SimpleNamespace(vendor_name="mthreads") + ) + COMPAT.pytest_collection_modifyitems([function]) + self.assertEqual(module.own_markers, [keep]) + + def test_benchmark_selector_requires_a_stated_reason(self): + manifest = copy.deepcopy(_manifest()) + target = next( + row for row in manifest["workloads"] if row["id"] == "grid_sample" + ) + self.assertEqual(target["benchmark_selector"], "dtype0") + target.pop("benchmark_scope_reason") + self.assertIn( + "workload grid_sample narrows the benchmark and must state " + "benchmark_scope_reason", + MODULE.validate_manifest(manifest), + ) + + def test_declared_benchmark_dtypes_must_exist_in_the_cli_vocabulary(self): + manifest = copy.deepcopy(_manifest()) + self.assertEqual( + next(row for row in manifest["workloads"] if row["id"] == "randperm")[ + "benchmark_dtypes" + ], + ["int32"], + ) + # int64 is not a --dtypes choice; the sealed run only found that out on + # the device, as an argparse error inside a screen cell. + next(row for row in manifest["workloads"] if row["id"] == "randperm")[ + "benchmark_dtypes" + ] = ["int32", "int64"] + self.assertIn( + "workload randperm declares benchmark dtype(s) the benchmark CLI " + "does not accept: int64", + MODULE.validate_manifest(manifest), + ) + + def test_stale_inventory_needs_an_override_reason(self): + manifest = copy.deepcopy(_manifest()) + marlin = next( + row + for row in manifest["workloads"] + if row["id"] == "fused_marlin_moe_mxfp4" + ) + self.assertEqual(marlin["op_file"], "fused_marlin_moe_w4a16_mxfp4") + self.assertTrue(marlin["inventory_override_reason"]) + + with tempfile.TemporaryDirectory() as temporary: + inventory = Path(temporary) + (inventory / "inventory_main.csv").write_text( + "repo,op,op_file,file,marked_skip,collected,op_names\n" + + "".join( + f"main,{row['op']},{row['op_file']},f.py,,yes,{row['id']}\n" + for row in manifest["workloads"] + if row["id"] != "fused_marlin_moe_mxfp4" + ), + encoding="utf-8", + ) + # The stated override lets the missing row through. + self.assertEqual( + MODULE.validate_inventory(manifest["workloads"], inventory), [] + ) + marlin.pop("inventory_override_reason") + self.assertEqual( + MODULE.validate_inventory(manifest["workloads"], inventory), + [ + "inventory missing ('main', 'fused_marlin_moe_w4a16_mxfp4', " + "'fused_marlin_moe_w4a16_mxfp4')" + ], + ) + + def test_missing_benchmark_target_fails_before_the_campaign_starts(self): + matrix = { + "tasks": [ + { + "task_id": "present", + "op_file": "sort", + "correctness_file": "tests/test_sort.py", + }, + { + # The shape of the sealed marlin defect: the inventory named + # a merged benchmark file that upstream had already split. + "task_id": "fused_marlin_moe_mxfp4", + "op_file": "fused_marlin_moe", + "correctness_file": "tests/test_fused_marlin_moe.py", + }, + ] + } + with tempfile.TemporaryDirectory() as temporary: + flaggems = Path(temporary) + (flaggems / "benchmark").mkdir() + (flaggems / "tests").mkdir() + (flaggems / "benchmark" / "test_sort.py").write_text("", encoding="utf-8") + (flaggems / "tests" / "test_sort.py").write_text("", encoding="utf-8") + (flaggems / "tests" / "test_fused_marlin_moe.py").write_text( + "", encoding="utf-8" + ) + missing = RUNNER.missing_targets(matrix, flaggems) + + self.assertEqual(len(missing), 1) + self.assertIn("fused_marlin_moe_mxfp4 benchmark", missing[0]) + self.assertIn("test_fused_marlin_moe.py", missing[0]) + + def test_worker_names_the_failed_preflight_check(self): + worker = (CAMPAIGN / "run_pytest_musa_rlc_product.sh").read_text(encoding="utf-8") + self.assertIn("MUSA_RLC_PRODUCT_PREFLIGHT_FAILED", worker) + self.assertIn( + 'preflight "benchmark file exists: ' + '$FLAGGEMS_DIR/benchmark/test_${OP_FILE}.py"', + worker, + ) + # The benchmark selector must reach pytest. + self.assertIn('selector_args+=(-k "$BENCH_SELECTOR")', worker) + self.assertIn('"${selector_args[@]}"', worker) + + def test_correctness_runs_with_the_vendor_compat_plugin(self): + command = RUNNER.correctness_command( + {"correctness_file": "tests/test_top_k_per_row_decode.py"}, + Path("/usr/bin/python3"), + ) + self.assertEqual( + command[command.index("-p") + 1], "_musa_benchmark_compat_plugin" + ) + + def test_valid_manifest_and_plan_sizes(self): + manifest = _manifest() + self.assertEqual(MODULE.validate_manifest(manifest), []) + canary, canary_budget = MODULE.build_plan(manifest, "canary") + full, full_budget = MODULE.build_plan(manifest, "all") + self.assertEqual(len(canary["tasks"]), 8) + self.assertEqual(len(full["tasks"]), 32) + self.assertEqual(canary["tasks"][0]["task_id"], "dropout") + self.assertEqual(canary["tasks"][1]["task_id"], "softmax") + self.assertEqual(canary["tasks"][0]["benchmark_dtypes"], ["float16"]) + self.assertEqual( + next(task for task in canary["tasks"] if task["task_id"] == "fft")[ + "benchmark_dtypes" + ], + ["complex64"], + ) + self.assertEqual(canary_budget["gpu_count"], 1) + self.assertEqual(full_budget["gpu_count"], 1) + self.assertTrue(canary_budget["serial_measurement"]) + + def test_invalid_device_and_canary_count_fail_closed(self): + manifest = copy.deepcopy(_manifest()) + manifest["device"]["physical_device"] = "0" + manifest["workloads"][0]["canary"] = False + errors = MODULE.validate_manifest(manifest) + self.assertIn("device.physical_device must equal '7'", errors) + self.assertIn("exactly eight workloads must be canaries", errors) + + def test_invalid_timer_and_default_policy_fail_closed(self): + manifest = copy.deepcopy(_manifest()) + manifest["protocol"]["timer"] = "host wall clock" + manifest["production_default_policy"] = "enabled" + errors = MODULE.validate_manifest(manifest) + self.assertIn("protocol.timer must bind the MUSA native-event path", errors) + self.assertIn("production_default_policy must be unchanged-off", errors) + + def test_runtime_classifier_positive_and_fail_closed_paths(self): + def cell(convert_count, latency, status="PASS", replay=0): + return { + "status": status, + "convert_count": convert_count, + "median_gems_latency_ms": latency, + "replay_fallback_count": replay, + } + + correct = {"status": "PASS"} + cells = { + "a_off": cell(8, 0.14), + "a_mask0": cell(8, 0.14), + "a_candidate": cell(0, 0.12), + "b_candidate": cell(0, 0.121), + "b_off": cell(8, 0.141), + } + classification, _ = RUNNER.classify_task(correct, cells) + self.assertEqual(classification, "G") + + invalid = copy.deepcopy(cells) + invalid["a_mask0"]["convert_count"] = 7 + classification, detail = RUNNER.classify_task(correct, invalid) + self.assertEqual(classification, "X") + self.assertEqual(detail["reason"], "mask0 topology differs from off") + + classification, _ = RUNNER.classify_task({"status": "FAIL"}, cells) + self.assertEqual(classification, "K") + + classification, detail = RUNNER.classify_task( + {"status": "FAIL"}, + cells, + { + "candidate": {"status": "FAIL"}, + "mask0": {"status": "FAIL"}, + "off": {"status": "FAIL"}, + }, + ) + self.assertEqual(classification, "X") + self.assertEqual( + detail["reason"], + "baseline correctness fails under off/mask0/candidate", + ) + + def test_full_preparation_requires_passing_canary_receipt(self): + manifest = _manifest() + self.assertEqual( + MODULE.validate_canary_receipt(manifest, "all", None), + ["full campaign requires --canary-receipt"], + ) + with tempfile.TemporaryDirectory() as temporary: + receipt = Path(temporary) / "campaign-receipt.json" + receipt.write_text( + json.dumps( + { + "campaign_id": manifest["campaign_id"], + "selection": "canary", + "status": "TERMINAL", + "canary_gate_pass": True, + } + ), + encoding="utf-8", + ) + self.assertEqual( + MODULE.validate_canary_receipt(manifest, "all", receipt), [] + ) + + +if __name__ == "__main__": + unittest.main()