Cover x_QUEUE_DEPTH and x_QUEUE_FULL CSRs

Author: mszalkowski-ant

verification/cocotb/top/lib_i3c_top/test_bus_stall.py

@@ -65,35 +65,68 @@ async def test_empty_indirect_fifo_read(dut):
 @cocotb.test()
 async def test_full_tx_desc_write(dut):
     tb = await initialize(dut)
-    for _ in range(TRANSACTION_COUNT):
+    QSIZE = tb.reg_map.I3C_EC.TTI.QUEUE_SIZE
+    QDEPTH = tb.reg_map.I3C_EC.TTI.DESC_QUEUE_DEPTH
+    QSTAT = tb.reg_map.I3C_EC.TTI.QUEUE_STATUS
+    MAX_DEPTH = 2 ** (await tb.read_csr_field(QSIZE.base_addr, QSIZE.TX_DESC_BUFFER_SIZE) + 1)
+
+    for i in range(TRANSACTION_COUNT):
         await tb.write_csr(tb.reg_map.I3C_EC.TTI.TX_DESC_QUEUE_PORT.base_addr, int2dword(random.randint(0, 0xffffffff)), 4)
+        if i < MAX_DEPTH:
+            depth = await tb.read_csr_field(QDEPTH.base_addr, QDEPTH.TX_DESC_QUEUE_DEPTH)
+            assert i + 1 == depth, "Transaction did not increase queue depth"
+
     # Bus must not stall — verify it's still responsive
     data = dword2int(await tb.read_csr(tb.reg_map.I3CBASE.HCI_VERSION.base_addr, 4))
     assert data == 0x120, f"Bus stalled: HCI_VERSION read returned 0x{data:X} after {TRANSACTION_COUNT} TX_DESC writes"
 
+    assert await tb.read_csr_field(QSTAT.base_addr, QSTAT.TX_DESC_QUEUE_FULL) == 1, "Queue is not full"
+
     await reset_n(tb.clk, tb.rst_n, cycles=5)
     await tb.teardown()
 
 @cocotb.test()
 async def test_full_tx_data_write(dut):
     tb = await initialize(dut)
-    for _ in range(TRANSACTION_COUNT):
+    QSIZE = tb.reg_map.I3C_EC.TTI.QUEUE_SIZE
+    QDEPTH = tb.reg_map.I3C_EC.TTI.DATA_QUEUE_DEPTH
+    QSTAT = tb.reg_map.I3C_EC.TTI.QUEUE_STATUS
+    MAX_DEPTH = 2 ** (await tb.read_csr_field(QSIZE.base_addr, QSIZE.TX_DATA_BUFFER_SIZE) + 1)
+
+    for i in range(TRANSACTION_COUNT):
         await tb.write_csr(tb.reg_map.I3C_EC.TTI.TX_DATA_PORT.base_addr, int2dword(random.randint(0, 0xffffffff)), 4)
+        if i <= MAX_DEPTH+1 and i > 0:
+            depth = await tb.read_csr_field(QDEPTH.base_addr, QDEPTH.TX_DATA_QUEUE_DEPTH)
+            assert i == depth + 1, "Transaction did not increase queue depth"
+
     # Bus must not stall — verify it's still responsive
     data = dword2int(await tb.read_csr(tb.reg_map.I3CBASE.HCI_VERSION.base_addr, 4))
     assert data == 0x120, f"Bus stalled: HCI_VERSION read returned 0x{data:X} after {TRANSACTION_COUNT} TX_DATA writes"
 
+    assert await tb.read_csr_field(QSTAT.base_addr, QSTAT.TX_DATA_QUEUE_FULL) == 1, "Queue is not full"
+
     await reset_n(tb.clk, tb.rst_n, cycles=5)
     await tb.teardown()
 
 @cocotb.test()
 async def test_full_ibi_write(dut):
     tb = await initialize(dut)
-    for _ in range(TRANSACTION_COUNT):
+    QSIZE = tb.reg_map.I3C_EC.TTI.IBI_QUEUE_SIZE
+    QDEPTH = tb.reg_map.I3C_EC.TTI.IBI_QUEUE_DEPTH
+    QSTAT = tb.reg_map.I3C_EC.TTI.QUEUE_STATUS
+    MAX_DEPTH = 2 ** (await tb.read_csr_field(QSIZE.base_addr, QSIZE.IBI_QUEUE_SIZE) + 1)
+
+    for i in range(TRANSACTION_COUNT):
         await tb.write_csr(tb.reg_map.I3C_EC.TTI.IBI_PORT.base_addr, int2dword(random.randint(0, 0xffffffff)), 4)
+        if i < MAX_DEPTH:
+            depth = await tb.read_csr_field(QDEPTH.base_addr, QDEPTH.IBI_QUEUE_DEPTH)
+            assert i + 1 == depth, "Transaction did not increase queue depth"
+
     # Bus must not stall — verify it's still responsive
     data = dword2int(await tb.read_csr(tb.reg_map.I3CBASE.HCI_VERSION.base_addr, 4))
     assert data == 0x120, f"Bus stalled: HCI_VERSION read returned 0x{data:X} after {TRANSACTION_COUNT} IBI writes"
 
+    assert await tb.read_csr_field(QSTAT.base_addr, QSTAT.IBI_QUEUE_FULL) == 1, "Queue is not full"
+
     await reset_n(tb.clk, tb.rst_n, cycles=5)
     await tb.teardown()

verification/cocotb/top/lib_i3c_top/test_i3c_target.py

@@ -90,6 +90,11 @@ async def test_setup(dut, fclk=333.0, fbus=12.5,
 
 # Receiver agent (firmware side)
 async def rx_agent(tb, data_transfers):
+    QSIZE = tb.reg_map.I3C_EC.TTI.QUEUE_SIZE
+    QDEPTH = tb.reg_map.I3C_EC.TTI.DATA_QUEUE_DEPTH
+    QSTAT = tb.reg_map.I3C_EC.TTI.QUEUE_STATUS
+    MAX_DEPTH = 2 ** (await tb.read_csr_field(QSIZE.base_addr, QSIZE.RX_DATA_BUFFER_SIZE) + 1)
+
     recv_data = []
 
     # Enable RX descriptor interrupt
@@ -134,8 +139,14 @@ async def rx_agent(tb, data_transfers):
 
         # Read RX data
         data_len = ceil(desc_len / 4)
+        assert data_len <= MAX_DEPTH
+        if data_len == MAX_DEPTH:
+            assert await tb.read_csr_field(QSTAT.base_addr, QSTAT.RX_DATA_QUEUE_FULL) == 1, "Queue is not full, but it should be"
+
         rx_data = []
-        for _ in range(data_len):
+        for i in range(data_len):
+            depth = await tb.read_csr_field(QDEPTH.base_addr, QDEPTH.RX_DATA_QUEUE_DEPTH)
+            assert depth == data_len - i, f"Unexpected queue depth: {depth}"
             data = dword2int(await tb.read_csr(tb.reg_map.I3C_EC.TTI.RX_DATA_PORT.base_addr, 4))
             for k in range(4):
                 rx_data.append((data >> (k * 8)) & 0xFF)

verification/cocotb/top/lib_i3c_top/test_interrupts.py

@@ -340,17 +340,23 @@ async def test_rx_desc_overflow(dut):
 
     # Setup
     i3c_controller, _, tb = await test_setup(dut, timeout_us=500)
+    QSIZE = tb.reg_map.I3C_EC.TTI.QUEUE_SIZE
+    QDEPTH = tb.reg_map.I3C_EC.TTI.DESC_QUEUE_DEPTH
+    QSTAT = tb.reg_map.I3C_EC.TTI.QUEUE_STATUS
+    MAX_DEPTH = 2 ** (await tb.read_csr_field(QSIZE.base_addr, QSIZE.RX_DESC_BUFFER_SIZE) + 1)
 
-    for _ in range(65):
+    for i in range(MAX_DEPTH):
         data = [random.randint(0, 255) for i in range(4)]
         await i3c_controller.i3c_write(TARGET_ADDRESS, data)
+        depth = await tb.read_csr_field(QDEPTH.base_addr, QDEPTH.RX_DESC_QUEUE_DEPTH)
+        assert i + 1 == depth, "Transaction did not increase queue depth"
 
-    assert tb.dut.xi3c_wrapper.i3c.tti_rx_desc_full.value == 1
+    assert await tb.read_csr_field(QSTAT.base_addr, QSTAT.RX_DESC_QUEUE_FULL) == 1, "Queue is not full, but it should be"
 
     await tb.read_csr(tb.reg_map.I3C_EC.TTI.RX_DESC_QUEUE_PORT.base_addr, 4)
 
     await ClockCycles(tb.clk, 50)
 
-    assert tb.dut.xi3c_wrapper.i3c.tti_rx_desc_full.value == 0
+    assert await tb.read_csr_field(QSTAT.base_addr, QSTAT.RX_DESC_QUEUE_FULL) == 0, "Queue is full, but it should not be"
 
     await tb.teardown()