[issue2] GPU 内通信方式最佳通信区间阈值拟合器（One-Shot/P2P/NVLS）

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 的多卡机器上用目标实现重新标定。
+

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

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"
+}

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"}
+

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())