diff --git a/src/code/issue2/README-2026.md b/src/code/issue2/README-2026.md new file mode 100644 index 0000000..1461e51 --- /dev/null +++ b/src/code/issue2/README-2026.md @@ -0,0 +1,19 @@ +# Issue 2 单机可完成部分:AllReduce 切换阈值拟合器 + +`threshold_model_2026.py` 从以下标定列生成连续消息区间: + +```text +message_bytes,algorithm,latency_us,sm_occupancy_pct,active_warps +``` + +它既可选择纯时延最优算法,也可用 `--max-sm-occupancy-pct` 在 SM 预算内选择最快项。 + +```bash +python -m pytest src/code/issue2/test_threshold_model_2026.py +python src/code/issue2/threshold_model_2026.py --synthetic-demo +python src/code/issue2/threshold_model_2026.py --synthetic-demo --max-sm-occupancy-pct 10 +``` + +合成数据只用于验证 One-Shot→P2P Two-Shot→Multimem/NVLS 的三段拟合逻辑。 +真实阈值和 SM/Warp 数据必须在含 NVSwitch 的多卡机器上用目标实现重新标定。 + diff --git a/src/code/issue2/results-2026/synthetic_measurements.csv b/src/code/issue2/results-2026/synthetic_measurements.csv new file mode 100644 index 0000000..5cdde75 --- /dev/null +++ b/src/code/issue2/results-2026/synthetic_measurements.csv @@ -0,0 +1,64 @@ +message_bytes,algorithm,latency_us,sm_occupancy_pct,active_warps +1024,one_shot,3.02048,72.0,96 +1024,two_shot_p2p,8.011946666666667,48.0,64 +1024,two_shot_multimem_nvls,30.003413333333334,4.0,4 +2048,one_shot,3.04096,72.0,96 +2048,two_shot_p2p,8.023893333333334,48.0,64 +2048,two_shot_multimem_nvls,30.006826666666665,4.0,4 +4096,one_shot,3.08192,72.0,96 +4096,two_shot_p2p,8.047786666666667,48.0,64 +4096,two_shot_multimem_nvls,30.013653333333334,4.0,4 +8192,one_shot,3.16384,72.0,96 +8192,two_shot_p2p,8.095573333333334,48.0,64 +8192,two_shot_multimem_nvls,30.027306666666668,4.0,4 +16384,one_shot,3.32768,72.0,96 +16384,two_shot_p2p,8.191146666666667,48.0,64 +16384,two_shot_multimem_nvls,30.054613333333332,4.0,4 +32768,one_shot,3.65536,72.0,96 +32768,two_shot_p2p,8.382293333333333,48.0,64 +32768,two_shot_multimem_nvls,30.109226666666668,4.0,4 +65536,one_shot,4.31072,72.0,96 +65536,two_shot_p2p,8.764586666666666,48.0,64 +65536,two_shot_multimem_nvls,30.218453333333333,4.0,4 +131072,one_shot,5.62144,72.0,96 +131072,two_shot_p2p,9.529173333333333,48.0,64 +131072,two_shot_multimem_nvls,30.436906666666665,4.0,4 +262144,one_shot,8.24288,72.0,96 +262144,two_shot_p2p,11.058346666666667,48.0,64 +262144,two_shot_multimem_nvls,30.873813333333334,4.0,4 +524288,one_shot,13.485759999999999,72.0,96 +524288,two_shot_p2p,14.116693333333334,48.0,64 +524288,two_shot_multimem_nvls,31.747626666666665,4.0,4 +1048576,one_shot,23.971519999999998,72.0,96 +1048576,two_shot_p2p,20.233386666666668,48.0,64 +1048576,two_shot_multimem_nvls,33.49525333333333,4.0,4 +2097152,one_shot,44.943039999999996,72.0,96 +2097152,two_shot_p2p,32.466773333333336,48.0,64 +2097152,two_shot_multimem_nvls,36.99050666666667,4.0,4 +4194304,one_shot,86.88607999999999,72.0,96 +4194304,two_shot_p2p,56.93354666666667,48.0,64 +4194304,two_shot_multimem_nvls,43.98101333333334,4.0,4 +8388608,one_shot,170.77215999999999,72.0,96 +8388608,two_shot_p2p,105.86709333333334,48.0,64 +8388608,two_shot_multimem_nvls,57.96202666666667,4.0,4 +16777216,one_shot,338.54431999999997,72.0,96 +16777216,two_shot_p2p,203.7341866666667,48.0,64 +16777216,two_shot_multimem_nvls,85.92405333333333,4.0,4 +33554432,one_shot,674.0886399999999,72.0,96 +33554432,two_shot_p2p,399.4683733333334,48.0,64 +33554432,two_shot_multimem_nvls,141.84810666666667,4.0,4 +67108864,one_shot,1345.1772799999999,72.0,96 +67108864,two_shot_p2p,790.9367466666667,48.0,64 +67108864,two_shot_multimem_nvls,253.69621333333333,4.0,4 +134217728,one_shot,2687.3545599999998,72.0,96 +134217728,two_shot_p2p,1573.8734933333335,48.0,64 +134217728,two_shot_multimem_nvls,477.39242666666667,4.0,4 +268435456,one_shot,5371.7091199999995,72.0,96 +268435456,two_shot_p2p,3139.746986666667,48.0,64 +268435456,two_shot_multimem_nvls,924.7848533333333,4.0,4 +536870912,one_shot,10740.418239999999,72.0,96 +536870912,two_shot_p2p,6271.493973333334,48.0,64 +536870912,two_shot_multimem_nvls,1819.5697066666667,4.0,4 +1073741824,one_shot,21477.836479999998,72.0,96 +1073741824,two_shot_p2p,12534.987946666668,48.0,64 +1073741824,two_shot_multimem_nvls,3609.1394133333333,4.0,4 diff --git a/src/code/issue2/results-2026/thresholds.json b/src/code/issue2/results-2026/thresholds.json new file mode 100644 index 0000000..01c6410 --- /dev/null +++ b/src/code/issue2/results-2026/thresholds.json @@ -0,0 +1,103 @@ +{ + "schema_version": 1, + "sm_constraint_pct": null, + "intervals": [ + { + "min_message_bytes": 0, + "max_message_bytes": 741454, + "algorithm": "one_shot", + "calibrated_sizes": [ + 1024, + 2048, + 4096, + 8192, + 16384, + 32768, + 65536, + 131072, + 262144, + 524288 + ], + "latency_us_at_calibrated_sizes": [ + 3.02048, + 3.04096, + 3.08192, + 3.16384, + 3.32768, + 3.65536, + 4.31072, + 5.62144, + 8.24288, + 13.485759999999999 + ], + "sm_occupancy_pct_at_calibrated_sizes": [ + 72.0, + 72.0, + 72.0, + 72.0, + 72.0, + 72.0, + 72.0, + 72.0, + 72.0, + 72.0 + ] + }, + { + "min_message_bytes": 741455, + "max_message_bytes": 2965820, + "algorithm": "two_shot_p2p", + "calibrated_sizes": [ + 1048576, + 2097152 + ], + "latency_us_at_calibrated_sizes": [ + 20.233386666666668, + 32.466773333333336 + ], + "sm_occupancy_pct_at_calibrated_sizes": [ + 48.0, + 48.0 + ] + }, + { + "min_message_bytes": 2965821, + "max_message_bytes": null, + "algorithm": "two_shot_multimem_nvls", + "calibrated_sizes": [ + 4194304, + 8388608, + 16777216, + 33554432, + 67108864, + 134217728, + 268435456, + 536870912, + 1073741824 + ], + "latency_us_at_calibrated_sizes": [ + 43.98101333333334, + 57.96202666666667, + 85.92405333333333, + 141.84810666666667, + 253.69621333333333, + 477.39242666666667, + 924.7848533333333, + 1819.5697066666667, + 3609.1394133333333 + ], + "sm_occupancy_pct_at_calibrated_sizes": [ + 4.0, + 4.0, + 4.0, + 4.0, + 4.0, + 4.0, + 4.0, + 4.0, + 4.0 + ] + } + ], + "warning": "synthetic data validates fitting only; replace it with NCCL/NVLS measurements" +} diff --git a/src/code/issue2/test_threshold_model_2026.py b/src/code/issue2/test_threshold_model_2026.py new file mode 100644 index 0000000..d75aa55 --- /dev/null +++ b/src/code/issue2/test_threshold_model_2026.py @@ -0,0 +1,17 @@ +from threshold_model_2026 import choose_at_each_size, fit_intervals, synthetic_measurements + + +def test_unconstrained_demo_has_three_message_regions(): + winners = choose_at_each_size(synthetic_measurements(), None) + intervals = fit_intervals(winners) + assert [item["algorithm"] for item in intervals] == [ + "one_shot", + "two_shot_p2p", + "two_shot_multimem_nvls", + ] + + +def test_strict_sm_constraint_prefers_nvls(): + winners = choose_at_each_size(synthetic_measurements(), 10.0) + assert {item.algorithm for item in winners} == {"two_shot_multimem_nvls"} + diff --git a/src/code/issue2/threshold_model_2026.py b/src/code/issue2/threshold_model_2026.py new file mode 100644 index 0000000..8a88684 --- /dev/null +++ b/src/code/issue2/threshold_model_2026.py @@ -0,0 +1,172 @@ +#!/usr/bin/env python3 +"""Fit message-size switching thresholds for intra-node AllReduce modes.""" + +from __future__ import annotations + +import argparse +import csv +import json +import math +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Iterable + + +@dataclass(frozen=True) +class Measurement: + message_bytes: int + algorithm: str + latency_us: float + sm_occupancy_pct: float + active_warps: int + + def __post_init__(self) -> None: + if self.message_bytes <= 0 or self.latency_us <= 0 or self.active_warps < 0: + raise ValueError("message size/latency must be positive and warps non-negative") + if not self.algorithm or not 0 <= self.sm_occupancy_pct <= 100: + raise ValueError("invalid algorithm or SM occupancy") + + +def load_measurements(path: Path) -> list[Measurement]: + with path.open("r", encoding="utf-8-sig", newline="") as handle: + reader = csv.DictReader(handle) + required = { + "message_bytes", + "algorithm", + "latency_us", + "sm_occupancy_pct", + "active_warps", + } + missing = required - set(reader.fieldnames or []) + if missing: + raise ValueError(f"missing CSV columns: {sorted(missing)}") + result = [ + Measurement( + int(row["message_bytes"]), + row["algorithm"], + float(row["latency_us"]), + float(row["sm_occupancy_pct"]), + int(row["active_warps"]), + ) + for row in reader + ] + if not result: + raise ValueError("measurement CSV is empty") + return result + + +def choose_at_each_size( + records: Iterable[Measurement], max_sm_occupancy_pct: float | None +) -> list[Measurement]: + by_size: dict[int, list[Measurement]] = {} + for record in records: + by_size.setdefault(record.message_bytes, []).append(record) + winners = [] + for size in sorted(by_size): + candidates = by_size[size] + if max_sm_occupancy_pct is not None: + constrained = [ + item for item in candidates if item.sm_occupancy_pct <= max_sm_occupancy_pct + ] + if constrained: + candidates = constrained + winners.append(min(candidates, key=lambda item: item.latency_us)) + return winners + + +def fit_intervals(winners: list[Measurement]) -> list[dict]: + if not winners: + raise ValueError("no winners to fit") + intervals = [] + start = 0 + while start < len(winners): + end = start + while end + 1 < len(winners) and winners[end + 1].algorithm == winners[start].algorithm: + end += 1 + minimum = ( + 0 + if start == 0 + else int(round(math.sqrt(winners[start - 1].message_bytes * winners[start].message_bytes))) + ) + maximum = ( + None + if end == len(winners) - 1 + else int(round(math.sqrt(winners[end].message_bytes * winners[end + 1].message_bytes))) - 1 + ) + intervals.append( + { + "min_message_bytes": minimum, + "max_message_bytes": maximum, + "algorithm": winners[start].algorithm, + "calibrated_sizes": [item.message_bytes for item in winners[start : end + 1]], + "latency_us_at_calibrated_sizes": [ + item.latency_us for item in winners[start : end + 1] + ], + "sm_occupancy_pct_at_calibrated_sizes": [ + item.sm_occupancy_pct for item in winners[start : end + 1] + ], + } + ) + start = end + 1 + return intervals + + +def synthetic_measurements() -> list[Measurement]: + records = [] + sizes = [2**power for power in range(10, 31)] + # Explicitly synthetic roofline curves: One-Shot wins small messages, + # SM-based P2P wins the middle, and NVLS/Multimem wins large messages. + models = [ + ("one_shot", 3.0, 50.0, 72.0, 96), + ("two_shot_p2p", 8.0, 150.0 / 1.75, 48.0, 64), + ("two_shot_multimem_nvls", 30.0, 300.0, 4.0, 4), + ] + for size in sizes: + for algorithm, base_us, effective_gbytes_s, sm_pct, warps in models: + latency = base_us + size / (effective_gbytes_s * 1e9) * 1e6 + records.append(Measurement(size, algorithm, latency, sm_pct, warps)) + return records + + +def write_measurements(records: list[Measurement], path: Path) -> None: + with path.open("w", encoding="utf-8-sig", newline="") as handle: + writer = csv.DictWriter(handle, list(asdict(records[0]).keys())) + writer.writeheader() + for record in records: + writer.writerow(asdict(record)) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Fit AllReduce message-size thresholds") + parser.add_argument("--measurements", type=Path) + parser.add_argument("--synthetic-demo", action="store_true") + parser.add_argument("--max-sm-occupancy-pct", type=float) + parser.add_argument("--output-dir", type=Path, default=Path("src/code/issue2/results-2026")) + return parser + + +def main() -> int: + args = build_parser().parse_args() + if bool(args.measurements) == bool(args.synthetic_demo): + raise SystemExit("choose exactly one of --measurements or --synthetic-demo") + args.output_dir.mkdir(parents=True, exist_ok=True) + if args.synthetic_demo: + records = synthetic_measurements() + write_measurements(records, args.output_dir / "synthetic_measurements.csv") + else: + records = load_measurements(args.measurements) + winners = choose_at_each_size(records, args.max_sm_occupancy_pct) + report = { + "schema_version": 1, + "sm_constraint_pct": args.max_sm_occupancy_pct, + "intervals": fit_intervals(winners), + "warning": "synthetic data validates fitting only; replace it with NCCL/NVLS measurements", + } + output = args.output_dir / "thresholds.json" + output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + print(f"Wrote {output}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())