umi2apb bug fix

Author: RiceShelley

pyproject.toml

@@ -37,7 +37,7 @@ test = [
     "cocotb-bus==0.3.0",
     "cocotbext-axi==0.1.28",
     "cocotbext-umi==0.0.3",
-    "cocotbext-apb==0.7.4"
+    "cocotbext-apb==0.11.0"
 ]
 
 [tool.check-wheel-contents]

tests/adapters/umi2apb/__init__.py

@@ -2,23 +2,40 @@
 # and provides common functionality for the tests.
 
 from cocotb.clock import Clock
-from cocotb.triggers import ClockCycles
+from cocotb.triggers import ClockCycles, Timer
 
 from cocotb_bus.scoreboard import Scoreboard
 from cocotbext.apb import ApbBus, ApbSlave, MemoryRegion
 
-from cocotblib.umi.drivers.sumi_driver import SumiDriver
-from cocotblib.umi.monitors.sumi_monitor import SumiMonitor
-from cocotblib.umi.sumi import SumiTransaction, SumiCmdType, SumiCmd
-from cocotblib.common import drive_reset
+from cocotbext.umi.drivers.sumi_driver import SumiDriver
+from cocotbext.umi.monitors.sumi_monitor import SumiMonitor
+from cocotbext.umi.sumi import SumiTransaction, SumiCmdType, SumiCmd
+
+
+async def do_reset(reset, time_ns: int, active_level: bool = False):
+    """Drive an asynchronous reset pulse on *reset*."""
+    reset.value = not active_level
+    await Timer(1, unit="step")
+    reset.value = active_level
+    await Timer(time_ns, "ns")
+    reset.value = not active_level
+    await Timer(1, unit="step")
 
 
 # Creates the umi2apb test environment
 class UMI2APBEnv:
-    def __init__(self, dut, clk_period_ns=10, mem_size=2**16):
+    def __init__(
+        self,
+        dut,
+        clk_period_ns=10,
+        mem_size=2**16,
+        umi_valid_gen=None,
+        backpressure_on_apb=True
+    ):
         self.dut = dut
         self.clk_period_ns = clk_period_ns
         self.mem_size = mem_size
+        self.umi_valid_gen = umi_valid_gen
 
         self.data_width = int(dut.RW.value)  # default of 64
         self.addr_width = int(dut.AW.value)  # default of 64
@@ -29,22 +46,20 @@ def __init__(self, dut, clk_period_ns=10, mem_size=2**16):
         self.clk = dut.apb_pclk
         self.nreset = dut.apb_nreset
 
-        self._build()
-
-    def _build(self):
-        dut = self.dut
-
         # Instantiates UMI driver
         self.sumi_driver = SumiDriver(
             entity=dut,
             name="udev_req",
             clock=self.clk,
-            bus_separator="_"
+            bus_separator="_",
+            valid_generator=self.umi_valid_gen,
         )
 
         # Instantiates APB slave and memory region
         apb_bus = ApbBus.from_prefix(dut, "apb")
         self.apb_slave = ApbSlave(apb_bus, self.clk, self.nreset)
+        if backpressure_on_apb:
+            self.apb_slave.enable_backpressure()
         self.region = MemoryRegion(self.mem_size)
         self.apb_slave.target = self.region
 
@@ -62,21 +77,9 @@ def _build(self):
 
     # Prerequisites for starting tests
     async def start(self):
+        await do_reset(self.nreset, self.clk_period_ns)
         Clock(self.clk, self.clk_period_ns, unit="ns").start()
-        await drive_reset(self.nreset, self.clk_period_ns)
-        self.dut.udev_resp_ready.value = 1
-
-    # Waits for umi responses
-    async def wait_for_responses(self, max_cycles):
-        cycles = 0
-        while self.expected_responses:
-            await ClockCycles(self.clk, 1)
-            cycles += 1
-            if cycles > max_cycles:
-                raise TimeoutError(
-                    f"Timeout waiting for responses "
-                    f"({len(self.expected_responses)} remaining)"
-                )
+        await ClockCycles(self.clk, 10)
 
 
 # Creates an ideal umi write response

tests/adapters/umi2apb/env.py

@@ -4,11 +4,11 @@
 from cocotb.handle import SimHandleBase
 from cocotb.triggers import ClockCycles
 
-from cocotblib.umi.sumi import SumiTransaction, SumiCmdType, SumiCmd
-from adapters.umi2apb.env import UMI2APBEnv, create_expected_write_response
+from cocotbext.umi.sumi import SumiTransaction, SumiCmdType, SumiCmd
+from env import UMI2APBEnv, create_expected_write_response
 
 
-@cocotb.test(timeout_time=50, timeout_unit="ms")
+@cocotb.test(timeout_time=1, timeout_unit="ms")
 async def test_backpressure(dut: SimHandleBase):
     """
     Test response backpressure handling:
@@ -54,7 +54,7 @@ async def test_backpressure(dut: SimHandleBase):
     env.sumi_driver.append(write_txn)
     print(f"Sent write: addr=0x{test_addr:x}, data=0x{test_data:08x}")
 
-    await ClockCycles(env.clk, 20)
+    await ClockCycles(env.clk, 200)
 
     # Verify response is being held
     assert dut.udev_resp_valid.value == 1, "Response should be valid"
@@ -65,7 +65,7 @@ async def test_backpressure(dut: SimHandleBase):
     dut.udev_resp_ready.value = 1
     print("Re-enabled udev_resp_ready")
 
-    await env.wait_for_responses(max_cycles=10)
+    await ClockCycles(env.clk, 100)
 
     # Verify mem was written correctly
     mem_data = await env.region.read(test_addr, env.data_size)

tests/adapters/umi2apb/test_backpressure.py

@@ -2,9 +2,12 @@
 import cocotb
 
 from cocotb.handle import SimHandleBase
+from cocotb.triggers import ClockCycles
+from cocotb_bus.drivers import BitDriver
 
-from cocotblib.umi.sumi import SumiTransaction, SumiCmdType, SumiCmd
-from adapters.umi2apb.env import UMI2APBEnv, create_expected_write_response
+from cocotbext.umi.sumi import SumiTransaction, SumiCmdType, SumiCmd
+from cocotbext.umi.utils.generators import random_toggle_generator
+from env import UMI2APBEnv, create_expected_write_response
 
 
 @cocotb.test(timeout_time=50, timeout_unit="ms")
@@ -21,6 +24,10 @@ async def test_basic_WR(dut: SimHandleBase):
     env = UMI2APBEnv(dut)
     await env.start()
 
+    BitDriver(signal=dut.udev_resp_ready, clk=env.clk).start(
+        generator=random_toggle_generator()
+    )
+
     umi_size = int(math.log2(env.data_size))
     test_addr = 0x100
     test_data = 0xDEADBEEF
@@ -47,10 +54,8 @@ async def test_basic_WR(dut: SimHandleBase):
         )
     )
 
-    env.sumi_driver.append(write_txn)
-
-    # Wait for write response
-    await env.wait_for_responses(max_cycles=100)
+    await env.sumi_driver.send(write_txn)
+    await ClockCycles(env.clk, 20)
 
     # Verify APB memory contents
     mem_data = await env.region.read(test_addr, env.data_size)
@@ -89,10 +94,13 @@ async def test_basic_WR(dut: SimHandleBase):
     )
 
     env.expected_responses.append(expected_read_resp)
-    env.sumi_driver.append(read_txn)
+    await env.sumi_driver.send(read_txn)
+
+    # Wait for scoreboard to consume all expected outputs
+    while len(env.expected_responses) != 0:
+        await ClockCycles(env.clk, 1)
 
-    # Wait for read response
-    await env.wait_for_responses(max_cycles=100)
+    await ClockCycles(env.clk, 10)
 
     print("Read response verified by scoreboard")
 

tests/adapters/umi2apb/test_basic_WR.py

@@ -1,8 +1,10 @@
 import math
 import cocotb
 
-from adapters.umi2apb.env import UMI2APBEnv
-from cocotblib.umi.sumi import SumiTransaction, SumiCmdType, SumiCmd
+from cocotb.triggers import Event, Combine, ClockCycles
+
+from env import UMI2APBEnv
+from cocotbext.umi.sumi import SumiTransaction, SumiCmdType, SumiCmd
 
 
 @cocotb.test(timeout_time=50, timeout_unit="ms")
@@ -16,6 +18,9 @@ async def test_full_throughput(dut):
     env = UMI2APBEnv(dut)
     await env.start()
 
+    # Always ready to accept responses
+    dut.udev_resp_ready.value = 1
+
     data_size = env.data_size
     addr_width = env.addr_width
     umi_size = int(math.log2(data_size))
@@ -24,6 +29,7 @@ async def test_full_throughput(dut):
 
     print("=== Back-to-Back Full Throughput Test ===")
 
+    send_events = []
     for i in range(num_transactions):
         txn_size = i % (umi_size + 1)
         txn_bytes = 1 << txn_size
@@ -81,10 +87,15 @@ async def test_full_throughput(dut):
             )
 
         env.expected_responses.append(expected_resp)
-        env.sumi_driver.append(txn)
+        evt = Event()
+        env.sumi_driver.append(txn, event=evt)
+        send_events.append(evt)
 
     # Wait for all responses
-    await env.wait_for_responses(max_cycles=num_transactions * 50)
+    await Combine(*(e.wait() for e in send_events))
+
+    await ClockCycles(env.clk, 100)
 
     print(f" All {num_transactions} back-to-back transactions completed successfully!")
+
     raise env.scoreboard.result

tests/adapters/umi2apb/test_full_throughput.py

@@ -2,10 +2,12 @@
 import cocotb
 
 from cocotb.handle import SimHandleBase
+from cocotb_bus.drivers import BitDriver
 from cocotb.triggers import ClockCycles
 
-from cocotblib.umi.sumi import SumiTransaction, SumiCmdType, SumiCmd
-from adapters.umi2apb.env import UMI2APBEnv
+from cocotbext.umi.sumi import SumiTransaction, SumiCmdType, SumiCmd
+from cocotbext.umi.utils.generators import random_toggle_generator
+from env import UMI2APBEnv
 
 
 @cocotb.test(timeout_time=50, timeout_unit="ms")
@@ -19,6 +21,10 @@ async def test_posted_write(dut: SimHandleBase):
     env = UMI2APBEnv(dut)
     await env.start()
 
+    BitDriver(signal=dut.udev_resp_ready, clk=env.clk).start(
+        generator=random_toggle_generator()
+    )
+
     umi_size = int(math.log2(env.data_size))
 
     print("=== Posted Write Test ===")