diff --git a/hw/sim/Makefile b/hw/sim/Makefile index 8ee4b56..2a7535b 100644 --- a/hw/sim/Makefile +++ b/hw/sim/Makefile @@ -1,22 +1,31 @@ PYTHON ?= python MODULE_TESTS := cam_core_banked candidate_fifo match_engine_pipeline cam_write_noise cam_read_noise popcount_pipeline topk_tracker result_serializer +TOP_CONFIGS := no_noise write_noise read_noise -.PHONY: help test-all test-top test-modules test-module test-model test-perf clean $(MODULE_TESTS:%=test-module-%) +.PHONY: help test-all test-top test-top-all test-modules test-module test-model test-perf clean $(MODULE_TESTS:%=test-module-%) $(TOP_CONFIGS:%=test-top-%) help: @echo "Available hw/sim targets:" @echo " make test-model" - @echo " make test-top" + @echo " make test-top # 只运行默认顶层配置 (no_noise)" + @echo " make test-top-all # 运行所有顶层噪声配置" + @echo " make test-top-no_noise # 无噪声集成测试" + @echo " make test-top-write_noise # 写入噪声集成测试" + @echo " make test-top-read_noise # 读取噪声集成测试" @echo " make test-module MODULE=cam_core_banked" @echo " make test-modules" @echo " make test-perf" @echo " make test-all" @echo " make clean" -test-all: test-model test-top test-modules +test-all: test-model test-top-all test-modules -test-top: - $(MAKE) -C tests/top +test-top: test-top-no_noise + +test-top-all: $(TOP_CONFIGS:%=test-top-%) + +$(TOP_CONFIGS:%=test-top-%): + $(MAKE) -C tests/top/$(@:test-top-%=%) test-modules: $(MODULE_TESTS:%=test-module-%) @@ -34,7 +43,9 @@ test-perf: $(MAKE) -C tests/perf clean: - $(MAKE) -C tests/top clean + @for config in $(TOP_CONFIGS); do \ + $(MAKE) -C tests/top/$$config clean || exit $$?; \ + done @for module in $(MODULE_TESTS); do \ $(MAKE) -C tests/modules/$$module clean || exit $$?; \ done diff --git a/hw/sim/tests/model/test_ref_model_noise.py b/hw/sim/tests/model/test_ref_model_noise.py index bfe5714..d824eaf 100644 --- a/hw/sim/tests/model/test_ref_model_noise.py +++ b/hw/sim/tests/model/test_ref_model_noise.py @@ -1,3 +1,17 @@ +# -*- coding: utf-8 -*- +""" +参考模型（ref_model）的纯 Python 单元测试。 + +本文件不涉及任何 RTL / Verilator 仿真，仅验证 Python 参考模型的正确性。 +所有 RTL-vs-模型 的对比测试（如顶层 test_cam_basic.py）都依赖此参考模型， +因此这里是整个测试体系的「基石」——参考模型如果有 bug，所有对比测试都将失效。 + +测试覆盖： + 1. 分组翻转掩码 — 完全速率 (rate=1/1) 的正确位翻转模式 + 2. 分组翻转掩码 — 零速率 (rate=0/100) 不应产生任何翻转 + 3. 评分函数语义 — 确认是「匹配位数」而非「汉明距离」 + 4. 读取噪声模型 — 相同输入 + 相同种子 = 可复现结果 +""" from __future__ import annotations from model.ref_model import ( @@ -7,7 +21,26 @@ from model.ref_model import ( ) +# ============================================================================== +# 测试 1：完全速率下的分组翻转掩码生成 +# ============================================================================== + + def test_grouped_flip_mask_full_rate_one_bit_per_64_bit_group(): + """ + 验证 generate_grouped_flip_mask 在 rate_num=1, rate_den=1 时的行为。 + + 背景： + - CAM 的 write noise 模块将 512-bit 哈希按 64-bit 分组，每组最多翻转 1 位。 + - random_value 的位域含义（每 group 14 bits）： + bits [5:0] → sample（未使用） + bits [13:6] → bit_idx（选择该组内翻转哪一位） + + 本测试： + - 构造一个 random_value，使每个 group 的 bit_idx = group+1 + - 断言生成的 mask 恰好有 8 个位被置位（每 group 一个） + - 断言每个被翻转的位位置与预期一致 + """ random_value = 0 for group in range(8): bit_idx = group + 1 @@ -17,38 +50,88 @@ def test_grouped_flip_mask_full_rate_one_bit_per_64_bit_group(): mask = generate_grouped_flip_mask( random_value=random_value, - hash_bits=512, - noise_bits=8, - rate_num=1, - rate_den=1, + hash_bits=512, # 8 组 × 64 bits/组 + noise_bits=8, # 每组的 bit_idx 位宽 + rate_num=1, # 分子 = 1 + rate_den=1, # 分母 = 1 → 100% 概率，每组都翻转 ) + # 预期：每组的 bit_idx 位被翻转 expected = 0 for group in range(8): expected |= 1 << (group * 64 + group + 1) assert mask == expected - assert mask.bit_count() == 8 + assert mask.bit_count() == 8 # 恰好 8 位被翻转（每组一位） + + +# ============================================================================== +# 测试 2：零速率下不应产生任何翻转 +# ============================================================================== def test_grouped_flip_mask_zero_rate_no_flips(): + """ + 验证 rate_num=0 时，无论 random_value 为何值，mask 都应为 0。 + + 这是写入噪声的「零噪声」配置边界测试—— + 确保 RTL 参数 WRITE_NOISE_RATE_NUM=0 能真正关闭噪声注入。 + """ mask = generate_grouped_flip_mask( - random_value=(1 << 128) - 1, + random_value=(1 << 128) - 1, # 全 1 的 random_value hash_bits=512, noise_bits=8, - rate_num=0, + rate_num=0, # 分子 = 0 → 翻转概率为 0 rate_den=100, ) - assert mask == 0 + assert mask == 0 # mask 必须全 0，一个位都不翻 + + +# ============================================================================== +# 测试 3：评分函数语义 — 确认是「XNOR 匹配位数」而非「汉明距离」 +# ============================================================================== def test_score_is_bit_match_popcount_not_hamming_distance(): + """ + 关键语义验证：xnor_popcount_score 计算的是匹配位的数量，不是汉明距离。 + + 示例： + query = 0b1010 + stored = 0b1000 + XNOR = 0b1101 → popcount = 3（有 3 个位匹配） + 汉明距离 = 1 → 只有一个位不同 + + 为什么这个区分很重要： + - 如果 RTL 或模型错误地使用了汉明距离作为分数，则： + 完全匹配的分数会是 0 而非 hash_bits（512） + Top-K 排序会颠倒（分数低的反而排前面） + - 这会导致整个 CAM 检索系统返回错误结果 + """ query = 0b1010 stored = 0b1000 assert xnor_popcount_score(query, stored, width=4) == 3 +# ============================================================================== +# 测试 4：读取噪声模型的可复现性（确定性种子） +# ============================================================================== + + def test_read_noise_model_is_reproducible_after_reset_seed(): + """ + 验证 match_top1_with_read_noise 在相同参数下产生相同结果。 + + 为什么这个测试至关重要： + - RTL 中的 read noise PRNG 使用固定种子 (0x6A09E667F3BCC909) + - 参考模型必须使用相同的种子来复现 RTL 的噪声行为 + - 如果两次调用结果不同，说明模型存在非确定性 bug + （如未重置 PRNG 状态、或使用了非确定性随机源） + + 测试数据： + - 8 行不同模式的 512-bit 哈希（全0、全1、稀疏值） + - 噪声配置：rate=1%, lanes=8, noise_bits=8 + """ rows = [0, (1 << 512) - 1, 0x1234, 0x5678, 0x9ABC, 0xDEF0, 0x1357, 0x2468] query = rows[2] kwargs = dict( @@ -65,6 +148,7 @@ def test_read_noise_model_is_reproducible_after_reset_seed(): first = match_top1_with_read_noise(**kwargs) second = match_top1_with_read_noise(**kwargs) + # 两次调用的 Top-1 结果和分数数组必须完全一致 assert first.top1_index == second.top1_index assert first.top1_score == second.top1_score assert first.scores.tolist() == second.scores.tolist() diff --git a/hw/sim/tests/modules/result_serializer/__init__.py b/hw/sim/tests/modules/result_serializer/__init__.py new file mode 100644 index 0000000..4e3cd26 --- /dev/null +++ b/hw/sim/tests/modules/result_serializer/__init__.py @@ -0,0 +1 @@ +# Result serializer module tests diff --git a/hw/sim/tests/modules/topk_tracker/__init__.py b/hw/sim/tests/modules/topk_tracker/__init__.py new file mode 100644 index 0000000..a1c8618 --- /dev/null +++ b/hw/sim/tests/modules/topk_tracker/__init__.py @@ -0,0 +1 @@ +# Top-K tracker module tests diff --git a/hw/sim/tests/top/Makefile b/hw/sim/tests/top/no_noise/Makefile similarity index 53% rename from hw/sim/tests/top/Makefile rename to hw/sim/tests/top/no_noise/Makefile index 1aca101..ad64f98 100644 --- a/hw/sim/tests/top/Makefile +++ b/hw/sim/tests/top/no_noise/Makefile @@ -1,9 +1,13 @@ -SIM_ROOT := $(abspath ../..) +SIM_ROOT := $(abspath ../../..) RTL_ROOT := $(abspath $(SIM_ROOT)/../rtl) include $(SIM_ROOT)/mk/rtl-sources.mk TOPLEVEL := cam_top -COCOTB_TEST_MODULES := tests.top.test_cam_basic +COCOTB_TEST_MODULES := tests.top.no_noise.test_no_noise VERILOG_SOURCES := $(RTL_CAM_TOP) +# 禁用所有噪声模块 +WRITE_NOISE_EN := 0 +READ_NOISE_EN := 0 + include $(SIM_ROOT)/mk/cocotb-common.mk diff --git a/hw/sim/tests/top/no_noise/__init__.py b/hw/sim/tests/top/no_noise/__init__.py new file mode 100644 index 0000000..4798807 --- /dev/null +++ b/hw/sim/tests/top/no_noise/__init__.py @@ -0,0 +1 @@ +# CAM top-level no-noise integration tests diff --git a/hw/sim/tests/top/no_noise/test_no_noise.py b/hw/sim/tests/top/no_noise/test_no_noise.py new file mode 100644 index 0000000..d35706f --- /dev/null +++ b/hw/sim/tests/top/no_noise/test_no_noise.py @@ -0,0 +1,348 @@ +# -*- coding: utf-8 -*- +""" +CAM 顶层（cam_top）no_noise 配置集成测试（WRITE_NOISE_EN=0, READ_NOISE_EN=0）。 + +所有噪声模块禁用，验证 CAM 在无噪声下的标准行为。 + +=== 测试清单 === + + - compile_includes_grouped_noise_helper — 编译冒烟 + - baseline_no_noise — 基线检索正确性 + - known_hamming_distance — 汉明距离验证 + - tie_break_policy — 平局决胜 + - all_zero_all_one_boundary — 全 0 / 全 1 边界 + - half_duplex_write_priority — 半双工写入优先 + - banked_pipeline_no_noise_top1 — 分块流水线 Top-1 + - query_scan_blocks_writes_until_result_consumed — 查询阻塞写入 + +=== 配置背景 === + + 本目录固定使用 WRITE_NOISE_EN=0 和 READ_NOISE_EN=0 编译， + 因此所有测试无需运行时参数门控——Makefile 保证配置正确。 +""" + +from __future__ import annotations + +import cocotb +import numpy as np +from cocotb.clock import Clock +from cocotb.triggers import RisingEdge +from model.ref_model import ( + match_top1, + random_hashes, +) + +from tests.top.utils import ( + collect_topk, + dut_hash_bits, + dut_num_rows, + query_once, + query_topk_once, + reset_dut, + wait_idle, + write_row, + write_rows, +) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 编译冒烟测试 — 验证 cam_top 能正确 elaborate +# ═══════════════════════════════════════════════════════════════════════════════ + + +@cocotb.test() +async def compile_includes_grouped_noise_helper(dut): + """编译测试：验证群组噪声辅助模块被正确包含在 cam_top 中。 + + 不验证功能，只确保 Verilator elaboration 不会报错。 + """ + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + assert int(dut.wr_ready.value) in (0, 1) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 测试 A：基线（WRITE_NOISE_EN=0, READ_NOISE_EN=0） +# ── 验证写+查在噪声关闭时与旧 CAM 行为完全一致 +# ═══════════════════════════════════════════════════════════════════════════════ + + +@cocotb.test() +async def baseline_no_noise(dut): + """基线测试：噪声全部关闭时，检索结果必须与 Python 参考模型完全一致。 + + 验证内容： + 1. Top-1 索引和分数与 match_top1 模型一致 + 2. 查询自身所在行 → 分数必须等于 HASH_BITS（完全匹配） + 3. 串行 Top-K 流的第一个 beat rank=0 + 4. 最后一个 beat 的 result_last=1（流终止） + 5. top1_index/top1_score 别名与第一个 beat 的值一致 + 6. score_debug（如果存在）与模型分数数组逐元素一致 + """ + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + + num_rows = dut_num_rows(dut) + hash_bits = dut_hash_bits(dut) + rng = np.random.default_rng(1) + rows = random_hashes(rng, num_rows, width=hash_bits) + query_index = min(123, num_rows - 1) + query = rows[query_index] + + await write_rows(dut, rows) + beats, top1_index, top1_score, score_debug = await query_topk_once(dut, query) + + expected = match_top1(query, rows, width=hash_bits) + + assert top1_index == expected.top1_index + assert top1_score == expected.top1_score + assert top1_index == query_index + assert top1_score == hash_bits + + # Serial Top-K stream verification: beats from the single query above + assert beats[0][0] == 0, "First beat must have rank 0" + assert beats[-1][3] == 1, "Last beat must assert result_last" + # Verify top1 aliases match first beat after stream fully consumed + assert int(dut.top1_index.value) == beats[0][1] + assert int(dut.top1_score.value) == beats[0][2] + # Verify returned top1 matches first beat rank0 + assert top1_index == beats[0][1] + assert top1_score == beats[0][2] + + if score_debug is not None: + assert np.array_equal(score_debug, expected.scores) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 遗留测试 — 仅在噪声关闭时有意义（精确分数才有效） +# ═══════════════════════════════════════════════════════════════════════════════ + + +@cocotb.test() +async def known_hamming_distance(dut): + """汉明距离验证：写入已知模式的哈希，验证分数计算正确。 + + 测试数据： + - Row 0: 全 0 + - Row 10: 低 7 位为 1 → score = hash_bits - 7 + - Row 11: 低 31 位为 1 → score = hash_bits - 31 + - Row 12: 低 128 位为 1→ score = hash_bits - 128 + - query = 0 + + 验证：Top-1 为 row 0（完全匹配），各行的 score_debug + 精确等于理论汉明距离对应的匹配位数。 + """ + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + + num_rows = dut_num_rows(dut) + hash_bits = dut_hash_bits(dut) + + query = 0 + rows = [0] * num_rows + rows[min(10, num_rows - 1)] = (1 << 7) - 1 + rows[min(11, num_rows - 1)] = (1 << 31) - 1 + rows[min(12, num_rows - 1)] = (1 << 128) - 1 + + await write_rows(dut, rows) + top1_index, top1_score, score_debug = await query_once(dut, query) + + assert top1_index == 0 + assert top1_score == hash_bits + + if score_debug is not None: + assert int(score_debug[min(10, num_rows - 1)]) == hash_bits - 7 + assert int(score_debug[min(11, num_rows - 1)]) == hash_bits - 31 + assert int(score_debug[min(12, num_rows - 1)]) == hash_bits - 128 + + +@cocotb.test() +async def tie_break_policy(dut): + """平局决胜策略：分数相同时，行号最小的获胜。 + + 设置： + - row 10, 20, 200 都存储了 query 的值（满分匹配） + - 其余行随机填充 + + 预期：top1_index = 10（不是 20 或 200） + """ + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + + num_rows = dut_num_rows(dut) + hash_bits = dut_hash_bits(dut) + rng = np.random.default_rng(2) + rows = random_hashes(rng, num_rows, width=hash_bits) + query = rows[min(200, num_rows - 1)] + rows[10] = query + rows[20] = query + rows[min(200, num_rows - 1)] = query + + await write_rows(dut, rows) + top1_index, top1_score, _ = await query_once(dut, query) + + assert top1_index == 10 + assert top1_score == hash_bits + + +@cocotb.test() +async def all_zero_all_one_boundary(dut): + """全 0 / 全 1 边界测试：验证极端哈希值的检索正确性。 + + 存储： + - row 0: 全 0 (0x000...000) + - row 1: 全 1 (0xFFF...FFF) + - 查询 = 全 0 + + 预期：Top-1 = row 0, score = hash_bits + row 1 的 score = 0（全 0 与全 1 无任何匹配位） + """ + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + + num_rows = dut_num_rows(dut) + hash_bits = dut_hash_bits(dut) + + rows = [0] * num_rows + rows[0] = 0 + rows[1] = (1 << hash_bits) - 1 + + query = 0 + await write_rows(dut, rows) + top1_index, top1_score, score_debug = await query_once(dut, query) + + assert top1_score == hash_bits + assert top1_index == 0 + + if score_debug is not None: + assert int(score_debug[0]) == hash_bits + assert int(score_debug[1]) == 0 + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 测试 F：半双工写入优先级仲裁 +# ── wr_valid 和 query_valid 同时有效 → 写入优先，查询被暂缓 +# ═══════════════════════════════════════════════════════════════════════════════ + + +@cocotb.test() +async def half_duplex_write_priority(dut): + """半双工仲裁：同时发起写入和查询 → 写入必须胜出。 + + 流程： + 1. 预先写入 row 0（值为 test_val） + 2. 同时驱动 wr_valid 和 query_valid，写入 row 1，查询 test_val + 3. 断言：写入被接受，row 1 被正确写入 + 4. 随后查询 test_val → 应返回 row 0 和 row 1 都是满分 + + 如果仲裁逻辑错误（查询胜出或同时处理），两个行中至少有一个会丢失。 + """ + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + + hash_bits = dut_hash_bits(dut) + test_val = (1 << hash_bits) - 1 + await write_row(dut, 0, test_val) + + await wait_idle(dut) + assert int(dut.wr_ready.value) == 1 + assert int(dut.query_ready.value) == 1 + + dut.wr_valid.value = 1 + dut.wr_addr.value = 1 + dut.write_hash.value = 0 + dut.query_valid.value = 1 + dut.query_hash.value = test_val + + await RisingEdge(dut.clk) + + dut.wr_valid.value = 0 + dut.query_valid.value = 0 + + await wait_idle(dut) + + top1_index, top1_score, _ = await query_once(dut, test_val) + assert top1_index == 0 + assert top1_score == hash_bits + + top1_index, top1_score, _ = await query_once(dut, 0) + assert top1_index == 1 + assert top1_score == hash_bits + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 测试 G：分块流水线无噪声 Top-1 +# ── 验证分块存储架构在无噪声时返回正确的 Top-1 +# ═══════════════════════════════════════════════════════════════════════════════ + + +@cocotb.test() +async def banked_pipeline_no_noise_top1(dut): + """分块流水线 Top-1：无噪声时，分块架构的结果必须与纯模型一致。 + + 这是 banked_pipeline 的冒烟测试——验证分块存储核心、 + 通道合并逻辑、以及匹配引擎流水线在端到端场景中协同工作。 + """ + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + + num_rows = dut_num_rows(dut) + hash_bits = dut_hash_bits(dut) + rng = np.random.default_rng(7) + rows = random_hashes(rng, num_rows, width=hash_bits) + query_index = min(17, num_rows - 1) + query = rows[query_index] + + await write_rows(dut, rows) + top1_index, top1_score, score_debug = await query_once(dut, query) + expected = match_top1(query, rows, width=hash_bits) + + assert top1_index == expected.top1_index + assert top1_score == expected.top1_score + assert top1_index == query_index + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 测试 H：查询扫描期间阻塞写入 +# ── 活跃的查询扫描会撤销 wr_ready，直到结果被消费完毕 +# ═══════════════════════════════════════════════════════════════════════════════ + + +@cocotb.test() +async def query_scan_blocks_writes_until_result_consumed(dut): + """半双工阻塞：查询扫描活跃期间，wr_ready 必须保持低电平。 + + 流程： + 1. 写入全部行 + 2. 发起查询（query_valid=1） + 3. 在结果被消费之前，尝试写入 → 断言 wr_ready=0 + 4. 消费完整结果流 → DUT 回到空闲 + + 这验证了半双工协议中「读期间禁止写」的约束。 + """ + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + + num_rows = dut_num_rows(dut) + hash_bits = dut_hash_bits(dut) + rows = [0] * num_rows + rows[0] = (1 << hash_bits) - 1 + await write_rows(dut, rows) + + await wait_idle(dut) + dut.query_hash.value = rows[0] + dut.query_valid.value = 1 + await RisingEdge(dut.clk) + dut.query_valid.value = 0 + + dut.wr_valid.value = 1 + dut.wr_addr.value = 1 + dut.write_hash.value = 0 + await RisingEdge(dut.clk) + assert int(dut.wr_ready.value) == 0 + dut.wr_valid.value = 0 + + # Consume full serial stream so the DUT returns idle + beats = await collect_topk(dut, timeout_cycles=2000) + assert len(beats) > 0 + assert beats[-1][3] == 1 # last asserted diff --git a/hw/sim/tests/top/read_noise/Makefile b/hw/sim/tests/top/read_noise/Makefile new file mode 100644 index 0000000..aacad2a --- /dev/null +++ b/hw/sim/tests/top/read_noise/Makefile @@ -0,0 +1,25 @@ +SIM_ROOT := $(abspath ../../..) +RTL_ROOT := $(abspath $(SIM_ROOT)/../rtl) +include $(SIM_ROOT)/mk/rtl-sources.mk + +TOPLEVEL := cam_top +COCOTB_TEST_MODULES := tests.top.read_noise.test_read_noise +VERILOG_SOURCES := $(RTL_CAM_TOP) + +# 读取噪声开启，写入噪声默认关闭 +READ_NOISE_EN := 1 +READ_NOISE_RATE_NUM := 1 +READ_NOISE_RATE_DEN := 100 +WRITE_NOISE_EN := 0 + +include $(SIM_ROOT)/mk/cocotb-common.mk + +# ── 写入+读取双重噪声子目标 ───────────────────────────────────────── +.PHONY: test-with-write-noise + +test-with-write-noise: + $(MAKE) -B -f Makefile results.xml WRITE_NOISE_EN=1 \ + COCOTB_TEST_FILTER=read_noise_model_match + +clean:: + rm -rf sim_build diff --git a/hw/sim/tests/top/read_noise/__init__.py b/hw/sim/tests/top/read_noise/__init__.py new file mode 100644 index 0000000..126363f --- /dev/null +++ b/hw/sim/tests/top/read_noise/__init__.py @@ -0,0 +1 @@ +# CAM top-level read-noise integration tests diff --git a/hw/sim/tests/top/read_noise/test_read_noise.py b/hw/sim/tests/top/read_noise/test_read_noise.py new file mode 100644 index 0000000..199f345 --- /dev/null +++ b/hw/sim/tests/top/read_noise/test_read_noise.py @@ -0,0 +1,115 @@ +# -*- coding: utf-8 -*- +""" +CAM 读取噪声（read_noise）集成测试。 + +本文件针对 READ_NOISE_EN=1 的编译配置，验证 RTL 的读取噪声行为 +与 Python 参考模型（ref_model）一致。 + +=== 测试内容 === + + read_noise_model_match — 读取噪声模型匹配： + 写入原始哈希，预测含写入噪声（如果 WRITE_NOISE_EN=1）的存储值， + 再用 match_top1_with_read_noise 计算含读取噪声的期望结果， + 与 RTL 实际 Top-1 进行比对。 + +=== 架构背景 === + + CAM 硬件由以下流水线组成： + Write Noise → Banked Core Storage → Read Noise → Match Engine Pipeline + ↓ + Top-K Tracker → Result Serializer + + 本测试覆盖的是 Read Noise → Match Engine 段。 + 写入噪声（WRITE_NOISE_EN）通过 Makefile 的 test-with-write-noise 子目标 + 启用，测试代码内部已兼容两种配置。 +""" + +from __future__ import annotations + +import cocotb +import numpy as np +from cocotb.clock import Clock +from cocotb.triggers import RisingEdge +from model.ref_model import ( + generate_write_flip_mask, + match_top1_with_read_noise, + random_hashes, + unpack_score_debug_flat, +) +from tests.top.utils import ( + dut_hash_bits, + dut_lanes, + dut_num_rows, + get_param, + query_once, + reset_dut, + write_rows, +) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 测试：读取噪声模型匹配 +# ── READ_NOISE_EN=1 由 Makefile 保证，测试代码中不再重复门控 +# ═══════════════════════════════════════════════════════════════════════════════ + + +@cocotb.test() +async def read_noise_model_match(dut): + """读取噪声模型匹配：验证 RTL 的读取噪声行为与 Python 参考模型一致。 + + 与写入噪声不同，读取噪声发生在查询阶段（每次查询向哈希值注入噪声）， + 因此： + - 如果先有写入噪声，存储行已经被翻转过一次 + - 然后查询时还会再注入一层读取噪声 + - 两层噪声使用不同的种子（写: 0xB504..., 读: 0x6A09...） + + 本测试： + 1. 用 Python 模型预计算存储后的哈希（含写入噪声） + 2. 用 match_top1_with_read_noise 预计算含读取噪声的期望结果 + 3. 写入原始值到 RTL，查询，比对结果 + """ + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + + num_rows = dut_num_rows(dut) + hash_bits = dut_hash_bits(dut) + lanes = dut_lanes(dut) + rng = np.random.default_rng(123) + rows = random_hashes(rng, num_rows, width=hash_bits) + + # If write noise is enabled, apply write flip masks to predict stored rows + stored_rows = list(rows) + if get_param(dut, "WRITE_NOISE_EN", 0): + seed = 0xB504_F32D_B504_F32D + prng_state = (seed << 64) | seed + stored_rows = [] + for row in rows: + flip, prng_state = generate_write_flip_mask( + prng_state, + hash_bits, + get_param(dut, "WRITE_NOISE_BITS", 8), + get_param(dut, "WRITE_NOISE_RATE_NUM", 1), + get_param(dut, "WRITE_NOISE_RATE_DEN", 100), + ) + stored_rows.append(row ^ flip) + + query = rows[min(5, num_rows - 1)] + + await write_rows(dut, rows) + top1_index, top1_score, score_debug = await query_once(dut, query) + + expected = match_top1_with_read_noise( + query, + stored_rows, + width=hash_bits, + lanes=lanes, + noise_bits=get_param(dut, "READ_NOISE_BITS", 8), + rate_num=get_param(dut, "READ_NOISE_RATE_NUM", 1), + rate_den=get_param(dut, "READ_NOISE_RATE_DEN", 100), + seed=0x6A09_E667_F3BC_C909, + ) + + assert top1_index == expected.top1_index + assert top1_score == expected.top1_score + if score_debug is not None: + assert np.array_equal(score_debug, expected.scores) diff --git a/hw/sim/tests/top/test_cam_basic.py b/hw/sim/tests/top/test_cam_basic.py deleted file mode 100644 index ba2148e..0000000 --- a/hw/sim/tests/top/test_cam_basic.py +++ /dev/null @@ -1,664 +0,0 @@ -from __future__ import annotations - -import cocotb -import numpy as np -from cocotb.clock import Clock -from cocotb.triggers import RisingEdge -from model.ref_model import ( # noqa: E402 - generate_write_flip_mask, - match_top1, - match_top1_with_read_noise, - random_hashes, - unpack_score_debug_flat, -) - -DEFAULT_NUM_ROWS = 4096 -DEFAULT_HASH_BITS = 512 -DEFAULT_LANES = 8 -DEFAULT_SCORE_BITS = 10 - - -def _get_param(dut, name, default=None): - """Read a Verilator-exposed parameter from the DUT.""" - try: - val = getattr(dut, name, None) - if val is not None: - return int(val.value) - except Exception: - pass - return default - - -def dut_num_rows(dut): - val = _get_param(dut, "NUM_ROWS", None) - if val is not None: - return val - # Derive from wr_addr width (ROW_BITS): NUM_ROWS = 2^ROW_BITS - return 1 << len(dut.wr_addr) - - -def dut_hash_bits(dut): - val = _get_param(dut, "HASH_BITS", None) - if val is not None: - return val - # Derive from write_hash signal width - return len(dut.write_hash) - - -def dut_lanes(dut): - val = _get_param(dut, "LANES", None) - if val is not None: - return val - # Derive from rd_resp_row_ids width / ROW_BITS - return len(dut.rd_resp_row_ids) // len(dut.wr_addr) - - -def dut_score_bits(dut): - val = _get_param(dut, "SCORE_BITS", None) - if val is not None: - return val - # Derive from top1_score signal width - return len(dut.top1_score) - - -# ── Helpers ────────────────────────────────────────────────────────────────── - - -async def collect_topk(dut, timeout_cycles: int = 2000): - """Collect all serial Top-K result beats with result_ready held high. - - Returns list of (rank, row, score, last) tuples. - Raises AssertionError if the stream does not complete within timeout. - """ - dut.result_ready.value = 1 - beats = [] - for _ in range(timeout_cycles): - if int(dut.result_valid.value): - rank = int(dut.result_rank.value) - row = int(dut.result_row.value) - score = int(dut.result_score.value) - last = int(dut.result_last.value) - beats.append((rank, row, score, last)) - if last: - return beats - await RisingEdge(dut.clk) - raise AssertionError("Top-K result stream did not finish") - - -async def query_topk_once(dut, query, timeout_cycles=2000): - """Issue a query, collect full serial Top-K stream, and return beats + Top-1 metadata. - - Returns (beats, top1_index, top1_score, score_debug). - After this call the full result stream has been consumed and the DUT is idle. - """ - await wait_idle(dut) - - dut.query_hash.value = int(query) - dut.query_valid.value = 1 - - # Wait for handshake - while True: - await RisingEdge(dut.clk) - if int(dut.query_ready.value): - break - - dut.query_valid.value = 0 - - # Consume full serial result stream - beats = await collect_topk(dut, timeout_cycles=timeout_cycles) - - # score_debug is available after query completes - num_rows = dut_num_rows(dut) - score_bits = dut_score_bits(dut) - score_debug = None - if hasattr(dut, "score_debug_flat"): - score_debug = unpack_score_debug_flat( - int(dut.score_debug_flat.value), - num_rows, - score_bits, - ) - - return beats, beats[0][1], beats[0][2], score_debug - - -async def reset_dut(dut): - """Reset the DUT with new handshake interface.""" - dut.rst_n.value = 0 - dut.wr_valid.value = 0 - dut.wr_addr.value = 0 - dut.write_hash.value = 0 - dut.query_valid.value = 0 - dut.query_hash.value = 0 - dut.result_ready.value = 1 - - for _ in range(5): - await RisingEdge(dut.clk) - - dut.rst_n.value = 1 - - for _ in range(2): - await RisingEdge(dut.clk) - - -async def wait_idle(dut): - """Wait until both wr_ready=1 and query_ready=1 (system fully idle).""" - while not (int(dut.wr_ready.value) and int(dut.query_ready.value)): - await RisingEdge(dut.clk) - - -async def write_row(dut, addr, value): - """Write a single row using wr_valid/wr_ready handshake.""" - await wait_idle(dut) - - dut.wr_addr.value = addr - dut.write_hash.value = int(value) - dut.wr_valid.value = 1 - - # Wait for handshake - while True: - await RisingEdge(dut.clk) - if int(dut.wr_ready.value): - break - - dut.wr_valid.value = 0 - - # Wait for write pipeline to drain - await wait_idle(dut) - - -async def write_rows(dut, rows): - """Write all rows sequentially.""" - for idx, value in enumerate(rows): - await write_row(dut, idx, value) - - -async def query_once(dut, query): - """Issue a query and return (top1_index, top1_score, score_debug). - - Consumes the full serial Top-K stream and returns rank-0 data. - """ - _, top1_index, top1_score, score_debug = await query_topk_once(dut, query) - return top1_index, top1_score, score_debug - - -# ── Compile smoke test ──────────────────────────────────────────────────────── - - -@cocotb.test() -async def compile_includes_grouped_noise_helper(dut): - """Compilation test: new grouped noise helper must elaborate with cam_top.""" - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - assert int(dut.wr_ready.value) in (0, 1) - - -# ── Test A: Baseline (WRITE_NOISE_EN=0) ───────────────────────────────────── - - -@cocotb.test() -async def baseline_no_noise(dut): - """Verify write+query works exactly like the old CAM when noise disabled.""" - noise_en = _get_param(dut, "WRITE_NOISE_EN", 0) - read_noise_en = _get_param(dut, "READ_NOISE_EN", 0) - if noise_en or read_noise_en: - dut._log.info("Skipping baseline_no_noise: requires noise disabled.") - return - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - - num_rows = dut_num_rows(dut) - hash_bits = dut_hash_bits(dut) - rng = np.random.default_rng(1) - rows = random_hashes(rng, num_rows, width=hash_bits) - query_index = min(123, num_rows - 1) - query = rows[query_index] - - await write_rows(dut, rows) - beats, top1_index, top1_score, score_debug = await query_topk_once(dut, query) - - expected = match_top1(query, rows, width=hash_bits) - - assert top1_index == expected.top1_index - assert top1_score == expected.top1_score - assert top1_index == query_index - assert top1_score == hash_bits - - # Serial Top-K stream verification: beats from the single query above - assert beats[0][0] == 0, "First beat must have rank 0" - assert beats[-1][3] == 1, "Last beat must assert result_last" - # Verify top1 aliases match first beat after stream fully consumed - assert int(dut.top1_index.value) == beats[0][1] - assert int(dut.top1_score.value) == beats[0][2] - # Verify returned top1 matches first beat rank0 - assert top1_index == beats[0][1] - assert top1_score == beats[0][2] - - if score_debug is not None: - assert np.array_equal(score_debug, expected.scores) - - -# ── Test B: Zero noise rate (WRITE_NOISE_EN=1, RATE_NUM=0) ────────────────── - - -@cocotb.test() -async def zero_rate_noise(dut): - """Noise module connected but THRESHOLD=0 → no flips.""" - noise_en = _get_param(dut, "WRITE_NOISE_EN", 1) - rate_num = _get_param(dut, "WRITE_NOISE_RATE_NUM", 1) - read_noise_en = _get_param(dut, "READ_NOISE_EN", 0) - if not noise_en or rate_num != 0 or read_noise_en: - dut._log.info("Skipping zero_rate_noise: requires WRITE_NOISE_EN=1, RATE_NUM=0, READ_NOISE_EN=0.") - return - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - - num_rows = dut_num_rows(dut) - hash_bits = dut_hash_bits(dut) - rng = np.random.default_rng(1) - rows = random_hashes(rng, num_rows, width=hash_bits) - query_index = min(123, num_rows - 1) - query = rows[query_index] - - await write_rows(dut, rows) - top1_index, top1_score, score_debug = await query_once(dut, query) - - expected = match_top1(query, rows, width=hash_bits) - - assert top1_index == expected.top1_index - assert top1_score == expected.top1_score - assert top1_index == query_index - assert top1_score == hash_bits - - if score_debug is not None: - assert np.array_equal(score_debug, expected.scores) - - -# ── Test C: 100% noise rate (RATE_NUM=1, RATE_DEN=1) ─────────────────────── - - -@cocotb.test() -async def full_rate_noise(dut): - """WRITE_NOISE_RATE_NUM=1, WRITE_NOISE_RATE_DEN=1 → every group flips.""" - noise_en = _get_param(dut, "WRITE_NOISE_EN", 1) - rate_num = _get_param(dut, "WRITE_NOISE_RATE_NUM", 1) - rate_den = _get_param(dut, "WRITE_NOISE_RATE_DEN", 100) - if not noise_en or rate_num != 1 or rate_den != 1: - dut._log.info("Skipping full_rate_noise: requires WRITE_NOISE_EN=1, RATE_NUM=1, RATE_DEN=1.") - return - if not hasattr(dut, "score_debug_flat"): - dut._log.info("Skipping full_rate_noise: requires SIM_DEBUG (score_debug_flat).") - return - - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - - hash_bits = dut_hash_bits(dut) - num_rows = dut_num_rows(dut) - noise_bits = _get_param(dut, "WRITE_NOISE_BITS", 8) - all_zero = 0 - all_one = (1 << hash_bits) - 1 - - RTL_SEED = 0xB504_F32D_B504_F32D - prng_state = (RTL_SEED << 64) | RTL_SEED - - flip0, prng_state = generate_write_flip_mask( - prng_state, hash_bits, noise_bits, rate_num, rate_den, - ) - expected_row0 = all_zero ^ flip0 - - flip1, prng_state = generate_write_flip_mask( - prng_state, hash_bits, noise_bits, rate_num, rate_den, - ) - expected_row1 = all_one ^ flip1 - - rows = [0] * num_rows - rows[0] = all_zero - rows[1] = all_one - - await write_rows(dut, rows) - - top1_index, top1_score, score_debug = await query_once(dut, expected_row0) - assert score_debug is not None, "score_debug required for full_rate_noise" - assert int(score_debug[0]) == hash_bits, ( - f"Row 0: expected exact match, score={score_debug[0]} != {hash_bits}" - ) - - top1_index, top1_score, score_debug = await query_once(dut, expected_row1) - assert score_debug is not None - assert int(score_debug[1]) == hash_bits, ( - f"Row 1: expected exact match, score={score_debug[1]} != {hash_bits}" - ) - - -# ── Test D: Default ~1% noise, reproducible ──────────────────────────────── - - -@cocotb.test() -async def default_noise_reproducible(dut): - """Fixed seed → deterministic write noise. Two identical runs produce same results.""" - noise_en = _get_param(dut, "WRITE_NOISE_EN", 1) - if not noise_en: - dut._log.info("Skipping default_noise_reproducible: requires WRITE_NOISE_EN=1.") - return - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - - num_rows = dut_num_rows(dut) - hash_bits = dut_hash_bits(dut) - rng = np.random.default_rng(42) - rows = random_hashes(rng, num_rows, width=hash_bits) - - await write_rows(dut, rows) - query = rows[min(50, num_rows - 1)] - top1_index_1, top1_score_1, _ = await query_once(dut, query) - - await reset_dut(dut) - - await write_rows(dut, rows) - top1_index_2, top1_score_2, _ = await query_once(dut, query) - - assert top1_index_1 == top1_index_2 - assert top1_score_1 == top1_score_2 - - -# ── Preserved legacy tests (only meaningful for noise disabled) ────────────── - - -@cocotb.test() -async def known_hamming_distance(dut): - """Hamming distance verification — exact scores only valid without noise.""" - if _get_param(dut, "WRITE_NOISE_EN", 1) or _get_param(dut, "READ_NOISE_EN", 0): - dut._log.info("Skipping known_hamming_distance: requires noise disabled.") - return - - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - - num_rows = dut_num_rows(dut) - hash_bits = dut_hash_bits(dut) - - query = 0 - rows = [0] * num_rows - rows[min(10, num_rows - 1)] = (1 << 7) - 1 - rows[min(11, num_rows - 1)] = (1 << 31) - 1 - rows[min(12, num_rows - 1)] = (1 << 128) - 1 - - await write_rows(dut, rows) - top1_index, top1_score, score_debug = await query_once(dut, query) - - assert top1_index == 0 - assert top1_score == hash_bits - - if score_debug is not None: - assert int(score_debug[min(10, num_rows - 1)]) == hash_bits - 7 - assert int(score_debug[min(11, num_rows - 1)]) == hash_bits - 31 - assert int(score_debug[min(12, num_rows - 1)]) == hash_bits - 128 - - -@cocotb.test() -async def tie_break_policy(dut): - """Tie-break: lowest row index wins — only verified without noise.""" - if _get_param(dut, "WRITE_NOISE_EN", 1) or _get_param(dut, "READ_NOISE_EN", 0): - dut._log.info("Skipping tie_break_policy: requires noise disabled.") - return - - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - - num_rows = dut_num_rows(dut) - hash_bits = dut_hash_bits(dut) - rng = np.random.default_rng(2) - rows = random_hashes(rng, num_rows, width=hash_bits) - query = rows[min(200, num_rows - 1)] - rows[10] = query - rows[20] = query - rows[min(200, num_rows - 1)] = query - - await write_rows(dut, rows) - top1_index, top1_score, _ = await query_once(dut, query) - - assert top1_index == 10 - assert top1_score == hash_bits - - -@cocotb.test() -async def all_zero_all_one_boundary(dut): - """All-zero / all-one boundary — only verified without noise.""" - if _get_param(dut, "WRITE_NOISE_EN", 1) or _get_param(dut, "READ_NOISE_EN", 0): - dut._log.info("Skipping all_zero_all_one_boundary: requires noise disabled.") - return - - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - - num_rows = dut_num_rows(dut) - hash_bits = dut_hash_bits(dut) - - rows = [0] * num_rows - rows[0] = 0 - rows[1] = (1 << hash_bits) - 1 - - query = 0 - await write_rows(dut, rows) - top1_index, top1_score, score_debug = await query_once(dut, query) - - assert top1_score == hash_bits - assert top1_index == 0 - - if score_debug is not None: - assert int(score_debug[0]) == hash_bits - assert int(score_debug[1]) == 0 - - -# ── Test E: Exact RTL-vs-model PRNG mask match ────────────────────────────── - - -@cocotb.test() -async def exact_noise_model_match(dut): - """Verify RTL stored hashes match ref_model.py for a known seed and rate.""" - noise_en = _get_param(dut, "WRITE_NOISE_EN", 1) - rate_num = _get_param(dut, "WRITE_NOISE_RATE_NUM", 1) - rate_den = _get_param(dut, "WRITE_NOISE_RATE_DEN", 100) - if not noise_en or rate_num == 0: - dut._log.info("Skipping exact_noise_model_match: requires WRITE_NOISE_EN=1, RATE_NUM>0.") - return - if _get_param(dut, "READ_NOISE_EN", 0): - dut._log.info("Skipping exact_noise_model_match: requires READ_NOISE_EN=0 (read noise corrupts score comparison).") - return - if not hasattr(dut, "score_debug_flat"): - dut._log.info("Skipping exact_noise_model_match: requires SIM_DEBUG.") - return - - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - - hash_bits = dut_hash_bits(dut) - noise_bits = _get_param(dut, "WRITE_NOISE_BITS", 8) - - n_test_rows = 4 - rng = np.random.default_rng(99) - rows = random_hashes(rng, n_test_rows, width=hash_bits) - - RTL_SEED = 0xB504_F32D_B504_F32D - prng_state = (RTL_SEED << 64) | RTL_SEED - expected_stored = [] - for row in rows: - flip, prng_state = generate_write_flip_mask( - prng_state, hash_bits, noise_bits, rate_num, rate_den, - ) - expected_stored.append(row ^ flip) - - for idx, val in enumerate(rows): - await write_row(dut, idx, val) - - for idx, expected in enumerate(expected_stored): - top1_index, top1_score, score_debug = await query_once(dut, expected) - assert score_debug is not None, "score_debug required for mask match verification" - assert int(score_debug[idx]) == hash_bits, ( - f"Row {idx}: expected stored hash to match model prediction, " - f"score={score_debug[idx]} != {hash_bits}" - ) - - -# ── Test F: Half-duplex write-priority arbitration ─────────────────────────── - - -@cocotb.test() -async def half_duplex_write_priority(dut): - """When wr_valid and query_valid are both high, write wins and query is held off.""" - if _get_param(dut, "WRITE_NOISE_EN", 1) or _get_param(dut, "READ_NOISE_EN", 0): - dut._log.info("Skipping half_duplex_write_priority: requires noise disabled.") - return - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - - hash_bits = dut_hash_bits(dut) - test_val = (1 << hash_bits) - 1 - await write_row(dut, 0, test_val) - - await wait_idle(dut) - assert int(dut.wr_ready.value) == 1 - assert int(dut.query_ready.value) == 1 - - dut.wr_valid.value = 1 - dut.wr_addr.value = 1 - dut.write_hash.value = 0 - dut.query_valid.value = 1 - dut.query_hash.value = test_val - - await RisingEdge(dut.clk) - - dut.wr_valid.value = 0 - dut.query_valid.value = 0 - - await wait_idle(dut) - - top1_index, top1_score, _ = await query_once(dut, test_val) - assert top1_index == 0 - assert top1_score == hash_bits - - top1_index, top1_score, _ = await query_once(dut, 0) - assert top1_index == 1 - assert top1_score == hash_bits - - -# ── Test G: Banked pipeline no-noise Top-1 ─────────────────────────────────── - - -@cocotb.test() -async def banked_pipeline_no_noise_top1(dut): - """No-noise banked pipeline returns the same Top-1 as the pure model.""" - if _get_param(dut, "WRITE_NOISE_EN", 0) or _get_param(dut, "READ_NOISE_EN", 0): - dut._log.info("Skipping banked_pipeline_no_noise_top1: requires noise disabled.") - return - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - - num_rows = dut_num_rows(dut) - hash_bits = dut_hash_bits(dut) - rng = np.random.default_rng(7) - rows = random_hashes(rng, num_rows, width=hash_bits) - query_index = min(17, num_rows - 1) - query = rows[query_index] - - await write_rows(dut, rows) - top1_index, top1_score, score_debug = await query_once(dut, query) - expected = match_top1(query, rows, width=hash_bits) - - assert top1_index == expected.top1_index - assert top1_score == expected.top1_score - assert top1_index == query_index - - -# ── Test H: Query scan blocks writes until result consumed ─────────────────── - - -@cocotb.test() -async def query_scan_blocks_writes_until_result_consumed(dut): - """Half-duplex: active query scan deasserts wr_ready.""" - if _get_param(dut, "WRITE_NOISE_EN", 0) or _get_param(dut, "READ_NOISE_EN", 0): - dut._log.info("Skipping query_scan_blocks_writes: requires noise disabled.") - return - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - - num_rows = dut_num_rows(dut) - hash_bits = dut_hash_bits(dut) - rows = [0] * num_rows - rows[0] = (1 << hash_bits) - 1 - await write_rows(dut, rows) - - await wait_idle(dut) - dut.query_hash.value = rows[0] - dut.query_valid.value = 1 - await RisingEdge(dut.clk) - dut.query_valid.value = 0 - - dut.wr_valid.value = 1 - dut.wr_addr.value = 1 - dut.write_hash.value = 0 - await RisingEdge(dut.clk) - assert int(dut.wr_ready.value) == 0 - dut.wr_valid.value = 0 - - # Consume full serial stream so the DUT returns idle - beats = await collect_topk(dut, timeout_cycles=2000) - assert len(beats) > 0 - assert beats[-1][3] == 1 # last asserted - - -# ── Test I: Read noise model match ────────────────────────────────────────── - - -@cocotb.test() -async def read_noise_model_match(dut): - """Read noise uses grouped masks and matches the Python model for one query.""" - if not _get_param(dut, "READ_NOISE_EN", 0): - dut._log.info("Skipping read_noise_model_match: requires READ_NOISE_EN=1.") - return - - cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) - await reset_dut(dut) - - num_rows = dut_num_rows(dut) - hash_bits = dut_hash_bits(dut) - lanes = dut_lanes(dut) - rng = np.random.default_rng(123) - rows = random_hashes(rng, num_rows, width=hash_bits) - - # If write noise is enabled, apply write flip masks to predict stored rows - stored_rows = list(rows) - if _get_param(dut, "WRITE_NOISE_EN", 0): - seed = 0xB504_F32D_B504_F32D - prng_state = (seed << 64) | seed - stored_rows = [] - for row in rows: - flip, prng_state = generate_write_flip_mask( - prng_state, - hash_bits, - _get_param(dut, "WRITE_NOISE_BITS", 8), - _get_param(dut, "WRITE_NOISE_RATE_NUM", 1), - _get_param(dut, "WRITE_NOISE_RATE_DEN", 100), - ) - stored_rows.append(row ^ flip) - - query = rows[min(5, num_rows - 1)] - - await write_rows(dut, rows) - top1_index, top1_score, score_debug = await query_once(dut, query) - - expected = match_top1_with_read_noise( - query, - stored_rows, - width=hash_bits, - lanes=lanes, - noise_bits=_get_param(dut, "READ_NOISE_BITS", 8), - rate_num=_get_param(dut, "READ_NOISE_RATE_NUM", 1), - rate_den=_get_param(dut, "READ_NOISE_RATE_DEN", 100), - seed=0x6A09_E667_F3BC_C909, - ) - - assert top1_index == expected.top1_index - assert top1_score == expected.top1_score - if score_debug is not None: - assert np.array_equal(score_debug, expected.scores) diff --git a/hw/sim/tests/top/utils.py b/hw/sim/tests/top/utils.py new file mode 100644 index 0000000..2ad95a2 --- /dev/null +++ b/hw/sim/tests/top/utils.py @@ -0,0 +1,256 @@ +# -*- coding: utf-8 -*- +""" +CAM 顶层测试的共享辅助函数。 + +被 no_noise/、write_noise/、read_noise/ 等配置目录的测试文件共同引用。 +提供： + - Verilator 参数读取（带回退推断） + - DUT 复位、空闲等待 + - 行写入 / 批量写入（握手协议） + - 查询发起 / Top-K 结果流收集 + - score_debug 解包（SIM_DEBUG 模式） + +所有函数都是 async，调用者需要处于 cocotb 协程上下文中。 +""" +from __future__ import annotations + +import numpy as np +from cocotb.triggers import RisingEdge +from model.ref_model import ( # noqa: E402 + match_top1, + unpack_score_debug_flat, +) + +# ── 默认拓扑参数（当 Verilator 参数不可用时使用） ─────────────────────────── +DEFAULT_NUM_ROWS = 4096 +DEFAULT_HASH_BITS = 512 +DEFAULT_LANES = 8 +DEFAULT_SCORE_BITS = 10 + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 参数读取（从 Verilator 参数或信号宽度推断） +# ═══════════════════════════════════════════════════════════════════════════════ + + +def get_param(dut, name, default=None): + """从 DUT 读取 Verilator 暴露的参数值，失败则返回 default。 + + 某些参数仅在特定编译配置下暴露（如 SIM_DEBUG 下的 score_debug_flat）， + 此时回退到 default 是预期行为而非错误。 + """ + try: + val = getattr(dut, name, None) + if val is not None: + return int(val.value) + except Exception: + pass + return default + + +def dut_num_rows(dut): + """获取 NUM_ROWS：优先读参数，否则从 wr_addr 位宽推断 (NUM_ROWS = 2^ROW_BITS)。""" + val = get_param(dut, "NUM_ROWS", None) + if val is not None: + return val + return 1 << len(dut.wr_addr) + + +def dut_hash_bits(dut): + """获取 HASH_BITS：优先读参数，否则从 write_hash 信号位宽推断。""" + val = get_param(dut, "HASH_BITS", None) + if val is not None: + return val + return len(dut.write_hash) + + +def dut_lanes(dut): + """获取 LANES：优先读参数，否则从 rd_resp_row_ids / wr_addr 位宽推断。""" + val = get_param(dut, "LANES", None) + if val is not None: + return val + return len(dut.rd_resp_row_ids) // len(dut.wr_addr) + + +def dut_score_bits(dut): + """获取 SCORE_BITS：优先读参数，否则从 top1_score 信号位宽推断。""" + val = get_param(dut, "SCORE_BITS", None) + if val is not None: + return val + return len(dut.top1_score) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 协议层辅助函数（复位 / 空闲等待 / 行写入 / 查询） +# ═══════════════════════════════════════════════════════════════════════════════ + + +async def reset_dut(dut): + """复位 DUT：rst_n 拉低 5 周期，释放后再等 2 周期。 + + 所有控制信号在复位期间保持无效电平。 + result_ready 初始化为 1（准备接收结果）。 + """ + dut.rst_n.value = 0 + dut.wr_valid.value = 0 + dut.wr_addr.value = 0 + dut.write_hash.value = 0 + dut.query_valid.value = 0 + dut.query_hash.value = 0 + dut.result_ready.value = 1 + + for _ in range(5): + await RisingEdge(dut.clk) + + dut.rst_n.value = 1 + + for _ in range(2): + await RisingEdge(dut.clk) + + +async def wait_idle(dut): + """等待 DUT 完全空闲：wr_ready=1 且 query_ready=1。 + + 这是发起新操作前的前置条件——CAM 是半双工的， + 同一时刻只能进行写入或查询。 + """ + while not (int(dut.wr_ready.value) and int(dut.query_ready.value)): + await RisingEdge(dut.clk) + + +async def write_row(dut, addr, value): + """写入单行：使用 wr_valid/wr_ready 握手协议。 + + 流程：等待空闲 → 驱动地址和数据 → 等待握手完成 → 等待写流水线排空。 + """ + await wait_idle(dut) + + dut.wr_addr.value = addr + dut.write_hash.value = int(value) + dut.wr_valid.value = 1 + + while True: + await RisingEdge(dut.clk) + if int(dut.wr_ready.value): + break + + dut.wr_valid.value = 0 + await wait_idle(dut) + + +async def write_rows(dut, rows): + """按顺序写入所有行（行索引 = 数组下标）。""" + for idx, value in enumerate(rows): + await write_row(dut, idx, value) + + +async def collect_topk(dut, timeout_cycles: int = 2000): + """收集串行 Top-K 结果流的所有 beat。 + + 保持 result_ready=1，逐个时钟周期采样 result_valid， + 直到 result_last 被断言。 + + 返回：[(rank, row, score, last), ...] 列表 + 超时则抛出 AssertionError。 + + 注意：此函数会「消耗」整个结果流，调用后 DUT 回到空闲状态。 + """ + dut.result_ready.value = 1 + beats = [] + for _ in range(timeout_cycles): + if int(dut.result_valid.value): + rank = int(dut.result_rank.value) + row = int(dut.result_row.value) + score = int(dut.result_score.value) + last = int(dut.result_last.value) + beats.append((rank, row, score, last)) + if last: + return beats + await RisingEdge(dut.clk) + raise AssertionError("Top-K result stream did not finish") + + +async def query_topk_once(dut, query, timeout_cycles=2000): + """发起一次查询并收集完整的串行 Top-K 结果流。 + + 完整流程： + 1. 等待 DUT 空闲 + 2. 通过 query_valid/query_ready 握手发送查询 + 3. 消费完整的结果流 + 4. 读取 score_debug_flat（如果存在） + + 返回：(beats, top1_index, top1_score, score_debug) + - beats: [(rank, row, score, last), ...] + - score_debug: np.ndarray 或 None（SIM_DEBUG 模式） + """ + await wait_idle(dut) + + dut.query_hash.value = int(query) + dut.query_valid.value = 1 + + # 等待查询握手完成 + while True: + await RisingEdge(dut.clk) + if int(dut.query_ready.value): + break + + dut.query_valid.value = 0 + + # 消费完整串行结果流 + beats = await collect_topk(dut, timeout_cycles=timeout_cycles) + + # score_debug 在查询完成后可用（需 SIM_DEBUG 编译） + num_rows = dut_num_rows(dut) + score_bits = dut_score_bits(dut) + score_debug = None + if hasattr(dut, "score_debug_flat"): + score_debug = unpack_score_debug_flat( + int(dut.score_debug_flat.value), + num_rows, + score_bits, + ) + + return beats, beats[0][1], beats[0][2], score_debug + + +async def query_once(dut, query): + """发起查询，返回 (top1_index, top1_score, score_debug)。 + + 内部调用 query_topk_once 并消费完整结果流，仅保留 rank-0 数据。 + """ + _, top1_index, top1_score, score_debug = await query_topk_once(dut, query) + return top1_index, top1_score, score_debug + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 便捷验证函数 +# ═══════════════════════════════════════════════════════════════════════════════ + + +def assert_baseline_top1(query, rows, top1_index, top1_score, hash_bits, + query_index, score_debug=None): + """验证无噪声场景下的基线 Top-1 结果。 + + 检查项： + 1. Top-1 与 match_top1 参考模型一致 + 2. 查询自身所在行 → score == hash_bits（完全匹配） + 3. score_debug 数组（如果存在）与模型一致 + """ + expected = match_top1(query, rows, width=hash_bits) + assert top1_index == expected.top1_index, ( + f"top1_index mismatch: {top1_index} != {expected.top1_index}" + ) + assert top1_score == expected.top1_score, ( + f"top1_score mismatch: {top1_score} != {expected.top1_score}" + ) + assert top1_index == query_index, ( + f"Expected query_index={query_index} to match self, got top1_index={top1_index}" + ) + assert top1_score == hash_bits, ( + f"Self-query should score {hash_bits}, got {top1_score}" + ) + + if score_debug is not None: + assert np.array_equal(score_debug, expected.scores), ( + "score_debug does not match model scores" + ) diff --git a/hw/sim/tests/top/write_noise/Makefile b/hw/sim/tests/top/write_noise/Makefile new file mode 100644 index 0000000..a23b387 --- /dev/null +++ b/hw/sim/tests/top/write_noise/Makefile @@ -0,0 +1,29 @@ +SIM_ROOT := $(abspath ../../..) +RTL_ROOT := $(abspath $(SIM_ROOT)/../rtl) +include $(SIM_ROOT)/mk/rtl-sources.mk + +TOPLEVEL := cam_top +COCOTB_TEST_MODULES := tests.top.write_noise.test_write_noise +VERILOG_SOURCES := $(RTL_CAM_TOP) + +# 写入噪声 ~1% 默认速率 +WRITE_NOISE_EN := 1 +WRITE_NOISE_RATE_NUM := 1 +WRITE_NOISE_RATE_DEN := 100 +READ_NOISE_EN := 0 + +include $(SIM_ROOT)/mk/cocotb-common.mk + +# ── 速率变体子目标 ───────────────────────────────────────────────── +.PHONY: test-zero-rate test-full-rate + +test-zero-rate: + $(MAKE) -B -f Makefile results.xml WRITE_NOISE_RATE_NUM=0 \ + COCOTB_TEST_FILTER=zero_rate_noise + +test-full-rate: + $(MAKE) -B -f Makefile results.xml WRITE_NOISE_RATE_NUM=1 WRITE_NOISE_RATE_DEN=1 \ + COCOTB_TEST_FILTER=full_rate_noise + +clean:: + rm -rf sim_build diff --git a/hw/sim/tests/top/write_noise/__init__.py b/hw/sim/tests/top/write_noise/__init__.py new file mode 100644 index 0000000..c25e921 --- /dev/null +++ b/hw/sim/tests/top/write_noise/__init__.py @@ -0,0 +1 @@ +# CAM top-level write-noise integration tests diff --git a/hw/sim/tests/top/write_noise/test_write_noise.py b/hw/sim/tests/top/write_noise/test_write_noise.py new file mode 100644 index 0000000..4e78b30 --- /dev/null +++ b/hw/sim/tests/top/write_noise/test_write_noise.py @@ -0,0 +1,274 @@ +# -*- coding: utf-8 -*- +""" +CAM 写入噪声（Write Noise）集成测试 —— 专用配置。 + +本文件测试 WRITE_NOISE_EN=1, READ_NOISE_EN=0 配置下， +写入噪声模块的正确性。默认噪声率约 1%（NUM=1, DEN=100）。 + +=== 测试列表 === + + 1. default_noise_reproducible — 固定种子 = 确定性噪声，两次运行结果一致 + 2. exact_noise_model_match — RTL 存储的哈希与 ref_model.py 的 PRNG 掩码逐位匹配 + 3. zero_rate_noise — 写入噪声模块连接但 RATE_NUM=0 → 无翻转 + 4. full_rate_noise — 100% 写入噪声率，与 Python 模型对比 + +=== 架构背景 === + + 写入噪声流水线位置：Write Noise → Banked Core Storage → Read Noise → Match Engine + 本测试覆盖完整的 cam_top 链路，写入噪声为唯一活跃噪声源。 + +=== Makefile 子目标 === + + test-zero-rate : make test-zero-rate (WRITE_NOISE_RATE_NUM=0) + test-full-rate : make test-full-rate (WRITE_NOISE_RATE_NUM=1, RATE_DEN=1, SIM_DEBUG) +""" + +from __future__ import annotations + +import cocotb +import numpy as np +from cocotb.clock import Clock +from cocotb.triggers import RisingEdge +from model.ref_model import ( + generate_write_flip_mask, + match_top1, + random_hashes, +) +from tests.top.utils import ( + collect_topk, + dut_hash_bits, + dut_num_rows, + get_param, + query_once, + query_topk_once, + reset_dut, + wait_idle, + write_row, + write_rows, +) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 测试 1：默认噪声 ~1%、可复现 +# ── 固定种子 → 两次相同写入产生相同结果 +# ═══════════════════════════════════════════════════════════════════════════════ + + +@cocotb.test() +async def default_noise_reproducible(dut): + """可复现性测试：相同种子、相同数据 → 两次独立运行结果一致。 + + 流程： + 1. 写入全部行，查询 row 50 → 记录 top1_index 和 top1_score + 2. 复位 + 3. 再次写入相同数据，查询相同行 → 记录结果 + 4. 断言两次结果完全一致 + + 如果结果不一致，说明 RTL 的 PRNG 状态没有正确复位， + 或存在跨运行的状态残留。 + """ + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + + num_rows = dut_num_rows(dut) + hash_bits = dut_hash_bits(dut) + rng = np.random.default_rng(42) + rows = random_hashes(rng, num_rows, width=hash_bits) + + await write_rows(dut, rows) + query = rows[min(50, num_rows - 1)] + top1_index_1, top1_score_1, _ = await query_once(dut, query) + + await reset_dut(dut) + + await write_rows(dut, rows) + top1_index_2, top1_score_2, _ = await query_once(dut, query) + + assert top1_index_1 == top1_index_2 + assert top1_score_1 == top1_score_2 + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 测试 2：精确 RTL-vs-模型 PRNG 掩码匹配 +# ── RTL 存储的哈希与 ref_model.py 生成的掩码逐位一致 +# ═══════════════════════════════════════════════════════════════════════════════ + + +@cocotb.test() +async def exact_noise_model_match(dut): + """精确噪声模型匹配：RTL 的 PRNG 输出必须与 Python 参考模型逐位一致。 + + 测试方法： + 1. 用固定 RTL seed 和已知噪声参数，在 Python 中预计算每行的 flip 掩码 + 2. 预期存储值 = 原始值 XOR flip_mask + 3. 写入原始值到 RTL，查询预期存储值 + 4. 断言每行 score = HASH_BITS（完全匹配） + + 这验证了 RTL 的 LFSR 实现与 Python 模型的 PRNG 使用相同的 + 多项式、相同的位宽、相同的种子初始化序列。 + """ + if not hasattr(dut, "score_debug_flat"): + dut._log.info("Skipping exact_noise_model_match: requires SIM_DEBUG.") + return + + rtol = None + atol = None + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + + hash_bits = dut_hash_bits(dut) + noise_bits = get_param(dut, "WRITE_NOISE_BITS", 8) + rate_num = get_param(dut, "WRITE_NOISE_RATE_NUM", 1) + rate_den = get_param(dut, "WRITE_NOISE_RATE_DEN", 100) + + n_test_rows = 4 + rng = np.random.default_rng(99) + rows = random_hashes(rng, n_test_rows, width=hash_bits) + + RTL_SEED = 0xB504_F32D_B504_F32D + prng_state = (RTL_SEED << 64) | RTL_SEED + expected_stored = [] + for row in rows: + flip, prng_state = generate_write_flip_mask( + prng_state, + hash_bits, + noise_bits, + rate_num, + rate_den, + ) + expected_stored.append(row ^ flip) + + for idx, val in enumerate(rows): + await write_row(dut, idx, val) + + for idx, expected in enumerate(expected_stored): + top1_index, top1_score, score_debug = await query_once(dut, expected) + assert score_debug is not None, ( + "score_debug required for mask match verification" + ) + assert int(score_debug[idx]) == hash_bits, ( + f"Row {idx}: expected stored hash to match model prediction, " + f"score={score_debug[idx]} != {hash_bits}" + ) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 测试 3：零噪声率（WRITE_NOISE_EN=1, RATE_NUM=0） +# ── 噪声模块已连接但翻转概率为 0 → 行为应与无噪声一致 +# ═══════════════════════════════════════════════════════════════════════════════ + + +@cocotb.test() +async def zero_rate_noise(dut): + """零速率噪声：WRITE_NOISE_RATE_NUM=0 时不应有任何位被翻转。 + + 这是噪声模块的边界测试——验证 RATE_NUM=0 确实禁用了翻转， + 而非产生「默认速率」的噪声。 + """ + rate_num = get_param(dut, "WRITE_NOISE_RATE_NUM", 1) + if rate_num != 0: + dut._log.info( + "Skipping zero_rate_noise: requires WRITE_NOISE_RATE_NUM=0." + ) + return + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + + num_rows = dut_num_rows(dut) + hash_bits = dut_hash_bits(dut) + rng = np.random.default_rng(1) + rows = random_hashes(rng, num_rows, width=hash_bits) + query_index = min(123, num_rows - 1) + query = rows[query_index] + + await write_rows(dut, rows) + top1_index, top1_score, score_debug = await query_once(dut, query) + + expected = match_top1(query, rows, width=hash_bits) + + assert top1_index == expected.top1_index + assert top1_score == expected.top1_score + assert top1_index == query_index + assert top1_score == hash_bits + + if score_debug is not None: + assert np.array_equal(score_debug, expected.scores) + + +# ═══════════════════════════════════════════════════════════════════════════════ +# 测试 4：100% 噪声率（RATE_NUM=1, RATE_DEN=1） +# ── 每组都翻转 → 精确验证 PRNG 掩码生成 +# ═══════════════════════════════════════════════════════════════════════════════ + + +@cocotb.test() +async def full_rate_noise(dut): + """完全速率噪声：每组 100% 翻转概率。 + + 使用固定 RTL seed (0xB504F32DB504F32D)，用 Python 模型预计算 + 写入全 0 和全 1 行后应存储的哈希值，然后验证 RTL 实际存储的哈希 + 与模型预测完全一致。 + + 这是最低容忍度的噪声测试——要求 score_debug_flat（SIM_DEBUG） + 且每行的分数必须精确等于 HASH_BITS。 + """ + rate_num = get_param(dut, "WRITE_NOISE_RATE_NUM", 1) + rate_den = get_param(dut, "WRITE_NOISE_RATE_DEN", 100) + if rate_num != 1 or rate_den != 1: + dut._log.info( + "Skipping full_rate_noise: requires WRITE_NOISE_RATE_NUM=1, RATE_DEN=1." + ) + return + if not hasattr(dut, "score_debug_flat"): + dut._log.info( + "Skipping full_rate_noise: requires SIM_DEBUG (score_debug_flat)." + ) + return + + cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start()) + await reset_dut(dut) + + hash_bits = dut_hash_bits(dut) + num_rows = dut_num_rows(dut) + noise_bits = get_param(dut, "WRITE_NOISE_BITS", 8) + all_zero = 0 + all_one = (1 << hash_bits) - 1 + + RTL_SEED = 0xB504_F32D_B504_F32D + prng_state = (RTL_SEED << 64) | RTL_SEED + + flip0, prng_state = generate_write_flip_mask( + prng_state, + hash_bits, + noise_bits, + rate_num, + rate_den, + ) + expected_row0 = all_zero ^ flip0 + + flip1, prng_state = generate_write_flip_mask( + prng_state, + hash_bits, + noise_bits, + rate_num, + rate_den, + ) + expected_row1 = all_one ^ flip1 + + rows = [0] * num_rows + rows[0] = all_zero + rows[1] = all_one + + await write_rows(dut, rows) + + top1_index, top1_score, score_debug = await query_once(dut, expected_row0) + assert score_debug is not None, "score_debug required for full_rate_noise" + assert int(score_debug[0]) == hash_bits, ( + f"Row 0: expected exact match, score={score_debug[0]} != {hash_bits}" + ) + + top1_index, top1_score, score_debug = await query_once(dut, expected_row1) + assert score_debug is not None + assert int(score_debug[1]) == hash_bits, ( + f"Row 1: expected exact match, score={score_debug[1]} != {hash_bits}" + ) diff --git a/scripts/run_cam_correctness.py b/scripts/run_cam_correctness.py index 1ef913c..4e12a8b 100644 --- a/scripts/run_cam_correctness.py +++ b/scripts/run_cam_correctness.py @@ -64,7 +64,7 @@ default_scenarios: list[Scenario] = [ Scenario( name="cam_top_basic", kind="cocotb", - module="tests.test_cam_basic", + module="tests.top.no_noise.test_no_noise", testcase=None, params={}, noise_mode="none", @@ -73,28 +73,18 @@ default_scenarios: list[Scenario] = [ Scenario( name="cam_top_write_noise", kind="cocotb", - module="tests.test_cam_basic", + module="tests.top.write_noise.test_write_noise", testcase=None, - params={ - "WRITE_NOISE_EN": "1", - "WRITE_NOISE_RATE_NUM": "1", - "WRITE_NOISE_RATE_DEN": "100", - "READ_NOISE_EN": "0", - }, + params={}, noise_mode="write_noise", expected="bit_flip_injected", ), Scenario( name="cam_top_read_noise", kind="cocotb", - module="tests.test_cam_basic", + module="tests.top.read_noise.test_read_noise", testcase=None, - params={ - "WRITE_NOISE_EN": "0", - "READ_NOISE_EN": "1", - "READ_NOISE_RATE_NUM": "1", - "READ_NOISE_RATE_DEN": "100", - }, + params={}, noise_mode="read_noise", expected="read_perturbed", ), @@ -150,19 +140,28 @@ default_scenarios: list[Scenario] = [ # Command builders # --------------------------------------------------------------------------- -# Internal: map unit scenario names to their Verilog TOPLEVEL module. -# cam_top_* scenarios use the Makefile default (cam_top) and are omitted. -_UNIT_TOPLEVEL: dict[str, str] = { - "cam_core_banked": "cam_core_banked", - "match_engine_pipeline": "match_engine_pipeline", - "cam_write_noise": "cam_write_noise", - "cam_read_noise": "cam_read_noise", +# Map scenario names to their Makefile directory under hw/sim/tests/. +# Top-level cam_top tests are now split by noise configuration. +_MAKE_DIR: dict[str, str] = { + "cam_top_basic": "top/no_noise", + "cam_top_write_noise": "top/write_noise", + "cam_top_read_noise": "top/read_noise", + "cam_core_banked": "modules/cam_core_banked", + "match_engine_pipeline": "modules/match_engine_pipeline", + "cam_write_noise": "modules/cam_write_noise", + "cam_read_noise": "modules/cam_read_noise", } -def _infer_toplevel(scenario: Scenario) -> str | None: - """Return the Verilog TOPLEVEL for *scenario* or None to use Makefile default.""" - return _UNIT_TOPLEVEL.get(scenario.name) +def _make_dir_for(scenario: Scenario) -> str: + """Return the relative Makefile directory under hw/sim/tests/ for *scenario*.""" + if scenario.name in _MAKE_DIR: + return _MAKE_DIR[scenario.name] + # Fallback: guess from module path (e.g. tests.top.foo → top/foo) + parts = scenario.module.split(".") if scenario.module else [] + if len(parts) >= 4 and parts[0] == "tests": + return "/".join(parts[1:-1]) # tests.top.no_noise → top/no_noise + return "." def build_scenario_command( @@ -178,15 +177,16 @@ def build_scenario_command( "uv", "run", "pytest", - "hw/sim/tests/test_ref_model_noise.py", + "hw/sim/tests/model/test_ref_model_noise.py", "-q", ] - # cocotb + # cocotb — run make in the specific test subdirectory + make_dir = f"hw/sim/tests/{_make_dir_for(scenario)}" cmd = [ "make", "-C", - "hw/sim", + make_dir, f"NUM_ROWS={num_rows}", f"HASH_BITS={hash_bits}", f"LANES={lanes}", @@ -196,9 +196,6 @@ def build_scenario_command( cmd.append(f"COCOTB_TESTCASE={scenario.testcase}") for key, value in scenario.params.items(): cmd.append(f"{key}={value}") - toplevel = _infer_toplevel(scenario) - if toplevel is not None: - cmd.append(f"TOPLEVEL={toplevel}") return cmd @@ -206,7 +203,8 @@ def build_clean_command(scenario: Scenario) -> list[str] | None: """Build a clean command or None if cleanup is not needed.""" if scenario.kind == "pytest": return None - return ["make", "-C", "hw/sim", "clean"] + make_dir = f"hw/sim/tests/{_make_dir_for(scenario)}" + return ["make", "-C", make_dir, "clean"] # ---------------------------------------------------------------------------