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(rtl): migrate CAM interface to handshake protocol with integrated noise generation

Browse Source 
BREAKING CHANGE: CAM write and query interface replaced with standard valid/ready
handshake. wr_en/wr_row/wr_hash → wr_valid/wr_ready/wr_addr/write_hash.
External noise_mask_lanes_flat removed; noise generation now handled internally
by cam_noisy module with configurable rate via parameters.

- cam_top: add parameters (NOISE_EN, NOISE_RATE_NUM/DEN, NOISE_GEN/SAMPLE_BITS, NOISE_SEED)
- cam_top: replace cam_core with cam_noisy (integrated noise generation)
- match_engine: remove external noise_mask_lanes_flat input
- hw/sim: update Makefile with noise parameters and compile args
- hw/sim/model: add generate_write_flip_mask() and xorshift64() matching RTL behavior
- hw/sim/tests: adapt testbench to new handshake protocol
This commit is contained in:
SikongJueluo
2026-05-04 18:02:34 +08:00
parent 0ae6d757dc
commit 2da17e101b
10 changed files with 723 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
hw/sim/Makefile
View File

@@ -9,6 +9,15 @@ NUM_ROWS ?= 512
HASH_BITS ?= 512
LANES ?= 16
# Noise parameters
NOISE_EN ?= 1
NOISE_RATE_NUM ?= 1
NOISE_RATE_DEN ?= 100
NOISE_GEN_BITS ?= 8
NOISE_SAMPLE_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_noisy.sv.
EXTRA_ARGS += +define+NUM_ROWS=$(NUM_ROWS) +define+HASH_BITS=$(HASH_BITS) +define+LANES=$(LANES)
COMPILE_ARGS += -Wall -Wno-fatal
@@ -16,6 +25,13 @@ 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_GEN_BITS=$(NOISE_GEN_BITS)
COMPILE_ARGS += -GNOISE_SAMPLE_BITS=$(NOISE_SAMPLE_BITS)
# Cleaner terminal output
export QUIET ?= 1
export VERBOSE ?= 0
@@ -32,6 +48,7 @@ endif
VERILOG_SOURCES += $(PWD)/../rtl/popcount.sv
VERILOG_SOURCES += $(PWD)/../rtl/argmax_update.sv
VERILOG_SOURCES += $(PWD)/../rtl/cam_core.sv
VERILOG_SOURCES += $(PWD)/../rtl/cam_noisy.sv
VERILOG_SOURCES += $(PWD)/../rtl/match_engine.sv
VERILOG_SOURCES += $(PWD)/../rtl/cam_top.sv

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

@@ -1,7 +1,7 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Iterable, Sequence
from typing import Sequence
import numpy as np
@@ -25,25 +25,20 @@ def xnor_popcount_score(query: int, stored: int, width: int = 512) -> int:
return popcount_int(same_bits)
def apply_noise(stored: int, noise_mask: int) -> int:
return stored ^ noise_mask
def match_top1(
query: int,
rows: Sequence[int],
*,
width: int = 512,
noise_masks: Sequence[int] | None = None,
) -> MatchResult:
"""Pure matching — noise is already baked into rows at write time."""
scores = np.zeros(len(rows), dtype=np.int32)
best_index = 0
best_score = -1
for idx, row in enumerate(rows):
effective = row if noise_masks is None else apply_noise(row, int(noise_masks[idx]))
score = xnor_popcount_score(int(query), int(effective), width)
score = xnor_popcount_score(int(query), int(row), width)
scores[idx] = score
# Tie-break: choose the smallest row index.
@@ -58,6 +53,54 @@ def match_top1(
)
def xorshift64(state: int) -> int:
"""64-bit XOR-shift PRNG, single step. Matches RTL behavior."""
mask64 = (1 << 64) - 1
s = state & mask64
s ^= (s << 13) & mask64
s ^= (s >> 7) & mask64
s ^= (s << 17) & mask64
return s
def generate_write_flip_mask(
prng_state: int,
hash_bits: int,
noise_gen_bits: int,
noise_sample_bits: int,
rate_num: int,
rate_den: int,
) -> tuple[int, int]:
"""
Generate write-noise flip mask.
Returns (flip_mask, next_prng_state).
Matches RTL multi-cycle GEN_MASK behavior.
Each cycle processes noise_gen_bits bit decisions:
- Advance xorshift64 → 64-bit output
- Split into noise_gen_bits x noise_sample_bits-bit samples
- Each sample < THRESHOLD → that bit flips
"""
assert hash_bits % noise_gen_bits == 0
assert noise_gen_bits * noise_sample_bits == 64
mask = 0
state = prng_state
sample_range = 1 << noise_sample_bits
threshold = (rate_num * sample_range) // rate_den
for bit_offset in range(0, hash_bits, noise_gen_bits):
# Advance PRNG
state = xorshift64(state)
# Split into noise_gen_bits independent samples
for b in range(noise_gen_bits):
sample_b = (state >> (b * noise_sample_bits)) & (sample_range - 1)
if sample_b < threshold:
mask |= (1 << (bit_offset + b))
return mask, state
def random_hashes(
rng: np.random.Generator,
n: int,
@@ -76,35 +119,6 @@ def random_hashes(
return out
def random_noise_masks(
rng: np.random.Generator,
n: int,
*,
width: int = 512,
bit_flip_rate: float = 0.0,
) -> list[int]:
if not (0.0 <= bit_flip_rate <= 1.0):
raise ValueError("bit_flip_rate must be in [0, 1]")
masks: list[int] = []
for _ in range(n):
bits = rng.random(width) < bit_flip_rate
value = 0
for i, bit in enumerate(bits):
if bool(bit):
value |= 1 << i
masks.append(value)
return masks
def pack_lanes_flat(masks: Sequence[int], *, width: int = 512) -> int:
flat = 0
lane_mask = mask_width(width)
for lane, mask in enumerate(masks):
flat |= (int(mask) & lane_mask) << (lane * width)
return flat
def unpack_score_debug_flat(flat: int, num_rows: int, score_bits: int) -> np.ndarray:
mask = (1 << score_bits) - 1
return np.array(

65
hw/sim/sweep_noise.py
View File

@@ -1,9 +1,41 @@
"""Sweep write-noise rates and measure top-1 stability.
Applies write-noise flip masks to stored rows (simulating noisy writes),
then queries the noisy rows and compares top-1 results against clean rows.
"""
from __future__ import annotations
import argparse
import numpy as np
from model.ref_model import match_top1, random_hashes, random_noise_masks
from model.ref_model import (
generate_write_flip_mask,
match_top1,
random_hashes,
)
def apply_write_noise(
rows: list[int],
*,
width: int,
rate_num: int,
rate_den: int,
noise_gen_bits: int = 8,
noise_sample_bits: int = 8,
seed: int = 0,
) -> list[int]:
"""Apply write-noise flip masks to every row, returning noisy copies."""
noisy: list[int] = []
state = seed
mask_w = (1 << width) - 1
for row in rows:
flip, state = generate_write_flip_mask(
state, width, noise_gen_bits, noise_sample_bits, rate_num, rate_den
)
noisy.append((row ^ flip) & mask_w)
return noisy
def main() -> None:
@@ -12,6 +44,8 @@ def main() -> None:
parser.add_argument("--queries", type=int, default=128)
parser.add_argument("--width", type=int, default=512)
parser.add_argument("--seed", type=int, default=1234)
parser.add_argument("--noise-gen-bits", type=int, default=8)
parser.add_argument("--noise-sample-bits", type=int, default=8)
parser.add_argument(
"--rates",
type=float,
@@ -29,19 +63,30 @@ def main() -> None:
clean_results = [match_top1(q, rows, width=args.width) for q in queries]
print("rate,top1_stability,avg_clean_margin")
# Use a fixed denominator matching the 8-bit sample space (2^8 = 256).
# Note: floor() is used, matching RTL threshold = (rate_num * 256) // rate_den.
# Rates below 1/256 (≈0.39%) collapse to zero under this scheme.
rate_den = 256
print("rate,rate_num,effective_prob,top1_stability,avg_clean_margin")
for rate in args.rates:
rate_num = int(rate * rate_den)
effective = rate_num / rate_den if rate_den > 0 else 0.0
stable = 0
margins = []
noisy_rows = apply_write_noise(
rows,
width=args.width,
rate_num=rate_num,
rate_den=rate_den,
noise_gen_bits=args.noise_gen_bits,
noise_sample_bits=args.noise_sample_bits,
seed=args.seed,
)
for q, clean in zip(queries, clean_results):
noise_masks = random_noise_masks(
rng,
args.rows,
width=args.width,
bit_flip_rate=rate,
)
noisy = match_top1(q, rows, width=args.width, noise_masks=noise_masks)
noisy = match_top1(q, noisy_rows, width=args.width)
if noisy.top1_index == clean.top1_index:
stable += 1
@@ -50,7 +95,7 @@ def main() -> None:
margin = int(sorted_scores[-1] - sorted_scores[-2])
margins.append(margin)
print(f"{rate},{stable / args.queries:.6f},{np.mean(margins):.3f}")
print(f"{rate},{rate_num},{effective:.6f},{stable / args.queries:.6f},{np.mean(margins):.3f}")
if __name__ == "__main__":

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

@@ -5,10 +5,11 @@ 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,
pack_lanes_flat,
random_hashes,
unpack_score_debug_flat,
xorshift64,
)
NUM_ROWS = 512
@@ -17,18 +18,30 @@ LANES = 16
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
# ── Helpers ──────────────────────────────────────────────────────────────────
async def reset_dut(dut):
"""Reset the DUT with new handshake interface."""
dut.rst_n.value = 0
dut.wr_en.value = 0
dut.wr_row.value = 0
dut.wr_hash.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
if hasattr(dut, "noise_mask_lanes_flat"):
dut.noise_mask_lanes_flat.value = 0
for _ in range(5):
await RisingEdge(dut.clk)
@@ -38,38 +51,70 @@ async def reset_dut(dut):
await RisingEdge(dut.clk)
async def write_rows(dut, rows):
for idx, value in enumerate(rows):
dut.wr_row.value = idx
dut.wr_hash.value = int(value)
dut.wr_en.value = 1
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)
dut.wr_en.value = 0
await RisingEdge(dut.clk)
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)
"""
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 cam_noisy to finish GEN_MASK/COMMIT
await wait_idle(dut)
async def query_once(dut, query, noise_masks=None):
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).
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
"""
await wait_idle(dut)
dut.query_hash.value = int(query)
dut.query_valid.value = 1
await RisingEdge(dut.clk)
# Wait for handshake
while True:
await RisingEdge(dut.clk)
if int(dut.query_ready.value):
break
dut.query_valid.value = 0
# Feed lane noise masks batch by batch while DUT is scanning.
# For no-noise builds this signal is absent and ignored.
base = 0
# Wait for result
while int(dut.result_valid.value) == 0:
if hasattr(dut, "noise_mask_lanes_flat") and noise_masks is not None:
lane_masks = []
for lane in range(LANES):
row = base + lane
lane_masks.append(noise_masks[row] if row < NUM_ROWS else 0)
dut.noise_mask_lanes_flat.value = pack_lanes_flat(
lane_masks, width=HASH_BITS
)
base += LANES
await RisingEdge(dut.clk)
top1_index = int(dut.top1_index.value)
@@ -83,12 +128,23 @@ async def query_once(dut, query, noise_masks=None):
SCORE_BITS,
)
dut.result_ready.value = 1
await RisingEdge(dut.clk)
dut.result_ready.value = 0
return top1_index, top1_score, score_debug
# ── Test A: Baseline (NOISE_EN=0) ────────────────────────────────────────────
@cocotb.test()
async def basic_write_query_no_noise(dut):
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.")
return
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_dut(dut)
@@ -111,8 +167,175 @@ async def basic_write_query_no_noise(dut):
assert np.array_equal(score_debug, expected.scores)
# ── Test B: Zero noise rate (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.")
return
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_dut(dut)
rng = np.random.default_rng(1)
rows = random_hashes(rng, NUM_ROWS, width=HASH_BITS)
query_index = 123
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):
"""THRESHOLD=256 → all bits flip. stored == ~written."""
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)
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.")
return
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_dut(dut)
all_zero = 0
all_one = (1 << HASH_BITS) - 1
# Write all-zero to row 0, all-one to row 1, rest zero
rows = [0] * NUM_ROWS
rows[0] = all_zero
rows[1] = all_one
await write_rows(dut, rows)
# After 100% flip: row 0 stored = ~0 = all_one, row 1 stored = ~all_one = all_zero
# Query all-zero → matches row 1 (stored all_zero) with score = HASH_BITS
top1_index, top1_score, _ = await query_once(dut, all_zero)
assert top1_index == 1
assert top1_score == HASH_BITS
# Query all-one → matches row 0 (stored all_one) with score = HASH_BITS
top1_index, top1_score, _ = await query_once(dut, all_one)
assert top1_index == 0
assert top1_score == HASH_BITS
# Query all-zero → score against row 0 (stored all_one) = 0
# Re-query to verify: row 0 stored is all_one, query all_zero → score 0
# But top1 picks row 1 with score HASH_BITS, so top1_index=1
# ── 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, "NOISE_EN", 1)
if not noise_en:
dut._log.info("Skipping default_noise_reproducible: requires NOISE_EN=1.")
return
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_dut(dut)
rng = np.random.default_rng(42)
rows = random_hashes(rng, NUM_ROWS, width=HASH_BITS)
# ── First run ──
await write_rows(dut, rows)
query = rows[50]
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) ──────────────────
@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.")
return
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_dut(dut)
query = 0
rows = [0] * NUM_ROWS
rows[10] = (1 << 7) - 1
rows[11] = (1 << 31) - 1
rows[12] = (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[10]) == HASH_BITS - 7
assert int(score_debug[11]) == HASH_BITS - 31
assert int(score_debug[12]) == 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.")
return
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_dut(dut)
rng = np.random.default_rng(2)
rows = random_hashes(rng, NUM_ROWS, width=HASH_BITS)
query = rows[200]
rows[10] = query
rows[20] = query
rows[200] = 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, "NOISE_EN", 1):
dut._log.info("Skipping all_zero_all_one_boundary: NOISE_EN=1, stored hashes may differ.")
return
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_dut(dut)
@@ -132,79 +355,117 @@ async def all_zero_all_one_boundary(dut):
assert int(score_debug[1]) == 0
# ── Test E: Exact RTL-vs-model PRNG mask match ──────────────────────────────
@cocotb.test()
async def known_hamming_distance(dut):
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)
if not noise_en or rate_num == 0:
dut._log.info("Skipping exact_noise_model_match: requires NOISE_EN=1, RATE_NUM>0.")
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)
query = 0
rows = [0] * NUM_ROWS
rows[10] = (1 << 7) - 1 # Hamming distance = 7
rows[11] = (1 << 31) - 1 # Hamming distance = 31
rows[12] = (1 << 128) - 1 # Hamming distance = 128
noise_gen_bits = _get_param(dut, "NOISE_GEN_BITS", 8)
noise_sample_bits = _get_param(dut, "NOISE_SAMPLE_BITS", 8)
await write_rows(dut, rows)
top1_index, top1_score, score_debug = await query_once(dut, query)
# 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)
# Many rows are identical to query; tie-break must select row 0.
# Predict stored hashes with Python model using the same seed
# RTL default seed: 64'hB504_F32D_B504_F32D
RTL_SEED = 0xB504_F32D_B504_F32D
prng_state = RTL_SEED
expected_stored = []
for row in rows:
flip, prng_state = generate_write_flip_mask(
prng_state, HASH_BITS, noise_gen_bits, noise_sample_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, (
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.
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.")
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
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.query_valid.value = 1
dut.query_hash.value = test_val # query for all-ones (in row 0)
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
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
@cocotb.test()
async def tie_break_policy(dut):
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_dut(dut)
rng = np.random.default_rng(2)
rows = random_hashes(rng, NUM_ROWS, width=HASH_BITS)
query = rows[200]
rows[10] = query
rows[20] = query
rows[200] = query
await write_rows(dut, rows)
top1_index, top1_score, _ = await query_once(dut, query)
assert top1_index == 10
# 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
@cocotb.test()
async def external_noise_mask(dut):
# This test is meaningful only when compiled with SIM_NOISE and SIM_DEBUG.
cocotb.start_soon(Clock(dut.clk, 10, unit="ns").start())
await reset_dut(dut)
if not hasattr(dut, "noise_mask_lanes_flat"):
dut._log.warning("SIM_NOISE not enabled; skipping exact noise-mask behavior.")
return
rng = np.random.default_rng(3)
rows = random_hashes(rng, NUM_ROWS, width=HASH_BITS)
query_index = 42
query = rows[query_index]
noise_masks = [0] * NUM_ROWS
noise_masks[query_index] = (1 << 13) - 1 # flip exactly 13 bits
await write_rows(dut, rows)
top1_index, top1_score, score_debug = await query_once(
dut,
query,
noise_masks=noise_masks,
)
expected = match_top1(query, rows, width=HASH_BITS, noise_masks=noise_masks)
assert top1_index == expected.top1_index
assert top1_score == expected.top1_score
if score_debug is not None:
assert int(score_debug[query_index]) == HASH_BITS - 13
assert np.array_equal(score_debug, expected.scores)