diff --git a/BMM/BMM_Theory/bmm_theory/branches/merge_batch.py b/BMM/BMM_Theory/bmm_theory/branches/merge_batch.py index 43038d2..f7142de 100644 --- a/BMM/BMM_Theory/bmm_theory/branches/merge_batch.py +++ b/BMM/BMM_Theory/bmm_theory/branches/merge_batch.py @@ -7,6 +7,8 @@ 核心思想: 合并 b0 个 batch 的 A'[b0*M,K] @ B'[K,b0*N] 为单次 DMA 搬入, 减少 GM->L1 搬移命令数 (省 b0 倍 T_cmd); 交叉项被算出但丢弃 (冗余比例 (b0-1)/b0), 进入条件 5 保证 case 为访存 Bound, 冗余算力被搬移时延掩盖. +T_cmd=0 时 (无命令固定开销) 合并失去唯一收益来源, 分界仲裁直接判 MergeBatch +不胜出 (beats_iterbatch 对 T_cmd<=0 做 +inf 处理, 见函数内说明). """ from __future__ import annotations @@ -222,6 +224,8 @@ class MergeBatchBranch(Branch): MergeBatch 最优 ⟺ K截断 (k_L1=K) 且 b_core > b0*(T_comp+T_write)/T_cmd L1 绑定情形 MergeBatch 恒劣于 IterBatch (搬移次数相同, 只放大 drain). + T_cmd=0 时阈值趋于无穷: 合并的唯一收益 (省 DMA 命令开销) 消失, + 只剩 drain 放大与冗余计算, MergeBatch 无胜出通道 (见 evaluate 对拍也一致). """ s = self.spec m, n, k = case.m, case.n, case.k @@ -237,12 +241,21 @@ class MergeBatchBranch(Branch): b0 = plan.merge_b0 t_comp = 2.0 * m * n * min(k_l1_iter, k) / s.q16 t_write = m * n * out_b / s.bw_pc + drain_pen = (b0 - 1) * (t_comp + t_write) + + if s.t_cmd <= 0: + # T_cmd=0: 阈值 b0*(T_comp+T_write)/T_cmd -> +inf, 不可除零, 直接判负 + detail = (f"k_L1={'K(截断)' if k_truncated else f'{k_l1_iter:.0f} threshold) detail = (f"k_L1={'K(截断)' if k_truncated else f'{k_l1_iter:.0f} +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()