Fix T_cmd=0 robustness: beats_iterbatch 阈值除零防护 (T_cmd<=0 判 MergeBatch 不胜出, 无命令节省) + 3 条回归测试

This commit is contained in:
2026-09-04 10:31:53 +08:00
parent 9afe6eec02
commit a071d87831
2 changed files with 64 additions and 1 deletions

View File

@@ -369,5 +369,55 @@ class TestTransposeModeling(unittest.TestCase):
self.assertFalse(c4.passed, f"A 不转置时 K=8 应报 dValue 违规: {c4.detail}")
class TestZeroCmdHandling(unittest.TestCase):
"""T_cmd=0 (无 DMA 命令固定开销) 时整链路不得除零/崩溃, 建模语义正确."""
def test_beats_iterbatch_no_division_error(self):
# T_cmd=0 时阈值 b0*(T_comp+T_write)/T_cmd -> +inf, MergeBatch 必须判不胜出
from bmm_theory.hardware import NpuSpec
from bmm_theory.branches.merge_batch import MergeBatchBranch
spec0 = NpuSpec(t_cmd_ns=0.0)
mb = MergeBatchBranch(spec0)
for b, m, n, k in [(2048, 32, 32, 256), (128, 64, 64, 512), (256, 128, 128, 4096)]:
win, detail = mb.beats_iterbatch(mkcase(b, m, n, k))
self.assertFalse(win, f"T_cmd=0 时 MergeBatch 不应胜出: {detail}")
self.assertIn("T_cmd=0", detail)
def test_route_with_zero_cmd_no_crash(self):
# 整条路由链 (含仲裁) 在 T_cmd=0 下无异常, 时延有限且 MergeBatch 让位 IterBatch
from bmm_theory.hardware import NpuSpec
router = BranchRouter(NpuSpec(t_cmd_ns=0.0))
r = router.route(mkcase(2048, 32, 32, 256)) # 原 MergeBatch 典型胜场形状
self.assertEqual(r["branch"], "IterBatch") # 命令节省消失 -> 无合并价值
self.assertIsNotNone(r["timing"])
shapes = [(128, 64, 64, 512), (64, 64, 64, 8192), (512, 128, 128, 128),
(128, 128, 128, 1024), (32, 4096, 4096, 4096)]
for b, m, n, k in shapes:
rr = router.route(mkcase(b, m, n, k))
t = rr["timing"]
self.assertIsNotNone(t, f"{b},{m},{n},{k} 应有 timing")
self.assertGreater(t.t_total, 0.0)
self.assertEqual(t.t_total, t.t_total) # 非 NaN
def test_zero_cmd_random_smoke(self):
# 随机小样本冒烟: 无异常/无 NaN
import random
from bmm_theory.hardware import NpuSpec
router = BranchRouter(NpuSpec(t_cmd_ns=0.0))
rng = random.Random(9)
for _ in range(300):
b = rng.choice([2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048])
m = rng.choice([8, 16, 32, 64, 128, 256, 512, 1024, 4096])
n = rng.choice([8, 16, 32, 64, 128, 256, 512, 1024, 4096])
k = rng.choice([16, 32, 64, 128, 256, 512, 1024, 4096, 16384])
c = BmmCase(case_id="z", batch_a=b, batch_b=b, m=m, n=n, k=k)
rr = router.route(c)
self.assertIsNotNone(rr["plan"])
t = rr["timing"]
self.assertIsNotNone(t)
self.assertGreater(t.t_total, 0.0)
self.assertEqual(t.t_total, t.t_total)
if __name__ == "__main__":
unittest.main()