#!/usr/bin/env python3 """test_pram_batch_sim.py - 主机侧仿真:验证 pramRead/pramWrite 批量化(1.6.0) 与旧实现(逐 DWORD 轮询)逐字节等价。 模型依据 LAN9252 DS00001909A §12.11.3/12.11.4: - PRAM FIFO 16 deep x 32bit;AVAIL_CNT(CMD bits12:8)=可连续读/写的 DWORD 数 - 读引擎按 DWORD 对齐向下取源:第 i 个 FIFO DWORD = 源地址 aligned(addr)+4i 起 4 字节; 线性有效窗口 = [0, len),即 DWORD i 的字节 k 有效 iff 0 <= 4i+k-lead < len - 写引擎按同一线性窗口落盘,主站推满一个 DWORD 计一次 - 引擎以有限速率搬运(rate),主站 respect AVAIL_CNT 就不会溢 FIFO/读空洞 对全部 (addr 对齐 x len 1..70/96/384/640 x 引擎节奏) 断言: 新旧算法产出逐字节一致,且主站从不超出 AVAIL_CNT(引擎内 assert)。 """ import random FIFO_DW = 16 # FIFO 深度 16 DWORD WINDOW_DW = 8 # 别名窗口一程 8 DWORD (0x00-0x1C / 0x20-0x3C) PRAM_BASE = 0x1000 PRAM_SIZE = 0x1000 def byte_valid(i, k, lead, ln): """DWORD i 的字节 k 是否落在线性有效窗口 [0, ln) 内。""" return 0 <= 4 * i + k - lead < ln class ReadEngine: def __init__(self, mem, addr, ln, rate): self.mem = mem self.lead = addr & 3 self.ln = ln self.rate = max(1, rate) self.total_dw = (ln + self.lead + 3) // 4 self.start = (addr & ~3) - PRAM_BASE # mem 内偏移 self.pushed = 0 self.popped = 0 self.data = [] def pump(self): n = 0 while (self.pushed - self.popped < FIFO_DW and self.pushed < self.total_dw and n < self.rate): i = self.pushed dw = bytearray(4) for k in range(4): if byte_valid(i, k, self.lead, self.ln): dw[k] = self.mem[self.start + 4 * i + k] self.data.append(bytes(dw)) self.pushed += 1 n += 1 return n def avail_cnt(self): return self.pushed - self.popped def pop(self): assert self.popped < self.pushed, "主站弹出超出 AVAIL_CNT!" dw = self.data[self.popped] self.popped += 1 return dw def pram_read_old(eng, addr, ln, poll_budget=2000): out = bytearray() done = 0 polls = 0 while done < ln: while eng.avail_cnt() == 0: polls += 1 if polls > poll_budget: return bytes(out), False eng.pump() dw = eng.pop() lead = (addr & 3) if done == 0 else 0 take = min(ln - done, 4 - lead) out += dw[lead:lead + take] done += take eng.pump() return bytes(out), True def pram_read_new(eng, addr, ln, poll_budget=2000): out = bytearray() done = 0 polls = 0 while done < ln: while eng.avail_cnt() == 0: polls += 1 if polls > poll_budget: return bytes(out), False eng.pump() batch = eng.avail_cnt() if batch == 0 or batch > WINDOW_DW: batch = 1 lead0 = (addr & 3) if done == 0 else 0 need_dw = (ln - done + lead0 + 3) // 4 batch = min(batch, need_dw) for _ in range(batch): dw = eng.pop() lead = (addr & 3) if done == 0 else 0 take = min(ln - done, 4 - lead) out += dw[lead:lead + take] done += take eng.pump() return bytes(out), True class WriteEngine: def __init__(self, addr, ln, rate): self.lead = addr & 3 self.ln = ln self.rate = max(1, rate) self.total_dw = (ln + self.lead + 3) // 4 self.start = (addr & ~3) - PRAM_BASE self.fifo = [] self.consumed_dw = 0 self.mem = bytearray(PRAM_SIZE) def pump(self): n = 0 while self.fifo and n < self.rate: dw = self.fifo.pop(0) i = self.consumed_dw for k in range(4): if byte_valid(i, k, self.lead, self.ln): self.mem[self.start + 4 * i + k] = dw[k] self.consumed_dw += 1 n += 1 return n def space_cnt(self): return FIFO_DW - len(self.fifo) def push(self, dw): assert len(self.fifo) < FIFO_DW, "主站写溢出 FIFO!" self.fifo.append(bytes(dw)) def _write_common(eng, addr, buf, poll_budget, batched): ln = len(buf) done = 0 polls = 0 while done < ln: while eng.space_cnt() == 0: polls += 1 if polls > poll_budget: return False eng.pump() n_iter = 1 if batched: batch = eng.space_cnt() if batch == 0 or batch > WINDOW_DW: batch = 1 lead0 = (addr & 3) if done == 0 else 0 need_dw = (ln - done + lead0 + 3) // 4 n_iter = min(batch, need_dw) for _ in range(n_iter): lead = (addr & 3) if done == 0 else 0 take = min(ln - done, 4 - lead) dw = bytearray(4) for i in range(take): dw[lead + i] = buf[done + i] eng.push(dw) done += take eng.pump() budget = poll_budget while eng.consumed_dw < eng.total_dw and budget: eng.pump() budget -= 1 return eng.consumed_dw == eng.total_dw def pram_write_old(eng, addr, buf, poll_budget=2000): return _write_common(eng, addr, buf, poll_budget, batched=False) def pram_write_new(eng, addr, buf, poll_budget=2000): return _write_common(eng, addr, buf, poll_budget, batched=True) def run(): random.seed(20260905) cases = 0 lens = list(range(1, 71)) + [96, 384, 640] # 覆盖 SM2(384B)/SM3(640B) 实际长度 for align in range(4): for ln in lens: for rate in (1, 3, 16, 64): src = bytes(random.randrange(1, 256) for _ in range(ln)) base = random.randrange(PRAM_SIZE - ln - 8) addr = PRAM_BASE + base + align mem = bytearray(PRAM_SIZE) mem[base + align: base + align + ln] = src # 读等价:新旧都须完整取回 src e1 = ReadEngine(mem, addr, ln, rate); e1.pump() r_old, ok1 = pram_read_old(e1, addr, ln) e2 = ReadEngine(mem, addr, ln, rate); e2.pump() r_new, ok2 = pram_read_new(e2, addr, ln) assert ok1 and ok2, f"read 未完成 align={align} len={ln} rate={rate}" assert r_old == src, f"read 旧实现不符 align={align} len={ln} rate={rate}" assert r_new == src, f"read 新实现不符 align={align} len={ln} rate={rate}" # 写等价:新旧都须把 src 落到同一片 PRAM w1 = WriteEngine(addr, ln, rate); w1.pump() assert pram_write_old(w1, addr, src), f"write 未完成 {align}/{ln}/{rate}" w2 = WriteEngine(addr, ln, rate); w2.pump() assert pram_write_new(w2, addr, src), f"write 未完成 {align}/{ln}/{rate}" assert bytes(w1.mem[base + align: base + align + ln]) == src, \ f"write 旧不符 {align}/{ln}/{rate}" assert bytes(w2.mem[base + align: base + align + ln]) == src, \ f"write 新不符 {align}/{ln}/{rate}" cases += 1 print(f"OK: {cases} 组 (对齐x长度x引擎节奏) 新旧算法逐字节等价") # 统计新实现的轮询次数收益(代表性场景:SM3 640B DWORD 对齐,引擎速率 16) for rate in (1, 3, 16): src = bytes(random.randrange(1, 256) for _ in range(640)) e = ReadEngine(bytearray(PRAM_SIZE), PRAM_BASE, 640, rate); e.pump() _, p1 = _count_polls(e, 640) print(f" 640B@rate{rate}: 新算法主站外圈轮询 {p1} 次") def _count_polls(eng, ln): done = 0 polls = 0 while done < ln: while eng.avail_cnt() == 0: polls += 1 eng.pump() batch = min(eng.avail_cnt(), WINDOW_DW) lead0 = 0 if done == 0 else 0 batch = min(batch, (ln - done + lead0 + 3) // 4) for _ in range(batch): eng.pop() done += 4 return None, polls if __name__ == "__main__": run()