From 79a0898d2a09af8ca39c59f08bc70c675946900e Mon Sep 17 00:00:00 2001 From: FU-max-boop Date: Mon, 31 Aug 2026 20:39:36 +0800 Subject: [PATCH] test(decode): verify backpressure and complex drain --- ut_backend/ctrl_block/decode/README.md | 38 +- .../ctrl_block/decode/env/decode_wrapper.py | 484 +++++++++++++++--- .../ctrl_block/decode/test_rv_decode.py | 159 ++++++ 3 files changed, 594 insertions(+), 87 deletions(-) diff --git a/ut_backend/ctrl_block/decode/README.md b/ut_backend/ctrl_block/decode/README.md index ff6c9bbc..5186e419 100644 --- a/ut_backend/ctrl_block/decode/README.md +++ b/ut_backend/ctrl_block/decode/README.md @@ -2,14 +2,14 @@ ## 测试目标 -解码单元的功能是对输入的指令进行解码，最终转换为后端可识别的微指令。输出的指令有两种类型：16位的压缩指令（RVC） 和 32位的普通指令（RVI）。本单元测试的主要目的是**检查Dcode模块是否能识别所有非法指令**。 +解码单元的功能是对输入的指令进行解码，最终转换为后端可识别的微指令。输出的指令有两种类型：16位的压缩指令（RVC） 和 32位的普通指令（RVI）。本单元测试检查 Decode 模块能否识别非法指令，并验证输出反压期间的 ready/valid 协议及复杂指令的多 uop 排空行为。 测试基本流程为： -1. 随机生成指令 -1. 把指令输给DUT，得到解码结果（结果种包含是否为异常指令） -1. 把指令输给disasm，判断能否正常解析（disasm为RISC-V官方反汇编工具，可以认为 Golden） -1. 对比DUT和disasm的结果判断是否一致 +1. 用固定编码验证合法/非法结果、反压和复杂指令排空 +1. 随机生成指令并输入 DUT，检查每条指令的握手、传输顺序和完成次数 +1. 用 disasm 提供随机指令的辅助诊断信息，但不把缺少架构状态的语法解析当作异常位 Golden +1. 对固定用例的异常位和多 uop 属性做严格比较 ## 测试环境 Env @@ -23,7 +23,7 @@ 预解码阶段，[TBD] - **ut.decodestage** -[TBD] +提供 6 路输入和 6 路输出。当前 RTL 以 `io_out_0_ready` 计算全局输出容量，因此 Env 将 6 路 output ready 作为一个整体驱动，不支持部分 lane ready。 依赖： @@ -39,7 +39,9 @@ |2|||检查是否能展开所有正常指令，
发现所有非法指令|RVC_EXPAND_ALL_16B|RANGE[`start~end`]: 16位压缩指令共有 2^16种可能，
通过不同的start-end指定输入指令的
遍历范围，遍历该范围内的输入是否
是合法或非法指令（start，end由用例指定）| |3||常规指令展开|遍历所有32bit指令，
检查是否合法|RVC_EXPAND_ALL_32B|检查项同上| |4|||随机生成N条32位指令，
检查是否合法|RVC_EXPAND_RANDOM_32B|POS_{i}：为了保证随机指令足够多，判断随机
生成的指令中第i位是否为1(0 <= i < 32)| -|5|TBD||||| +|5|decodestage|输出反压|检查全阻塞期间无握手且 payload 保持，恢复后仅传输一次|DECODE_OUTPUT_READY|ALL_STALLED、ALL_READY| +|6|decodestage|复杂指令多 uop 解码|检查输入结束后继续运行直至最后一个 uop 握手|DECODE_COMPLEX_DRAIN|COMPLEX_LAST_UOP_FIRED| +|7|decodestage|确定性非法指令|检查固定非法编码仅完成一次且置异常位|DECODE_KNOWN_ILLEGAL|KNOWN_ILLEGAL_COMPLETED| ## Env提供的验证接口(API) @@ -73,7 +75,27 @@ class RVCExpander def rvc_expander(request) -> RVCExpander: ``` -#### 2. TBD +#### 2. decodestage + +```python +# 同步驱动 6 路 output ready；不允许部分 lane ready +decoder.Set_output_ready(ready) + +# 驱动并采样一个周期，在上升沿前返回 ready/valid 握手快照 +cycle = decoder.Run_cycle(insts=(), output_ready=True) + +# 按输入握手推进，并在所有指令的 lastUop 握手后结束 +success = decode_run(decoder, inst_list, need_log_file=False) + +# 调试或协议检查可请求统计信息 +success, stats = decode_run( + decoder, inst_list, need_log_file=False, return_stats=True +) +``` + +`decode_run` 会在输入全部接收后清空 input valid，继续运行直至所有架构指令的 `lastUop` 完成；如果超过按输入数量计算的超时上限则失败。默认返回值仍为布尔值，以兼容已有用例。 + +随机指令用例中的 syntax-only disasm 没有当前扩展、CSR、VType 状态，也可能不了解 DUT 的自定义编码，因此只用于辅助诊断；随机结果严格检查指令传输与顺序，不把 disasm 当作异常位 Golden。异常位由固定合法序列和固定非法编码用例做严格比较。固定的复杂指令排空用例会先执行 `vsetvli`，并逐条检查多 uop 的 `firstUop`、`lastUop` 和 `numUops`。 ## 用例说明 diff --git a/ut_backend/ctrl_block/decode/env/decode_wrapper.py b/ut_backend/ctrl_block/decode/env/decode_wrapper.py index 70709f44..17fa6941 100644 --- a/ut_backend/ctrl_block/decode/env/decode_wrapper.py +++ b/ut_backend/ctrl_block/decode/env/decode_wrapper.py @@ -125,25 +125,63 @@ class Decode(toffee.Bundle): def __init__(self, dut: DUTDecodeStage): super().__init__() self.dut = dut - for i in range(6): - setattr(self, f"in_data_{i}", toffee.Bundle.from_prefix(f"io_in_{i}_", dut)) - setattr(self, f"out_data_{i}", toffee.Bundle.from_prefix(f"io_out_{i}_", dut)) - self.input_inst = [getattr(self, f"in_data_{i}") for i in range(6)] - self.output_instrution = [getattr(self, f"out_data_{i}") for i in range(6)] - self.io = toffee.Bundle.from_prefix(f"io_", dut) - self.bind(dut) + self._input_pin_names = tuple( + name + for name in dir(dut) + if getattr(getattr(dut, name), "mIOType", None) == 0 + and name not in {"clock", "reset"} + ) + for name in self._input_pin_names: + getattr(dut, name).AsImmWrite() + payload_prefix = "io_out_0_bits_" + self._output_payload_fields = tuple(sorted( + name[len(payload_prefix):] + for name in dir(dut) + if name.startswith(payload_prefix) + and hasattr(getattr(dut, name), "value") + )) + required_payload_fields = { + "instr", "firstUop", "lastUop", "uopIdx", "numUops" + } + missing_payload_fields = required_payload_fields.difference( + self._output_payload_fields + ) + if missing_payload_fields: + raise RuntimeError( + "DecodeStage output payload discovery missed: " + + ", ".join(sorted(missing_payload_fields)) + ) + self._last_cycle_state = {} + self._run_stats = {} def SetDefaultValue(self): - """Use Boundle to set DUT pin value""" - self.io.assign({ - "out_0_ready": 0b1, - "out_1_ready": 0b1, - "out_2_ready": 0b1, - "out_3_ready": 0b1, - "out_4_ready": 0b1, - "out_5_ready": 0b1, - "*": 0, # Set other pins to 0 - }) + """Drive every non-clock input to a deterministic default value.""" + for name in self._input_pin_names: + getattr(self.dut, name).value = 0 + self.Set_output_ready(True) + + def Set_output_ready(self, ready): + """Drive the six output ready signals with the same value. + + DecodeStage only uses lane 0 ready to calculate its global output + capacity. The integration contract therefore requires all six lanes + to be driven together. + """ + if type(ready) not in (bool, int) or ready not in (0, 1): + raise ValueError("output ready must be boolean") + ready_value = int(bool(ready)) + for lane in range(6): + getattr(self.dut, f"io_out_{lane}_ready").value = ready_value + + def Get_output_ready_state(self): + """Return the output ready state and reject partial-ready driving.""" + ready_state = tuple( + int(getattr(self.dut, f"io_out_{lane}_ready").value) + for lane in range(6) + ) + assert len(set(ready_state)) == 1, \ + "DecodeStage output ready must be identical on all lanes" + return ready_state def Reset(self): """Directly operate the dut pins to reset""" @@ -155,23 +193,36 @@ def Reset(self): self.dut.Step(1) def Input_instruction(self, i, valid, instr, isRVC, brType, isCall, isRet, pred_taken, instr_ex): - self.input_inst[i].valid.value = valid - self.input_inst[i].bits_instr.value = instr - if hasattr(self.input_inst[i], "bits_foldpc"): - self.input_inst[i].bits_foldpc.value = 0 - self.input_inst[i].bits_exceptionVec_2.value = instr_ex + getattr(self.dut, f"io_in_{i}_valid").value = valid + getattr(self.dut, f"io_in_{i}_bits_instr").value = instr + foldpc = getattr(self.dut, f"io_in_{i}_bits_foldpc", None) + if foldpc is not None: + foldpc.value = 0 for j in range(24): p = getattr(self.dut, f'io_in_{i}_bits_exceptionVec_{j}', None) - if p: + if p is not None: p.value = 0 - self.input_inst[i].bits_trigger.value = 0 - self.input_inst[i].bits_preDecodeInfo_isRVC.value = isRVC - self.input_inst[i].bits_preDecodeInfo_brType.value = brType - self.input_inst[i].bits_pred_taken.value = pred_taken - self.input_inst[i].bits_crossPageIPFFix.value = 0 - self.input_inst[i].bits_ftqPtr_flag.value = 0 - self.input_inst[i].bits_ftqPtr_value.value = 0 - self.input_inst[i].bits_ftqOffset.value = 0 + getattr(self.dut, f"io_in_{i}_bits_exceptionVec_2").value = instr_ex + getattr(self.dut, f"io_in_{i}_bits_trigger").value = 0 + getattr( + self.dut, f"io_in_{i}_bits_preDecodeInfo_isRVC" + ).value = isRVC + getattr( + self.dut, f"io_in_{i}_bits_preDecodeInfo_brType" + ).value = brType + getattr( + self.dut, f"io_in_{i}_bits_preDecodeInfo_isCall" + ).value = isCall + getattr( + self.dut, f"io_in_{i}_bits_preDecodeInfo_isRet" + ).value = isRet + getattr(self.dut, f"io_in_{i}_bits_pred_taken").value = pred_taken + getattr( + self.dut, f"io_in_{i}_bits_crossPageIPFFix" + ).value = 0 + getattr(self.dut, f"io_in_{i}_bits_ftqPtr_flag").value = 0 + getattr(self.dut, f"io_in_{i}_bits_ftqPtr_value").value = 0 + getattr(self.dut, f"io_in_{i}_bits_ftqOffset").value = 0 def FromCSR_illegalInst(self, sfenceVMA, sfencePart, hfenceGVMA, hfenceVVMA, hlsv, fsIsOff, vsIsOff, wfi, frm): self.dut.io_fromCSR_illegalInst_sfenceVMA.value = sfenceVMA @@ -192,50 +243,232 @@ def FromCSR_virtualInst(self, sfenceVMA, sfencePart, hfence, hlsv, wfi): self.dut.io_fromCSR_virtualInst_wfi.value = wfi def Get_input_ready(self, i): - return self.input_inst[i].ready.value + return getattr(self.dut, f"io_in_{i}_ready").value def Get_allow_input_number(self): - cnt = 0 - for i in range(6): - if self.input_inst[i].ready.value == 1: - cnt = i + 1 - else: - break - return cnt + """Return the contiguous prefix of inputs accepted this cycle.""" + fire_mask = self.Get_input_fire_mask() + return next( + (lane for lane, fired in enumerate(fire_mask) if not fired), + len(fire_mask), + ) + + def Get_input_fire_mask(self): + """Return input handshakes for the current combinational cycle.""" + return tuple( + int( + bool(getattr(self.dut, f"io_in_{lane}_valid").value) + and bool(getattr(self.dut, f"io_in_{lane}_ready").value) + ) + for lane in range(6) + ) + + def Get_input_valid_mask(self): + """Return the current input valid state.""" + return tuple( + int(bool(getattr(self.dut, f"io_in_{lane}_valid").value)) + for lane in range(6) + ) + + def Get_input_ready_state(self): + """Return the current input ready state.""" + return tuple( + int(bool(getattr(self.dut, f"io_in_{lane}_ready").value)) + for lane in range(6) + ) + + def Get_output_fire_mask(self): + """Return output handshakes for the current combinational cycle.""" + return tuple( + int( + bool(getattr(self.dut, f"io_out_{lane}_valid").value) + and bool(getattr(self.dut, f"io_out_{lane}_ready").value) + ) + for lane in range(6) + ) + + def Get_output_valid_mask(self): + """Return the current output valid state.""" + return tuple( + int(bool(getattr(self.dut, f"io_out_{lane}_valid").value)) + for lane in range(6) + ) + + def Get_output_payloads(self): + """Copy every output payload field into ordinary Python values.""" + output_payloads = [] + for lane in range(6): + if getattr(self.dut, f"io_out_{lane}_valid").value != 1: + continue + fields = tuple( + (field, int(getattr( + self.dut, + f"io_out_{lane}_bits_{field}" + ).value)) + for field in self._output_payload_fields + ) + output_payloads.append((lane, fields)) + return tuple(output_payloads) + + def Get_fired_uops(self): + """Return the uops transferred on the current output handshake.""" + fired_uops = [] + for lane in range(6): + if getattr(self.dut, f"io_out_{lane}_valid").value != 1 or \ + getattr(self.dut, f"io_out_{lane}_ready").value != 1: + continue + fired_uops.append({ + "lane": lane, + "instr": int(getattr( + self.dut, f"io_out_{lane}_bits_instr" + ).value), + "first_uop": int(getattr( + self.dut, f"io_out_{lane}_bits_firstUop" + ).value), + "last_uop": int(getattr( + self.dut, f"io_out_{lane}_bits_lastUop" + ).value), + "uop_idx": int(getattr( + self.dut, f"io_out_{lane}_bits_uopIdx" + ).value), + "num_uops": int(getattr( + self.dut, f"io_out_{lane}_bits_numUops" + ).value), + }) + return tuple(fired_uops) + + def Get_last_cycle_state(self): + """Return the most recent Run_cycle handshake snapshot.""" + return self._last_cycle_state + + def Set_run_stats(self, stats): + """Store the most recent decode_run statistics for functional coverage.""" + self._run_stats = dict(stats) + + def Get_run_stats(self): + """Return statistics from the most recent decode_run call.""" + return self._run_stats def Input_instruction_list(self, insts, valid): + if len(insts) > 6: + raise ValueError("DecodeStage accepts at most six instructions") for i in range(6): self.Input_instruction(i, 0, 0, 0, 0, 0, 0, 0, 0) for i, inst in enumerate(insts): self.Input_instruction(i, valid, inst[0], 0, 0, 0, 0, 0, inst[3]) - def Get_decode_result(self): + def Get_decode_result(self, fired_only=False): + """Return completed architectural instructions. + + The default preserves the historical valid-only behavior. Protocol + scoreboards should pass ``fired_only=True`` so a stalled output is not + consumed more than once. + """ insts_result = [] num = 0 for i in range(6): - if self.output_instrution[i].valid.value == 1 and self.output_instrution[i].bits_lastUop.value == 1: - insts_result.append((self.output_instrution[i].bits_instr.value, - self.output_instrution[i].bits_exceptionVec_2.value or self.output_instrution[ - i].bits_exceptionVec_22.value, - self.output_instrution[i].bits_firstUop.value)) + output_valid = getattr(self.dut, f"io_out_{i}_valid").value == 1 + output_ready = getattr(self.dut, f"io_out_{i}_ready").value == 1 + output_fired = output_valid and output_ready + output_last_uop = getattr( + self.dut, f"io_out_{i}_bits_lastUop" + ).value + if output_valid and output_last_uop == 1 and ( + not fired_only or output_fired): + output_instr = getattr( + self.dut, f"io_out_{i}_bits_instr" + ).value + output_exception = ( + getattr( + self.dut, f"io_out_{i}_bits_exceptionVec_2" + ).value + or getattr( + self.dut, f"io_out_{i}_bits_exceptionVec_22" + ).value + ) + output_first_uop = getattr( + self.dut, f"io_out_{i}_bits_firstUop" + ).value + insts_result.append(( + output_instr, + output_exception, + output_first_uop, + )) num = num + 1 return num, insts_result + def Run_cycle(self, insts=(), output_ready=True, cover_group=None): + """Drive and sample one ready/valid cycle before its rising edge.""" + insts = list(insts) + if len(insts) > 6: + raise ValueError("Run_cycle accepts at most six instructions") + self.Set_output_ready(output_ready) + self.Input_instruction_list(insts, int(bool(insts))) + self.dut.RefreshComb() + + input_fire_mask = self.Get_input_fire_mask() + seen_gap = False + for fired in input_fire_mask: + if not fired: + seen_gap = True + elif seen_gap: + raise AssertionError("DecodeStage input fire must be a prefix") + + _, completed = self.Get_decode_result(fired_only=True) + fired_uops = self.Get_fired_uops() + self._last_cycle_state = { + "accepted": sum(input_fire_mask), + "input_valid_mask": self.Get_input_valid_mask(), + "input_ready_state": self.Get_input_ready_state(), + "input_instructions": tuple( + int(getattr( + self.dut, f"io_in_{lane}_bits_instr" + ).value) + for lane in range(6) + ), + "input_fire_mask": input_fire_mask, + "output_ready_state": self.Get_output_ready_state(), + "output_valid_mask": self.Get_output_valid_mask(), + "output_fire_mask": self.Get_output_fire_mask(), + "output_payloads": self.Get_output_payloads(), + "fired_uops": fired_uops, + "completed": tuple(completed), + "complex_last_uop_fired": any( + uop["last_uop"] and not uop["first_uop"] + for uop in fired_uops + ), + } + if cover_group is not None: + cover_group.sample() + self.dut.Step(1) + return self._last_cycle_state + def Get_decode_checkpoint_illeagl_inst(self): illegal = 0 for i in range(6): - if self.output_instrution[i].valid.value == 1 and self.output_instrution[i].bits_lastUop.value == 1: - if (self.output_instrution[i].bits_exceptionVec_2.value or self.output_instrution[ - i].bits_exceptionVec_22.value): + if getattr(self.dut, f"io_out_{i}_valid").value == 1 and \ + getattr(self.dut, f"io_out_{i}_bits_lastUop").value == 1: + if getattr( + self.dut, + f"io_out_{i}_bits_exceptionVec_2" + ).value or getattr( + self.dut, + f"io_out_{i}_bits_exceptionVec_22" + ).value: illegal = 1 + if self._run_stats.get("saw_illegal_completion", False): + illegal = 1 return illegal def Get_decode_checkpoint_complex_inst(self): complex = 0 for i in range(6): - if self.output_instrution[i].valid.value == 1 and self.output_instrution[i].bits_lastUop.value == 1 and \ - self.output_instrution[i].bits_firstUop.value != 1: + if getattr(self.dut, f"io_out_{i}_valid").value == 1 and \ + getattr(self.dut, f"io_out_{i}_bits_lastUop").value == 1 and \ + getattr(self.dut, f"io_out_{i}_bits_firstUop").value != 1: complex = 1 + if self._run_stats.get("saw_complex_last_uop", False): + complex = 1 return complex @@ -259,7 +492,11 @@ def comapre_result(ref_value_list, dut_value_list, num): return None else: for i in range(num): - if ref_value_list[i][1] != dut_value_list[i][1]: + expected_exception = ( + ref_value_list[i][1] or ref_value_list[i][3] + ) + if ref_value_list[i][0] != dut_value_list[i][0] or \ + expected_exception != dut_value_list[i][1]: debug("================================") debug(ref_value_list[i]) debug(dut_value_list[i]) @@ -323,9 +560,16 @@ def comapre_result_in_text(ref_value_list, dut_value_list, num): return None else: for i in range(num): - # Do not change to assert for now. First, focus on implementing the functionality. Assert directly stops execution.。 - if (ref_value_list[i][2] == 0 or (ref_value_list[i][0] == dut_value_list[i][0])) and ( - ref_value_list[i][1] == dut_value_list[i][1]): + disasm_text = ref_value_list[i][2] + instruction_matches = ( + disasm_text == 0 + or ref_value_list[i][0] == dut_value_list[i][0] + ) + # Random syntax-only disassembly has no architectural extension, + # CSR or VType context, and may not know DUT-specific encodings. + # This path therefore checks transfer/order only; deterministic + # exception tests use comapre_result for an exact comparison. + if instruction_matches: # print("Meets expectations:（￣︶￣）↗") good_info = f"good ----- ref: {ref_value_list[i][0]}, {ref_value_list[i][1]}, {ref_value_list[i][2]}, {ref_value_list[i][3]}" write_all_info_to_file(good_info) @@ -333,8 +577,7 @@ def comapre_result_in_text(ref_value_list, dut_value_list, num): # print("Not meeting expectations: <(_ _)>") bad_info = f"bad ----- ref: {ref_value_list[i][0]}, {ref_value_list[i][1]}, {ref_value_list[i][2]}, {ref_value_list[i][3]}, old inst: {ref_value_list[i][4]}, dut: {dut_value_list[i][0]}, {dut_value_list[i][1]}, complex: {dut_value_list[i][2] == 0}" write_all_info_to_file(bad_info) - if (ref_value_list[i][2][0] != 'v'): - write_err_info_to_file(bad_info) + write_err_info_to_file(bad_info) eq = False return eq @@ -392,36 +635,119 @@ def convert_reference_format(rvc_expander, ref_insts, need_expand, disasm_func, # The main part of the test environment -def decode_run(decoder, inst_list, need_log_file, log_file_name=None): +def decode_run(decoder, inst_list, need_log_file, log_file_name=None, + timeout_cycles=None, return_stats=False, cover_group=None): if need_log_file: open_log_file(log_file_name) decoder.SetDefaultValue() decoder.Reset() + decoder.Set_run_stats({}) pos = 0 detect_pos = 0 insts_length = len(inst_list) - sub_valid = 1 - sub_list = inst_list[pos:pos + 6] success = True - while True: - decoder.Input_instruction_list(sub_list, sub_valid) - decoder.dut.Step(1) - allow_number = decoder.Get_allow_input_number() - if allow_number > 0: - pos = pos + allow_number - sub_list = inst_list[pos:pos + allow_number] - sub_valid = 1 - num, step_result_list = decoder.Get_decode_result() - if num > 0: - if need_log_file == True: - if comapre_result_in_text(inst_list[detect_pos:detect_pos + num], step_result_list, num) == False: - success = False - else: - if comapre_result(inst_list[detect_pos:detect_pos + num], step_result_list, num) == False: + if timeout_cycles is None: + timeout_cycles = max(64, 16 * insts_length + 16) + stats = { + "accepted": 0, + "completed": 0, + "cycles": 0, + "uops_fired": 0, + "uop_trace": [], + "last_input_fire_cycle": None, + "last_completion_cycle": None, + "timed_out": False, + "drained": False, + "saw_illegal_completion": False, + "saw_complex_last_uop": False, + "completed_instructions": [], + } + + try: + while pos < insts_length or detect_pos < insts_length: + if stats["cycles"] >= timeout_cycles: + stats["timed_out"] = True + success = False + debug(f"Decode timeout after {timeout_cycles} cycles") + break + + cycle = stats["cycles"] + cycle_state = decoder.Run_cycle( + inst_list[pos:pos + 6], + output_ready=True, + cover_group=cover_group, + ) + accepted = cycle_state["accepted"] + old_pos = pos + pos = pos + accepted + stats["accepted"] = pos + if old_pos < insts_length and pos >= insts_length: + stats["last_input_fire_cycle"] = cycle + + step_result_list = list(cycle_state["completed"]) + num = len(step_result_list) + stats["uops_fired"] += len(cycle_state["fired_uops"]) + stats["uop_trace"].extend( + dict(uop, cycle=cycle) + for uop in cycle_state["fired_uops"] + ) + stats["saw_complex_last_uop"] = ( + stats["saw_complex_last_uop"] + or cycle_state["complex_last_uop_fired"] + ) + stats["saw_illegal_completion"] = ( + stats["saw_illegal_completion"] + or any(bool(result[1]) for result in step_result_list) + ) + if detect_pos + num > pos: + success = False + debug("Decode produced a completion before input handshake") + if detect_pos + num > insts_length: + success = False + num = insts_length - detect_pos + step_result_list = step_result_list[:num] + if num > 0: + ref_results = inst_list[detect_pos:detect_pos + num] + if need_log_file: + if comapre_result_in_text( + ref_results, step_result_list, num) is False: + success = False + elif comapre_result( + ref_results, step_result_list, num) is False: success = False - detect_pos = detect_pos + num - if pos >= insts_length: - break - decoder.dut.Step(10) - close_log_file() + detect_pos = detect_pos + num + stats["completed"] = detect_pos + stats["last_completion_cycle"] = cycle + stats["completed_instructions"].extend( + int(result[0]) for result in step_result_list + ) + stats["cycles"] += 1 + + stats["drained"] = ( + pos == insts_length and detect_pos == insts_length + ) + if success and stats["drained"]: + quiet_state = decoder.Run_cycle( + (), + output_ready=True, + cover_group=cover_group, + ) + stats["cycles"] += 1 + stats["uops_fired"] += len(quiet_state["fired_uops"]) + stats["uop_trace"].extend( + dict(uop, cycle=stats["cycles"] - 1) + for uop in quiet_state["fired_uops"] + ) + if quiet_state["fired_uops"] or quiet_state["completed"]: + success = False + finally: + close_log_file() + + stats["completed_instructions"] = tuple( + stats["completed_instructions"] + ) + stats["uop_trace"] = tuple(stats["uop_trace"]) + decoder.Set_run_stats(stats) + if return_stats: + return success, stats return success diff --git a/ut_backend/ctrl_block/decode/test_rv_decode.py b/ut_backend/ctrl_block/decode/test_rv_decode.py index a457390b..38362652 100644 --- a/ut_backend/ctrl_block/decode/test_rv_decode.py +++ b/ut_backend/ctrl_block/decode/test_rv_decode.py @@ -18,6 +18,165 @@ from tools.disasm import disasmbly, libdisasm +def _decode_references(instructions, expected_exception=0): + """Build deterministic references without invoking the random generator.""" + return [ + (instruction, expected_exception, "", 0, instruction) + for instruction in instructions + ] + + +@toffee_test.testcase +async def test_decode_backpressure_holds_and_releases(decoder): + """Stalled outputs stay stable and transfer exactly once after release.""" + instructions = ( + 0x00100093, # addi x1, x0, 1 + 0x00200113, # addi x2, x0, 2 + 0x00300193, # addi x3, x0, 3 + 0x00400213, # addi x4, x0, 4 + 0x00500293, # addi x5, x0, 5 + 0x00600313, # addi x6, x0, 6 + ) + references = _decode_references(instructions) + g.add_cover_point(decoder, { + "ALL_STALLED": lambda env: env.Get_output_ready_state() == (0,) * 6, + "ALL_READY": lambda env: env.Get_output_ready_state() == (1,) * 6, + }, name="DECODE_OUTPUT_READY") + + decoder.SetDefaultValue() + decoder.Reset() + stalled_first = decoder.Run_cycle( + references, output_ready=False, cover_group=g + ) + stalled_second = decoder.Run_cycle( + references, output_ready=False, cover_group=g + ) + released = decoder.Run_cycle( + references, output_ready=True, cover_group=g + ) + quiet = decoder.Run_cycle((), output_ready=True, cover_group=g) + + assert stalled_first["input_fire_mask"] == (0,) * 6 + assert stalled_first["output_fire_mask"] == (0,) * 6 + assert stalled_first["output_valid_mask"] == \ + stalled_second["output_valid_mask"] + assert stalled_first["output_valid_mask"] == (1,) * 6 + assert stalled_second["input_fire_mask"] == (0,) * 6 + assert stalled_second["output_fire_mask"] == (0,) * 6 + assert stalled_first["output_payloads"] == stalled_second["output_payloads"] + assert released["input_fire_mask"] == (1,) * 6 + assert released["output_fire_mask"] == (1,) * 6 + assert tuple( + int(result[0]) for result in released["completed"] + ) == instructions + assert quiet["fired_uops"] == () + assert quiet["completed"] == () + + g.mark_function( + "DECODE_OUTPUT_READY", + test_decode_backpressure_holds_and_releases, + bin_name=["ALL_STALLED", "ALL_READY"], + ) + + +@toffee_test.testcase +async def test_decode_drains_complex_tail(decoder): + """The runner waits for a complex instruction's last uop after input ends.""" + instructions = ( + 0x00100093, # addi x1, x0, 1 + 0x0C007057, # vsetvli zero, zero, e8, m1, ta, ma + 0xC221A257, # vwaddu.vv v4, v2, v3 + ) + references = _decode_references(instructions) + g.add_cover_point(decoder, { + "COMPLEX_LAST_UOP_FIRED": lambda env: env.Get_last_cycle_state().get( + "complex_last_uop_fired", False + ), + }, name="DECODE_COMPLEX_DRAIN") + + success, stats = decode_run( + decoder, + references, + need_log_file=False, + return_stats=True, + cover_group=g, + ) + + assert success is True, ( + stats["completed_instructions"], + stats["uop_trace"], + ) + assert stats["timed_out"] is False + assert stats["drained"] is True + assert stats["accepted"] == len(instructions) + assert stats["completed"] == len(instructions) + assert stats["completed_instructions"] == instructions + assert stats["uops_fired"] == 5 + assert stats["saw_complex_last_uop"] is True + assert stats["last_completion_cycle"] > stats["last_input_fire_cycle"] + for instruction in instructions[1:]: + instruction_uops = tuple( + uop for uop in stats["uop_trace"] + if uop["instr"] == instruction + ) + assert instruction_uops + assert instruction_uops[0]["first_uop"] == 1 + assert all( + uop["first_uop"] == 0 for uop in instruction_uops[1:] + ) + assert instruction_uops[-1]["last_uop"] == 1 + assert all( + uop["last_uop"] == 0 for uop in instruction_uops[:-1] + ) + assert all( + uop["num_uops"] == len(instruction_uops) + for uop in instruction_uops + ) + assert all( + 0 <= uop["uop_idx"] < len(instruction_uops) + for uop in instruction_uops + ) + + g.mark_function( + "DECODE_COMPLEX_DRAIN", + test_decode_drains_complex_tail, + bin_name="COMPLEX_LAST_UOP_FIRED", + ) + + +@toffee_test.testcase +async def test_decode_known_illegal_instructions(decoder): + """Known invalid encodings complete once with the illegal exception set.""" + instructions = (0x00000000, 0xFFFFFFFF) + references = _decode_references( + instructions, expected_exception=1 + ) + g.add_cover_point(decoder, { + "KNOWN_ILLEGAL_COMPLETED": lambda env: any( + bool(result[1]) + for result in env.Get_last_cycle_state().get("completed", ()) + ), + }, name="DECODE_KNOWN_ILLEGAL") + + success, stats = decode_run( + decoder, + references, + need_log_file=False, + return_stats=True, + cover_group=g, + ) + + assert success is True, stats["completed_instructions"] + assert stats["drained"] is True + assert stats["completed_instructions"] == instructions + assert stats["saw_illegal_completion"] is True + g.mark_function( + "DECODE_KNOWN_ILLEGAL", + test_decode_known_illegal_instructions, + bin_name="KNOWN_ILLEGAL_COMPLETED", + ) + + @toffee_test.testcase async def test_rvc_inst(decoder, rvc_expander): """