This post is about one of the important and interesting problems while building verification environments: When to stop the simulation.

UVM provides some ways to control test termination (ie objections and drain time). With cocotb, I like to explore the idea of idle state of a cocotb component by keeping track of events and how long since the last active event.

The UVM Way Link to heading

In UVM, objections are used to keep track of the running process. If a process(ie run_phase) raises an objection, the simulation doesn’t finish until it’s dropped.

phase.raise_objection(this);

phase.drop_objection(this);

Another important point here is drain time, which is needed to keep the simulation running after dropping all objections, to allow in-flight transactions to finish.

Cocotb Way Link to heading

In cocotb, the test finishes if the test (decorated with @cocotb.test) finishes (nothing in the event loop). So, I recently started doing what I call test termination with idle detection for all components I write.

After driving stimulus through a driver (or whatever coroutine), I can await on idle (or a combination of several coroutines) with some soak time even after the coroutines go idle.

@cocotb.test()
async def test_component_structure(dut):

    # clock Generation
    clock = Clock(dut.clk, 10, unit="ns")
    cocotb.start_soon(clock.start())

    # Create Model, reset and start
    model = Model(dut)

    model.reset()
    model.start()

    # Drive some stimuli
    # ...

    # wait for idle
    await model.wait_for_idle(idle_after=50, unit="ns")

    # Cleanup
    model.stop()
    model.reset()

    cocotb.log.info("Test completed")

Let’s break down the important methods here. wait_for_idle waits for the idle_after time from the last component activity.

    async def wait_for_idle(self, idle_after=100, unit="ns", poll_step=10):
        if not self._running:
            return True

        while True:
            now = cocotb.utils.get_sim_time(unit=unit)
            if self._last_activity_time is None:
                await Timer(poll_step, unit=unit)
                cocotb.log.info("No activity detected yet")
                continue
            if (now - self._last_activity_time) >= idle_after:
                cocotb.log.info(f"Idle detected after {now - self._last_activity_time} {unit}")
                return True
            cocotb.log.info(f"Not idle yet. Last activity was {now - self._last_activity_time} {unit} ago")
            await Timer(poll_step, unit=unit)

This means we have to mark the activity. For this, I am using _mark_activity.

    # Coroutines
    async def _run(self):
        while self._running:
            await Timer(20, unit="ns")

            if self._last_activity_time == 0:
                # Mark activity at 30 ns
                await Timer(10, unit="ns")
                cocotb.log.info("1_mark_activity called in _run")
                self._mark_activity()

                # Mark activity at 70 ns
                await Timer(40, unit="ns")
                cocotb.log.info("2_mark_activity called in _run")
                self._mark_activity()

                # More than the drain time, so test will finish before hitting this.
                await Timer(60, unit="ns")
                cocotb.log.info("3_mark_activity called in _run")
                self._mark_activity()

            # More delay... Simulate some processing time. Will timeout before comes here
            await Timer(1000, unit="ns") # Simulate some processing time

_mark_activity is a simple method to get the current simtime and update _mark_activity.

    # Internal methods
    def _mark_activity(self, unit="ns"):
        self._mark_activity = cocotb.utils.get_sim_time(unit=unit)

Minor detail: the start of the component should be marked as activity to start keeping track from the first time the component coroutines start.

    def start(self):
        self._running = True
        self._mark_activity()
        self._monitor_task  = cocotb.start_soon(self._monitor())
        self._run_task      = cocotb.start_soon(self._run())