Merge branch 'schwarz-fix-mp_bram_verification' into 'devel'

Reimplementation of MP BRAM cocotb verification

See merge request ndk/ndk-fpga!244

Author: jakubcabal

comp/base/mem/mp_bram/cocotb/Makefile

@@ -1,15 +1,11 @@
-# Makefile: Makefile script to compile specified module
-# Copyright (C) 2023 CESNET z. s. p. o.
-# Author(s): Oliver Gurka <oliver.gurka@cesnet.cz>
+# Makefile: Makefile to compile module
+# Copyright (C) 2024 CESNET z. s. p. o.
+# Author(s): Ondřej Schwarz <OndrejSchwarz@cesnet.cz>
 #
 # SPDX-License-Identifier: BSD-3-Clause
 
 TOP_LEVEL_ENT=MP_BRAM
-
-SYNTH=quartus
-DEVICE=AGILEX
-CLK_PORTS=CLK
-CLK_PERIOD=4.1
+TARGET=cocotb
 
 .PHONY: all
 all: comp

comp/base/mem/mp_bram/cocotb/cocotb_test.py

@@ -1,124 +1,90 @@
-"""
-file: cocotb_test.py
-author(s): Oliver Gurka <oliver.gurka@cesnet.cz>
-description: Simple cocotb tests for multiport BRAM implemented using XOR
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (C) 2025 CESNET z. s. p. o.
+# Author(s): Ondrej Schwarz <ondrejschwarz@cesnet.cz>
 
-Copyright (C) 2024 CESNET z. s. p. o.
-SPDX-License-Identifier: BSD-3-Clause
-"""
 
 import cocotb
 from cocotb.clock import Clock
 from cocotb.triggers import RisingEdge, ClockCycles
+from cocotbext.ofm.mp_bram.controller import MP_BRAM_Controller
+from cocotbext.ofm.utils.ram import VWWRAM
+from cocotbext.ofm.ver.generators import random_packets
+from cocotb.result import TestFailure, TestSuccess
+from random import randint
 
 
-async def reset(dut):
-    # Reset the design
-    dut.RESET.value = 1
-    await ClockCycles(dut.CLK, 5)
-    dut.RESET.value = 0
-    await RisingEdge(dut.CLK)
+class testbench():
+    def __init__(self, dut, debug=False):
+        self.dut = dut
+        self.stream_in = MP_BRAM_Controller(dut, "", dut.CLK)
 
+        self.ref_ram = VWWRAM(capacity=2**len(self.stream_in.bus.WR_ADDR[0]),
+                              word_width=len(self.stream_in.bus.WR_DATA[0]),
+                              block_width=dut.BLOCK_WIDTH.value if dut.BLOCK_ENABLE.value else None)
 
-async def read(dut, port, addr):
-    re = RisingEdge(dut.CLK)
+        if debug:
+            self.stream_in.log.setLevel(cocotb.logging.DEBUG)
 
-    dut.RD_ADDR[port].value = addr
-    dut.RD_EN[port].value = 1
-    await re
-    dut.RD_EN[port].value = 0
+    async def reset(self):
+        self.dut.RESET.value = 1
+        await ClockCycles(self.dut.CLK, 2)
+        self.dut.RESET.value = 0
+        await RisingEdge(self.dut.CLK)
 
-    while dut.RD_DATA_VLD[port].value != 1:
-        await re
 
-    return dut.RD_DATA[port].value
+async def write_read_test(dut, pkt_count=1000, item_width_min=1, item_width_max=16, parallel_write=False, parallel_read=False):
+    # Start clock generator
+    cocotb.start_soon(Clock(dut.CLK, 5, units='ns').start())
+    tb = testbench(dut)
+    await tb.reset()
+    await tb.stream_in.clear_memory()
 
+    addr_width = len(tb.stream_in.bus.WR_ADDR[0])
 
-def start_clock(dut):
-    # Set up the clock
-    clock = Clock(dut.CLK, 10, units="ns")
-    cocotb.start_soon(clock.start())
+    cocotb.log.info("-------------------------------------Write and immediate read---------------------------------")
 
+    for transaction in random_packets(item_width_min, item_width_max, pkt_count):
+        cocotb.log.info(f"generated transaction: {transaction.hex()}")
 
-async def setup(dut):
-    start_clock(dut)
-    for i in range(dut.READ_PORTS.value):
-        dut.RD_EN[i].value = 0
-    for i in range(dut.WRITE_PORTS.value):
-        dut.WR_EN[i].value = 0
-    await reset(dut)
+        address = randint(0, (2**addr_width-1)-(len(transaction)*8))
 
+        await tb.stream_in.write(address, transaction, parallel=parallel_write)
 
-@cocotb.test()
-async def test_simple(dut):
-    await setup(dut)
-    re = RisingEdge(dut.CLK)
-    # Write some data
-    dut.WR_EN[0].value = 1
-    dut.WR_ADDR[0].value = 0
-    dut.WR_DATA[0].value = 0xAA
-    await re
-
-    dut.WR_ADDR[0].value = 1
-    dut.WR_DATA[0].value = 0xBB
-    await RisingEdge(dut.CLK)
-
-    dut.WR_EN.value = 0
-    await RisingEdge(dut.CLK)
+        tb.ref_ram.write(address, transaction) # writting transaction to reference RAM
 
-    assert (await read(dut, 0, 0)) == 0xAA
-    assert (await read(dut, 1, 1)) == 0xBB
+        output = await tb.stream_in.read(address, len(transaction), parallel=parallel_read)
+        cocotb.log.info(f"received transaction:  {output.hex()}")
 
+        if output != transaction:
+            raise TestFailure(f"Expected {transaction.hex()}, got {output.hex()}")
 
-@cocotb.test()
-async def test_parallel_write(dut):
-    await setup(dut)
-    re = RisingEdge(dut.CLK)
-
-    # Write some data
-    dut.WR_EN[0].value = 1
-    dut.WR_ADDR[0].value = 42
-    dut.WR_DATA[0].value = 0xAA
+    cocotb.log.info("--------------------Comparing end state of reference RAM and simulated RAM--------------------")
 
-    dut.WR_EN[1].value = 1
-    dut.WR_ADDR[1].value = 43
-    dut.WR_DATA[1].value = 0xFF
-    await re
-    dut.WR_EN.value = 0
+    for i in range(2**addr_width):
+        ref_word = tb.ref_ram.read_word(i)
+        sim_word = await tb.stream_in.read_word(i, 0)
 
-    if not dut.ONE_CLK_WRITE.value:
-        await re
+        if ref_word != sim_word:
+            raise TestFailure(f"On address {i} reference RAM has {ref_word}, but simulated RAM has {sim_word}.")
 
-    assert (await read(dut, 0, 42)) == 0xAA
-    assert (await read(dut, 1, 43)) == 0xFF
+    raise TestSuccess()
 
 
 @cocotb.test()
-async def test_parallel_write_read(dut):
-    await setup(dut)
-    re = RisingEdge(dut.CLK)
+async def test_simple(dut):
+    await write_read_test(dut)
 
-    # Write some data
-    dut.WR_EN[0].value = 1
-    dut.WR_ADDR[0].value = 52
-    dut.WR_DATA[0].value = 0xAA
 
-    dut.WR_EN[1].value = 1
-    dut.WR_ADDR[1].value = 53
-    dut.WR_DATA[1].value = 0xFF
-    await re
-    dut.WR_EN.value = 0
+@cocotb.test()
+async def test_parallel_write(dut):
+    await write_read_test(dut, parallel_write=True)
 
-    if not dut.ONE_CLK_WRITE.value:
-        await re
 
-    dut.RD_ADDR[0] = 52
-    dut.RD_ADDR[1] = 53
-    dut.RD_EN = 3
-    await re
+@cocotb.test()
+async def test_parallel_read(dut):
+    await write_read_test(dut, parallel_read=True)
 
-    while dut.RD_DATA_VLD[0].value != 1:
-        await re
 
-    assert (dut.RD_DATA[0].value == 0xAA)
-    assert (dut.RD_DATA[1].value == 0xFF)
+@cocotb.test()
+async def test_parallel_write_read(dut):
+    await write_read_test(dut, parallel_write=True, parallel_read=True)

comp/base/mem/mp_bram/cocotb/prepare.sh

@@ -0,0 +1,15 @@
+#!/bin/sh
+# SPDX-License-Identifier: BSD-3-Clause
+# Copyright (C) 2025 CESNET z. s. p. o.
+# Author(s): Ondřej Schwarz <ondrejschwarz@cesnet.cz>
+
+NDK_FPGA_PATH=../../../../..
+source $NDK_FPGA_PATH/env.sh
+
+ndk_fpga_venv_prepare "venv-mp_bram"
+
+pip install .
+
+echo ""
+echo "Now activate environment with:"
+echo "source venv-mp_bram/bin/activate"

comp/base/mem/mp_bram/cocotb/pyproject.toml

@@ -0,0 +1,10 @@
+[project]
+name = "cocotb-hash-table-simple-test"
+version = "0.1.0"
+dependencies = [
+    "cocotbext-ofm @ ${NDK_FPGA_COCOTBEXT_OFM_URL}",
+]
+
+[build-system]
+requires = ["pdm-backend"]
+build-backend = "pdm.backend"