FPGA/IC验证进阶：使用Cocotb构建面向对象的TestBench

对于许多 FPGA/IC 工程师而言，设计实现游刃有余，但自仿与验证往往成为短板：要么投入大量时间学习 UVM，要么在 Verilog 中反复造轮子，效率低下。
Python + Cocotb 提供了一条更适合硬件设计者的自仿路径——借助 Python 强大的数据处理能力，结合 Cocotb 框架，仅需少量代码即可构建高效、可扩展的自仿平台，大幅压缩仿真成本。
Cocotb 的核心价值在于：用 Python 解放验证生产力，让工程师专注于设计本身。

很多朋友在使用 Cocotb 做仿真时，常常感到无从下手，其根本原因往往并不在语法本身，而是 缺乏面向对象的思维。

在传统硬件验证体系中，SystemVerilog UVM 依靠面向对象与组件化的方法，构建了大规模、可复用的测试平台；而 Python 的 Cocotb 虽然轻量灵活，但很多人在使用时往往仅停留在写几个 await RisingEdge() 的脚本阶段。

实际上，Cocotb 完全可以用来构建类似 UVM 的面向对象 TestBench。
此时可能有人会疑问：为什么一定要面向对象？我直接在 @cocotb.test 里写测试逻辑不就行了吗？

接下来，我们将通过具体案例展示：面向对象的 Cocotb 不仅写法更优雅、更符合工程化思维，还能让仿真流程更可控，同时更易于扩展和复用。

什么是面向对象

可能有不少写 RTL 的工程师，一听到“面向对象”就会下意识地心生抗拒，觉得那是软件工程里的高级概念——对象、封装、继承、多态、抽象，这些听起来就很复杂，似乎和硬件设计关系不大。

但实际上，面向对象并不是某种晦涩难懂的语法。
对于硬件设计者来说，它更像是一种仿真系统的设计方法。

我们不妨用一个非常直观的例子来看。

用人来理解什么是对象

假设我们要在仿真中描述“人”，那么可以先定义一个“人”这个概念（类）：

class 人:
    pass

仅仅有概念还不够，一个人通常需要一些属性来描述，比如性别、年龄：

class 人:
    def __init__(self, 性别: int, 年龄: int):
        self.性别 = 性别
        self.年龄 = 年龄

除了属性，人还需要具备一些行为能力，比如吃饭、喝水，这些就是对象的方法：

class 人:
    def __init__(self, 性别: int, 年龄: int):
        self.性别 = 性别
        self.年龄 = 年龄

    async def 吃饭(self):
        ...

    async def 喝水(self):
        ...

创建对象并驱动行为

有了“人”这个类之后，我们就可以创建具体的对象：

张三 = 人(性别=男, 年龄=16)
李四 = 人(性别=女, 年龄=24)

想让张三去喝水：

cocotb.start_soon(张三.喝水())

想让李四一边吃饭一边喝水（两个并发行为）：

cocotb.start_soon(李四.喝水())
cocotb.start_soon(李四.吃饭())

面向对象，其实就这么简单

你会发现，这里并没有什么复杂的高级概念：

• 类：对某一类事物的抽象描述
• 对象：具体的实例
• 属性：对象携带的状态
• 方法：对象能够执行的行为

这套思维方式，本质上和我们在 RTL 里定义模块、信号、行为并没有太大区别。

那么问题来了：
既然人可以作为一个对象，那么我们的 TestBench，是否也可以作为一个对象来组织和管理呢？

仿真的两大类别

在仿真场景中，我们可以借鉴 PCIe 的定义方式，将数据流大致分为两类：Posted 类仿真和 Non-Posted 类仿真。这里的含义与 PCIe 协议中的 posted / non-posted 请求是一致的。

Posted 类仿真，指的是没有交互返回的单向数据流。
例如：

• UDP 模块
• AXIS-UART 模块
• PCIe Memory Write 模块

这类模块的典型特征是：数据直进直出。一旦数据被发送出去，发送端不需要等待对端的任何响应或确认，仿真逻辑相对简单，行为也更线性。

而 Non-Posted 类仿真 则属于存在交互和反馈的数据流。
例如：

• TCP 模块
• PCIe Memory Read 模块

在这类场景中，仿真通常表现为：先发出一个请求数据流，随后还需要接收对应的响应、返回数据或握手信号。请求与响应之间往往存在时序关系、乱序、等待、超时等复杂情况。

从仿真编写的难度来看：
Non-Posted 类仿真通常比 Posted 类仿真要复杂得多。

这是因为 Non-Posted 场景不仅需要驱动数据，还需要管理请求与响应之间的关联关系、状态机以及异常处理，而这些恰恰是面向对象仿真最能发挥优势的地方。

面向TestBench仿真

从Posted仿真开始

对于Posted仿真，Cocotb的编写可以非常简洁。以AXIS-SLICE模块为例，它的作用主要是将AXI-Stream的数据打一拍送出去。

module afx_axis_slice #(
    parameter DATA_W = 32,
    parameter USER_W = 2,
    // local param
    parameter KEEP_W = DATA_W/8
)(
    input  logic                     clk,
    input  logic                     rst_n,
    input  logic [DATA_W-1:0]        s_axis_tdata,
    input  logic [KEEP_W-1:0]        s_axis_tkeep,
    input  logic [USER_W-1:0]        s_axis_tuser,
    input  logic                     s_axis_tvalid,
    output logic                     s_axis_tready,
    output logic [DATA_W-1:0]        m_axis_tdata,
    output logic [KEEP_W-1:0]        m_axis_tkeep,
    output logic [USER_W-1:0]        m_axis_tuser,
    output logic                     m_axis_tvalid,
    input  logic                     m_axis_tready
);

我们仅需要一个TestBench对象：

class AXISSlice_TestBench:
    pass

TB的属性

那么，作为一个 TestBench 对象，它本身需要具备哪些属性？

首当其冲的，必然是DUT 对象以及与之交互的接口。无论多么复杂的仿真，最终都绕不开与 RTL 的信号交互。

其次，是模块相关的参数信息，例如 DATA_W、USER_W 等。这些参数决定了数据的格式、宽度以及行为边界。
接口与参数共同构成了TestBench 与 RTL 交互的基础对象。

除此之外，还有一类非常重要但常被忽略的内容：TestBench 自身的控制属性。
例如：

• 用于控制随机行为和压力分布的随机数种子（seed）
• 一些希望在不同测试场景下灵活配置的开关或阈值

这些属性并不直接对应 RTL 信号，但它们决定了仿真的“性格”和行为模式。

通过将这些内容统一纳入 TestBench 的属性中，我们才能真正构建出一个工程化、可复用、可演进的仿真框架。

class AXISSlice_TestBench:
    def __init__(self, dut, seed=0, data_w=256, user_w=2):
        # dut对象
        self.dut = dut
        # RTL参数
        self.dw = data_w
        self.uw = user_w
        # 随机数种子
        self.rng = random.Random(seed)
        # slave接口
        self.axis_source = AxiStreamSource(
            bus=AxiStreamBus.from_prefix(dut, "s_axis"),
            clock=dut.clk,
            reset=dut.rst_n,
            reset_active_level=False
        )
        # master接口
        self.axis_sink = AxiStreamSink(
            bus=AxiStreamBus.from_prefix(dut, "m_axis"),
            clock=dut.clk,
            reset=dut.rst_n,
            reset_active_level=False
        )

TB的方法

有了属性之后，接下来便是TestBench 对象的方法设计。
那么，一个 TestBench 究竟需要具备哪些能力？

首先，最基础、也是最核心的能力，是复位与初始化，以及发送数据和接收数据的能力。

其次，是随机化激励的生成，以及对应期望比对数据的构建能力。通过对数据内容、地址、长度乃至时序进行随机化，TestBench 可以更高效地覆盖各种边界条件和异常路径，从而显著提升验证的覆盖深度与效率。这在算法与数据结构的随机化测试思想中也是相通的。

再者，TestBench 还需要具备数据比对与结果校验的能力。无论是通过 scoreboard 将期望值与实际值进行对比，还是通过断言检测非法行为，验证的最终目标始终是判断 DUT 的功能和时序是否符合预期。

最后，如果仿真对象是一个AXI-Stream 类系统，TestBench 往往还需要提供压力控制（Backpressure Control）能力。例如调节 tvalid/tready 的概率分布、人为引入暂停与拥塞，模拟真实系统中的流控行为。

当这些能力被合理地封装为 TestBench 的方法之后，仿真代码就不再是一堆零散的脚本，而是一个具备完整行为模型、可配置、可复用的验证对象。

class AXISSlice_TestBench:
    ...
    # 复位能力
    async def reset(self):
        self.dut.rst_n.setimmediatevalue(1)
        await RisingEdge(self.dut.clk)
        await RisingEdge(self.dut.clk)
        self.dut.rst_n.value = 0
        await RisingEdge(self.dut.clk)
        await RisingEdge(self.dut.clk)
        self.dut.rst_n.value = 1
        await RisingEdge(self.dut.clk)
        await RisingEdge(self.dut.clk)
        for _ in range(10):
            await RisingEdge(self.dut.clk)

    # 初始化能力
    async def init(self):
        self.dut.s_axis_tvalid.value = 0
        self.dut.m_axis_tready.value = 1
        await ClockCycles(self.dut.clk, 10)

    # 发送数据能力
    async def sender(self):
        ...
        frame = AxiStreamFrame(
            tdata=payload,  # tdata
            tuser=tuser
        )
        await axis_source.send(frame)
        await RisingEdge(self.dut.clk)

    # 接收数据能力
    async def recver(self):
        frame = await axis_sink.recv()
        ...

    # 激励产生及期望数据产生能力
    def gen_chk(frame: bytearray):
        ...

    async def gen_frame(self):
        lens = self.rng.randint(1, 8192)
        sim_dat = bytearray(self.rng_sender.getrandbits(8) for _ in range(lens))
        chk_dat = gen_chk(sim_dat)
        ...

    # 数据比对与结果校验能力
    async def check_frame(self):
        ...
        if recv_dat != chk_dat:
            ...

    # 压力控制能力
    def random_backpressure(self, rng: random.Random, n: float) -> Generator[bool, None, None]:
        while True:
            yield rng.random() < n

    async def apply_backpressure(self, n: float) -> None:
        try:
            self.axis_sink.set_pause_generator(self.random_backpressure(self.rng, n))
        except Exception as e:
            self.dut._log.error(f"backpresse failed: {e}")
            raise

    async def apply_validpause(self, n: float) -> None:
        try:
            self.axis_source.set_pause_generator(self.random_backpressure(self.rng, n))
        except Exception as e:
            self.dut._log.error(f"backpresse failed: {e}")
            raise

TB的桥梁

写到这里，可能会有同学产生疑问：
通过 gen_frame 生成的激励数据和期望数据，以及 recver 接收到的实际数据，究竟是如何在 sender、check_frame 等方法之间传递的？

这里就必须引出一个在 Cocotb 仿真中最核心、也是几乎无法回避的概念——队列（Queue）。

可以说，队列是 Cocotb 仿真的桥梁。正是因为有了队列，才能在软件层面真实地模拟硬件系统中天然存在的并行性。
对于硬件设计者而言，队列更多是一个软件概念；而在硬件世界中，它对应的名字其实非常熟悉——FIFO。

在面向对象的 TestBench 设计中，我们可以将不同阶段的数据流统一通过队列进行解耦：

• 将生成的激励数据放入激励队列（sim_queue）
• 将对应的期望数据放入比对队列（cmp_queue）
• 将从 RTL 接收到的实际数据放入接收队列（recv_queue）

当需要发送数据时，sender从sim_queue中取出激励并驱动接口；
当需要进行结果校验时，check_frame分别从cmp_queue中取出期望数据、从recv_queue中取出实际数据进行比对。

此外：

• 若检测到错误，可以将错误信息放入 err_queue
• 当一帧或一个事务校验完成时，可以通过 event_queue 发送完成事件，用于同步或统计

通过这种方式，各个仿真组件之间不再直接耦合，而是通过队列进行通信，使得整体架构更加清晰、灵活，也更接近真实硬件系统的运行方式。这种组件间解耦与通信的思想，在网络与系统的并发编程模型中也十分常见。

因此，在TestBench 对象的属性中，对这些队列进行统一初始化和管理，是构建一个工程化、可扩展 Cocotb 仿真框架的关键一步。

class AXISSlice_TestBench:
    def __init__(self, dut, seed=0):
        # dut对象
        self.dut = dut
        # 随机数种子
        self.rng = random.Random(seed)
        # 队列
        self.sim_queue = Queue(1)
        self.cmp_queue = Queue()
        self.recv_queue = Queue()
        self.err_queue = Queue()
        self.event_queue = Queue()
        # slave接口
        self.axis_source = AxiStreamSource(
            bus=AxiStreamBus.from_prefix(dut, "s_axis"),
            clock=dut.clk,
            reset=dut.rst_n,
            reset_active_level=False
        )
        # master接口
        self.axis_sink = AxiStreamSink(
            bus=AxiStreamBus.from_prefix(dut, "m_axis"),
            clock=dut.clk,
            reset=dut.rst_n,
            reset_active_level=False
        )

在TestBench 方法中我们加上这些队列：

# 发送数据能力
async def sender(self):
    while True:
        payload = await self.sim_queue.get()
        ...
        frame = AxiStreamFrame(
            tdata=payload  # tdata
        )
        await axis_source.send(frame)
        await RisingEdge(self.dut.clk)

# 接收数据能力
async def recver(self):
    while True:
        frame = await axis_sink.recv()
        await self.recv_queue.put(frame.tdata)
        ...

# 激励产生及期望数据产生能力
def gen_chk(frame: bytearray):
    ...

async def gen_frame(self):
    while True:
        lens = self.rng.randint(1, 8192)
        sim_dat = bytearray(self.rng_sender.getrandbits(8) for _ in range(lens))
        chk_dat = gen_chk(sim_dat)
        await self.sim_queue.put(sim_dat)
        await self.cmp_queue.put(chk_dat)
        ...

# 数据比对与结果校验能力
async def check_frame(self):
    while True:
        recv_dat = await recv_queue.get()
        chk_dat = await cmp_queue.get()
        if recv_dat != chk_dat:
            await self.err_queue.put(f"CHECK DATA FAIL!!! \nthe recv is:{recv_dat.hex()}, \nthe chek is:{chk_dat.hex()}")
            ...
        await self.event_queue.put(1)

# 压力控制能力
def random_backpressure(self, rng: random.Random, n: float) -> Generator[bool, None, None]:
    while True:
        yield rng.random() < n

async def apply_backpressure(self, n: float) -> None:
    try:
        self.axis_sink.set_pause_generator(self.random_backpressure(self.rng, n))
    except Exception as e:
        self.dut._log.error(f"backpresse failed: {e}")
        raise

async def apply_validpause(self, n: float) -> None:
    try:
        self.axis_source.set_pause_generator(self.random_backpressure(self.rng, n))
    except Exception as e:
        self.dut._log.error(f"backpresse failed: {e}")
        raise

创建对象

当 AXISSlice_TestBench 的关键方法、属性以及各类“桥梁”（队列与事件）都搭建完成之后，测试用例本身就会变得异常简洁。在 @cocotb.test 中，我们只需要创建一个 tb 对象即可。

整个 TestBench 的运行由事件队列驱动流程推进，由错误队列驱动异常中断。从现在的视角来看，这种写法是否比将所有逻辑堆在一个 test 函数中更加清晰、结构也更加直观呢？

@cocotb.test()
async def axis_slice_test(dut):
    # init clock & reset
    clock = Clock(dut.clk, period=10, units="ns")
    cocotb.start_soon(clock.start())
    tb = AXISSlice_TestBench(dut, seed=0)

    await tb.reset()
    await tb.init()

    # 启动各个并发组件
    cocotb.start_soon(tb.sender())
    cocotb.start_soon(tb.recver())
    cocotb.start_soon(tb.gen_frame())
    cocotb.start_soon(tb.check_frame())
    cocotb.start_soon(tb.apply_backpressure(0.8))
    cocotb.start_soon(tb.apply_validpause(0.1))

    # 测试 1000 个包，若出现错误立即中断
    for _ in range(1000):
        await tb.event_queue.get()
        if not tb.err_queue.empty():
            err = await tb.error_event_queue.get()
            await RisingEdge(dut.clk)
            assert err == 0, f"TEST FAIL: {err}"
    await RisingEdge(dut.clk)

可以看到，test 函数本身已经不再关心具体的协议细节和时序行为，它只负责：

• 启动 TestBench
• 运行各个功能组件
• 监听事件与错误

这正是面向对象 Cocotb 仿真的核心价值所在：
测试用例只描述要测什么，而不是怎么测。

写在最后

通过这篇文章，相信大家已经对面向对象的仿真方法有了一个较为完整且直观的认识。需要说明的是，本文主要聚焦于 Posted 类仿真，而对于 Non-Posted 仿真，这里先按下不表。

实际上，如果前文的思路已经理解到位，Non-Posted 仿真是完全可以自行推导和实现的。这里简单提一句：Non-Posted 仿真相比 Posted 仿真，本质上只是多引入了一个对端设备（device）对象。

在这种模型中，TestBench 生成激励后，并不是简单地发出去就结束，而是由 device 对象处理 TB 接收的请求、根据协议语义进行处理，并返回相应的响应数据。例如：

• 在验证 SATA 协议时，可以构造一个简化的 SATA_DEVICE 类；
• 在验证 TCP 对端逻辑时，可以构造一个简化的 TCP_DEVICE 类；

device 对象的职责，本质上就是模拟真实系统中的对端行为，从而完成请求—响应式的交互仿真。

至此，希望这篇文章能帮助你建立起一个清晰的认知：
面向对象并不是为了写得更像软件，而是为了让复杂仿真问题变得更可拆解、更可复用，也更接近真实硬件系统的行为模型。