用Python给Verilog设计自仿（五）：阻塞与队列，如何实现FPGA仿真高吞吐数据校验
扫描二维码
随时随地手机看文章
1前言
对于许多FPGA/IC工程师而言，设计实现游刃有余，验证仿真却常成短板——传统验证方法面临两难困局：学习UVM需投入大量时间成本，而纯Verilog自仿又会陷入重复造轮子的低效循环。以通信协议仿真为例，仅报文解析就需要重写整套解析逻辑，相当于用Verilog再实现一次协议栈，耗时费力。
此时，Python的生态优势便锋芒尽显。其丰富的字符串处理库可直接解析报文，配合Cocotb框架，仅需少量Python代码即可构建高效测试平台，将验证工作量压缩70%以上。Cocotb的独特价值正在于此：用Python解放验证生产力，让工程师专注于设计创新而非重复劳动。
2阻塞与非阻塞
在cocotb中阻塞操作需要使用await，而非阻塞队列操作则不需要，如果阻塞不使用await，协程则会一直暂停，直到条件满足，我们可以类比为await操作相当于Verilog中的阻塞赋值=，而不使用await则类似于Verilog中的非阻塞赋值<=。当我们在cocotb中执行阻塞队列操作时，协程会在执行到队列操作时暂停，直到特定条件被满足为止，这时需要使用await来使协程等待。例如，await queue.get()会阻塞当前协程，直到队列中有数据可供获取。而如果没有使用await，协程将会一直停留在这个位置，无法继续执行后续操作，直到条件满足。可以类比为Verilog中的阻塞赋值=，即当某个操作完成之前，后续的语句都无法继续执行。
//Verilog中的阻塞赋值 always @(*)begin b = a; c = b; end //Verilog中的非阻塞赋值 always @(posedge clk)begin b <= a; c <= b; end
#python 阻塞 asyncdefassign_testa: b = a asyncdefassign_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
3Cocotb-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。
4Queue有什么用
其实他类似于硬件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"
5写在最后
本文为原创Cocotb技术专栏，欢迎工程师伙伴们留言讨论或交流，共同学习。若想第一时间获取更新，可点击下方「关注」或订阅Cocotb专题。
下载文档
