用Python给Verilog设计自仿（五）：阻塞与队列，如何实现FPGA仿真高吞吐数据校验

前言

对于许多FPGA/IC工程师而言，设计实现游刃有余，验证仿真却常成短板——传统验证方法面临两难困局：学习UVM需投入大量时间成本，而纯Verilog自仿又会陷入重复造轮子的低效循环。以通信协议仿真为例，仅报文解析就需要重写整套解析逻辑，相当于用Verilog再实现一次协议栈，耗时费力。
此时，Python的生态优势便锋芒尽显。其丰富的字符串处理库可直接解析报文，配合Cocotb框架，仅需少量Python代码即可构建高效测试平台，将验证工作量压缩70%以上。Cocotb的独特价值正在于此：用Python解放验证生产力，让工程师专注于设计创新而非重复劳动。

阻塞与非阻塞

在cocotb中，有一个非常重要的概念，即“阻塞与非阻塞”。这个概念与Verilog中的阻塞和非阻塞赋值有着紧密的对应关系，但在实际应用中会有一些细微的区别。在Verilog中，阻塞赋值是通过=来进行的，这种赋值会立即更新变量的值，并且会阻止后续语句的执行，直到当前语句完成。而非阻塞赋值则通过<=来实现，赋值操作会被安排在当前时间步结束时进行，从而允许后续的语句在同一时间步内继续执行。
在cocotb中，阻塞与非阻塞的概念同样适用，但由于cocotb是基于Python和协程的，它采用了异步编程模型，因此它的阻塞和非阻塞语义与Verilog有所不同。具体来说，cocotb中的“阻塞”通常指代协程的暂停执行，等待某个事件发生（如等待某个信号的变化或者模拟时间的推进）。而“非阻塞”则意味着即使某个操作正在进行，协程依然可以执行其他操作，不会停顿等待。

在cocotb中阻塞操作需要使用await，而非阻塞队列操作则不需要，如果阻塞不使用await，协程则会一直暂停，直到条件满足，我们可以类比为await操作相当于Verilog中的阻塞赋值=，而不使用await则类似于Verilog中的非阻塞赋值<=。当我们在cocotb中执行阻塞队列操作时，协程会在执行到队列操作时暂停，直到特定条件被满足为止，这时需要使用await来使协程等待。例如，await queue.get()会阻塞当前协程，直到队列中有数据可供获取。而如果没有使用await，协程将会一直停留在这个位置，无法继续执行后续操作，直到条件满足。可以类比为Verilog中的阻塞赋值=，即当某个操作完成之前，后续的语句都无法继续执行。
另一方面，非阻塞队列操作则不会使协程暂停，它会立即返回一个结果或者执行一个异步操作，而不会等待条件的满足。例如，非阻塞的队列操作如queue.get_nowait()就不会使协程阻塞，它会立即返回一个结果，或者在没有数据时抛出异常，允许协程继续执行其他任务。这类似于Verilog中的非阻塞赋值<=，即它不会导致执行暂停，而是让后续操作可以并行进行。

//Verilog中的阻塞赋值
always @(*)begin
    b = a;
    c = b;
end

//Verilog中的非阻塞赋值
always @(posedge clk)begin
    b <= a;
    c <= b;
end

#python 阻塞
async def assign_testa:
    b = a

async def assign_testb:
    c = b

await assign_testa #等待testa执行完，再执行testb
await assign_testa 

#python 非阻塞
task1 = asyncio.create_task(assign_testa())  # 不会阻塞，马上开始 assign_testa
task2 = asyncio.create_task(assign_testb())  # 不会阻塞，马上开始 assign_testb

Cocotb-Queue

我们可以把Queue类比为硬件中的FIFO，也是先进先出的结构。
Cocotb 提供了几种队列类型，用于协同多个生产者和消费者协程的操作。它们是基于 asyncio 的Queue类，具有异步操作特性，并且可以帮助你在测试中同步不同的协程之间的通信。

class cocotb.queue.Queue(maxsize=0)

maxsize：指定队列的最大容量。如果 maxsize 小于或等于 0，则队列大小无限制。如果 maxsize 大于 0，当队列达到该大小时，put() 方法会阻塞，直到有空间。

我们在使用队列时，需要首先定义一个Queue类

from cocotb.queue import Queue
rx_queue = Queue()

主要方法

async put(item)
将 item 放入队列。
如果队列已满，则会等待直到有空间可用。

put_nowait(item)
将 item 放入队列，不会阻塞。
如果队列已满，立即抛出 QueueFull 异常。

async get()
从队列中移除并返回一个项。
如果队列为空，会等待直到有项可以取出。

get_nowait()
从队列中移除并返回一个项。
如果队列为空，立即抛出 QueueEmpty 异常。

qsize()
返回队列中当前项的数量。

empty()
如果队列为空，返回 True，否则返回 False。

full()
如果队列已满，返回 True，否则返回 False。

Queue有什么用

其实他类似于硬件FIFO中的功能

生产者（例如monitor）负责从硬件或仿真环境中接收数据，并将这些数据放入队列中。消费者（例如校验器）则从队列中取出数据进行处理。这个过程的关键在于队列的使用，它能够确保生产者和消费者之间不会直接依赖于对方的执行进度，从而避免了一个操作（如生产者）阻塞整个协程的情况。

如果没有队列，消费者需要直接等待生产者生成数据，这就意味着消费者的协程会一直被阻塞，直到生产者完成数据的生成。而使用队列后，即使生产者暂时没有数据可供消费，消费者仍然可以继续执行其他操作，只有在队列中有数据时，消费者才会从队列中取出并进行处理。这种解耦使得系统的执行更为灵活和高效。

在cocotb仿真过程中，如果要校验的数据数据量很大，建议使用Queue来完成校验数据的存取。
我们通过receiver_monitor来入队响应报文，通过data_validator来出队响应报文，并校验报文是否正确

async def receiver_monitor(dut: SimHandle, sink: AxiStreamSink) -> None:

    """异步接收协程"""
    while True:
        try:
            frame = await sink.recv()  # 阻塞式接收
            # rx_queue.put_nowait(frame)  # 非阻塞入队
            await rx_queue.put(frame)  # 阻塞入队
        except Exception as e:
            dut._log.error(f"queue error: {e}")
            raise

errors = []
async def data_validator(dut: SimHandle, frame_count: int) -None:
    result = 0
    while(result < frame_count):
        frame = await rx_queue.get()  # 阻塞式出队
        data = (int.from_bytes(frame.tdata, byteorder='little'))
        if data != result:
            errors.append(f"CHECK ERROR, got {data},the result is {hex(result)}")
            dut._log.error(f"verilate fail :{data},the result is {hex(result)}")
        else:
            dut._log.info(f"CHECK PASS receive data is :{data}")
        result = result + 1

@cocotb.test()
async def axis_simple_test(dut: SimHandle):
    ......
    assert len(errors) == 0,"TEST FAIL"

写在最后

本文为原创Cocotb技术专栏，欢迎工程师伙伴们留言讨论或交流，共同学习。若想第一时间获取更新，可点击下方「关注」或订阅Cocotb专题。

本文由本账号所属公众号提供，欢迎关注微信公众号AdirtCoreFpga，获取第一时间更新