同一份采样数据需要同时送到计算模块和记录模块。计算模块多数时候能接收，记录模块偶尔停顿。如果把输入 valid 直接连到两个输出，只把上游 ready 写成两个 ready 的与，先就绪的一路可能反复收到同一份数据。
问题不在载荷复制，而在于每个接收端的进度需要分别记住。本文用一个载荷寄存器和两个 pending 位，实现一路输入、两路独立 ready 的精确复制：每个输入在每个输出上恰好握手一次，两路可以在不同的时钟周期完成。
实际验证环境：Windows 上的 Vivado/XSim 2022.1，SW Build 3526262，2026-11-06。 4组载荷宽度配置共完成48,408个自检周期，4个指定RTL变异与2个非法参数均被捕获；独立测试台另完成316,125次断言。以下时序表是契约推导，不冒充波形截图。验证仅为RTL二态行为，没有综合、实现、时序闭合或板级证据。
1. 先看会重复数据的直连方案
下面这组关系很常见，也很危险：
m0_valid = s_valid
m1_valid = s_valid
s_ready = m0_ready && m1_ready
假设输入持续给出 A，输出0一直 ready，输出1连续三拍不 ready。上游因为 s_ready=0 保持 A，但输出0每拍都看到自己的 valid 和 ready 同时为 1，于是把 A 接收了三次。
“上游没有握手”不代表“某个下游没有握手”。三个接口各自有自己的握手事件。ready/valid 的基本传输条件可参考 AMD PG256 的 AXI4-Stream Interface。将多个接口相连时，必须保证这些独立事件的数量关系正确。
也不要简单把输出0的 valid 与输出1的 ready 相与。那会让一个输出的 valid 依赖另一个接收端的即时 ready，并可能在已经有效但被背压时撤回 valid。这种接法很难维持清晰的端口约定。
2. 两个 pending 位就是两个独立收据
本模块保存一份 data_q，以及两位 pending：
pending[0]=1：输出0还没有收到当前项。pending[1]=1：输出1还没有收到当前项。- 两位都为 0：没有未完成的项，可以接收下一份输入。
输出 valid 直接取对应 pending 位。某路完成握手后，只清除自己的位，另一位不受影响。
以 pending[1:0] 的顺序表示状态，下面是预期时序：
| 边沿前 pending | 载荷 | ready0 | ready1 | 本拍输出握手 | 本拍输入 | 边沿后 pending/载荷 |
|---|---|---|---|---|---|---|
| 00 | 无 | 0 | 0 | 无 | 接受 A | 11 / A |
| 11 | A | 1 | 0 | 输出0收到 A | B 被背压 | 10 / A |
| 10 | A | 1 | 0 | 无 | B 被背压 | 10 / A |
| 10 | A | 1 | 1 | 输出1收到 A | 接受 B | 11 / B |
| 11 | B | 1 | 1 | 两路各收到 B | 无 | 00 / B |
第三行输出0虽然仍然 ready，但自己的 pending 位已经清零，所以不再握手。末行后 data_q 仍可以保留 B，valid 为 0 时数据内容不代表新的传输。
3. 最后一份副本取走时，可以同拍装入下一项
如果输入 ready 只在 pending==0 时拉高，每一份数据完全送出后都要再等一拍才能装入下一份，双方持续就绪时也会出现空泡。
本实现先计算本拍输出握手之后还剩哪些副本：
remaining = pending & ~{m1_ready, m0_ready}
s_ready = !rst && (remaining == 0)
pending 为 0 的分支无论 ready 如何都不会影响 remaining。因此，已经完成的分支后来拉低 ready，也不会阻止另外一份副本结束。
当 remaining 为 0，当前数据本拍结束，或者当前本来就是空闲，可以接受新输入。时序过程先清除旧副本的 pending，再由新输入握手把它设成 11。新输入的装载必须优先于旧项的清除。 两处非阻塞赋值写同一个寄存器时，代码中的后一次赋值定义了这个优先级。
这也解释了为什么载荷可以共用一个寄存器：只有所有旧副本都在本拍完成，才允许覆盖 data_q。被背压的接收端不会在未握手时看到载荷更换。
如果两个接收端持续 ready，填入第一项后，模块可以每拍完成一份旧项并装入一份新项。这里说的是协议吞吐能力，不是测得的时钟频率。输入 ready 有从两个下游 ready 经过组合逻辑回传的路径，接入长链路时仍需检查时序、组合环和必要的寄存切分。
4. 复位、慢分支与元数据要单独约定
本文采用单时钟、同步高有效复位。rst 为 1 时屏蔽三处握手；在上升沿清空 pending 与载荷寄存器。所有输入信号必须来自同一时钟域，异步输入需要另行处理。
复位会取消还没有送出的副本。例如输出0已经收到 A、输出1尚未收到时发生复位，系统不能撤回输出0的 A。因此“每个输入恰好复制两次”只对一个没有复位的运行区间成立。需要跨复位一致性时，应增加事务编号、重放或系统级恢复协议，不能靠两位 pending 自动补齐。
一个分支永久不 ready，最终就会阻止新输入。该模块只保存一项，并未为快分支提供无限队列。若需要让快分支继续运行，可以考虑给各分支增加 FIFO，并明确队列满后的整体策略。
DATA_W 允许 1～64。真正接 AXI4-Stream 时，帧结束、字节有效和用户标签也必须与数据一起复制。可以先把这些字段打包进载荷，再在各输出拆开；示例并没有单独提供 TLAST/TKEEP/TUSER 端口，不能直接当作完整 AXI4-Stream IP。
5. 完整 RTL
保存为 stream_fork2.sv。文章代码由同名文件直接嵌入，数据与控制的所有寄存状态都在下面。
本次原生结果绑定实际受测源码哈希；下面发布副本仅更新第2行过时注释，其他字节经独立核对一致。没有把新注释对应的文件哈希宣称为重新跑过XSim。
// One registered payload, two independently consumed copies.
// Behavioral example; synthesis and hardware validation not performed.
module stream_fork2 #(
parameter integer DATA_W=16,
localparam integer DW=(DATA_W<1 || DATA_W>64) ? 1 : DATA_W
) (
input logic clk, rst,
input logic s_valid,
output logic s_ready,
input logic [DW-1:0] s_data,
output logic m0_valid,
input logic m0_ready,
output logic [DW-1:0] m0_data,
output logic m1_valid,
input logic m1_ready,
output logic [DW-1:0] m1_data
);
logic [DW-1:0] data_q;
logic [1:0] pending, remaining;
assign remaining=pending & ~{m1_ready,m0_ready};
assign s_ready=!rst && (remaining == 2'b00);
assign m0_valid=!rst && pending[0];
assign m1_valid=!rst && pending[1];
assign m0_data=data_q;
assign m1_data=data_q;
always_ff @(posedge clk) begin
if(rst) begin
pending <= 2'b00;
data_q <= '0;
end else begin
pending <= remaining;
// Replacement must take priority over clearing the old copies.
if(s_valid && s_ready) begin
data_q <= s_data;
pending <= 2'b11;
end
end
end
// synthesis translate_off
initial begin
if(DATA_W<1 || DATA_W>64) $fatal(1,"PARAM_ERROR stream_fork2");
end
// synthesis translate_on
endmodule
6. 记分板按每个输出单独记账
只比较 m0_data==m1_data 不能发现重复传输：两路都反复输出 A，数据比较仍然成立。测试台需要按输入序列给两个输出分别计数。
完整自检使用一个期望数组和三个游标：
written：本轮复位以来接受的输入数量
read0：输出0已接受的数量
read1：输出1已接受的数量
当输出0握手，必须等于 expected[read0]，随后只增加 read0；输出1类似。输入握手才增加 written。由于模块只容纳一项，每一路还应满足：
0 <= written - read0 <= 1
0 <= written - read1 <= 1
一个周期可以既消费旧项，又接受新项。测试台必须先根据边沿前的 written/read 状态检查和消费旧输出，再把新输入追加到期望数组。顺序写反会把刚输入的数据当作可以同拍穿透输出的数据，掩盖延迟错误。
测试包含输出0先收、输出1先收、某一路连续停顿、两路连续就绪、最后一份副本与新输入同拍、部分副本已送出后复位，以及固定种子的独立随机背压。64位配置包含全1、最高位、交替位和高低两半分别生成的随机载荷，避免只测到低32位。
Python 参考检查把“待接收端集合减去本拍就绪端集合”与两位布尔表达式对照，遍历4种 pending、4种 ready 和2种输入 valid，共32种状态/输入组合；4种参考模型错误都找到了反例。这并不是 RTL 仿真，也不是形式化工具证明。
原生回归分别修改了“任意一路收到就清空两位”“先完成的分支不清位”“任意一路ready就接新输入”“旧状态清除覆盖新项装载”，四种变异均在成功编译展开后被对应断言捕获。源代码无法编译、仿真超时或不相关的错误，没有充当这些测试有效的证据。
7. 完整自检测试台
保存为 tb_stream_fork2.sv。每次 step 在下降沿驱动输入，检查边沿前的控制与数据，再等待上升沿更新。CSV记录每个分支自己的握手，便于直接数出 A 在每一路究竟出现了几次。
`timescale 1ns/1ps
module tb_stream_fork2 #(parameter integer DATA_W=16);
logic clk=0;
always #5 clk=~clk;
logic rst=1,s_valid=0,s_ready,m0_valid,m0_ready=0,m1_valid,m1_ready=0;
logic [DATA_W-1:0] s_data='0,m0_data,m1_data;
stream_fork2 #(.DATA_W(DATA_W)) dut(.*);
logic [DATA_W-1:0] expected[0:20000];
integer written=0, read0=0, read1=0, checks=0, accepted_total=0;
integer trace_fd;
bit last_accept=0, stall0=0, stall1=0;
logic [DATA_W-1:0] held0,held1;
integer unsigned rng=32'h71d602b3;
function automatic integer unsigned next_random();
rng=rng^(rng<<13); rng=rng^(rng>>17); rng=rng^(rng<<5); return rng;
endfunction
function automatic logic [DATA_W-1:0] next_payload();
logic [63:0] word;
word[31:0]=next_random(); word[63:32]=next_random();
return DATA_W'(word);
endfunction
task automatic step(input bit r,v, input logic[DATA_W-1:0] d, input bit r0,r1);
bit p0,p1, expected_ready;
@(negedge clk); rst=r;s_valid=v;s_data=d;m0_ready=r0;m1_ready=r1; #1;
p0=(read0<written); p1=(read1<written);
expected_ready=!r && ((!p0 || r0) && (!p1 || r1));
if(s_ready !== expected_ready || m0_valid !== (!r && p0) || m1_valid !== (!r && p1))
$fatal(1,"ASSERT_FORK_CONTROL checks=%0d wr=%0d r0=%0d r1=%0d",checks,written,read0,read1);
if(!r) begin
if(p0 && m0_data !== expected[read0]) $fatal(1,"ASSERT_FORK_DATA0");
if(p1 && m1_data !== expected[read1]) $fatal(1,"ASSERT_FORK_DATA1");
if(stall0 && (!m0_valid || m0_data!==held0)) $fatal(1,"ASSERT_FORK_STABILITY0");
if(stall1 && (!m1_valid || m1_data!==held1)) $fatal(1,"ASSERT_FORK_STABILITY1");
end
stall0=!r && p0 && !r0; stall1=!r && p1 && !r1;
held0=m0_data; held1=m1_data;
last_accept=!r && v && expected_ready;
$fdisplay(trace_fd,"%0d,%0d,%0d,%0d,%0h,%0d,%0d,%0h,%0d,%0d,%0h,%0d,%0d,%0d",checks,r,v,s_ready,d,m0_valid,r0,m0_data,m1_valid,r1,m1_data,written,read0,read1);
// Scoreboard consumes old outputs BEFORE enqueueing a new input.
if(r) begin written=0;read0=0;read1=0;end
else begin
if(p0 && r0) read0=read0+1;
if(p1 && r1) read1=read1+1;
if(last_accept) begin
if(written>20000) $fatal(1,"ASSERT_FORK_SCOREBOARD_SIZE");
expected[written]=d;written=written+1; accepted_total=accepted_total+1;
end
if(written-read0>1 || written-read1>1 || read0>written || read1>written)
$fatal(1,"ASSERT_FORK_CONSERVATION");
end
@(posedge clk); #1; checks=checks+1;
endtask
bit rv,r0,r1;
logic [DATA_W-1:0] rd;
initial begin
trace_fd=$fopen("trace.csv","w");
if(!trace_fd) $fatal(1,"TRACE_OPEN_ERROR");
$fdisplay(trace_fd,"cycle,rst,s_valid,s_ready,s_data_hex,m0_valid,m0_ready,m0_data_hex,m1_valid,m1_ready,m1_data_hex,inputs_before,out0_before,out1_before");
step(1,0,'0,0,0);
// Output0 accepts early and must not see the same item again.
step(0,1,DATA_W'(16'h1111),0,0);
step(0,1,DATA_W'(16'h2222),1,0);
repeat(8) step(0,1,DATA_W'(16'h2222),1,0);
// Last consumer and next input accept on this same edge.
step(0,1,DATA_W'(16'h2222),1,1);
step(0,0,'0,1,1);
step(0,0,'0,1,1);
// Symmetric branch order.
step(0,1,DATA_W'(16'h3333),0,0);
repeat(8) step(0,0,'0,0,1);
step(0,0,'0,1,1);
// Exercise every payload bit, including the DATA_W=64 high half.
step(0,1,{DATA_W{1'b1}},1,1);
step(0,1,(DATA_W'(1) << (DATA_W-1)),1,1);
step(0,1,DATA_W'(64'haaaaaaaaaaaaaaaa),1,1);
step(0,1,DATA_W'(64'h5555555555555555),1,1);
step(0,0,'0,1,1);
// Back-to-back inputs when both outputs are ready.
for(integer n=0;n<64;n=n+1) step(0,1,DATA_W'(n),1,1);
step(0,0,'0,1,1);
// Reset with only one copy outstanding starts a new epoch.
step(0,1,DATA_W'(16'h4444),0,0);
step(0,0,'0,1,0);
step(1,0,'0,1,1);
step(0,0,'0,1,1);
// Deterministic randomized independent backpressure.
rv=0;rd='0;last_accept=1;
for(integer n=0;n<12000;n=n+1) begin
if(!rv || last_accept) begin rv=((next_random()%5)!=0);rd=next_payload();end
r0=((next_random()%4)!=0);r1=((next_random()%3)!=0);
step(0,rv,rd,r0,r1);
end
if(rv && !last_accept) step(0,rv,rd,1,1);
repeat(3) step(0,0,'0,1,1);
if(read0!=written || read1!=written) $fatal(1,"ASSERT_FORK_DRAIN");
$display("PASS stream_fork2 checks=%0d accepted_total=%0d epoch_inputs=%0d out0=%0d out1=%0d DATA_W=%0d",checks,accepted_total,written,read0,read1,DATA_W);
$fclose(trace_fd);
$finish;
end
initial begin #10000000;$fatal(1,"ASSERT_FORK_TIMEOUT");end
endmodule
8. 运行入口与证据边界
将两个文件放到同一目录，在配置了 Vivado 环境的命令行中执行：
xvlog --sv stream_fork2.sv tb_stream_fork2.sv
xelab tb_stream_fork2 --timescale 1ns/1ps --debug typical -s stream_fork2_sim
xsim stream_fork2_sim -runall
本次默认16位配置的原始结果为：
PASS stream_fork2 checks=12102 accepted_total=6408 epoch_inputs=6336 out0=6336 out1=6336 DATA_W=16
1、8、16、64位四组配置每组都运行12,102个自检周期，总计48,408周期。checks按每个测试台step计一次，不是全部条件断言的执行次数；accepted_total包含复位前的输入，epoch_inputs/out0/out1只统计最后一次复位后的完整运行区间，不能跨字段直接比较为“丢失”。每组均保存真实trace.csv。仿真快照的生成与运行方式见 AMD UG900 xelab，命令选项另见 UG900 xsim Executable Options。文档版本不能代替实际工具版本日志。
源码包统一的 run_rtl.py 默认只打印计划；加 --execute --engine xsim 后才执行。复制器部分的4组载荷宽度正例、4个RTL变异和2个非法参数测试，本次全部完成。它拒绝把进程超时、异常退出、矛盾PASS或未知断言判作有效负例。本次XSim在$fatal后仍可返回进程码0，所以必须同时核对诊断文本与终态，不能只看退出码。
独立测试台另测1、7、64位配置，分别完成105,099、105,171、105,855次断言，共316,125次；“任一分支接收就清两份”和“丢掉64位载荷高半部”两个行为变异均被捕获，DATA_W=65也被明确拒绝。独立测试有一次只取消输出0待收副本的复位事件，因此每组记录discarded0=1、discarded1=0，核算时按复位契约记账，不把取消副本误报为传输丢失。
实际CSV还由独立双队列审计器重放，检查每个分支次数、顺序、背压载荷、同拍替换及复位边界。两篇共10组正例的137,523行记录逐行通过；这属于原生输出后处理，不另算RTL仿真次数。
本次最初的原生运行因时间单位设置在展开阶段失败；修复运行器并显式指定--timescale 1ns/1ps后完成全部回归，RTL行为与测试激励保持不变。最终运行保留命令日志、实际DUT信号WDB与CSV，并完成独立原生复核。行为仿真之后仍需要结合目标器件进行综合、时序和实现检查；本文没有下板证据，也没有资源或频率指标。
这个模块最值得保留的思路是：复制一份载荷时，也复制一份“尚未完成”的状态。把每个接收端的收据独立保存，才能在不同背压下保持次数、顺序和载荷一致。
附录：完整批跑器
下面是本次实际使用的 run_rtl.py，只依赖Python标准库，要求Python 3.10或更新版本；本次运行使用Python 3.14.5。将它放到工作根目录，把本文两份SV文件放入 stream_fork2/ 子目录。运行下面命令复现本文所属的全部正向、变异和非法参数配置：
python run_rtl.py --execute --engine xsim --only fork
如果两个题目的SV目录都已放齐，去掉--only即可运行两题共30项。筛选运行的结果会明确标记full_suite=false，不应把单题回归写成两题全部完成。--engine iverilog仅是备选入口，本次结果来自XSim，不宣称Icarus交叉验证已运行。
各项独立目录保存真实命令、UTC起止时间、stdout/stderr、退出状态、编译源哈希与WDB；正向用例另保存完整trace.csv。异常退出、超时、没有真实DUT波形对象、指定变异被其他断言击中，都会影响验收。版本检查必须识别实际工具版本。
#!/usr/bin/env python3
"""Native simulation runner. Default is plan-only; use --execute deliberately.
No third-party Python dependencies. Run in a Vivado environment with xvlog,
xelab and xsim on PATH. Icarus is an optional second simulator, not XSim evidence.
Compilation failure NEVER counts as a killed mutant or invalid-parameter pass.
"""
from pathlib import Path
import argparse, hashlib, json, os, re, shutil, subprocess, sys, datetime
ROOT=Path(__file__).resolve().parent
TOKEN_PARAMS=[
{'CAPACITY':1,'REFILL':1,'INITIAL_TOKENS':0,'DATA_W':1},
{'CAPACITY':2,'REFILL':1,'INITIAL_TOKENS':2,'DATA_W':8},
{'CAPACITY':7,'REFILL':3,'INITIAL_TOKENS':0,'DATA_W':16},
{'CAPACITY':8,'REFILL':8,'INITIAL_TOKENS':8,'DATA_W':32},
{'CAPACITY':17,'REFILL':5,'INITIAL_TOKENS':4,'DATA_W':64},
{'CAPACITY':65535,'REFILL':65535,'INITIAL_TOKENS':0,'DATA_W':16},
]
MUTATIONS={
'token_bucket':{
'wrap_credit':('assign available=(raw_credit > (CW+1)\'(CAPACITY)) ? CW\'(CAPACITY) : raw_credit[CW-1:0];','assign available=raw_credit[CW-1:0];'),
'old_credit':('(s_cost <= available)','(s_cost <= tokens)'),
'spend_on_valid':('else if(m_valid && m_ready)','else if(m_valid)'),
'allow_zero':("(s_cost != '0) && ",''),
'spend_before_saturate':('tokens <= available-s_cost;',"tokens <= ((raw_credit-{1'b0,s_cost}) > (CW+1)'(CAPACITY)) ? CW'(CAPACITY) : CW'(raw_credit-{1'b0,s_cost});"),
'valid_depends_ready':('assign m_valid=!rst && s_valid && can_send;','assign m_valid=!rst && s_valid && can_send && m_ready;'),
},
'stream_fork2':{
'clear_both_on_any':('assign remaining=pending & ~{m1_ready,m0_ready};',"assign remaining=(|(pending & {m1_ready,m0_ready})) ? 2'b00 : pending;"),
'repeat_early_receiver':('assign remaining=pending & ~{m1_ready,m0_ready};',"assign remaining=(|(pending & ~{m1_ready,m0_ready})) ? pending : 2'b00;"),
'accept_when_one_ready':("assign s_ready=!rst && (remaining == 2'b00);","assign s_ready=!rst && ((pending == 2'b00) || (|(pending & {m1_ready,m0_ready})));"),
'clear_overwrites_replace':(" pending <= 2'b11;\n end"," pending <= 2'b11;\n end\n pending <= remaining;"),
}
}
EXPECTED_MUTATION_DIAGNOSTICS={
'wrap_credit':'ASSERT_TOKEN_STATE',
'old_credit':'ASSERT_TOKEN_CONTROL',
'spend_on_valid':'ASSERT_TOKEN_STATE',
'allow_zero':'ASSERT_TOKEN_CONTROL',
'spend_before_saturate':'ASSERT_TOKEN_STATE',
'valid_depends_ready':'ASSERT_TOKEN_CONTROL',
'clear_both_on_any':'ASSERT_FORK_CONTROL',
'repeat_early_receiver':'ASSERT_FORK_CONTROL',
'accept_when_one_ready':'ASSERT_FORK_CONTROL',
'clear_overwrites_replace':'ASSERT_FORK_CONTROL',
}
def cases():
result=[]
for i,p in enumerate(TOKEN_PARAMS):
result.append({'id':f'token_positive_{i}','module':'token_bucket','kind':'positive','parameters':p})
for w in (1,8,16,64):
result.append({'id':f'fork_positive_w{w}','module':'stream_fork2','kind':'positive','parameters':{'DATA_W':w}})
for module,mutants in MUTATIONS.items():
for mutation in mutants:
result.append({'id':f'{module}_{mutation}','module':module,'kind':'mutation','mutation':mutation,'parameters':{}})
for key,value in [('CAPACITY',0),('CAPACITY',65536),('REFILL',0),('REFILL',8),('INITIAL_TOKENS',-1),('INITIAL_TOKENS',8),('DATA_W',0),('DATA_W',65)]:
result.append({'id':f'token_invalid_{key}_{value}','module':'token_bucket','kind':'invalid','parameters':{key:value}})
for w in (0,65):
result.append({'id':f'fork_invalid_w{w}','module':'stream_fork2','kind':'invalid','parameters':{'DATA_W':w}})
return result
def tool_path(name):
# Vivado Windows commonly exposes .bat wrappers.
value=shutil.which(name) or shutil.which(name+'.bat')
if not value:
raise RuntimeError(f'{name} not found on PATH; native stage remains NOT RUN')
return value
def run_command(command, cwd, label, timeout=120):
started=datetime.datetime.now(datetime.timezone.utc).isoformat()
timed_out=False
try:
result=subprocess.run(command,cwd=cwd,stdout=subprocess.PIPE,
stderr=subprocess.PIPE,timeout=timeout,shell=False)
stdout=result.stdout; stderr=result.stderr
code=result.returncode
except subprocess.TimeoutExpired as error:
timed_out=True
stdout=error.stdout or b''; stderr=error.stderr or b''
code=124
(cwd/(label+'.stdout.log')).write_bytes(stdout)
(cwd/(label+'.stderr.log')).write_bytes(stderr)
text=stdout.decode('utf-8',errors='replace')+stderr.decode('utf-8',errors='replace')
if timed_out:text+='\nRUNNER_TIMEOUT\n'
(cwd/(label+'.log')).write_text(text,encoding='utf-8',newline='\n')
record={'command':command,'cwd':str(cwd),'start_utc':started,
'end_utc':datetime.datetime.now(datetime.timezone.utc).isoformat(),
'exit_code':code,'timed_out':timed_out,'log':str(cwd/(label+'.log')),
'stdout':str(cwd/(label+'.stdout.log')),'stderr':str(cwd/(label+'.stderr.log'))}
(cwd/(label+'.command.json')).write_text(json.dumps(record,indent=2)+'\n',encoding='utf-8',newline='\n')
return record,text
def prepare(case, folder):
module=case['module']
source=(ROOT/module/(module+'.sv')).read_text(encoding='utf-8')
if case['kind']=='mutation':
old,new=MUTATIONS[module][case['mutation']]
if source.count(old)!=1:raise RuntimeError(f"mutation anchor not unique: {case['id']}")
source=source.replace(old,new)
(folder/(module+'.sv')).write_text(source,encoding='utf-8',newline='\n')
params=', '.join(f'.{key}({value})' for key,value in case['parameters'].items())
parameter_clause=f'#({params})' if params else ''
if case['kind']=='invalid':
wrapper=f'''`timescale 1ns/1ps
module tb_case;
{module} {parameter_clause} dut();
initial begin #10; $fatal(1,"INVALID_PARAMETER_ACCEPTED"); end
endmodule
'''
sources=[module+'.sv','tb_case.sv']
else:
tb='tb_'+module
shutil.copy2(ROOT/module/(tb+'.sv'),folder/(tb+'.sv'))
wrapper=f'module tb_case; {tb} {parameter_clause} test(); endmodule\n'
sources=[module+'.sv',tb+'.sv','tb_case.sv']
(folder/'tb_case.sv').write_text(wrapper,encoding='utf-8',newline='\n')
hashes={name:hashlib.sha256((folder/name).read_bytes()).hexdigest() for name in sources}
(folder/'compiled-source-hashes.json').write_text(json.dumps(hashes,indent=2)+'\n',encoding='utf-8',newline='\n')
return sources
def execute(case, run_root, engine, executables):
folder=run_root/case['id'];folder.mkdir()
sources=prepare(case,folder)
logs=[]
if engine=='xsim':
compile_commands=[('xvlog',[executables['xvlog'],'--sv',*sources]),
('xelab',[executables['xelab'],'tb_case','--timescale','1ns/1ps','--debug','typical','-s','sim'])]
scope='/tb_case/dut/*' if case['kind']=='invalid' else '/tb_case/test/dut/*'
tcl=f'set objects [get_objects {scope}]\nif {{[llength $objects] == 0}} {{error "WAVE_SCOPE_EMPTY"}}\nputs "WAVE_OBJECTS=[llength $objects]"\nlog_wave $objects\nrun all\nquit\n'
(folder/'run.tcl').write_text(tcl,encoding='utf-8',newline='\n')
sim_command=[executables['xsim'],'sim','--tclbatch','run.tcl','--wdb','simulation.wdb']
else:
compile_commands=[('iverilog',[executables['iverilog'],'-g2012','-s','tb_case','-o','sim.vvp',*sources])]
sim_command=[executables['vvp'],'sim.vvp']
for label,command in compile_commands:
info,text=run_command(command,folder,label)
logs.append(info)
if info['exit_code']!=0:
return {**case,'result':'FAIL_COMPILE','commands':logs}
info,text=run_command(sim_command,folder,'simulation')
logs.append(info)
# A diagnostic printed before a timeout/crash is not a completed test.
process_ok=(not info.get('timed_out',False) and info['exit_code'] in (0,1)
and 'RUNNER_TIMEOUT' not in text and 'TRACE_OPEN_ERROR' not in text)
if engine=='xsim':
# Tcl echoes source lines: inspect the emitted count, not an error string
# that also appears in an unexecuted guard's echoed source.
process_ok=process_ok and bool(re.search(r'^WAVE_OBJECTS=[1-9]\d*\s*$',text,re.M))
if not process_ok:
status='FAIL_EXECUTION'
elif case['kind']=='positive':
ok=(info['exit_code']==0 and f"PASS {case['module']} " in text
and 'ASSERT_' not in text and 'PARAM_ERROR' not in text)
status='PASS_NATIVE' if ok else 'FAIL_NATIVE'
elif case['kind']=='mutation':
expected=EXPECTED_MUTATION_DIAGNOSTICS[case['mutation']]
# A different assertion is a test-design issue, not the expected kill.
diagnostics=set(re.findall(r'ASSERT_[A-Z0-9_]+',text))
ok=(diagnostics=={expected} and f"PASS {case['module']} " not in text
and 'PARAM_ERROR' not in text)
status='KILLED_NATIVE' if ok else 'SURVIVED_OR_INVALID_FAILURE'
else:
ok=(f"PARAM_ERROR {case['module']}" in text
and 'INVALID_PARAMETER_ACCEPTED' not in text and 'PASS ' not in text
and 'ASSERT_' not in text)
status='REJECTED_NATIVE' if ok else 'FAIL_PARAMETER_CHECK'
return {**case,'result':status,'commands':logs}
def main():
parser=argparse.ArgumentParser(description=__doc__)
parser.add_argument('--engine',choices=['xsim','iverilog'],default='xsim')
parser.add_argument('--execute',action='store_true')
parser.add_argument('--only',default='',help='optional case-id substring')
parser.add_argument('--output',type=Path,default=ROOT/'native_runs')
args=parser.parse_args()
selected=[c for c in cases() if args.only in c['id']]
if not selected:raise SystemExit('No matching cases')
if not args.execute:
print(json.dumps({'status':'NOT RUN','engine':args.engine,'cases':selected},ensure_ascii=False,indent=2))
return
names=['xvlog','xelab','xsim'] if args.engine=='xsim' else ['iverilog','vvp']
executables={name:tool_path(name) for name in names}
stamp=datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%dT%H%M%S.%fZ')
run_root=args.output.resolve()/f'{args.engine}_{stamp}';run_root.mkdir(parents=True)
versions=[]
for name,exe in executables.items():
option='-V' if name in ('iverilog','vvp') else '-version'
record,version_text=run_command([exe,option],run_root,name+'_version',30)
pattern=r'(?<!\d)(?:19|20)\d{2}\.\d+(?:\.\d+)?\b' if args.engine=='xsim' else r'(?i)\bversion\s+\d+'
record['version_identified']=bool(record['exit_code']==0 and not record.get('timed_out',False) and re.search(pattern,version_text))
versions.append(record)
hashes={str(p.relative_to(ROOT)):hashlib.sha256(p.read_bytes()).hexdigest()
for p in sorted(ROOT.rglob('*.sv')) if 'native_runs' not in p.parts}
results=[execute(case,run_root,args.engine,executables) for case in selected]
okay={'PASS_NATIVE','KILLED_NATIVE','REJECTED_NATIVE'}
report={'engine':args.engine,'utc_start':stamp,'versions':versions,'source_sha256':hashes,
'result':('FAIL_VERSION_EVIDENCE' if not all(v['version_identified'] for v in versions)
else 'PASS_SELECTED_NATIVE' if all(r['result'] in okay for r in results) else 'FAIL_NATIVE'),
'full_suite':len(selected)==len(cases()),'cases':results,
'limits':'RTL behavioral simulation only. No synthesis, implementation, board or timing evidence.'}
report['utc_end']=datetime.datetime.now(datetime.timezone.utc).isoformat()
report['runner_sha256']=hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
(run_root/'results.json').write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n',encoding='utf-8',newline='\n')
print(json.dumps(report,ensure_ascii=False,indent=2))
if report['result']!='PASS_SELECTED_NATIVE':raise SystemExit(1)
if __name__=='__main__':main()
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/hxy_programmer/article/details/170230808
