Share the same software routiines to configure system in both cocotb simulation and actual hardware

[mvsoliveira]

I was wondering about the best way to reuse hardware configuration software routines for both cocotb simulation and actual hardware configuration and testing.

In my case, the system is configured exclusively through a memory-mapped interface (AXI-4).
The software routines to be shared are expressed as several AXI-4 read and write calls executed sequentially (no concurrence is needed). My first guess is to add as an argument to these software routines the AXIMaster object carrying the read and write methods.

Therefore, the cocotbext-axi AXIMaster object would be passed as an argument to these software routines for cocotb simulations. Similarly, an AXIMaster object carrying read and write methods pointing to /dev/mem would be passed to the software routines while running the software on the target hardware (Zynq Ultrascale+ MPSoC). This way the same software would be adopted while changing only the AXIMaster input argument.

My hesitation starts from the fact that the cocotb simulation is based on async coroutines invoked using the await keyword. On the other side, the on-the-target software is based on normal sequential Python calls.

I assume other users already did something similar. Would anyone like to share some tips or an example of how this can be done?

[mvsoliveira]

I am sharing a bit more details to see if the question becomes more clear.

Imagine one has a low-level software featuring thousands of lines using sequential coding, which is represented in a simplistic way in function shared_software.

def shared_software(interface_obj):
    # assert reset
    interface_obj.write(0x0,0x1)
    interface_obj.write(0x0,0x0)

The question is how one can import the same shared_software function to be used for both configuring the hardware (running in the target hardware, e.g. Zynq MPSoC) and also for driving the sequential part of a cocotb testbench (running in the simulator, e.g. Siemens Questa).

For configuring the hardware, the shared_software function currently receives as an argument an object that features a method called write that receives an address offset and data input as input arguments and writes the data to /dev/mem using the respective address offset. For the cocotb testbench, I had in mind a wrapper function around the write method of the cocotbext-axi AxiMaster class.

Please see how the cocotb testbench would look like:

@cocotb.coroutine
async def chip2chip_test(dut):
    # Starting clock and AXI-4 master
    cocotb.start_soon(Clock(dut.ref_clk, 3.571, units="ns").start(start_high=True))
    axi_master = AxiMaster(AxiBus.from_prefix(dut, "s_axi"), dut.s_aclk, dut.s_aresetn,
                                   reset_active_level=False, max_burst_len=1)
   # Creating a sequential wrapper for AXI master
   sequential_axi_master = some_wrapper_class(axi_master)
   # Calling the same low-level software that run in the embedded system
   shared_software(sequential_axi_master)

In essence, this low-level software is sequential, so no parallelism is needed other than making sure the other threads initiated before, e.g. clock, keep running during the simulation. I am aware one can do it using something like greenback. But given that this is a very generic problem, I was wondering if cocotb already envisages a way of doing this.

[ktbarrett]

greenback only supports asyncio and Trio, not cocotb. It cannot be used.

[marlonjames]

You should be able to accomplish this with cocotb.external and cocotb.function decorators.

result = await cocotb.external(shared_software(sequential_axi_master))

where any functions called in shared_software that need to "yield" time to the simulator are decorated with cocotb.function:

@cocotb.function
async def write(address, data):
    ...

A caveat with this approach:
cocotb.external runs the function in a separate thread each time it is called. If you need to run lots of external functions there will be overhead from the thread creation and management.

[ktbarrett]

Another answer is to use asyncio in standalone mode and make the calls into the interface use awaits. That's what we did at a previous employer.

[themperek]

Something like this maybe (for UART):

import serial
import asyncio
import cocotb

from cocotbext.uart import UartSource, UartSink

class UartInterface:
    def __init__(self, baud=115200):
        self.port = serial.Serial('/dev/ttyUSB0', baudrate=baud, timeout=1)

    async def write(self, address, data):
        cmd = b"\x10" + pack("u8u32", len(data), address) + bytearray(data)
        self.port.write(cmd)

    async def read(self, address, length):
        cmd = b"\x11" + pack("u8u32", length, address)
        self.port.write(cmd)
        ret = self.port.read(length)
        return array.array("B", ret)


class UartSimInterface:
    def __init__(self, dut, baud=115200):
        self.uart_source = UartSource(dut.uart_rxd, baud=baud, bits=8)
        self.uart_sink = UartSink(dut.uart_txd, baud=baud, bits=8)

    async def write(self, address, data):
        cmd = b"\x10" + pack("u8u32", len(data), address) + bytearray(data)
        await self.uart_source.write(cmd)
        await self.uart_source.wait()

    async def read(self, address, length):
        cmd = b"\x11" + pack("u8u32", length, address)
        await self.uart_source.write(cmd)
        await self.uart_source.wait()
        ret = b""
        for _ in range(length):
            w = await self.uart_sink.read(1)
            ret += w
        return array.array("B", ret)

@cocotb.test()
async def test_some(dut):
     
        if self._dut is not None:
            baud = 12500000
            intf = UartSimInterface(dut, baud=baud)
        else:
            baud = 115200
            intf = UartInterface(baud=baud)

        await intf.write(0, [0,1,2,3])
        await intf.read(0, 4)
        
 if __name__ == "__main__":       
     asyncio.run(test_some(dut=None))

[ktbarrett]

@themperek that's a pretty good example of what we did, but we used a special test decorator to wrap the test function in asyncio and just used pyest for regression management. Also we had several compat task management and synchronization primitives.

Ultimately though we went away from that since C tests at this level had more value.

[timothyscherer]

@ktbarrett Could you provide more information about how you wrapped the test function in asyncio? I believe I have done exactly this, and I have gotten it to work correctly when using cocotb. I am trying to also make these lab tests compatible with pyuvm, which depends on the cocotb executor (and thus, a simulator, which I would like to avoid). The uart example above requires a simulator to be running, even though it does not interact with the simulator at all.

So I have a proposition I would like to hear your thoughts about:
Could one write a "simulator" that uses one of the interfaces (VPI/VHPI/FLI) with its own event loop. This "simulator" would be used for lab testing only, so no actual simulator. And it would only be intelligent enough to spit out the real current time (gpi_get_sim_time) and allow real time to pass (gpi_iterate?). It would be on the testbench architect to swap out the implementation based on this special "simulator" being selected.

[jgovers12]

I was looking for a solution for an almost identical problem. After a lot of experimentation I came up with a solution.

The test code should not directly execute your sequential application code. Instead run this in a new thread. Furthermore the sequential thread should not directly change the dut. Instead convey requests/responses through queue's. See below for a simplified example:

import cocotb
import threading
import queue

req_q = queue.Queue()
rsp_q = queue.Queue()

async def sim_driver(dut):
    while True:
        await cocotb.triggers.FallingEdge(dut.s_axi_aclk)
        if not req_q.empty():
            req = req_q.get()
            # do something depending on req
            if req["op"] == "axi":
                dut.driver_axi_request.value = 1
                await cocotb.triggers.RisingEdge(dut.driver_axi_ready)
                dut.driver_axi_request.value = 0
                rsp_q.put(None)
            elif req["op"] == "stop":
                break


def app_task():
    for n in range(4):
        req_q.put({"op": "axi"})
        x = rsp_q.get()
    req_q.put({"op": "stop"})

@cocotb.test()
def system_test(dut):
    app_thread = threading.Thread(target=app_task)
    app_thread.start()

    drv_task = cocotb.start_soon(sim_driver(dut))

    yield drv_task.join()
    app_thread.join()

[lucasmossrfteq]

This is also an excellent solution.

[mvsoliveira]

That answers the question. I tested it and works perfectly. Many thanks @marlonjames and @ktbarrett !