SikongJueluo/Mini-Nav
1
0
Fork 0
mirror of https://github.com/SikongJueluo/Mini-Nav.git synced 2026-07-12 20:15:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(hw): add banked CAM pipeline with grouped read/write noise

Browse Source 
- Add cam_core_banked.sv with 8-lane banked CAM core
- Add cam_write_noise.sv and cam_read_noise.sv for grouped noise injection
- Add noise_mask_grouped.sv generating grouped flip masks from 128-bit PRNG
- Add match_engine_pipeline.sv with multi-stage pipelined top-1 selection
- Add popcount_pipeline.sv for pipelined popcount operations
- Refactor test_cam_basic.py with parametrized DUT introspection helpers
- Add Python ref_model match_top1_with_read_noise() for read noise verification
- Update Makefile with separate WRITE_NOISE_* and READ_NOISE_* parameter groups
- Add new testbenches: test_cam_core_banked, test_cam_read_noise,
  test_cam_write_noise, test_match_engine_pipeline, test_ref_model_noise
  
breaking change hint: NUM_ROWS default changed from 512→4096, LANES from 16→8
This commit is contained in:
SikongJueluo
2026-05-13 16:21:27 +08:00
parent c41e64d1c6
commit 8f59a287c4
17 changed files with 1331 additions and 384 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
hw/sim/Makefile
View File

@@ -1,22 +1,12 @@
SIM ?= verilator
TOPLEVEL_LANG ?= verilog
TOPLEVEL := cam_top
TOPLEVEL ?= cam_top
# MODULE ?= tests.test_cam_basic
COCOTB_TEST_MODULES ?= tests.test_cam_basic
NUM_ROWS ?= 512
NUM_ROWS ?= 4096
HASH_BITS ?= 512
LANES ?= 16
# Noise parameters
NOISE_EN ?= 1
NOISE_RATE_NUM ?= 1
NOISE_RATE_DEN ?= 100
NOISE_BITS ?= 8
# NOISE_SEED cannot be overridden via Makefile due to Verilator -G quoting issues
# with 64'h hex literals. To change the seed, edit the default in cam_top.sv
# (sim top is cam_top, which passes it to cam_noisy).
LANES ?= 8
EXTRA_ARGS += +define+NUM_ROWS=$(NUM_ROWS) +define+HASH_BITS=$(HASH_BITS) +define+LANES=$(LANES)
@@ -25,11 +15,42 @@ COMPILE_ARGS += -I$(PWD)/../rtl
COMPILE_ARGS += +define+SIM_DEBUG
COMPILE_ARGS += $(EXTRA_DEFINES)
# Noise parameter overrides via Verilator -G
COMPILE_ARGS += -GNOISE_EN=$(NOISE_EN)
COMPILE_ARGS += -GNOISE_RATE_NUM=$(NOISE_RATE_NUM)
COMPILE_ARGS += -GNOISE_RATE_DEN=$(NOISE_RATE_DEN)
COMPILE_ARGS += -GNOISE_BITS=$(NOISE_BITS)
# Noise parameters — only passed when explicitly set (non-empty).
# For cam_noisy/cam_top tests, pass WRITE_NOISE_*=... READ_NOISE_*=...
# For individual module tests, leave them unset to skip.
WRITE_NOISE_EN ?= $(NOISE_EN)
WRITE_NOISE_RATE_NUM ?= $(NOISE_RATE_NUM)
WRITE_NOISE_RATE_DEN ?= $(NOISE_RATE_DEN)
WRITE_NOISE_BITS ?= $(NOISE_BITS)
READ_NOISE_EN ?=
READ_NOISE_RATE_NUM ?=
READ_NOISE_RATE_DEN ?=
READ_NOISE_BITS ?=
ifneq ($(strip $(WRITE_NOISE_EN)),)
COMPILE_ARGS += -GWRITE_NOISE_EN=$(WRITE_NOISE_EN)
endif
ifneq ($(strip $(WRITE_NOISE_RATE_NUM)),)
COMPILE_ARGS += -GWRITE_NOISE_RATE_NUM=$(WRITE_NOISE_RATE_NUM)
endif
ifneq ($(strip $(WRITE_NOISE_RATE_DEN)),)
COMPILE_ARGS += -GWRITE_NOISE_RATE_DEN=$(WRITE_NOISE_RATE_DEN)
endif
ifneq ($(strip $(WRITE_NOISE_BITS)),)
COMPILE_ARGS += -GWRITE_NOISE_BITS=$(WRITE_NOISE_BITS)
endif
ifneq ($(strip $(READ_NOISE_EN)),)
COMPILE_ARGS += -GREAD_NOISE_EN=$(READ_NOISE_EN)
endif
ifneq ($(strip $(READ_NOISE_RATE_NUM)),)
COMPILE_ARGS += -GREAD_NOISE_RATE_NUM=$(READ_NOISE_RATE_NUM)
endif
ifneq ($(strip $(READ_NOISE_RATE_DEN)),)
COMPILE_ARGS += -GREAD_NOISE_RATE_DEN=$(READ_NOISE_RATE_DEN)
endif
ifneq ($(strip $(READ_NOISE_BITS)),)
COMPILE_ARGS += -GREAD_NOISE_BITS=$(READ_NOISE_BITS)
endif
# Cleaner terminal output
export QUIET ?= 1
@@ -48,6 +69,12 @@ VERILOG_SOURCES += $(PWD)/../rtl/popcount.sv
VERILOG_SOURCES += $(PWD)/../rtl/argmax_update.sv
VERILOG_SOURCES += $(PWD)/../rtl/cam_core.sv
VERILOG_SOURCES += $(PWD)/../rtl/random/random128.sv
VERILOG_SOURCES += $(PWD)/../rtl/noise_mask_grouped.sv
VERILOG_SOURCES += $(PWD)/../rtl/cam_core_banked.sv
VERILOG_SOURCES += $(PWD)/../rtl/cam_write_noise.sv
VERILOG_SOURCES += $(PWD)/../rtl/cam_read_noise.sv
VERILOG_SOURCES += $(PWD)/../rtl/popcount_pipeline.sv
VERILOG_SOURCES += $(PWD)/../rtl/match_engine_pipeline.sv
VERILOG_SOURCES += $(PWD)/../rtl/cam_noisy.sv
VERILOG_SOURCES += $(PWD)/../rtl/match_engine.sv
VERILOG_SOURCES += $(PWD)/../rtl/cam_top.sv

124
hw/sim/model/ref_model.py
View File

@@ -101,6 +101,130 @@ def generate_write_flip_mask(
return mask, state
def generate_grouped_flip_mask(
*,
random_value: int,
hash_bits: int,
noise_bits: int,
rate_num: int,
rate_den: int,
) -> int:
"""Generate a grouped flip mask from one 128-bit value.
This is the shared write/read noise model: 8 default 64-bit groups, one
candidate flip per group, 6-bit bit index and 8-bit threshold sample.
It is not independent Bernoulli sampling over all 512 bits.
"""
assert noise_bits > 0
assert hash_bits % noise_bits == 0
group_bits = hash_bits // noise_bits
bit_index_bits = 6
sample_bits = 8
group_random_bits = bit_index_bits + sample_bits
assert group_bits == 64
assert noise_bits * group_random_bits <= 128
assert rate_den > 0
assert 0 <= rate_num <= rate_den
sample_range = 1 << sample_bits
threshold = (rate_num * sample_range) // rate_den
mask = 0
for group_idx in range(noise_bits):
group_rand = (random_value >> (group_idx * group_random_bits)) & ((1 << group_random_bits) - 1)
bit_idx = group_rand & ((1 << bit_index_bits) - 1)
sample = (group_rand >> bit_index_bits) & (sample_range - 1)
if sample < threshold:
mask |= 1 << (group_idx * group_bits + bit_idx)
return mask
def lane_seed_128(seed: int, lane: int) -> int:
"""Derive a nonzero 128-bit lane seed matching the RTL salt convention."""
mask128 = (1 << 128) - 1
salt = ((lane + 1) * 0x9E37_79B9_7F4A_7C15) & ((1 << 64) - 1)
mixed64 = (int(seed) ^ salt) & ((1 << 64) - 1)
state = ((mixed64 << 64) | mixed64) & mask128
assert state != 0
return state
def generate_read_lane_masks(
lane_states: list[int],
*,
hash_bits: int,
noise_bits: int,
rate_num: int,
rate_den: int,
lane_valid: list[bool],
) -> tuple[list[int], list[int]]:
"""Advance valid lane PRNG states once and return one mask per lane."""
next_states: list[int] = []
masks: list[int] = []
for lane, state in enumerate(lane_states):
if lane_valid[lane]:
next_state = xorshift128(state)
mask = generate_grouped_flip_mask(
random_value=next_state,
hash_bits=hash_bits,
noise_bits=noise_bits,
rate_num=rate_num,
rate_den=rate_den,
)
else:
next_state = state
mask = 0
next_states.append(next_state)
masks.append(mask)
return masks, next_states
def match_top1_with_read_noise(
query: int,
rows: Sequence[int],
*,
width: int = 512,
lanes: int = 8,
noise_bits: int = 8,
rate_num: int = 1,
rate_den: int = 100,
seed: int = 0x6A09_E667_F3BC_C909,
) -> MatchResult:
"""Top-1 matching with dynamic read noise, one query in flight."""
assert lanes > 0
assert len(rows) % lanes == 0
scores = np.zeros(len(rows), dtype=np.int32)
best_index = 0
best_score = -1
lane_states = [lane_seed_128(seed, lane) for lane in range(lanes)]
for base in range(0, len(rows), lanes):
lane_valid = [True] * lanes
masks, lane_states = generate_read_lane_masks(
lane_states,
hash_bits=width,
noise_bits=noise_bits,
rate_num=rate_num,
rate_den=rate_den,
lane_valid=lane_valid,
)
for lane in range(lanes):
row_idx = base + lane
noisy_row = int(rows[row_idx]) ^ masks[lane]
score = xnor_popcount_score(int(query), noisy_row, width)
scores[row_idx] = score
if score > best_score:
best_score = score
best_index = row_idx
return MatchResult(top1_index=int(best_index), top1_score=int(best_score), scores=scores)
def random_hashes(
rng: np.random.Generator,
n: int,

431
hw/sim/tests/test_cam_basic.py
View File

@@ -7,14 +7,15 @@ 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,
)
NUM_ROWS = 512
HASH_BITS = 512
LANES = 16
SCORE_BITS = 10
DEFAULT_NUM_ROWS = 4096
DEFAULT_HASH_BITS = 512
DEFAULT_LANES = 8
DEFAULT_SCORE_BITS = 10
def _get_param(dut, name, default=None):
@@ -28,6 +29,38 @@ def _get_param(dut, name, default=None):
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 ──────────────────────────────────────────────────────────────────
@@ -57,14 +90,7 @@ async def wait_idle(dut):
async def write_row(dut, addr, value):
"""Write a single row using wr_valid/wr_ready handshake.
1. Wait for system idle
2. Assert wr_valid + set addr/hash
3. Wait for handshake (wr_ready=1 on clock edge)
4. Deassert wr_valid
5. Wait for wr_ready to return 1 (commit complete)
"""
"""Write a single row using wr_valid/wr_ready handshake."""
await wait_idle(dut)
dut.wr_addr.value = addr
@@ -79,7 +105,7 @@ async def write_row(dut, addr, value):
dut.wr_valid.value = 0
# Wait for cam_noisy to finish GEN_MASK/COMMIT
# Wait for write pipeline to drain
await wait_idle(dut)
@@ -90,15 +116,7 @@ async def write_rows(dut, rows):
async def query_once(dut, query):
"""Issue a query and return (top1_index, top1_score, score_debug).
1. Wait for system idle
2. Assert query_valid + set query_hash
3. Wait for query_ready handshake
4. Deassert query_valid
5. Wait for result_valid
6. Read result, pulse result_ready to consume
"""
"""Issue a query and return (top1_index, top1_score, score_debug)."""
await wait_idle(dut)
dut.query_hash.value = int(query)
@@ -119,12 +137,14 @@ async def query_once(dut, query):
top1_index = int(dut.top1_index.value)
top1_score = int(dut.top1_score.value)
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,
num_rows,
score_bits,
)
dut.result_ready.value = 1
@@ -134,66 +154,83 @@ async def query_once(dut, query):
return top1_index, top1_score, score_debug
# ── Test A: Baseline (NOISE_EN=0) ────────────────────────────────────────────
# ── 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_EN=0."""
noise_en = _get_param(dut, "NOISE_EN", 0)
if noise_en:
dut._log.info("Skipping baseline_no_noise: requires NOISE_EN=0.")
"""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 = 123
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)
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
assert top1_score == hash_bits
if score_debug is not None:
assert np.array_equal(score_debug, expected.scores)
# ── Test B: Zero noise rate (NOISE_EN=1, RATE_NUM=0) ────────────────────────
# ── 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, equivalent to NOISE_EN=0."""
noise_en = _get_param(dut, "NOISE_EN", 1)
rate_num = _get_param(dut, "NOISE_RATE_NUM", 1)
if not noise_en or rate_num != 0:
dut._log.info("Skipping zero_rate_noise: requires NOISE_EN=1, RATE_NUM=0.")
"""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 = 123
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)
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
assert top1_score == hash_bits
if score_debug is not None:
assert np.array_equal(score_debug, expected.scores)
@@ -204,17 +241,12 @@ async def zero_rate_noise(dut):
@cocotb.test()
async def full_rate_noise(dut):
"""NOISE_RATE_NUM=1, NOISE_RATE_DEN=1 → every group flips its selected bit per write.
At full rate with default NOISE_BITS=8, exactly 8 deterministic bits flip per write
(one selected bit per group), not all 512 bits. We use ref_model.py PRNG to predict
the exact stored rows.
"""
noise_en = _get_param(dut, "NOISE_EN", 1)
rate_num = _get_param(dut, "NOISE_RATE_NUM", 1)
rate_den = _get_param(dut, "NOISE_RATE_DEN", 100)
"""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 NOISE_EN=1, RATE_NUM=1, 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).")
@@ -223,47 +255,41 @@ async def full_rate_noise(dut):
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_dut(dut)
noise_bits = _get_param(dut, "NOISE_BITS", 8)
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
all_one = (1 << hash_bits) - 1
# Predict stored rows using the same RTL seed convention as exact_noise_model_match.
RTL_SEED = 0xB504_F32D_B504_F32D
prng_state = (RTL_SEED << 64) | RTL_SEED
# Flip mask for row 0 (all-zero written)
flip0, prng_state = generate_write_flip_mask(
prng_state, HASH_BITS, noise_bits, rate_num, rate_den,
prng_state, hash_bits, noise_bits, rate_num, rate_den,
)
expected_row0 = all_zero ^ flip0 # stored value after noise
expected_row0 = all_zero ^ flip0
# Flip mask for row 1 (all-one written)
flip1, prng_state = generate_write_flip_mask(
prng_state, HASH_BITS, noise_bits, rate_num, rate_den,
prng_state, hash_bits, noise_bits, rate_num, rate_den,
)
expected_row1 = all_one ^ flip1 # stored value after noise
expected_row1 = all_one ^ flip1
# Write all-zero to row 0, all-one to row 1, rest zero
rows = [0] * NUM_ROWS
rows = [0] * num_rows
rows[0] = all_zero
rows[1] = all_one
await write_rows(dut, rows)
# Query expected_row0 → should exactly match row 0's stored value
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 for predicted stored value, "
f"score={score_debug[0]} != {HASH_BITS}"
assert int(score_debug[0]) == hash_bits, (
f"Row 0: expected exact match, score={score_debug[0]} != {hash_bits}"
)
# Query expected_row1 → should exactly match row 1's stored value
top1_index, top1_score, score_debug = await query_once(dut, expected_row1)
assert score_debug is not None, "score_debug required for full_rate_noise"
assert int(score_debug[1]) == HASH_BITS, (
f"Row 1: expected exact match for predicted stored value, "
f"score={score_debug[1]} != {HASH_BITS}"
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}"
)
@@ -273,111 +299,117 @@ async def full_rate_noise(dut):
@cocotb.test()
async def default_noise_reproducible(dut):
"""Fixed seed → deterministic write noise. Two identical runs produce same results."""
noise_en = _get_param(dut, "NOISE_EN", 1)
noise_en = _get_param(dut, "WRITE_NOISE_EN", 1)
if not noise_en:
dut._log.info("Skipping default_noise_reproducible: requires NOISE_EN=1.")
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)
rows = random_hashes(rng, num_rows, width=hash_bits)
# ── First run ──
await write_rows(dut, rows)
query = rows[50]
query = rows[min(50, num_rows - 1)]
top1_index_1, top1_score_1, _ = await query_once(dut, query)
# Reset for second run
await reset_dut(dut)
# ── Second run with same data ──
await write_rows(dut, rows)
top1_index_2, top1_score_2, _ = await query_once(dut, query)
# Deterministic: same seed → same PRNG sequence → same stored hashes → same result
assert top1_index_1 == top1_index_2
assert top1_score_1 == top1_score_2
# ── Preserved legacy tests (only meaningful for NOISE_EN=0) ──────────────────
# ── 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, "NOISE_EN", 1):
dut._log.info("Skipping known_hamming_distance: NOISE_EN=1, stored hashes may differ.")
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[10] = (1 << 7) - 1
rows[11] = (1 << 31) - 1
rows[12] = (1 << 128) - 1
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
assert top1_score == hash_bits
if score_debug is not None:
assert int(score_debug[10]) == HASH_BITS - 7
assert int(score_debug[11]) == HASH_BITS - 31
assert int(score_debug[12]) == HASH_BITS - 128
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, "NOISE_EN", 1):
dut._log.info("Skipping tie_break_policy: NOISE_EN=1, stored hashes may differ.")
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[200]
rows = random_hashes(rng, num_rows, width=hash_bits)
query = rows[min(200, num_rows - 1)]
rows[10] = query
rows[20] = query
rows[200] = 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
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, "NOISE_EN", 1):
dut._log.info("Skipping all_zero_all_one_boundary: NOISE_EN=1, stored hashes may differ.")
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)
rows = [0] * NUM_ROWS
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
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_score == hash_bits
assert top1_index == 0
if score_debug is not None:
assert int(score_debug[0]) == HASH_BITS
assert int(score_debug[0]) == hash_bits
assert int(score_debug[1]) == 0
@@ -386,16 +418,15 @@ async def all_zero_all_one_boundary(dut):
@cocotb.test()
async def exact_noise_model_match(dut):
"""Verify RTL stored hashes match ref_model.py for a known seed and rate.
Writes rows with noise enabled, then queries back via score_debug to
reconstruct stored hashes, and compares against Python model predictions.
"""
noise_en = _get_param(dut, "NOISE_EN", 1)
rate_num = _get_param(dut, "NOISE_RATE_NUM", 1)
rate_den = _get_param(dut, "NOISE_RATE_DEN", 100)
"""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 NOISE_EN=1, 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.")
@@ -404,39 +435,31 @@ async def exact_noise_model_match(dut):
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_dut(dut)
noise_bits = _get_param(dut, "NOISE_BITS", 8)
hash_bits = dut_hash_bits(dut)
noise_bits = _get_param(dut, "WRITE_NOISE_BITS", 8)
# Use a small subset to keep test fast
n_test_rows = 4
rng = np.random.default_rng(99)
rows = random_hashes(rng, n_test_rows, width=HASH_BITS)
rows = random_hashes(rng, n_test_rows, width=hash_bits)
# Predict stored hashes with Python model using the same seed.
# RTL random128 seed: {NOISE_SEED, NOISE_SEED}, default 64'hB504_F32D_B504_F32D.
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,
prng_state, hash_bits, noise_bits, rate_num, rate_den,
)
expected_stored.append(row ^ flip)
# Write only test rows (rest stay at 0 from reset)
for idx, val in enumerate(rows):
await write_row(dut, idx, val)
# Query all-zero to get Hamming distances (= HASH_BITS - popcount(stored ^ 0) = HASH_BITS - popcount(stored))
# So popcount(stored) = HASH_BITS - score
# This gives us the number of set bits but not the exact value.
# Instead, query each expected_stored value — it should score HASH_BITS if match is exact.
for idx, expected in enumerate(expected_stored):
top1_index, top1_score, score_debug = await query_once(dut, expected)
# The stored hash at idx should exactly match expected, so score == HASH_BITS
assert score_debug is not None, "score_debug required for mask match verification"
assert int(score_debug[idx]) == HASH_BITS, (
assert int(score_debug[idx]) == hash_bits, (
f"Row {idx}: expected stored hash to match model prediction, "
f"score={score_debug[idx]} != {HASH_BITS}"
f"score={score_debug[idx]} != {hash_bits}"
)
@@ -445,52 +468,160 @@ async def exact_noise_model_match(dut):
@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.
Only runs with NOISE_EN=0 so stored hashes are predictable.
"""
if _get_param(dut, "NOISE_EN", 1):
dut._log.info("Skipping half_duplex_write_priority: requires NOISE_EN=0 for exact scores.")
"""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)
# Write a known value to row 0
test_val = (1 << HASH_BITS) - 1 # all-ones
hash_bits = dut_hash_bits(dut)
test_val = (1 << hash_bits) - 1
await write_row(dut, 0, test_val)
# Now system is idle: wr_ready=1, query_ready=1
await wait_idle(dut)
assert int(dut.wr_ready.value) == 1
assert int(dut.query_ready.value) == 1
# Drive both wr_valid and query_valid simultaneously
dut.wr_valid.value = 1
dut.wr_addr.value = 1
dut.write_hash.value = 0 # write all-zeros to row 1
dut.write_hash.value = 0
dut.query_valid.value = 1
dut.query_hash.value = test_val # query for all-ones (in row 0)
dut.query_hash.value = test_val
await RisingEdge(dut.clk)
# Write should have been accepted (wr_ready was 1), query should NOT have been accepted
# because write-priority gates query_ready when wr_valid=1
wr_accepted = int(dut.wr_ready.value) == 0 # after handshake, wr_ready drops
# query_ready should have been 0 during the simultaneous assertion
# (it's !wr_valid gated), so query was blocked
# Deassert both
dut.wr_valid.value = 0
dut.query_valid.value = 0
# Wait for write to complete (noise generation + commit)
await wait_idle(dut)
# Now query should work — row 0 has all-ones (written first)
top1_index, top1_score, _ = await query_once(dut, test_val)
assert top1_index == 0
assert top1_score == HASH_BITS
assert top1_score == hash_bits
# Verify row 1 was written (all-zeros) — query all-zeros
top1_index, top1_score, _ = await query_once(dut, 0)
assert top1_index == 1
assert top1_score == HASH_BITS
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
while int(dut.result_valid.value) == 0:
await RisingEdge(dut.clk)
dut.result_ready.value = 1
await RisingEdge(dut.clk)
# ── 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)

48
hw/sim/tests/test_cam_core_banked.py Normal file
View File

@@ -0,0 +1,48 @@
from __future__ import annotations
import cocotb
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge
async def reset_core(dut):
dut.rst_n.value = 0
dut.wr_valid.value = 0
dut.wr_row.value = 0
dut.wr_hash.value = 0
dut.rd_valid_i.value = 0
dut.rd_base_row_i.value = 0
for _ in range(3):
await RisingEdge(dut.clk)
dut.rst_n.value = 1
for _ in range(2):
await RisingEdge(dut.clk)
@cocotb.test()
async def banked_core_reads_aligned_eight_row_batch_after_one_cycle(dut):
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_core(dut)
LANES = 8
ROW_BITS = len(dut.rd_row_ids_o) // LANES
for row in range(LANES):
dut.wr_valid.value = 1
dut.wr_row.value = row
dut.wr_hash.value = row + 0x100
await RisingEdge(dut.clk)
dut.wr_valid.value = 0
dut.rd_base_row_i.value = 0
dut.rd_valid_i.value = 1
await RisingEdge(dut.clk)
dut.rd_valid_i.value = 0
await RisingEdge(dut.clk)
assert int(dut.rd_valid_o.value) == 1
for lane in range(LANES):
got_row = (int(dut.rd_row_ids_o.value) >> (lane * ROW_BITS)) & ((1 << ROW_BITS) - 1)
got_hash = (int(dut.rd_hashes_o.value) >> (lane * 512)) & ((1 << 512) - 1)
assert got_row == lane
assert got_hash == lane + 0x100

48
hw/sim/tests/test_cam_read_noise.py Normal file
View File

@@ -0,0 +1,48 @@
from __future__ import annotations
import cocotb
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge
async def reset_read_noise(dut):
dut.rst_n.value = 0
dut.valid_i.value = 0
dut.row_ids_i.value = 0
dut.hashes_i.value = 0
dut.lane_valid_i.value = 0
for _ in range(3):
await RisingEdge(dut.clk)
dut.rst_n.value = 1
for _ in range(2):
await RisingEdge(dut.clk)
@cocotb.test()
async def read_noise_disabled_forwards_hashes_after_one_stage(dut):
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_read_noise(dut)
LANES = 8
ROW_BITS = len(dut.row_ids_i) // LANES
HASH_BITS_PER_LANE = len(dut.hashes_i) // LANES
hashes = 0
rows = 0
for lane in range(LANES):
hashes |= (lane + 0x55) << (lane * HASH_BITS_PER_LANE)
rows |= lane << (lane * ROW_BITS)
dut.hashes_i.value = hashes
dut.row_ids_i.value = rows
dut.lane_valid_i.value = 0xFF
dut.valid_i.value = 1
await RisingEdge(dut.clk)
dut.valid_i.value = 0
await RisingEdge(dut.clk)
await RisingEdge(dut.clk)
assert int(dut.valid_o.value) == 1
assert int(dut.hashes_noisy_o.value) == hashes
assert int(dut.row_ids_o.value) == rows
assert int(dut.lane_valid_o.value) == 0xFF

39
hw/sim/tests/test_cam_write_noise.py Normal file
View File

@@ -0,0 +1,39 @@
from __future__ import annotations
import cocotb
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge
from model.ref_model import generate_write_flip_mask
async def reset_write_noise(dut):
dut.rst_n.value = 0
dut.wr_valid.value = 0
dut.wr_row.value = 0
dut.wr_hash.value = 0
for _ in range(3):
await RisingEdge(dut.clk)
dut.rst_n.value = 1
for _ in range(2):
await RisingEdge(dut.clk)
@cocotb.test()
async def write_noise_outputs_grouped_noisy_hash(dut):
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_write_noise(dut)
value = 0x123456789ABCDEF
dut.wr_row.value = 3
dut.wr_hash.value = value
dut.wr_valid.value = 1
await RisingEdge(dut.clk)
dut.wr_valid.value = 0
while int(dut.core_wr_valid.value) == 0:
await RisingEdge(dut.clk)
seed = 0xB504_F32D_B504_F32D
flip, _ = generate_write_flip_mask((seed << 64) | seed, 512, 8, 1, 100)
assert int(dut.core_wr_row.value) == 3
assert int(dut.core_wr_hash.value) == (value ^ flip)

66
hw/sim/tests/test_match_engine_pipeline.py Normal file
View File

@@ -0,0 +1,66 @@
from __future__ import annotations
import cocotb
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge
async def reset_match(dut):
dut.rst_n.value = 0
dut.query_valid.value = 0
dut.query_hash.value = 0
dut.result_ready.value = 1
dut.rd_valid_i.value = 0
dut.rd_row_ids_i.value = 0
dut.rd_hashes_i.value = 0
dut.rd_lane_valid_i.value = 0
for _ in range(3):
await RisingEdge(dut.clk)
dut.rst_n.value = 1
for _ in range(2):
await RisingEdge(dut.clk)
@cocotb.test()
async def match_engine_returns_top1_after_pipeline_drain(dut):
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_match(dut)
LANES = 8
ROW_BITS = len(dut.rd_row_ids_i) // LANES
HASH_BITS = len(dut.rd_hashes_i) // LANES
NUM_ROWS = 1 << len(dut.rd_base_row_o)
TARGET_ROW = 9
query = (1 << HASH_BITS) - 1
dut.query_hash.value = query
dut.query_valid.value = 1
await RisingEdge(dut.clk)
dut.query_valid.value = 0
for base in range(0, NUM_ROWS, LANES):
while int(dut.rd_valid_o.value) == 0:
await RisingEdge(dut.clk)
assert int(dut.rd_base_row_o.value) == base
rows = 0
hashes = 0
for lane in range(LANES):
row_id = base + lane
rows |= row_id << (lane * ROW_BITS)
row_hash = query if row_id == TARGET_ROW else 0
hashes |= row_hash << (lane * HASH_BITS)
dut.rd_row_ids_i.value = rows
dut.rd_hashes_i.value = hashes
dut.rd_lane_valid_i.value = 0xFF
dut.rd_valid_i.value = 1
await RisingEdge(dut.clk)
dut.rd_valid_i.value = 0
for _ in range(20):
if int(dut.result_valid.value):
break
await RisingEdge(dut.clk)
assert int(dut.result_valid.value) == 1
assert int(dut.result_row.value) == TARGET_ROW
assert int(dut.result_score.value) == HASH_BITS

70
hw/sim/tests/test_ref_model_noise.py Normal file
View File

@@ -0,0 +1,70 @@
from __future__ import annotations
from model.ref_model import (
generate_grouped_flip_mask,
match_top1_with_read_noise,
xnor_popcount_score,
)
def test_grouped_flip_mask_full_rate_one_bit_per_64_bit_group():
random_value = 0
for group in range(8):
bit_idx = group + 1
sample = 0
random_value |= bit_idx << (group * 14)
random_value |= sample << (group * 14 + 6)
mask = generate_grouped_flip_mask(
random_value=random_value,
hash_bits=512,
noise_bits=8,
rate_num=1,
rate_den=1,
)
expected = 0
for group in range(8):
expected |= 1 << (group * 64 + group + 1)
assert mask == expected
assert mask.bit_count() == 8
def test_grouped_flip_mask_zero_rate_no_flips():
mask = generate_grouped_flip_mask(
random_value=(1 << 128) - 1,
hash_bits=512,
noise_bits=8,
rate_num=0,
rate_den=100,
)
assert mask == 0
def test_score_is_bit_match_popcount_not_hamming_distance():
query = 0b1010
stored = 0b1000
assert xnor_popcount_score(query, stored, width=4) == 3
def test_read_noise_model_is_reproducible_after_reset_seed():
rows = [0, (1 << 512) - 1, 0x1234, 0x5678, 0x9ABC, 0xDEF0, 0x1357, 0x2468]
query = rows[2]
kwargs = dict(
query=query,
rows=rows,
width=512,
lanes=8,
noise_bits=8,
rate_num=1,
rate_den=100,
seed=0x6A09_E667_F3BC_C909,
)
first = match_top1_with_read_noise(**kwargs)
second = match_top1_with_read_noise(**kwargs)
assert first.top1_index == second.top1_index
assert first.top1_score == second.top1_score
assert first.scores.tolist() == second.scores.tolist()