Add initial AXI bus stress tests

Author: robertszczepanski

verification/cocotb/block/axi_adapter/Makefile

@@ -16,6 +16,8 @@ MODULE      ?= $(subst $(space),$(comma),$(subst .py,,$(TEST_FILES)))
 TOPLEVEL     = axi_adapter_wrapper
 
 VERILOG_SOURCES  = \
+	$(CALIPTRA_ROOT)/src/caliptra_prim/rtl/caliptra_prim_pkg.sv \
+	$(CALIPTRA_ROOT)/src/caliptra_prim/rtl/caliptra_prim_util_pkg.sv \
     $(SRC_DIR)/i3c_pkg.sv \
     $(SRC_DIR)/libs/axi/axi_pkg.sv \
     $(SRC_DIR)/libs/axi/axi_if.sv \

verification/cocotb/block/axi_adapter/axi_adapter_wrapper.sv

@@ -58,7 +58,9 @@ module axi_adapter_wrapper
     output logic [           1:0] bresp,
     output logic [AxiIdWidth-1:0] bid,
     output logic                  bvalid,
-    input  logic                  bready
+    input  logic                  bready,
+
+    output logic [           6:0] fifo_depth_o
 );
   // I3C SW CSR access interface
   logic                    s_cpuif_req;
@@ -164,6 +166,35 @@ module axi_adapter_wrapper
       .hwif_out(hwif_out)
   );
 
+  logic fifo_wvalid;
+  logic fifo_wready;
+  logic [31:0] fifo_wdata;
+  logic fifo_rvalid;
+  logic fifo_rready;
+  logic [31:0] fifo_rdata;
+
+  logic unused_err, unused_full;
+
+  // FIFO for stress testing AXI
+  caliptra_prim_fifo_sync #(
+      .Width(32),
+      .Pass (1'b0),
+      .Depth(64)
+  ) fifo (
+      .clk_i(aclk),
+      .rst_ni(areset_n),
+      .clr_i('0),
+      .wvalid_i(fifo_wvalid),
+      .wready_o(fifo_wready),
+      .wdata_i(fifo_wdata),
+      .depth_o(fifo_depth_o),
+      .rvalid_o(fifo_rvalid),
+      .rready_i(fifo_rready),
+      .rdata_o(fifo_rdata),
+      .full_o(unused_full),
+      .err_o(unused_err)
+  );
+
   // TODO: These write-enable signals were not combo-driven or initialized on reset.
   // This is a placeholder driver. They require either unimplemented drivers or changes in RDL.
   always_comb begin : missing_csr_we_inits
@@ -214,13 +245,21 @@ module axi_adapter_wrapper
     hwif_in.I3C_EC.TTI.TX_DESC_QUEUE_PORT.wr_ack = 0;
     hwif_in.I3C_EC.TTI.TX_DATA_PORT.wr_ack = 0;
     hwif_in.I3C_EC.TTI.IBI_PORT.wr_ack = 0;
-    hwif_in.I3C_EC.SecFwRecoveryIf.INDIRECT_FIFO_DATA.rd_ack = '0;
-    hwif_in.I3C_EC.SecFwRecoveryIf.INDIRECT_FIFO_DATA.wr_ack = '0;
 
     // Unhandled wr/rd_ack (drivers are mising)
     hwif_in.DAT.rd_ack = 0;
     hwif_in.DAT.wr_ack = 0;
     hwif_in.DCT.rd_ack = 0;
     hwif_in.DCT.wr_ack = 0;
   end : other_uninit_signals
+
+  always_comb begin : connect_inidrect_fifo
+    fifo_wvalid = hwif_out.I3C_EC.SecFwRecoveryIf.INDIRECT_FIFO_DATA.req & hwif_out.I3C_EC.SecFwRecoveryIf.INDIRECT_FIFO_DATA.req_is_wr;
+    fifo_wdata = hwif_out.I3C_EC.SecFwRecoveryIf.INDIRECT_FIFO_DATA.wr_data;
+    hwif_in.I3C_EC.SecFwRecoveryIf.INDIRECT_FIFO_DATA.wr_ack = fifo_wvalid & fifo_wready;
+
+    fifo_rready = hwif_out.I3C_EC.SecFwRecoveryIf.INDIRECT_FIFO_DATA.req & ~hwif_out.I3C_EC.SecFwRecoveryIf.INDIRECT_FIFO_DATA.req_is_wr;
+    hwif_in.I3C_EC.SecFwRecoveryIf.INDIRECT_FIFO_DATA.rd_data = fifo_rdata;
+    hwif_in.I3C_EC.SecFwRecoveryIf.INDIRECT_FIFO_DATA.rd_ack = fifo_rvalid & fifo_rready;
+  end
 endmodule

verification/cocotb/block/axi_adapter/test_bus_stress.py

@@ -0,0 +1,162 @@
+# SPDX-License-Identifier: Apache-2.0
+
+import logging
+import random
+
+from bus2csr import (
+    FrontBusTestInterface,
+    compare_values,
+    dword2int,
+    get_frontend_bus_if,
+    int2bytes,
+    int2dword,
+)
+from utils import mask_bits, rand_bits, rand_bits32
+
+import cocotb
+from cocotb.handle import SimHandleBase
+from cocotb.triggers import ClockCycles, Combine, Event, RisingEdge, Timer, with_timeout
+from cocotb_helpers import reset_n
+
+
+async def timeout_task(timeout):
+    await Timer(timeout, "us")
+    raise RuntimeError("Test timeout!")
+
+
+async def initialize(dut, timeout=50):
+    """
+    Common test initialization routine which sets up environment and starts a timeout coroutine
+    to observe whether the test did not fall in infinite loop.
+    """
+
+    cocotb.log.setLevel(logging.DEBUG)
+
+    # Start the background timeout task
+    await cocotb.start(timeout_task(timeout))
+
+    # Initialize inputs
+    dut.araddr.value = 0
+    dut.arburst.value = 0
+    dut.arsize.value = 0
+    dut.arlen.value = 0
+    dut.aruser.value = 0
+    dut.arid.value = 0
+    dut.arlock.value = 0
+    dut.arvalid.value = 0
+    dut.rready.value = 0
+    dut.awaddr.value = 0
+    dut.awburst.value = 0
+    dut.awsize.value = 0
+    dut.awlen.value = 0
+    dut.awuser.value = 0
+    dut.awid.value = 0
+    dut.awlock.value = 0
+    dut.awvalid.value = 0
+    dut.wdata.value = 0
+    dut.wstrb.value = 0
+    dut.wlast.value = 0
+    dut.wvalid.value = 0
+    dut.bready.value = 0
+
+    # Configure testbench
+    tb = get_frontend_bus_if()(dut)
+    tb.log = dut._log
+    await tb.register_test_interfaces()
+    await ClockCycles(tb.clk, 20)
+    await reset_n(tb.clk, tb.rst_n, cycles=5)
+
+    data_len = random.randint(10, 100)
+    test_data = [random.randint(0, 2**32 - 1) for _ in range(data_len)]
+
+    return tb, data_len, test_data
+
+
+@cocotb.test()
+async def test_collision_with_write(dut):
+    tb, data_len, test_data = await initialize(dut)
+
+    fifo_addr = tb.reg_map.I3C_EC.SECFWRECOVERYIF.INDIRECT_FIFO_DATA.base_addr
+
+    tb.log.info(f"Generated {data_len} dwords to transfer.")
+
+    async def writer():
+        # Write sequence should just write data
+        for d in test_data:
+            await tb.write_csr(fifo_addr, int2bytes(d))
+
+    async def reader(return_data):
+        # Read sequence should read data on each write data
+        for _ in test_data:
+            # Awaiting `awvalid` causes reading simultaneously with write data channel activity
+            await RisingEdge(dut.awvalid)
+            return_data.append(dword2int(await tb.read_csr(fifo_addr)))
+
+    received_data = []
+    w = cocotb.start_soon(writer())
+    r = cocotb.start_soon(reader(received_data))
+
+    await Combine(w, r)
+
+    assert received_data == test_data, "Recieved data does not match sent data!"
+
+    tb.log.info("Test finished!")
+
+
+@cocotb.test(skip=True)
+async def test_collision_with_read(dut):
+    tb, data_len, test_data = await initialize(dut)
+
+    fifo_addr = tb.reg_map.I3C_EC.SECFWRECOVERYIF.INDIRECT_FIFO_DATA.base_addr
+
+    tb.log.info(f"Generated {data_len} dwords to transfer.")
+
+    async def writer():
+        # Write sequence should write data on each read data
+        for i, d in enumerate(test_data):
+            # Awaiting `arvalid` causes writing simultaneously with read data channel activity
+            if i >= 2:
+                await RisingEdge(dut.arvalid)
+            await tb.write_csr(fifo_addr, int2bytes(d))
+
+    async def reader(return_data):
+        # Wait until there is data in FIFO
+        while dut.fifo_depth_o.value < 2:
+            continue
+
+        # Read sequence should just read data
+        for i in range(data_len):
+            wvalid = RisingEdge(dut.wvalid)
+            wready = RisingEdge(dut.wready)
+            await Combine(wvalid, wready)
+            return_data.append(dword2int(await tb.read_csr(fifo_addr)))
+
+    received_data = []
+    w = cocotb.start_soon(writer())
+    r = cocotb.start_soon(reader(received_data))
+
+    await Combine(w, r)
+
+    assert received_data == test_data, "Recieved data does not match sent data!"
+
+    tb.log.info("Test finished!")
+
+
+@cocotb.test(skip=True)
+async def test_write_burst(dut):
+    tb = await initialize(dut)
+
+
+@cocotb.test(skip=True)
+async def test_read_burst(dut):
+    tb = await initialize(dut)
+
+
+@cocotb.test(skip=True)
+async def test_read_burst_collision_with_write(dut):
+    tb = await initialize(dut)
+
+
+@cocotb.test(skip=True)
+async def test_write_burst_collision_with_read(dut):
+    tb = await initialize(dut)

verification/cocotb/noxfile.py

@@ -100,6 +100,7 @@ def ahb_if_verify(session, test_group, test_name, coverage, simulator):
     "test_name",
     [
         "test_csr_sw_access",
+        "test_bus_stress",
     ],
 )
 @nox.parametrize("coverage", coverage_types)