diff --git a/BMM/BMM_Theory/bmm_theory/branches/merge_batch.py b/BMM/BMM_Theory/bmm_theory/branches/merge_batch.py index f7142de..bbc61a7 100644 --- a/BMM/BMM_Theory/bmm_theory/branches/merge_batch.py +++ b/BMM/BMM_Theory/bmm_theory/branches/merge_batch.py @@ -7,8 +7,9 @@ 核心思想: 合并 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 处理, 见函数内说明). +T_cmd=0 时 (无命令固定时延/未标定) 命令时延收益不可量化, 但合并仍省 b0 倍 +搬移命令/主机指令数 (指令发射/调度/同步收益未建模) —— beats_iterbatch 按既定 +策略裁决: K截断即优先 MergeBatch (模型内代价仅 drain 惩罚), L1 绑定恒劣. """ from __future__ import annotations @@ -224,8 +225,11 @@ 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 对拍也一致). + + T_cmd=0 (无命令时延/未标定) 时阈值趋于 +inf, 但 MergeBatch 还有**未量化的 + 结构性收益**: 大 B 小 MN 时搬移命令数/主机指令数省 b0 倍 (每条命令的指令 + 发射/调度/同步开销未建模). 因此 T_cmd<=0 采用既定策略: K截断即可胜 + (模型内代价仅为 drain 惩罚, 访存 Bound case 下小且方向已知); L1 绑定仍恒劣. """ s = self.spec m, n, k = case.m, case.n, case.k @@ -244,10 +248,18 @@ class MergeBatchBranch(Branch): 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} 阈值 +inf; 按策略裁决 (见 docstring). + # 大 B 小 MN 时合并把 b_core 条搬移/计算命令序列并为 b_core/b0 条, + # 指令发射/调度/同步收益存在但未量化 —— K截断时判胜, 由路由层以策略覆盖 + # 时延模型比较; L1 绑定 (搬移次数与 IterBatch 相同) 仍恒劣. + if k_truncated: + detail = (f"k_L1=K(截断); T_cmd=0: 命令时延收益不可量化(阈值=+inf), " + f"但合并省 {b0} 倍搬移命令/指令数 (结构性收益, 未量化) -> " + f"策略优先 MergeBatch; 模型内代价 drain 惩罚=" + f"{drain_pen*1e6:.2f}us") + return True, detail + detail = (f"k_L1={k_l1_iter:.0f} 恒劣") return False, detail threshold = b0 * (t_comp + t_write) / s.t_cmd diff --git a/BMM/BMM_Theory/bmm_theory/router.py b/BMM/BMM_Theory/bmm_theory/router.py index 86e7f09..ab876e0 100644 --- a/BMM/BMM_Theory/bmm_theory/router.py +++ b/BMM/BMM_Theory/bmm_theory/router.py @@ -7,7 +7,9 @@ 仲裁规则 (v1.1 §4.5 统一分界): MergeBatch 最优 <=> K截断(k_L1=K) 且 b_core > b0*(T_comp+T_write)/T_cmd L1 绑定时 MergeBatch 恒劣于 IterBatch; - 两分支同时合法时用端到端时延模型 T_total 仲裁. + 两分支同时合法时用端到端时延模型 T_total 仲裁; + 例外: T_cmd<=0 (命令时延不可量化/未标定) 时, 指令级收益未建模, + 按既定策略: K截断即可优先 MergeBatch (覆盖时延模型仅来自 drain/冗余的差额). 3. StreamK 检查: P <= C/2 且满足切K条件 -> StreamK (B/M/N 买不满时买 K) 4. 兜底: ASW_Basic 切 M/N (含降核模式) """ @@ -90,14 +92,28 @@ class BranchRouter: t_ib = ib.timing.t_total lat_win = self.merge_batch.name if t_mb <= t_ib else self.iter_batch.name win = self.merge_batch.name if mb_win else self.iter_batch.name - arbitration = ( - f"两分支均合法, 仲裁: " - f"[分界条件] MergeBatch最优={mb_win} ({detail}); " - f"[时延模型] T_MergeBatch={t_mb*1e6:.2f}us vs T_IterBatch={t_ib*1e6:.2f}us -> {lat_win}更优; " - f"[裁决] {win}" + ("" if win == lat_win else f" (分界条件与时延模型不一致, 以时延模型为准: {lat_win})") - ) - if win != lat_win: - win = lat_win # 时延模型为最终裁决 + # 冲突解决: 默认"时延模型为最终裁决"; 例外是 T_cmd<=0 且分界条件判 + # MergeBatch 胜 (K截断) 的情形 —— 此时时延模型不含指令级收益 + # (MergeBatch 搬移命令数/主机指令数省 b0 倍, 未量化), 按既定策略 + # 优先 MergeBatch (时延模型内的差额只是 drain 惩罚/冗余, 方向已知小量). + policy_merge = (self.spec.t_cmd <= 0 and mb_win and + win != lat_win and win == self.merge_batch.name) + if policy_merge: + arbitration = ( + f"两分支均合法, 仲裁: " + f"[分界条件] MergeBatch最优={mb_win} ({detail}); " + f"[时延模型] T_MergeBatch={t_mb*1e6:.2f}us vs T_IterBatch={t_ib*1e6:.2f}us -> {lat_win}更优; " + f"[裁决] {win} (T_cmd<=0 策略: 命令/指令级收益未建模, 时延模型差异仅来自 " + f"drain/冗余, 以 MergeBatch 优先策略裁决)") + else: + arbitration = ( + f"两分支均合法, 仲裁: " + f"[分界条件] MergeBatch最优={mb_win} ({detail}); " + f"[时延模型] T_MergeBatch={t_mb*1e6:.2f}us vs T_IterBatch={t_ib*1e6:.2f}us -> {lat_win}更优; " + f"[裁决] {win}" + ("" if win == lat_win else f" (分界条件与时延模型不一致, 以时延模型为准: {lat_win})") + ) + if win != lat_win: + win = lat_win # 时延模型为最终裁决 elif any(capable.values()): win = next(n for n, v in capable.items() if v) arbitration = f"仅 {win} 条件满足" diff --git a/BMM/BMM_Theory/tests/test_branches.py b/BMM/BMM_Theory/tests/test_branches.py index 5c1cfd1..d1eda2d 100644 --- a/BMM/BMM_Theory/tests/test_branches.py +++ b/BMM/BMM_Theory/tests/test_branches.py @@ -370,25 +370,30 @@ class TestTransposeModeling(unittest.TestCase): class TestZeroCmdHandling(unittest.TestCase): - """T_cmd=0 (无 DMA 命令固定开销) 时整链路不得除零/崩溃, 建模语义正确.""" + """T_cmd=0 (无命令时延/未标定) 时整链路不得除零/崩溃, 按策略优先 MergeBatch.""" - def test_beats_iterbatch_no_division_error(self): - # T_cmd=0 时阈值 b0*(T_comp+T_write)/T_cmd -> +inf, MergeBatch 必须判不胜出 + def test_beats_iterbatch_policy(self): + # T_cmd<=0: 阈值 +inf 不可除零; K截断按策略判 MergeBatch 胜 (指令级收益未量化), + # L1 绑定仍恒劣 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)]: + # (b, m, n, k, K截断与否) + cases = [(2048, 32, 32, 256, True), (128, 64, 64, 512, True), + (256, 128, 128, 4096, False)] + for b, m, n, k, truncated in cases: win, detail = mb.beats_iterbatch(mkcase(b, m, n, k)) - self.assertFalse(win, f"T_cmd=0 时 MergeBatch 不应胜出: {detail}") + self.assertEqual(win, truncated, f"{b},{m},{n},{k}: {detail}") self.assertIn("T_cmd=0", detail) - def test_route_with_zero_cmd_no_crash(self): - # 整条路由链 (含仲裁) 在 T_cmd=0 下无异常, 时延有限且 MergeBatch 让位 IterBatch + def test_route_with_zero_cmd_prefers_merge(self): + # T_cmd=0 且 K截断时路由应优先 MergeBatch (命令/指令次数少 b0 倍, 结构性收益) 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") # 命令节省消失 -> 无合并价值 + r = router.route(mkcase(2048, 32, 32, 256)) # 大 B 小 MN 典型合并场景 + self.assertEqual(r["branch"], "MergeBatch") + self.assertIn("策略", r["arbitration"]) 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)]