"""BBA register file — EXI domain. Decodes EXI transactions (2-byte header + N data bytes), reads/writes the BBA register space, and owns all AsyncFIFO / PulseSynchronizer CDC primitives. Transaction header format -------------------------- Byte 0 [7] write_flag [6:0] addr[12:6] Byte 1 [7:2] addr[5:0] [1:0] xfer_len−1 (0=1B, 1=2B, 2=3B, 3=4B) Addresses 0x0000–0x00FF : register file (sparse individual Signals, exi domain). Addresses 0x0100–0x1FFF : SPRAM ring buffer (sync domain, prefetch FIFOs). """ from amaranth import * from amaranth.lib.fifo import SyncFIFO __all__ = ["BBARegisterFile"] # Register addresses _NCRA = 0x00 _IMR = 0x08 _IR = 0x09 _RWP_LO = 0x16 _RWP_HI = 0x17 _RRP_LO = 0x18 _RRP_HI = 0x19 # First page of the SPRAM RX ring (pages 1-15; page 0 is the register region). # Must match rx_frame_assembler._RX_PAGE_FIRST. _RX_PAGE_FIRST = 0x01 _PAR0 = 0x20 _PAR1 = 0x21 _PAR2 = 0x22 _PAR3 = 0x23 _PAR4 = 0x24 _PAR5 = 0x25 _NWAYS = 0x31 _HIPR = 0x3A _TWD_LO = 0x34 _TWD_HI = 0x35 _TXDATA = 0x48 # Read-only hardcoded values _NWAYS_VAL = 0x17 _HIPR_VAL = 0x01 # Device ID returned on first 4-byte read of addr 0x0000 _DEVICE_ID = [0x04, 0x02, 0x02, 0x00] class BBARegisterFile(Elaboratable): """EXI transaction decoder and BBA register file with CDC bridges. Sync-domain FIFO/pulse ports are wired by BBATop to the sync-domain modules. """ def __init__(self): # ── EXI byte-stream interface (exi domain, from/to ExiCapture) ──── # RX: received bytes (header + write data + read dummies) — FWFT read # side of ExiCapture's rx_fifo. self.rx_data = Signal(8) self.rx_rdy = Signal() self.rx_en = Signal() # TX: response bytes pushed proactively into ExiCapture's tx_fifo. self.tx_data = Signal(8) self.tx_en = Signal() self.tx_rdy = Signal() # High while an EXI transaction is in progress (from ExiCapture). # SPRAM reads stream until this deasserts → supports variable-length # (DMA) bulk reads, not just ≤4-byte immediate transfers. self.cs_active = Signal() # One-shot: set at reset, cleared by the GC's first NCRA reset. While # set, a command-space read of register 0 returns the device ID -- but # the REPLY is produced by SPIMode3Slave's fast path, not here, because # a 2-byte command header leaves too little time for a round trip # through both byte FIFOs. This output is what arms that path. self.id_phase = Signal(init=1) # ── Ring-overflow protection (RBFI) ─────────────────────────────── # RRP out to RXFrameAssembler so it can tell how much of the ring is # still unread and DROP frames instead of overwriting them. self.rrp = Signal(8, init=_RX_PAGE_FIRST) # Pulse in: a frame was dropped because the ring was full. Sets # IR[7] (RBFI), which the GC's IMR = 0x86 enables alongside RI and TI, # so the driver learns it lost a frame instead of silently reading # corrupted data. self.rx_overflow = Signal() # ── Interrupt (exi domain) ──────────────────────────────────────── self.exi_int_n = Signal(init=1) # ── PAR output (for forwarding to W5500 as source MAC) ─────────── self.par = Signal(48) # PAR0-5 packed: PAR0 in low byte par[0:8] # NCRA[3] = SR (start receive) bit — gates the RX ring-buffer path. self.ncra_sr = Signal() # ── CDC FIFO sync-domain sides (wired by BBATop) ────────────────── # SPRAM request exi→sync: sync reads these self.spram_req_r_data = Signal(16) self.spram_req_r_en = Signal() self.spram_req_r_rdy = Signal() # SPRAM response sync→exi: sync writes these # SPRAM response, DIRECT from the arbiter -- no FIFO. That FIFO cost # one 24 MHz cycle (42 ns), and the round trip only fits inside the # protocol's turnaround byte without it (333 ns -> 292 ns against a # 315 ns deadline). Responses must therefore be consumed the cycle # they are valid, which is why only ONE request may be outstanding. self.spram_rsp_data = Signal(8) self.spram_rsp_valid = Signal() # TX bytes exi→sync: sync reads these self.tx_bytes_r_data = Signal(8) self.tx_bytes_r_en = Signal() self.tx_bytes_r_rdy = Signal() # TX ctrl (frame length) exi→sync: sync reads these self.tx_ctrl_r_data = Signal(16) self.tx_ctrl_r_en = Signal() self.tx_ctrl_r_rdy = Signal() # RX write-pointer update sync→exi: sync writes these self.rx_wptr_w_data = Signal(8) self.rx_wptr_w_en = Signal() self.rx_wptr_w_rdy = Signal() # RX read-pointer update exi→sync: sync reads these self.rx_rptr_r_data = Signal(8) self.rx_rptr_r_en = Signal() self.rx_rptr_r_rdy = Signal() # PulseSynchronizer ports (exi↔sync) self.ncra_rst_o = Signal() # exi→sync self.rx_irq_i = Signal() # sync→exi self.tx_irq_i = Signal() # sync→exi def elaborate(self, platform): m = Module() # ── CDC FIFOs ──────────────────────────────────────────────────── # Plain SyncFIFOs: `exi` and `sync` were always the SAME clock net # (bba_top wired ClockSignal("sync") to ClockSignal("exi")), so the # AsyncFIFOs between them bought nothing and cost ~3 cycles each way. # On the SPRAM path that latency exceeded the protocol's turnaround # byte and shifted every DMA read by one byte. One slow domain now. spram_req = SyncFIFO(width=16, depth=4) # TX byte FIFO: deep enough for a WHOLE maximum-size frame. # # SPI cannot back-pressure, so BBARegisterFile silently drops bytes # when w_rdy is low -- there is no mechanism to tell the GC to wait. # The depth is therefore not a buffering choice, it is the only thing # standing between a drain stall and a corrupted frame. # # At 16 deep it held 16 bytes = ~4.7 us at the EXI byte rate (296 # ns/byte). TXFrameDrain shares the W5100 bus arbiter with RX and the # socket-3 UDP test, so any stall longer than that dropped bytes # mid-frame. Swiss writes an ENTIRE frame in one register-space # transaction (bba_outs to BBA_WRTXFIFOD), so 16 could never work for # real traffic; the on-console test app only got away with it by # writing one byte per transaction, which paced the drain by accident. # # 2048 covers the 1518-byte maximum frame with margin, so a frame # cannot be truncated even if the drain makes no progress at all for # the whole transaction. Costs ~4 of 27 free BRAMs. tx_bytes = SyncFIFO(width=8, depth=2048) tx_ctrl = SyncFIFO(width=16, depth=4) rx_wptr = SyncFIFO(width=8, depth=4) rx_rptr = SyncFIFO(width=8, depth=4) m.submodules.spram_req = spram_req m.submodules.tx_bytes = tx_bytes m.submodules.tx_ctrl = tx_ctrl m.submodules.rx_wptr = rx_wptr m.submodules.rx_rptr = rx_rptr # Expose sync-domain FIFO sides m.d.comb += [ self.spram_req_r_data .eq(spram_req.r_data), spram_req.r_en .eq(self.spram_req_r_en), self.spram_req_r_rdy .eq(spram_req.r_rdy), self.tx_bytes_r_data .eq(tx_bytes.r_data), tx_bytes.r_en .eq(self.tx_bytes_r_en), self.tx_bytes_r_rdy .eq(tx_bytes.r_rdy), self.tx_ctrl_r_data .eq(tx_ctrl.r_data), tx_ctrl.r_en .eq(self.tx_ctrl_r_en), self.tx_ctrl_r_rdy .eq(tx_ctrl.r_rdy), rx_wptr.w_data .eq(self.rx_wptr_w_data), rx_wptr.w_en .eq(self.rx_wptr_w_en), self.rx_wptr_w_rdy .eq(rx_wptr.w_rdy), self.rx_rptr_r_data .eq(rx_rptr.r_data), rx_rptr.r_en .eq(self.rx_rptr_r_en), self.rx_rptr_r_rdy .eq(rx_rptr.r_rdy), ] # ── PulseSynchronizers ─────────────────────────────────────────── # Single slow domain, so these are plain wires rather than # PulseSynchronizers. A pulse generated here is already in the only # domain that exists. ncra_rst_pulse = Signal() m.d.comb += self.ncra_rst_o.eq(ncra_rst_pulse) # ── Register file (sparse individual Signals, exi domain) ──────── # Only the registers actually read/written by the GC or sync domain. # Writes to unknown addresses are silently ignored; reads return 0. r_ncra = Signal(8) r_imr = Signal(8) r_ir = Signal(8) # Both ring pointers reset to the ring's FIRST PAGE, not to 0. # The SPRAM ring is pages 1-15 (page 0 holds the register region), and # RXFrameAssembler's own write pointer resets to page 1 -- so a 0 here # disagreed with the assembler and, worse, told the GC driver to read # its first frame from page 0. The frame is at page 1; page 0 is the # register page. Hardware symptom: "RRP page 0 outside ring pages # 1-15" from the test app's DMA descriptor check, on a board whose RX # path was otherwise working. # RRP is driver-owned (the GC writes it after consuming frames), but # its RESET value is ours, and RRP == RWP == 1 is the correct initial # state: empty, and pointing at the start of the ring. r_rwp_lo = Signal(8, init=_RX_PAGE_FIRST) r_rrp_lo = Signal(8, init=_RX_PAGE_FIRST) # PAR0–5 reset to a valid Nintendo OUI MAC (00:09:BF:00:00:01) so the # device has a sane source MAC even before the GC driver programs its # own. PAR0 is the first MAC octet. _par_reset = [0x00, 0x09, 0xBF, 0x00, 0x00, 0x01] r_par = Array([Signal(8, name=f"par{i}", init=_par_reset[i]) for i in range(6)]) r_twd_lo = Signal(8) r_twd_hi = Signal(8) # PAR packed output: PAR0 in the LOW byte (par[0:8]). The W5500 master # reads mac_shadow[i] = par[i*8:(i+1)*8], so this puts PAR0 first in the # SHAR write — i.e. PAR0 is the first MAC octet on the wire. m.d.comb += self.par.eq(Cat( r_par[0], r_par[1], r_par[2], r_par[3], r_par[4], r_par[5], )) m.d.comb += self.ncra_sr.eq(r_ncra[3]) # start-receive bit # ── Transaction state ──────────────────────────────────────────── # Real GC<->BBA header (see docs/exi_protocol_real.md), verified # against libogc and Swiss. Byte 0 bit7 selects the address space and # bit6 is the write flag; there is NO length field -- the data phase # runs until CS. # command (bit7=0): [reg&0x3F, 00, 00, data] one 4-byte transfer # register (bit7=1): [0x80|w<<6, ah, al, 00] + data phase until CS addr = Signal(16) is_write = Signal() cmd_space = Signal() # this transaction is a command access cmd_reg = Signal(6) cmd_wreg = Signal(6) # target of a pending command-space write byte_ctr = Signal(3) # response-byte index (device-ID sequence) cmd_wr = Signal() # command-space write pending prefetched = Signal() # first response byte already pushed cmd_wdata = Signal(8) # EXI-side command registers. Swiss uses exactly two: 0x02 is the EXI # interrupt mask and 0x03 the EXI interrupt status (write-1-to-clear). r_cmd_imr = Signal(8) r_cmd_ir = Signal(8) tx_frame_len = Signal(16) # TX trigger bookkeeping. `twd_armed` fires one cycle AFTER TWD_HI is # written, so tx_frame_len has both halves by then; `tx_queued` records # that the drain already has a length for this frame. twd_armed = Signal() tx_queued = Signal() # Bytes actually streamed into TXDATA since the last trigger. # # ⚠️ Needed because libogc DOES NOT WRITE TWD AT ALL. Its # __bba_link_tx is four transactions -- read NWAYS, one register-space # write of the whole frame to 0x0048 (EXI_Dma for the 32-byte-aligned # bulk plus EXI_ImmEx for the remainder), read NCRA, write NCRA = 0x04 # -- with no TWD_LO/TWD_HI anywhere. The real device must therefore # count what it was given, which is what this does. Using # tx_frame_len on the ST fallback path would transmit a stale or zero # length for any driver that skips TWD. txdata_cnt = Signal(16) # True until first NCRA reset write: return device ID on addr=0 reads id_phase = self.id_phase # Per-byte SPRAM read handshake (register-read path): sp_req marks a # request in flight; drain_ctr counts the read-phase dummy bytes. sp_req = Signal() drain_ctr = Signal(2) # SPRAM streaming-read state (DMA / variable-length reads): # sp_addr — next SPRAM byte address to request (auto-increments) # outstanding — SPRAM requests issued but whose responses are not yet # popped (bounds prefetch and is drained at end) sp_addr = Signal(16) outstanding = Signal(4) sp_hold = Signal(8) # response held when tx_fifo was full sp_pending = Signal() SP_LIMIT = 4 # max prefetch depth in flight # Effective address of the current data byte — a REGISTERED running # pointer (set to the base in HEADER1, incremented per byte). Keeping # it registered keeps the 13-bit adder off the combinational path that # feeds the read-response mux → tx_fifo write data. eff_addr = Signal(16) rd_sel = eff_addr[0:8] # ── Combinational read-response value (non-SPRAM) ──────────────── reg_rdval = Signal(8) with m.Switch(rd_sel): with m.Case(_NCRA): m.d.comb += reg_rdval.eq(r_ncra) with m.Case(_IMR): m.d.comb += reg_rdval.eq(r_imr) with m.Case(_IR): m.d.comb += reg_rdval.eq(r_ir) with m.Case(_RWP_LO): m.d.comb += reg_rdval.eq(r_rwp_lo) with m.Case(_RRP_LO): m.d.comb += reg_rdval.eq(r_rrp_lo) with m.Case(_PAR0, _PAR1, _PAR2, _PAR3, _PAR4, _PAR5): m.d.comb += reg_rdval.eq(r_par[eff_addr[0:3]]) with m.Case(_TWD_LO): m.d.comb += reg_rdval.eq(r_twd_lo) with m.Case(_TWD_HI): m.d.comb += reg_rdval.eq(r_twd_hi) with m.Case(_NWAYS): m.d.comb += reg_rdval.eq(_NWAYS_VAL) with m.Case(_HIPR): m.d.comb += reg_rdval.eq(_HIPR_VAL) with m.Default(): m.d.comb += reg_rdval.eq(0) # The device-ID constant lives in SPIMode3Slave now (it must be # answered with zero CDC latency), so there is no copy here. cmd_rdval = Signal(8) with m.Switch(cmd_reg): with m.Case(0x02): m.d.comb += cmd_rdval.eq(r_cmd_imr) with m.Case(0x03): m.d.comb += cmd_rdval.eq(r_cmd_ir) with m.Default(): m.d.comb += cmd_rdval.eq(0) rd_val = Signal(8) # response for the current non-SPRAM read byte with m.If(cmd_space): m.d.comb += rd_val.eq(cmd_rdval) with m.Else(): m.d.comb += rd_val.eq(reg_rdval) # ── Default strobes ────────────────────────────────────────────── m.d.sync += [ tx_bytes.w_en .eq(0), tx_ctrl.w_en .eq(0), rx_rptr.w_en .eq(0), rx_wptr.r_en .eq(0), ncra_rst_pulse .eq(0), ] # The SPRAM request write port is COMBINATIONAL (see RHDR_LO): a # registered w_en costs a 24 MHz cycle the turnaround byte cannot # spare. Default it off here and let the states assert it. m.d.comb += [ spram_req.w_en .eq(0), spram_req.w_data.eq(0), self.rx_en .eq(0), self.tx_en .eq(0), self.tx_data.eq(0), # Combinational so the FIFO advances in the SAME cycle as the pop — # a registered r_en would let `pop` re-fire on the same byte. ] # ── Transaction FSM (proactive push/pull over byte FIFOs) ──────── # The SPI bit cadence lives in the capture domain; here we just consume # received bytes and, for reads, push response bytes into tx_fifo during # the EXI clock-idle gap before the GC clocks the data phase. with m.FSM(domain="sync", name="exi_fsm"): with m.State("HEADER0"): # Byte 0 selects the space and direction. The REAL protocol # (docs/exi_protocol_real.md) has no length field: the GC # clocks as many data bytes as it wants and ends with CS. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) m.d.sync += [is_write.eq(self.rx_data[6]), prefetched.eq(0), byte_ctr.eq(0), sp_req.eq(0), outstanding.eq(0), # sp_pending too: a byte still held from the # PREVIOUS transaction would be pushed as this # one's first response and shift the whole # read by one. sp_pending.eq(0)] with m.If(self.rx_data[7]): # register space m.d.sync += cmd_space.eq(0) m.next = "RHDR_HI" with m.Else(): # command space m.d.sync += [cmd_space.eq(1), cmd_reg.eq(self.rx_data[0:6]), addr.eq(self.rx_data[0:6]), eff_addr.eq(self.rx_data[0:6])] # NOTE: the device-ID reply is NOT produced here. # A 2-byte command header leaves one byte time for a # round trip through both AsyncFIFOs, which measured # too slow on a console (the ID came back one byte # late). SPIMode3Slave answers it from the EXI clock # domain instead -- see its device-ID fast path. m.next = "CHDR_1" # ── register space: [0x80|w<<6, ah, al, 0x00] ────────────────── with m.State("RHDR_HI"): with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) m.d.sync += addr[8:16].eq(self.rx_data) m.next = "RHDR_LO" with m.State("RHDR_LO"): # The address completes HERE, one whole byte before the data # phase. That trailing 0x00 byte is the device's prefetch # window -- the old 2-byte-plus-length format spent those bits # on the length field and left zero slack, which is why the # first on-console read came back 22 bit-times late. Kick the # fetch off now so the answer is ready for the first data slot. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) full = Cat(self.rx_data, addr[8:16]) # 16-bit address m.d.sync += [addr.eq(full), eff_addr.eq(full), sp_addr.eq(full)] with m.If(~is_write & (full >= 0x100) & spram_req.w_rdy): # COMBINATIONAL write: the turnaround byte gives only # ~296 ns and the SPRAM round trip needs most of it, so # a registered w_en (one more 24 MHz cycle, ~42 ns) is # enough to miss the window and shift the whole DMA # read by a byte. Same-domain SyncFIFO now, so this is # legal. m.d.comb += [spram_req.w_data.eq(full), spram_req.w_en.eq(1)] m.d.sync += [sp_addr.eq(full + 1), outstanding.eq(1)] m.next = "RHDR_PAD" with m.State("RHDR_PAD"): # Trailing header byte. For a register read, push the first # response byte during this byte-time so it has crossed the # tx CDC before the GC clocks data slot 0. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) with m.If(is_write): m.next = "WRITE" with m.Elif(addr >= 0x100): m.next = "SPRAM_STREAM" with m.Else(): m.next = "REG_READ" with m.Elif(~is_write & (addr < 0x100) & self.tx_rdy & ~prefetched): # Prefetch the FIRST response byte (once!) while the pad # byte is still on the wire, so it has crossed the tx CDC # before the GC clocks data slot 0. Must be one-shot: this # branch is re-evaluated every cycle the pad byte has not # arrived, and repeating it would consume the register # stream early (symptom: reads shifted by a byte). m.d.comb += [self.tx_data.eq(rd_val), self.tx_en.eq(1)] m.d.sync += [eff_addr.eq(eff_addr + 1), prefetched.eq(1)] # ── command space: a 2-BYTE header, then a data phase ───────── # # [0x40|reg, 0x00] + N data bytes written, or # [reg, 0x00] + N data bytes read. # # ⚠️ NOT a single 4-byte transfer. This file used to decode # [reg, 00, 00, data] and wait for a fourth byte, taking the DATA # byte as header byte 2 -- so every command-space WRITE was # silently dropped. libogc's own driver (see the init map in # CLAUDE.md) does, for a write of 0x00 to command register 2: # # EXI_Imm (len=2, write) -> 42 00 the whole header # EXI_Sync # EXI_ImmEx(len=1, write) -> 00 the data # # i.e. three bytes between CS fall and CS rise, not four. The # device-ID read survived the old decode only because # SPIMode3Slave answers it on the fast path, independently of # this FSM, which is why nothing caught it. with m.State("CHDR_1"): with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) # The device-ID query (command reg 0 while id_phase) is # answered by SPIMode3Slave, not here: pushing anything # for it would leave stale bytes in the tx FIFO that a # 2-byte header has too few pops to drain. with m.If(~is_write & (cmd_reg == 0) & id_phase): m.next = "DEVID" with m.Elif(is_write): m.next = "CMD_WRITE" with m.Else(): # Read: the response was prefetched below and the pop # at the end of this byte lands it in data slot 0. m.next = "TRAILER" with m.Elif(~is_write & self.tx_rdy & ~prefetched): # Ordinary command read: push during header byte 1 so the # byte-boundary pop presents it in data slot 0. m.d.comb += [self.tx_data.eq(rd_val), self.tx_en.eq(1)] m.d.sync += prefetched.eq(1) with m.State("CMD_WRITE"): # Data phase. Each byte writes the running command register, # which auto-increments so a multi-byte command write behaves # like the register-space path. cmd_wreg latches the target # separately because the handler below applies the write a # cycle later, by which time cmd_reg has already moved on. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) m.d.sync += [cmd_wr.eq(1), cmd_wdata.eq(self.rx_data), cmd_wreg.eq(cmd_reg), cmd_reg.eq(cmd_reg + 1)] with m.If(~self.cs_active): m.next = "TRAILER" with m.State("TRAILER"): # Soak up anything else the GC clocks, then wait for CS to rise. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) with m.If(~self.cs_active): m.next = "HEADER0" with m.State("WRITE"): # Consume xfer_len+1 data bytes, writing the register file. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) with m.Switch(rd_sel): with m.Case(_NCRA): m.d.sync += r_ncra.eq(self.rx_data) with m.If(self.rx_data[0]): m.d.sync += r_ncra[0].eq(0) # RESET self-clears m.d.sync += ncra_rst_pulse.eq(1) m.d.sync += id_phase.eq(0) # An NCRA reset resets the RX ring, so the # pointers the GC reads MUST follow -- the # assembler's own write pointer goes back to # page 1, and without this the register file # keeps whatever the last frame left. # # The GC then reads a stale RWP pointing at a # page the assembler is not using, so its # first RX after a reset reads the wrong page. # Found via netboot: after Swiss's own reset, # syncing RRP to RWP landed on a stale page # and the descriptor there was garbage. This # affects any driver that resets and then # trusts RWP, Swiss included. m.d.sync += [ r_rwp_lo.eq(_RX_PAGE_FIRST), r_rrp_lo.eq(_RX_PAGE_FIRST), ] # Fallback trigger: TWD_HI normally queues the # frame (see _TWD_HI). Only fire here if it did # not, so software that skips TWD still transmits # and we never double-queue one frame. with m.If(self.rx_data[1:3].any() & ~tx_queued): with m.If(tx_ctrl.w_rdy): # The COUNTED length, not tx_frame_len: # a driver reaching this path never wrote # TWD, so tx_frame_len is stale or zero. m.d.sync += tx_ctrl.w_data.eq(txdata_cnt) m.d.sync += tx_ctrl.w_en.eq(1) m.d.sync += tx_queued.eq(1) m.d.sync += txdata_cnt.eq(0) with m.Case(_IMR): m.d.sync += r_imr.eq(self.rx_data) with m.Case(_IR): m.d.sync += r_ir.eq(r_ir & ~self.rx_data) # write-1-clear with m.Case(_RRP_LO): m.d.sync += r_rrp_lo.eq(self.rx_data) with m.If(rx_rptr.w_rdy): m.d.sync += rx_rptr.w_data.eq(self.rx_data) m.d.sync += rx_rptr.w_en.eq(1) with m.Case(_PAR0, _PAR1, _PAR2, _PAR3, _PAR4, _PAR5): m.d.sync += r_par[eff_addr[0:3]].eq(self.rx_data) with m.Case(_TWD_LO): m.d.sync += r_twd_lo.eq(self.rx_data) m.d.sync += tx_frame_len[0:8].eq(self.rx_data) with m.Case(_TWD_HI): m.d.sync += r_twd_hi.eq(self.rx_data) m.d.sync += tx_frame_len[8:16].eq(self.rx_data) # Hand the length to TXFrameDrain NOW, not at the ST # trigger. The drain blocks on tx_bytes_r_rdy, so # starting it here lets it consume bytes AS the GC # streams them and keeps tx_bytes a rate-matching # buffer. Waiting for ST meant the whole frame had # to fit in the 16-deep FIFO, and anything past 16 # bytes was silently dropped (hardware 2026-09-30: # frames truncated at exactly 16 bytes on the wire). m.d.sync += twd_armed.eq(1) with m.Case(_TXDATA): with m.If(tx_bytes.w_rdy): m.d.sync += tx_bytes.w_data.eq(self.rx_data) m.d.sync += tx_bytes.w_en.eq(1) m.d.sync += txdata_cnt.eq(txdata_cnt + 1) # All other addresses silently ignored # No length field in the real protocol: consume data # bytes until CS rises. # # ⚠️ TXDATA IS A FIFO PORT, NOT A REGISTER. A whole frame # is written to that ONE address, so the address must not # advance across its data phase. Swiss does exactly this # (bba_outs to BBA_WRTXFIFOD streams the entire frame in a # single register-space write), so incrementing here # delivered only the FIRST byte of every frame and silently # dropped the rest into 0x49+ -- which the switch above # ignores. # # It went unnoticed because the on-console test app wrote # TXDATA one byte per EXI transaction, so every byte got a # fresh header at 0x48. The moment it batched 8 bytes per # transaction, 7 of every 8 vanished: the frame was an # eighth of its declared length, TXFrameDrain blocked # waiting for the rest and TI never set (hardware # 2026-10-09: txto=16, "no ACK after 20 tries"). # # Every other register IS a register and must advance, so # that bba_outs can write PAR0-5 or MAR0-7 as a block. with m.If(rd_sel != _TXDATA): m.d.sync += eff_addr.eq(eff_addr + 1) with m.If(~self.cs_active): m.next = "HEADER0" with m.State("REG_READ"): # Register read (addr < 0x100): stream until CS, like the SPRAM # path. The real protocol has no length field. with m.If(self.tx_rdy & self.cs_active): m.d.comb += [self.tx_data.eq(rd_val), self.tx_en.eq(1)] m.d.sync += eff_addr.eq(eff_addr + 1) with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) # drain dummies with m.If(~self.cs_active): m.next = "HEADER0" with m.State("DEVID"): # The reply comes from SPIMode3Slave's device-ID fast path; # this state only drains the GC's dummy bytes and waits for # CS, so NOTHING is pushed into the tx FIFO here. Pushing # would leave stale bytes a 2-byte header cannot drain. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) with m.If(~self.cs_active): m.next = "HEADER0" with m.State("SPRAM_STREAM"): # Stream SPRAM bytes until CS deasserts — handles both ≤4-byte # immediate reads and arbitrary-length DMA reads uniformly. # Issue read requests ahead (prefetch, bounded by SP_LIMIT) and # push responses into tx_fifo; the capture domain pops them as # the GC clocks. Drain rx dummies as they arrive. # With no response FIFO the reply must be taken the cycle it # is valid, so keep exactly ONE request in flight and only # issue when tx_fifo has somewhere to put the answer. The GC # consumes a byte every ~296 ns and the SPRAM round trip is a # couple of 24 MHz cycles, so serialising costs nothing. # Only ONE request in flight (no response FIFO), and only # when tx_fifo can take the answer. with m.If(self.cs_active & spram_req.w_rdy & self.tx_rdy & (outstanding == 0) & ~sp_pending): m.d.comb += spram_req.w_data.eq(sp_addr) m.d.comb += spram_req.w_en.eq(1) m.d.sync += [sp_addr.eq(sp_addr + 1), outstanding.eq(outstanding + 1)] with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) # drain dummy bytes with m.If(~self.cs_active): m.next = "SPRAM_END" with m.State("SPRAM_END"): # CS deasserted: drain in-flight SPRAM responses and rx dummies, # then idle. Leftover prefetch in tx_fifo is flushed by # ExiCapture on the next CS assertion. with m.If(self.rx_rdy): m.d.comb += self.rx_en.eq(1) with m.If((outstanding == 0) & ~sp_pending & ~self.rx_rdy): m.next = "HEADER0" # ── SPRAM response: consumed in ANY state ──────────────────────── # The request is issued in RHDR_LO but the reply lands ~2 cycles later, # while the FSM is still in RHDR_PAD waiting for the turnaround byte. # With no response FIFO there is nothing to catch it there, so handle # it outside the FSM: push straight to tx_fifo if it has room, else # hold it one cycle. SPRAM reads and register reads never overlap, so # driving tx here cannot collide with the FSM's own pushes. # ⚠️ A response is only OURS if a request is actually in flight. # # ETH writes preempt EXI reads in SPRAMArbiter (ETH has absolute # priority and asserts eth_wr_ready combinationally), so a read # response can arrive later than expected -- including after the # transaction that asked for it has ended. Pushing it then INSERTS A # BYTE at the start of the NEXT DMA read and shifts everything the GC # reads by one. It also underflowed `outstanding` (4-bit: 0 - 1 = 15), # which broke the prefetch bound for the rest of that transaction. # # Hardware symptom (2026-10-09): the SPRAM ring descriptor appearing at # page offset +1 instead of +0 for 17 of 22 frames, VARYING frame to # frame, while the assembler was writing it correctly all along. It # only happens when the GC reads the ring while frames are being # written into it, which no simulation did. with m.If(self.spram_rsp_valid & (outstanding != 0) & self.tx_rdy & ~sp_pending): m.d.comb += [self.tx_data.eq(self.spram_rsp_data), self.tx_en.eq(1)] m.d.sync += outstanding.eq(outstanding - 1) with m.Elif(self.spram_rsp_valid & (outstanding != 0)): m.d.sync += [sp_hold.eq(self.spram_rsp_data), sp_pending.eq(1), outstanding.eq(outstanding - 1)] with m.Elif(sp_pending & self.tx_rdy): m.d.comb += [self.tx_data.eq(sp_hold), self.tx_en.eq(1)] m.d.sync += sp_pending.eq(0) # ── Command-space write, applied the cycle after the data byte ─── with m.If(cmd_wr): m.d.sync += cmd_wr.eq(0) with m.Switch(cmd_wreg): with m.Case(0x02): m.d.sync += r_cmd_imr.eq(cmd_wdata) with m.Case(0x03): m.d.sync += r_cmd_ir.eq(r_cmd_ir & ~cmd_wdata) # ── Interrupt output ───────────────────────────────────────────── m.d.comb += self.rrp.eq(r_rrp_lo) m.d.sync += self.exi_int_n.eq(~(r_ir & r_imr).any()) # ── Consume RWP updates from sync domain ────────────────────────── with m.If(rx_wptr.r_rdy): m.d.sync += rx_wptr.r_en.eq(1) m.d.sync += r_rwp_lo.eq(rx_wptr.r_data) # ── PulseSynchronizer arrivals ──────────────────────────────────── with m.If(self.rx_irq_i): m.d.sync += r_ir[1].eq(1) # RI bit # ── Deferred TX queue (armed by a TWD_HI write) ────────────────── # One cycle after TWD_HI so tx_frame_len holds both bytes. Held until # tx_ctrl has room so a full control FIFO cannot drop the trigger. with m.If(twd_armed & tx_ctrl.w_rdy): m.d.sync += tx_ctrl.w_data.eq(tx_frame_len) m.d.sync += txdata_cnt.eq(0) m.d.sync += tx_ctrl.w_en.eq(1) m.d.sync += twd_armed.eq(0) m.d.sync += tx_queued.eq(1) # TX done. Clearing the ST bits and tx_queued belongs HERE, with the # completion that they describe. Adding the RBFI handler below split # this block and left those two lines attached to rx_overflow -- an RX # event -- so after the first frame tx_queued stayed set forever, which # disables the NCRA ST fallback trigger entirely (it is guarded on # ~tx_queued) and leaves the ST bits asserted in NCRA. Software that # drives TX via ST instead of TWD would transmit exactly once. with m.If(self.tx_irq_i): m.d.sync += r_ir[2].eq(1) # TI bit m.d.sync += r_ncra[1:3].eq(0) # clear ST bits m.d.sync += tx_queued.eq(0) # ready for the next frame # RBFI: the ring was full and a frame was discarded. Reported rather # than hidden -- the previous behaviour was to overwrite unread pages, # so the driver got corrupted frames with no signal that anything had # gone wrong. with m.If(self.rx_overflow): m.d.sync += r_ir[7].eq(1) return m # ── Testbench ───────────────────────────────────────────────────────────── if __name__ == "__main__": import sys, os sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) from amaranth.sim import Simulator, Period reg = BBARegisterFile() # Drive the byte-stream interface directly (the SPI bit cadence and FIFOs # live in ExiCapture; here we model the byte producer/consumer). # Responses can be pushed WHILE the header is still being clocked -- the # device prefetches during the header's turnaround byte, and the device-ID # reply starts from header byte 0. So every helper that advances time has # to record tx pushes, or the early ones are silently lost and reads look # shifted. tx_log is that shared record; collect_tx drains it first. tx_log = [] def sample_tx(ctx): if ctx.get(reg.tx_en): tx_log.append(ctx.get(reg.tx_data)) async def push_rx(ctx, b): """Present one received byte and wait for the register file to pop it.""" ctx.set(reg.rx_data, b) ctx.set(reg.rx_rdy, 1) while True: en = ctx.get(reg.rx_en) sample_tx(ctx) await ctx.tick("sync") if en: break ctx.set(reg.rx_rdy, 0) async def collect_tx(ctx, n): """Collect n response bytes, including any pushed during the header.""" out = [] while tx_log and len(out) < n: out.append(tx_log.pop(0)) for _ in range(3000): if len(out) >= n: break if ctx.get(reg.tx_en): out.append(ctx.get(reg.tx_data)) await ctx.tick("sync") return out # Real protocol (docs/exi_protocol_real.md): register space is a 4-byte # header [0x80|w<<6, ah, al, 0x00] followed by a data phase that runs # until CS rises -- there is no length field. async def exi_read(ctx, addr, length=1): tx_log.clear() ctx.set(reg.cs_active, 1) await push_rx(ctx, 0x80) await push_rx(ctx, (addr >> 8) & 0xFF) await push_rx(ctx, addr & 0xFF) await push_rx(ctx, 0x00) # turnaround / prefetch window result = await collect_tx(ctx, length) ctx.set(reg.cs_active, 0) await ctx.tick("sync").repeat(6) return result async def exi_write(ctx, addr, data): ctx.set(reg.cs_active, 1) await push_rx(ctx, 0xC0) await push_rx(ctx, (addr >> 8) & 0xFF) await push_rx(ctx, addr & 0xFF) await push_rx(ctx, 0x00) for b in data: await push_rx(ctx, b) ctx.set(reg.cs_active, 0) await ctx.tick("sync").repeat(6) async def drain_tx_bytes(ctx, limit=80): """Pop everything the register file pushed into tx_bytes.""" out = [] for _ in range(limit): if not ctx.get(reg.tx_bytes_r_rdy): break out.append(ctx.get(reg.tx_bytes_r_data)) ctx.set(reg.tx_bytes_r_en, 1) await ctx.tick("sync").repeat(1) ctx.set(reg.tx_bytes_r_en, 0) await ctx.tick("sync").repeat(1) return out async def exi_devid(ctx): """EXI_GetID: write 0x0000 (2 bytes), then clock 4 read bytes.""" tx_log.clear() ctx.set(reg.cs_active, 1) await push_rx(ctx, 0x00) # command space, reg 0, read await push_rx(ctx, 0x00) result = await collect_tx(ctx, 4) ctx.set(reg.cs_active, 0) await ctx.tick("sync").repeat(6) return result async def cmd_write(ctx, creg, data): """Command-space write in libogc's REAL shape: a 2-byte header [0x40|reg, 0x00] then the data bytes -- three bytes total for a single-byte write, NOT a 4-byte transfer.""" ctx.set(reg.cs_active, 1) await push_rx(ctx, 0x40 | (creg & 0x3F)) await push_rx(ctx, 0x00) for b in data: await push_rx(ctx, b) ctx.set(reg.cs_active, 0) await ctx.tick("sync").repeat(6) async def cmd_read(ctx, creg, length=1): """Command-space read: 2-byte header, then clock `length` bytes.""" tx_log.clear() ctx.set(reg.cs_active, 1) await push_rx(ctx, creg & 0x3F) # Idle while header byte 1 is "on the wire". The device must push its # response DURING that byte: the engine's byte-boundary pop for data # slot 0 happens at the end of byte 1, which is before the FSM has even # seen byte 1 (the rx CDC is ~125 ns). A response pushed in reaction # to consuming byte 1 would land in slot 1 instead, so pushing both # header bytes back-to-back here would not model the real transfer. for _ in range(4): sample_tx(ctx) # the push lands HERE; a bare tick loses it await ctx.tick("sync") await push_rx(ctx, 0x00) result = await collect_tx(ctx, length) ctx.set(reg.cs_active, 0) await ctx.tick("sync").repeat(6) return result # SPRAM contents the streaming-read test reads back (byte i = 0xA0+i). spram_mem = {0x100 + i: (0xA0 + i) & 0xFF for i in range(64)} async def spram_model(ctx): """Model the SPRAM + arbiter as a request/grant port. The response FIFO is gone: the arbiter drives spram_rsp_valid directly for exactly ONE cycle, one cycle after it takes a request. Mirror that precisely -- a response held high for two cycles makes the register file push the same byte twice. """ busy = False held = 0 async for vals in ctx.tick("sync").sample( reg.spram_req_r_rdy, reg.spram_req_r_data): rdy, addr = vals[-2:] ctx.set(reg.spram_req_r_en, 0) ctx.set(reg.spram_rsp_valid, 0) if busy: # deliver, one cycle after accept ctx.set(reg.spram_rsp_data, held) ctx.set(reg.spram_rsp_valid, 1) busy = False elif rdy: # accept one request held = spram_mem.get(addr, 0) ctx.set(reg.spram_req_r_en, 1) busy = True errors = [] async def testbench(ctx): ctx.set(reg.tx_rdy, 1) # tx_fifo always has room in this model await ctx.tick("sync").repeat(8) # T1: a device-ID query must push NOTHING. # The reply comes from SPIMode3Slave's fast path (a 2-byte command # header leaves too little time for a round trip through both byte # FIFOs), so anything pushed here is stale data the header has too few # pops to drain -- which is exactly how an earlier version made the # NEXT enumeration come back a slot late. The ID value itself is # tested in spi_mode3_slave (T7) and end-to-end in bba_top. result = await exi_devid(ctx) if result != []: errors.append(f"T1 device-ID query must push nothing, got {result}") print(f"T1 device-ID query pushes nothing: {result == []}") # T1b: both ring pointers must come out of reset at the ring's FIRST # PAGE, not 0. RRP == RWP == 1 is "empty, pointing at the start of # the ring"; 0 points at the register page, and a driver that read its # first frame from RRP=0 would get the register region instead of the # frame -- which is what the DMA descriptor check caught on hardware. for name, a in (("RWP", _RWP_LO), ("RRP", _RRP_LO)): got = await exi_read(ctx, a, 1) if got != [_RX_PAGE_FIRST]: errors.append(f"T1b {name} reset: expected " f"[{_RX_PAGE_FIRST}], got {got}") print(f"T1b {name} at reset: {got} (want [{_RX_PAGE_FIRST}])") # T1c: an NCRA reset must put BOTH pointers back to page 1, even after # they have been moved. The assembler's own write pointer returns to # page 1 on reset, so a register file that keeps the old value hands # the GC a RWP pointing at a page the assembler is not using -- and # its first RX after a reset then reads the wrong page. That is a # real-driver bug, not just a test-app one: Swiss resets and then # trusts RWP. await exi_write(ctx, _RRP_LO, [7]) # move RRP off page 1 got = await exi_read(ctx, _RRP_LO, 1) if got != [7]: errors.append(f"T1c RRP should be writable, got {got}") # Move RWP too, the way the assembler does -- otherwise this half of # the test has no teeth, because RWP would read 1 whether or not the # reset clears it. ctx.set(reg.rx_wptr_w_data, 9) ctx.set(reg.rx_wptr_w_en, 1) await ctx.tick("sync").repeat(1) ctx.set(reg.rx_wptr_w_en, 0) await ctx.tick("sync").repeat(4) got = await exi_read(ctx, _RWP_LO, 1) if got != [9]: errors.append(f"T1c RWP should track the assembler, got {got}") await exi_write(ctx, _NCRA, [0x01]) # NCRA reset await ctx.tick("sync").repeat(4) for name, a in (("RWP", _RWP_LO), ("RRP", _RRP_LO)): got = await exi_read(ctx, a, 1) if got != [_RX_PAGE_FIRST]: errors.append(f"T1c {name} after NCRA reset: expected " f"[{_RX_PAGE_FIRST}], got {got}") print(f"T1c {name} after NCRA reset: {got} " f"(want [{_RX_PAGE_FIRST}])") # ...and the constant must not drift from the assembler's, which owns # the write pointer. Two copies of a page number is exactly how the # off-by-one got in. from exi_bba.rx_frame_assembler import _RX_PAGE_FIRST as _ASM_FIRST if _RX_PAGE_FIRST != _ASM_FIRST: errors.append(f"T1b ring-first-page mismatch: register file " f"{_RX_PAGE_FIRST} vs assembler {_ASM_FIRST}") print(f"T1b ring first page agrees with assembler: " f"{_RX_PAGE_FIRST == _ASM_FIRST}") await ctx.tick("sync").repeat(4) # T2: Write and read back PAR0-PAR3 await exi_write(ctx, _PAR0, [0xDE, 0xAD, 0xBE, 0xEF]) await ctx.tick("sync").repeat(4) result = await exi_read(ctx, _PAR0, length=4) if result != [0xDE, 0xAD, 0xBE, 0xEF]: errors.append(f"T2 PAR readback: {result}") print(f"T2 PAR0-3: {[f'0x{b:02X}' for b in result]}") await ctx.tick("sync").repeat(4) # T3: NWAYS hardcoded 0x17 result = await exi_read(ctx, _NWAYS, length=1) if result != [0x17]: errors.append(f"T3 NWAYS: expected 0x17, got {result}") print(f"T3 NWAYS: 0x{result[0]:02X}") await ctx.tick("sync").repeat(4) # T4: HIPR hardcoded 0x01 result = await exi_read(ctx, _HIPR, length=1) if result != [0x01]: errors.append(f"T4 HIPR: expected 0x01, got {result}") print(f"T4 HIPR: 0x{result[0]:02X}") await ctx.tick("sync").repeat(4) # T5: IMR write, rx_irq pulse, INT_N asserts, then IR clear await exi_write(ctx, _IMR, [0x02]) # enable RI (bit 1) await ctx.tick("sync").repeat(4) ctx.set(reg.rx_irq_i, 1) await ctx.tick("sync").repeat(1) ctx.set(reg.rx_irq_i, 0) await ctx.tick("sync").repeat(12) # wait for PS propagation int_n = ctx.get(reg.exi_int_n) if int_n != 0: errors.append(f"T5 INT_N after RI: expected 0, got {int_n}") print(f"T5 INT_N after RI pulse: {int_n} (want 0)") await exi_write(ctx, _IR, [0x02]) # write-1-to-clear RI await ctx.tick("sync").repeat(4) int_n = ctx.get(reg.exi_int_n) if int_n != 1: errors.append(f"T5 INT_N after clear: expected 1, got {int_n}") print(f"T5 INT_N after IR clear: {int_n} (want 1)") # T6: streaming SPRAM read (DMA) — the real protocol has no length # field, so the data phase simply runs for as long as CS is held. # Swiss reads the packet buffer exactly this way (bba_ins -> DMA). N = 12 tx_log.clear() ctx.set(reg.cs_active, 1) await push_rx(ctx, 0x80) # register space, read await push_rx(ctx, 0x01) # addr[15:8] -> 0x0100 await push_rx(ctx, 0x00) # addr[7:0] await push_rx(ctx, 0x00) # turnaround byte got = [] for _ in range(5000): if ctx.get(reg.tx_en): got.append(ctx.get(reg.tx_data)) if len(got) >= N: break await ctx.tick("sync") ctx.set(reg.cs_active, 0) # end the transaction await ctx.tick("sync").repeat(40) # let SPRAM_END drain/clean up want = [spram_mem[0x100 + i] for i in range(N)] print(f"T6 DMA read {N}B: {[f'0x{b:02X}' for b in got]}") if got != want: errors.append(f"T6 streaming SPRAM read: got {got}, want {want}") # T7: a normal register read still works after the streaming transaction # (FSM cleaned up and returned to HEADER0) result = await exi_read(ctx, _NWAYS, length=1) if result != [0x17]: errors.append(f"T7 NWAYS after DMA: got {result}") print(f"T7 NWAYS after DMA read: 0x{result[0]:02X}") # T8: a MULTI-BYTE write to TXDATA must push EVERY byte. # # TXDATA (0x48) is a FIFO port, not a register: a whole frame goes to # that one address. The write address used to advance across the data # phase unconditionally, so byte 0 landed in TXDATA and bytes 1..N-1 # went to 0x49+ and were silently ignored. # # Swiss writes an entire frame this way (bba_outs to BBA_WRTXFIFOD), so # its TX would have delivered ONE BYTE per frame. It went unnoticed # because the on-console test app wrote TXDATA one byte per EXI # transaction; the moment it batched 8 bytes per transaction, 7 of # every 8 vanished and TXFrameDrain blocked waiting for a frame that # never completed (hardware: txto=16, no ACK). # 64 bytes, not 8: this also exercises the deepened FIFO and matches # the largest chunk bba_write() can send in one transaction. payload = [(0x11 + i * 3) & 0xFF for i in range(64)] await exi_write(ctx, _TXDATA, payload) got = await drain_tx_bytes(ctx) if got != payload: errors.append(f"T8 multi-byte TXDATA write: got {got}, " f"want {payload}") print(f"T8 multi-byte TXDATA write: {len(got)}/{len(payload)} bytes " f"{'ok' if got == payload else 'MISMATCH'}") # T9: TX completion must clear the ST bits and re-arm for the next # frame. Those two actions used to sit in the rx_overflow handler by # accident, so after ONE transmit tx_queued stayed set forever: the # NCRA ST fallback trigger is guarded on ~tx_queued, so software that # drives TX via ST rather than TWD transmitted exactly once and then # silently stopped. await exi_write(ctx, _NCRA, [0x08 | 0x02]) # SR + ST0 await ctx.tick("sync").repeat(2) ctx.set(reg.tx_irq_i, 1) # frame completed await ctx.tick("sync").repeat(1) ctx.set(reg.tx_irq_i, 0) await ctx.tick("sync").repeat(2) got = await exi_read(ctx, _NCRA, 1) if got and (got[0] & 0x06): errors.append(f"T9 ST bits not cleared on TX done: NCRA=0x{got[0]:02X}") print(f"T9 NCRA after TX done: 0x{got[0]:02X} (ST bits must be clear)") # ...and a SECOND ST-triggered transmit must still queue, which it # cannot if tx_queued was left set. tx_ctrl_before = ctx.get(reg.tx_ctrl_r_rdy) await exi_write(ctx, _NCRA, [0x08 | 0x02]) await ctx.tick("sync").repeat(4) if not ctx.get(reg.tx_ctrl_r_rdy): errors.append("T9 second ST trigger did not queue a frame") print(f"T9 second ST trigger queues: {bool(ctx.get(reg.tx_ctrl_r_rdy))} " f"(was {bool(tx_ctrl_before)})") # ── T10: command-space write in libogc's real 3-byte shape ─────── # The decoder used to expect [reg, 00, 00, data] and wait for a fourth # byte, consuming the DATA byte as header byte 2 -- so every # command-space write was silently DROPPED. libogc writes command # register 2 (its interrupt mask) at both the start and the end of # init, so a dropped write means that mask is never applied. # # Nothing caught this before because no test issued a command-space # write at all: the device-ID READ is answered by SPIMode3Slave's fast # path, independently of this FSM, so the only command-space traffic # under test bypassed the decoder being tested. await cmd_write(ctx, 0x02, [0xF8]) # libogc's IRMASKNONE got = await cmd_read(ctx, 0x02, 1) print(f"T10 cmd write reg2=0xF8 -> read back {[hex(x) for x in got]}") if got != [0xF8]: errors.append(f"T10 command-space write dropped: read back {got}") await cmd_write(ctx, 0x02, [0x00]) # ... and IRMASKALL got2 = await cmd_read(ctx, 0x02, 1) print(f"T10b cmd write reg2=0x00 -> read back {[hex(x) for x in got2]}") if got2 != [0x00]: errors.append(f"T10b command-space rewrite failed: {got2}") # ── T11: TX in libogc's shape -- NO TWD WRITE AT ALL ───────────── # __bba_link_tx is four transactions: read NWAYS, ONE register-space # write of the whole frame to 0x0048, read NCRA, write NCRA = 0x04. # There is no TWD_LO/TWD_HI anywhere, so the device has to count what # it was handed. The ST fallback used to queue tx_frame_len, which a # driver that skips TWD never sets -- so the frame would go out with a # stale or zero length. await drain_tx_bytes(ctx) # clear anything pending while ctx.get(reg.tx_ctrl_r_rdy): # and any queued length ctx.set(reg.tx_ctrl_r_en, 1) await ctx.tick("sync") ctx.set(reg.tx_ctrl_r_en, 0) await ctx.tick("sync") ctx.set(reg.tx_irq_i, 1) # TX done -> clears tx_queued await ctx.tick("sync") ctx.set(reg.tx_irq_i, 0) await ctx.tick("sync").repeat(2) frame11 = [0x40 + i for i in range(24)] await exi_write(ctx, _TXDATA, frame11) # the frame, in ONE write await exi_write(ctx, _NCRA, [0x04]) # ST trigger, no TWD qlen = ctx.get(reg.tx_ctrl_r_data) if ctx.get(reg.tx_ctrl_r_rdy) else None body = await drain_tx_bytes(ctx) print(f"T11 libogc-shape TX (no TWD): queued len={qlen} " f"bytes={len(body)}") if qlen != len(frame11): errors.append(f"T11 queued length {qlen}, want {len(frame11)} " f"(ST path must use the COUNTED length)") if body != frame11: errors.append(f"T11 frame bytes wrong: {len(body)} of " f"{len(frame11)}") # NO T10 FOR THE LATE-RESPONSE LEAK -- and that is deliberate. # # The guard added to the response handler (outstanding != 0) and the # sp_pending clear in HEADER0 fix a real latent bug: a response that # arrives after its transaction ended used to be pushed anyway, and # `outstanding - 1` from zero underflowed a 4-bit counter to 15, # breaking the prefetch bound for the rest of the transaction. # # But this testbench CANNOT demonstrate the consequence. It models the # TX side as tx_rdy/tx_en with a fresh capture list per transaction -- # there is no persistent tx FIFO and no header drain pops, which is # exactly the mechanism by which a leaked byte shifts the NEXT read. A # test written here passes whether or not the guard is present (checked # both ways), and a test that cannot fail is worse than no test. # # WHAT IS ACTUALLY MISSING: a bba_top-level test that performs an EXI # DMA read from the SPRAM ring WHILE the ethernet side writes frames # into it. ETH has absolute priority in SPRAMArbiter, so only then can # a read response be delayed past the end of its transaction. No # simulation does this today, which is why the +1 descriptor offset # (17 of 22 frames on hardware, 2026-10-09) is still unexplained. sim = Simulator(reg) sim.add_clock(Period(MHz=24), domain="sync") sim.add_testbench(testbench) sim.add_process(spram_model) sim.run() if errors: print("\nFAILURES:") for e in errors: print(" ", e) sys.exit(1) else: print("\nAll tests passed.")