FPGA基础 -- cocotb仿真之任务调度cocotb.start_soon与asyncio的使用注意事项
为什么在 cocotb 里不要用 asyncio、而要用 cocotb.start_soon()”讲透彻——从调度机制、时间语义、线程安全、异常传播与收尾、以及可替代方案全覆盖。
结论先行
- cocotb 有自己的协程调度器，由仿真器（VPI/VHPI/FLI）事件驱动；不是
asyncio的事件循环。 - 仿真时间≠真实时间：
cocotb.Timer("10 ns")推进的是模拟时间；asyncio.sleep()走的是墙钟时间，与仿真推进完全脱钩。 cocotb.start_soon()把协程注册到 cocotb 调度器，能感知RisingEdge/ReadOnly/ReadWrite等仿真相位；asyncio完全不知道这些相位。- 生命周期与异常：
start_soon()启的任务在测试结束会被自动取消/收尾、异常会正确上抛；asyncio任务不会被 cocotb 管理，容易泄漏、卡住、吞异常。 - 单线程要求：大多数仿真器 API 只能在主仿真线程调用。
asyncio常见的跑法（单独事件循环/线程）会越线程调用仿真对象，直接未定义行为。
1) 调度模型：两个“世界”的事件循环不兼容
- cocotb 调度器：当你
await RisingEdge(sig)、await Timer(10, "ns")、await ReadOnly()时，本质是把协程挂到仿真器的事件队列上。仿真器到点触发回调，cocotb 才恢复协程。唯一时基是模拟时间/相位。 - asyncio 调度器：管理
Future/Task的是 Python 的墙钟事件循环。它既不认识RisingEdge，也不认识ReadOnly/ReadWrite，更无法在NBA 提交后保证你再读取信号。
结果：你用 asyncio.create_task() 启的任务无法由仿真事件唤醒；反之，await 了 RisingEdge 的 cocotb 协程也不会被 asyncio 驱动。两边是两套互不相干的“时空”。
2) 时间语义：模拟时间 vs 真实时间
Timer(100, "ns")：推进仿真 100ns，期间墙钟可能 0ms（Verilator 的零延时步进）或若干 ms。asyncio.sleep(0.1)：墙钟 100ms，仿真时间可能没动（尤其是事件驱动仿真）。
→ 常见灾难：你在asyncio.sleep()等“1ms 后再读数据”，但仿真里下一拍都没到；或者你在 monitor 里用asyncio.sleep(0)当让步，结果仍在错相位读取，读不到 SOT/DT 等一次性字样。
3) 仿真相位：ReadOnly/ReadWrite 只能由 cocotb 感知
- 你已经踩过一次坑：不加
ReadOnly()在 NBA 落地前读取hs_data_out，错过0xB8B8B8B8。 - 这些相位触发（ReadOnly/ReadWrite/NextTimeStep）是 cocotb 对仿真器调度队列的包装；
asyncio完全不认识，无法保证“在 NBA 之后读”、“在驱动区写”。
4) 线程与仿真器 API：只能主线程
- 大多数 VPI/VHPI/FLI 后端必须在仿真主线程调用信号读写。
asyncio常见做法是起一个独立 loop（甚至独立线程）→ 这会导致你在非仿真线程里碰dut.sig.value，未定义行为（轻则读脏，重则崩仿真）。
5) 生命周期与异常传播
-
cocotb.start_soon(coro)→ 返回cocotb.task.Task：- 随测试生命周期自动管理：测试结束时自动取消未结束的子任务；
- 子任务抛出的异常会关联到当前测试，让回归正确 fail；
- 你还能
await task做显式 join。
-
asyncio.create_task(coro)：- cocotb 看不见它，测试结束不回收；
- 异常可能被吞或仅在
Task回收时打印 warning； - 容易在 CI 里造成间歇性挂死/泄漏。
6) 可观测性与可重复性
- cocotb 的调度顺序（不同 Trigger、不同 Task 之间）是为可重复/可验证设计的；
asyncio的调度顺序受墙钟/事件循环实现影响，不可复现→ 回归不稳定。
7) 正确并发姿势：cocotb.start_soon() 模式
典型写法（也是你项目里应该遵循的套路）：
# 时钟
cocotb.start_soon(Clock(dut.tx_byte_clk, 8, units="ns").start())
cocotb.start_soon(Clock(dut.pixclk_o, 13.888, units="ns").start())# Monitor（注意每拍 ReadOnly）
async def hs_monitor(dut, out_q):prev_active = Falsecur = []while True:await RisingEdge(dut.tx_byte_clk)await ReadOnly()active = int(dut.txclk_hsen.value) and (int(dut.txclk_hsgate.value) == 0)if active:cur.append(int(dut.hs_data_out.value))if prev_active and not active and cur:await out_q.put(cur) # cocotb.queue.Queuecur = []prev_active = activemon_task = cocotb.start_soon(hs_monitor(dut, my_queue))# Driver（注意 ReadWrite/相位）
async def driver(dut):await RisingEdge(dut.tx_byte_clk) # 对齐# 驱动 fv/lv/dvalid/pixdata...# 写之前可 await ReadWrite()；或直接赋值，让 cocotb 放到写相位
drv_task = cocotb.start_soon(driver(dut))# 等待或 join
await Timer(200, "us")
for t in (mon_task, drv_task):t.kill() # 或 await t
8) 如果你必须用 asyncio（极少见）的替代方案
有些人想在 TB 里连网络/串口/文件异步 IO。建议：
- 子线程跑
asyncioloop，与 cocotb 通过线程安全队列通信； - 子线程禁止直接读写
dut；把需要的操作封装成回调，回到主仿真线程用 cocotb 触发（例如cocotb.triggers.Timer(0)安排一个“下一拍执行”的安全回调）； - 或者更简单：把外部 IO 做成同步阻塞，丢给
cocotb.extern包一层，到主线程里再await返回值（仍需谨慎，避免卡住仿真推进）。
总之：任何触达仿真对象的操作都必须回到 cocotb 的调度上下文。
9) 历史兼容：fork vs start_soon
- 旧 API
cocotb.fork()早期就能用，但不返回 Task 对象（可控性差）； - 新 API
cocotb.start_soon()返回Task，支持join()/kill()，更安全、可管理。 - 现在统一用
start_soon()。
小结
asyncio 和 cocotb 是两台互不连接的引擎：一个按墙钟调度，另一个按仿真事件调度。把 TB 的并发交给 cocotb.start_soon()，才能获得正确的相位语义、可重复的调度、可控的生命周期与异常传播。
当你采样像 0xB8B8B8B8 这种“一拍即逝”的字样时，RisingEdge + ReadOnly 搭配 start_soon() 的监控模式，才是可靠的专业做法。