admin/matmul-analysis
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix review issues #17-#22: 恢复 #11/#12/#14 (StreamK fixpipe 单次计账/K=1 AIV单缓冲/advice) + 占位方案不可评估 + 转置 dValue 判据三处同源(form c 双缓冲适配修复) + 恢复 #13/#15 回归测试 + 清理临时 csv/.gitignore + 文档同步

Browse Source
This commit is contained in:
admin
2026-09-03 21:10:05 +08:00
parent f5a0b6fe81
commit 9afe6eec02
22 changed files with 219 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

108
BMM/BMM_Theory/tests/test_branches.py
View File

@@ -150,12 +150,17 @@ class TestIssueRegression(unittest.TestCase):
def setUp(self):
self.router = BranchRouter()
def test_issue4_k1_small_batch_no_crash(self):
# issue#4 P0: K=1 且 B<128 不得崩溃, 应标注"暂无理论方案"
r = self.router.route(mkcase(64, 8192, 32, 1, dtype_a="int8", dtype_b="int8"))
self.assertIsNotNone(r["plan"]) # 占位方案, 不为 None
self.assertEqual(r["plan"].used_core_num, 0) # 标注无方案
self.assertIn("暂无理论方案", r["arbitration"])
def test_issue4_k1_small_batch_real_plan(self):
# issue#4/#12/#17: K=1 且 B<128 不崩溃, 且给出 AIV 单缓冲真实方案 (不再是无方案占位)
case = mkcase(64, 8192, 32, 1, dtype_a="int8", dtype_b="int8")
r = self.router.route(case)
self.assertIsNotNone(r["plan"])
self.assertEqual(r["branch"], "特殊分支")
self.assertEqual(r["plan"].used_core_num, 64) # AIV 核
self.assertNotIn("暂无理论方案", r["arbitration"])
self.assertIn("单缓冲", r["plan"].note)
from bmm_theory.constraints import check_plan_constraints
self.assertEqual(check_plan_constraints(case, r["plan"]), [])
def test_issue5_asw_reduced_core_base_k_dtype_aware(self):
# issue#5: ASW 降核 base_k 按 dtype 反推, fp32 不再 L0A 溢出
@@ -204,6 +209,97 @@ class TestIssueRegression(unittest.TestCase):
self.assertNotEqual(r["branch"], "StreamK")
class TestIssueRegression2(unittest.TestCase):
"""第三轮复评问题 (#17-#20) + 恢复 #11-#15 回归."""
def setUp(self):
self.router = BranchRouter()
def test_issue11_streamk_fixpipe_no_double_count(self):
# issue#11/#17: 部分和写出只经 t_reduce 计账一次; 稳态 fixpipe 不得再计
from bmm_theory.branches.stream_k import StreamKBranch
case = mkcase(4, 128, 128, 10240)
sk = StreamKBranch().analyze(case)
self.assertTrue(sk.capable)
t = sk.timing
self.assertAlmostEqual(t.t_fixpipe, 0.0) # 归约串行口径下无稳态 fixpipe 账
self.assertAlmostEqual(t.fixpipe_bytes, 0.0)
# 端到端 = max(MTE2, MMAD) + 归约, 不再虚高到 55us/FIXPIPE
expect = max(t.t_mte2, t.t_mmad) + t.t_reduce
self.assertAlmostEqual(t.t_total, expect)
self.assertEqual(t.bottleneck, "MTE2_GM")
def test_issue12_k1_pingpong_still_ok(self):
# issue#12/#17: K=1 且 B>=128 仍走 UB 乒乓 (原行为不变)
r = self.router.route(mkcase(128, 256, 256, 1))
self.assertEqual(r["branch"], "特殊分支")
self.assertIn("乒乓", r["plan"].l1_form)
self.assertIsNotNone(r["timing"])
def test_issue13_merge_b0_l0ab_capped(self):
# issue#13: MergeBatch 瘦长 case 的 b0 受 L0A/L0B 容量约束 (B=811 M=33 N=1 fp32)
from bmm_theory.constraints import check_plan_constraints
case = mkcase(811, 33, 1, 2459, dtype_a="fp32", dtype_b="fp32", dtype_c="fp32")
r = self.router.route(case)
self.assertEqual(r["branch"], "MergeBatch")
p = r["plan"]
self.assertLessEqual(p.base_m * p.base_k * 4 * 2, 64 * 1024) # L0A 容量内
self.assertLessEqual(p.base_n * p.base_k * 4 * 2, 64 * 1024) # L0B 容量内
self.assertEqual(check_plan_constraints(case, p), [])
def test_issue13_router_fallback_when_winner_infeasible(self):
# issue#13: 仲裁胜出的 MergeBatch 自检违规时, 回退到可行候选 IterBatch
from bmm_theory.constraints import check_plan_constraints
case = mkcase(256, 1, 256, 4096, dtype_a="int8", dtype_b="int8") # 原 0.2% 违规样例
r = self.router.route(case)
self.assertEqual(r["branch"], "IterBatch") # 回退
self.assertIn("自检违规", r["arbitration"])
self.assertIn("回退", r["arbitration"])
self.assertEqual(check_plan_constraints(case, r["plan"]), [])
def test_issue15_input_validation(self):
# issue#15: 非法维度/负值必须抛错, 不再静默产出伪方案
for kw in (dict(m=0), dict(m=-5), dict(n=0), dict(k=-1),
dict(batch_a=0), dict(batch_b=-3)):
with self.assertRaises(ValueError, msg=str(kw)):
BmmCase(case_id="bad", **kw)
with self.assertRaises(ValueError):
BmmCase(case_id="bad", m=64, n=64, k=1, dtype_a="xxx")
def test_issue18_placeholder_plan_infeasible_in_evaluate(self):
# issue#18: 占位方案(used_core_num=0)在 evaluate 中必须不可行,
# 不得被当作可行方案给出正常时延
from bmm_theory.evaluator import PlanEvaluator
from bmm_theory.models import ImplPlan
# K=1 且 B<128 已恢复真实单缓冲方案 (#17), 故直接构造占位 plan 验证约束层
case = mkcase(64, 8192, 32, 1, dtype_a="int8", dtype_b="int8")
r = self.router.route(case)
self.assertGreater(r["plan"].used_core_num, 0) # 真实方案
ph = ImplPlan(case_id="ph", branch="特殊分支", used_core_num=0,
note="该区域暂无理论方案(进入条件不满足)")
er = PlanEvaluator().evaluate(case, ph)
self.assertFalse(er.feasible, "占位方案应判不可行")
self.assertIn("used_core_num", er.violations)
def test_issue19_transpose_dvalue_guard_effective(self):
# issue#19: 转置感知 dValue 判据在生成守卫/条件4/约束三处同源后真正生效.
# 判别形状: B=64 M=4096 N=64 K=4096 bf16 —— d 形态 k_l1=16,
# 均不转置时 dv_a=k_l1*2=32B <128 挡下 (B 侧 dv_b=N*2=128B 恰好达标也不放行,
# 因为两侧切 K 两侧都要高效);
# A 转置后 dv_a=M*2=8192B, 应能走 IterBatch 形态 d.
from bmm_theory.constraints import check_plan_constraints
c_not = BmmCase(case_id="x", batch_a=64, batch_b=64, m=4096, n=64, k=4096,
trans_a=False, trans_b=False)
r_not = self.router.route(c_not)
self.assertNotEqual(r_not["branch"], "IterBatch") # 非转置被 dValue 守卫挡下
c_tr = BmmCase(case_id="x", batch_a=64, batch_b=64, m=4096, n=64, k=4096,
trans_a=True, trans_b=False)
r_tr = self.router.route(c_tr)
self.assertEqual(r_tr["branch"], "IterBatch") # A 转置 M 向连续, 守卫放行
self.assertIn("d_", r_tr["plan"].l1_form)
self.assertEqual(check_plan_constraints(c_tr, r_tr["plan"]), [])
class TestFp4Support(unittest.TestCase):
"""fp4 (0.5B) dtype 支持 (对齐 bmmv3, 2026-09-03)."""