Roflin/rebbarb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
dac9c0c0650b6db69479ffc86dfe6aae5e349904
rebbarb/exi_bba/bba_register_file.py
T
Roflin dac9c0c065 bba_register_file: delete the dead rx_rptr FIFO and unused constants 
An audit for signals that are written but never read, ports nothing wires, and
constants only the tests touch turned up something bigger than stray signals:
the whole rx_rptr SyncFIFO was dead.

asm.rrp is wired from reg.rrp -- the combinational tap -- and the FIFO's far
side (rx_rptr_r_data / _r_en / _r_rdy) is wired NOWHERE in BBATop.  It is a
leftover from when the register file and the frame assembler sat in different
clock domains; when those merged the direct tap replaced it and the FIFO was
never removed.  Gone: a 4x8 SyncFIFO, its write logic in the RRP case, its
default strobe and three ports.

Also removed _RWP_HI and _RRP_HI, which were never referenced.  The register
table now says why instead: the ring is 15 pages, so the high bytes of both
pointers are always zero and the read mux's default already returns that.

NCRB_PR went the other way -- it was used only by a test while BBATop steered
the ethernet address filter from a bare `reg.ncrb[0]`.  It is now public and
BBATop imports it, so the constant is live and the magic index is gone.

Kept deliberately: _NCRB_PM and _NCRB_AB are referenced only by T15, but they
document the filtering bits of a real register and make that assertion
readable, so they stay.

Area went DOWN, which is the evidence the FIFO was real logic rather than
something yosys had already pruned:

    ICESTORM_LC   4350 original -> 4380 after the docs pass -> 4328 now

Full suite passes, bba_top passes at EXI_HALF=1/2/8.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-10-10 19:25:51 +02:00

1259 lines
57 KiB
Python
Raw Blame History

"""BBA register file — the GameCube-facing side of the adapter.
Sits between the EXI byte stream (`ExiCapture`) and the ethernet side, and does
three jobs:
* decode EXI transactions and drive the register space;
* buffer TX frame bytes and RX ring pointers across to the ethernet engines
through `SyncFIFO`s;
* raise the two levels of BBA interrupt.
Everything here runs in the `sync` domain (24 MHz). The per-bit SPI cadence is
`SPIMode3Slave`'s problem; this module only ever sees whole bytes.
EXI transaction formats
-----------------------
Byte 0 selects the address space (bit 7) and the direction (bit 6). There is
no length field: the data phase runs for as long as the host keeps the clock
running, and ends when CS rises.
command space (6-bit register) 2-byte header, then a data phase
read [reg, 0x00] + N bytes out
write [0x40|reg, 0x00] + N bytes in
register space (16-bit register) 4-byte header, then a data phase
read [0x80, hi, lo, 0x00] + N bytes out
write [0xC0, hi, lo, 0x00] + N bytes in
The trailing `0x00` of a register-space header is the device's prefetch window:
the address is complete one whole byte before the data phase, which is the only
slack available to fetch the first response byte.
Address spaces
--------------
command 0x00-0x3F EXI-level registers (interrupt mask/status, revision)
register 0x0000-0x00FF MAC registers (this file)
register 0x0100-0x1FFF SPRAM RX ring; any read here streams until CS
Device-ID queries (command space, read, register 0) are answered by
`SPIMode3Slave`, not here — a 2-byte header leaves too little time to fetch a
reply through the byte FIFOs. This file only consumes the bytes.
"""
from types import SimpleNamespace
from amaranth import *
from amaranth.lib.fifo import SyncFIFO
__all__ = ["BBARegisterFile", "NCRB_PR"]
# ─────────────────────────── register space ──────────────────────────────
# MAC-level registers. Multi-byte registers are a LO/HI pair of consecutive
# addresses; only the low byte of each page pointer is ever non-zero here
# because the ring is 15 pages.
_NCRA = 0x00 # Network Control A: [0] RESET (self-clearing),
# [1] ST0 / [2] ST1 start transmit, [3] SR start receive
_NCRB = 0x01 # Network Control B: RX filtering (PR/PM/AB) and the
# receive-interrupt packet counter
_IMR = 0x08 # Interrupt Mask: enables the matching IR bits
_IR = 0x09 # Interrupt status, write-1-to-clear:
# [1] RI frame received, [2] TI frame transmitted,
# [7] RBFI receive buffer full (a frame was dropped)
_BP_LO = 0x0A # Boundary Page: first page of the RX ring
_BP_HI = 0x0B
_TLBP_LO = 0x0C # TX Low Boundary Page: start of the TX buffer region
_TLBP_HI = 0x0D
_RXINTT_LO = 0x14 # RX Interrupt Timer: interrupt coalescing delay
_RXINTT_HI = 0x15
_RWP_LO = 0x16 # RX Write Pointer (page) — device-owned, follows the frame
# assembler. The high byte (0x17) is not decoded: the
# ring is 15 pages, so it is always zero and the read
# mux's default already returns that.
_RRP_LO = 0x18 # RX Read Pointer (page) — host-owned, written once it has
# consumed a frame. High byte (0x19) as above.
_RHBP_LO = 0x1A # RX High Boundary Page: last page of the RX ring
_RHBP_HI = 0x1B
_PAR0 = 0x20 # Physical Address Registers 0-5: the MAC, PAR0 first on
_PAR1 = 0x21 # the wire. Also forwarded to the ethernet chip as its
_PAR2 = 0x22 # own source address.
_PAR3 = 0x23
_PAR4 = 0x24
_PAR5 = 0x25
_NWAYC = 0x30 # NWAY (autonegotiation) configuration
_NWAYS = 0x31 # NWAY status — read-only, see _NWAYS_VAL
_GCA = 0x32 # GMAC Configuration A
_TWD_LO = 0x34 # Transmit frame length, written before streaming TXDATA.
_TWD_HI = 0x35 # Optional: see the TX length note on `txdata_cnt`.
_HIPR = 0x3A # Adapter-present sentinel — read-only, see _HIPR_VAL
_MISC = 0x3D # MISC control 1
_TXDATA = 0x48 # TX FIFO data port: the host streams frame bytes here
_MISC2 = 0x50 # MISC control 2
# First page of the SPRAM RX ring. Page 0 holds the register region, so the
# ring is pages 1-15. Must match rx_frame_assembler._RX_PAGE_FIRST.
_RX_PAGE_FIRST = 0x01
# Lowest address that reads from SPRAM rather than the register file.
_SPRAM_BASE = 0x100
# Read-only register values.
_NWAYS_VAL = 0x17 # 100 Mbit full duplex, link up, autonegotiation complete
_HIPR_VAL = 0x01 # adapter present
# Reset values. NWAYC and MISC are the documented hardware defaults; BP and
# RHBP describe OUR ring layout, so a host that reads them without writing
# sees bounds that match where frames actually land.
_NWAYC_VAL = 0x84
_MISC_VAL = 0x3C
_BP_VAL = _RX_PAGE_FIRST
_RHBP_VAL = 0x0F
# NCRB bit positions. PR is public because BBATop steers the ethernet chip's
# address filter from it.
NCRB_PR = 0 # promiscuous: accept every frame
_NCRB_PM = 2 # pass multicast
_NCRB_AB = 4 # accept broadcast
# IR / IMR bit positions.
_IR_RI = 1 # frame received
_IR_TI = 2 # frame transmitted
_IR_RBFI = 7 # receive buffer full, frame dropped
# ─────────────────────────── command space ──────────────────────────────
# EXI-level registers, addressed with the 2-byte header. Only three have
# behaviour; the rest are storage so that a write is not silently lost.
_CMD_REVID = 0x01 # revision ID — read-only
_CMD_IMR = 0x02 # EXI interrupt mask, gates INT_N
_CMD_IR = 0x03 # EXI interrupt status, write-1-to-clear.
# Bit 7 is a live level: "a MAC interrupt is pending".
_CMD_LAST = 0x0F # highest command register with storage; above this reads 0
_CMD_REVID_VAL = 0x00 # the BBA was never revised past 0
_CMD_IR_MAC = 7 # bit 7 of _CMD_IR
# Interrupt-mask reset value: unmasked. Hosts set this explicitly during
# init, so a permissive default costs them nothing and leaves INT_N usable for
# software that never writes it.
_CMD_IMR_VAL = 0xF8
class BBARegisterFile(Elaboratable):
"""EXI transaction decoder, BBA register space, and the FIFO bridges.
Ports fall into four groups:
EXI byte stream — `rx_*`/`tx_*`/`cs_active`, driven by `ExiCapture`.
`rx_*` is a first-word-fallthrough read side; `tx_*` is a write side.
`cs_active` is high for the duration of one EXI transaction and is what
ends every data phase.
Ethernet side — the `*_r_*` / `*_w_*` groups are the far sides of this
module's FIFOs, wired by `BBATop` to the frame assembler, the TX drain
and the SPRAM arbiter.
Register taps — `par`, `ncra_sr`, `ncrb`, `bp`, `rhbp`, `rrp` expose
register contents that other modules need directly.
Events — `rx_irq_i`, `tx_irq_i`, `rx_overflow` in; `ncra_rst_o` and
`exi_int_n` out.
"""
def __init__(self):
# ── EXI byte stream (from/to ExiCapture) ─────────────────────────
self.rx_data = Signal(8)
self.rx_rdy = Signal()
self.rx_en = Signal()
self.tx_data = Signal(8)
self.tx_en = Signal()
self.tx_rdy = Signal()
self.cs_active = Signal()
# ── Register taps consumed by other modules ──────────────────────
self.par = Signal(48) # PAR0-5 packed, PAR0 in the low byte
self.ncra_sr = Signal() # NCRA[3]: gates the RX ring path
self.ncrb = Signal(8) # RX filtering / interrupt-counter bits
self.bp = Signal(12) # first page of the RX ring
self.rhbp = Signal(12) # last page of the RX ring
self.rrp = Signal(8, init=_RX_PAGE_FIRST) # host read pointer
# ── Events ───────────────────────────────────────────────────────
self.rx_overflow = Signal() # in: a frame was dropped (RBFI)
self.rx_irq_i = Signal() # in: frame received (RI)
self.tx_irq_i = Signal() # in: frame transmitted (TI)
self.ncra_rst_o = Signal() # out: host wrote the NCRA reset bit
self.exi_int_n = Signal(init=1) # out: EXI interrupt, active low
# ── FIFO far sides, wired by BBATop ──────────────────────────────
# SPRAM read requests out to the arbiter.
self.spram_req_r_data = Signal(16)
self.spram_req_r_en = Signal()
self.spram_req_r_rdy = Signal()
# SPRAM read data back in. No FIFO: the response must be consumed in
# the cycle it is valid, which is why only one request is in flight.
self.spram_rsp_data = Signal(8)
self.spram_rsp_valid = Signal()
# TX frame bytes and the frame length, out to TXFrameDrain.
self.tx_bytes_r_data = Signal(8)
self.tx_bytes_r_en = Signal()
self.tx_bytes_r_rdy = Signal()
self.tx_ctrl_r_data = Signal(16)
self.tx_ctrl_r_en = Signal()
self.tx_ctrl_r_rdy = Signal()
# RX ring write pointer in from the assembler. The read pointer goes
# back out through the `rrp` tap above, not a FIFO.
self.rx_wptr_w_data = Signal(8)
self.rx_wptr_w_en = Signal()
self.rx_wptr_w_rdy = Signal()
# ──────────────────────────── structure ─────────────────────────────
def _make_fifos(self, m):
"""Create the FIFOs and connect their far sides to this module's ports."""
f = SimpleNamespace(
spram_req=SyncFIFO(width=16, depth=4),
# Deep enough for a whole maximum-size frame: SPI cannot
# back-pressure, so a byte that arrives when the FIFO is full is
# dropped and the frame goes out corrupt. The depth is what
# stands between a drain stall and a bad frame, not a tuning knob.
tx_bytes=SyncFIFO(width=8, depth=2048),
tx_ctrl=SyncFIFO(width=16, depth=4),
rx_wptr=SyncFIFO(width=8, depth=4),
)
m.submodules.spram_req = f.spram_req
m.submodules.tx_bytes = f.tx_bytes
m.submodules.tx_ctrl = f.tx_ctrl
m.submodules.rx_wptr = f.rx_wptr
m.d.comb += [
self.spram_req_r_data.eq(f.spram_req.r_data),
f.spram_req.r_en .eq(self.spram_req_r_en),
self.spram_req_r_rdy .eq(f.spram_req.r_rdy),
self.tx_bytes_r_data.eq(f.tx_bytes.r_data),
f.tx_bytes.r_en .eq(self.tx_bytes_r_en),
self.tx_bytes_r_rdy .eq(f.tx_bytes.r_rdy),
self.tx_ctrl_r_data.eq(f.tx_ctrl.r_data),
f.tx_ctrl.r_en .eq(self.tx_ctrl_r_en),
self.tx_ctrl_r_rdy .eq(f.tx_ctrl.r_rdy),
f.rx_wptr.w_data .eq(self.rx_wptr_w_data),
f.rx_wptr.w_en .eq(self.rx_wptr_w_en),
self.rx_wptr_w_rdy .eq(f.rx_wptr.w_rdy),
]
return f
def _make_registers(self, m):
"""Create register storage and the taps that read straight out of it."""
# PAR resets to a valid Nintendo OUI so the adapter has a usable
# source MAC before any host writes one. Hosts generally READ the MAC
# from here rather than setting it, so this value is the one that ends
# up on the wire.
par_reset = [0x00, 0x09, 0xBF, 0x00, 0x00, 0x01]
r = SimpleNamespace(
ncra=Signal(8),
ncrb=Signal(8),
imr=Signal(8),
ir=Signal(8),
# Both ring pointers reset to the ring's first page, matching the
# frame assembler. RRP == RWP == first page means "empty".
rwp_lo=Signal(8, init=_RX_PAGE_FIRST),
rrp_lo=Signal(8, init=_RX_PAGE_FIRST),
par=Array([Signal(8, name=f"par{i}", init=par_reset[i])
for i in range(6)]),
twd_lo=Signal(8),
twd_hi=Signal(8),
bp_lo=Signal(8, init=_BP_VAL),
bp_hi=Signal(8),
tlbp_lo=Signal(8),
tlbp_hi=Signal(8),
rxintt_lo=Signal(8),
rxintt_hi=Signal(8),
rhbp_lo=Signal(8, init=_RHBP_VAL),
rhbp_hi=Signal(8),
nwayc=Signal(8, init=_NWAYC_VAL),
gca=Signal(8),
misc=Signal(8, init=_MISC_VAL),
misc2=Signal(8),
# Command-space registers.
cmd_imr=Signal(8, init=_CMD_IMR_VAL),
cmd_ir=Signal(8),
cmd_misc=Array([Signal(8, name=f"cmd_misc_{i}") for i in range(16)]),
)
return r
def _register_taps(self, m, r):
"""Drive the ports that mirror register contents."""
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])),
self.ncra_sr.eq(r.ncra[3]),
self.ncrb .eq(r.ncrb),
self.rrp .eq(r.rrp_lo),
self.bp .eq(Cat(r.bp_lo, r.bp_hi[0:4])),
self.rhbp .eq(Cat(r.rhbp_lo, r.rhbp_hi[0:4])),
]
def _read_mux(self, m, r, t):
"""Combinational read data for the current register address.
Returns `rd_val`, valid for both address spaces: `t.cmd_space` selects
which decode applies. SPRAM reads do not come through here.
"""
reg_rd = Signal(8, name="reg_rdval")
with m.Switch(t.rd_sel):
with m.Case(_NCRA): m.d.comb += reg_rd.eq(r.ncra)
with m.Case(_NCRB): m.d.comb += reg_rd.eq(r.ncrb)
with m.Case(_IMR): m.d.comb += reg_rd.eq(r.imr)
with m.Case(_IR): m.d.comb += reg_rd.eq(r.ir)
with m.Case(_BP_LO): m.d.comb += reg_rd.eq(r.bp_lo)
with m.Case(_BP_HI): m.d.comb += reg_rd.eq(r.bp_hi)
with m.Case(_TLBP_LO): m.d.comb += reg_rd.eq(r.tlbp_lo)
with m.Case(_TLBP_HI): m.d.comb += reg_rd.eq(r.tlbp_hi)
with m.Case(_RXINTT_LO): m.d.comb += reg_rd.eq(r.rxintt_lo)
with m.Case(_RXINTT_HI): m.d.comb += reg_rd.eq(r.rxintt_hi)
with m.Case(_RWP_LO): m.d.comb += reg_rd.eq(r.rwp_lo)
with m.Case(_RRP_LO): m.d.comb += reg_rd.eq(r.rrp_lo)
with m.Case(_RHBP_LO): m.d.comb += reg_rd.eq(r.rhbp_lo)
with m.Case(_RHBP_HI): m.d.comb += reg_rd.eq(r.rhbp_hi)
with m.Case(_PAR0, _PAR1, _PAR2, _PAR3, _PAR4, _PAR5):
# PAR0 is 0x20, so the low three address bits are the index.
m.d.comb += reg_rd.eq(r.par[t.eff_addr[0:3]])
with m.Case(_NWAYC): m.d.comb += reg_rd.eq(r.nwayc)
with m.Case(_NWAYS): m.d.comb += reg_rd.eq(_NWAYS_VAL)
with m.Case(_GCA): m.d.comb += reg_rd.eq(r.gca)
with m.Case(_TWD_LO): m.d.comb += reg_rd.eq(r.twd_lo)
with m.Case(_TWD_HI): m.d.comb += reg_rd.eq(r.twd_hi)
with m.Case(_HIPR): m.d.comb += reg_rd.eq(_HIPR_VAL)
with m.Case(_MISC): m.d.comb += reg_rd.eq(r.misc)
with m.Case(_MISC2): m.d.comb += reg_rd.eq(r.misc2)
with m.Default(): m.d.comb += reg_rd.eq(0)
cmd_rd = Signal(8, name="cmd_rdval")
with m.Switch(t.cmd_reg):
with m.Case(_CMD_REVID): m.d.comb += cmd_rd.eq(_CMD_REVID_VAL)
with m.Case(_CMD_IMR): m.d.comb += cmd_rd.eq(r.cmd_imr)
with m.Case(_CMD_IR): m.d.comb += cmd_rd.eq(t.cmd_ir_val)
with m.Default():
with m.If(t.cmd_reg <= _CMD_LAST):
m.d.comb += cmd_rd.eq(r.cmd_misc[t.cmd_reg[0:4]])
with m.Else():
m.d.comb += cmd_rd.eq(0)
rd_val = Signal(8, name="rd_val")
with m.If(t.cmd_space):
m.d.comb += rd_val.eq(cmd_rd)
with m.Else():
m.d.comb += rd_val.eq(reg_rd)
return rd_val
def _register_write(self, m, r, f, t):
"""Apply one register-space write of `rx_data` to address `rd_sel`.
Called from the WRITE state. Addresses with no case are ignored.
"""
with m.Switch(t.rd_sel):
with m.Case(_NCRA):
m.d.sync += r.ncra.eq(self.rx_data)
with m.If(self.rx_data[0]): # RESET: self-clearing
m.d.sync += [
r.ncra[0].eq(0),
t.ncra_rst.eq(1),
# The reset rewinds the ring, so both pointers must
# follow the assembler back to the first page.
r.rwp_lo.eq(_RX_PAGE_FIRST),
r.rrp_lo.eq(_RX_PAGE_FIRST),
]
with m.If(self.rx_data[1:3].any() & ~t.tx_queued):
# ST0/ST1: start transmit. Either bit will do; the buffer
# select they encode is not implemented. The length is
# the number of bytes the host actually streamed, since a
# host that reaches this path never wrote TWD.
with m.If(f.tx_ctrl.w_rdy):
m.d.sync += [
f.tx_ctrl.w_data.eq(t.txdata_cnt),
f.tx_ctrl.w_en.eq(1),
t.tx_queued.eq(1),
t.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-to-clear
with m.Case(_RRP_LO):
# The assembler sees this through the `rrp` tap.
m.d.sync += r.rrp_lo.eq(self.rx_data)
with m.Case(_PAR0, _PAR1, _PAR2, _PAR3, _PAR4, _PAR5):
m.d.sync += r.par[t.eff_addr[0:3]].eq(self.rx_data)
with m.Case(_TWD_LO):
m.d.sync += [r.twd_lo.eq(self.rx_data),
t.tx_frame_len[0:8].eq(self.rx_data)]
with m.Case(_TWD_HI):
# Arms the deferred queue one cycle later, by which point both
# halves of the length are present.
m.d.sync += [r.twd_hi.eq(self.rx_data),
t.tx_frame_len[8:16].eq(self.rx_data),
t.twd_armed.eq(1)]
with m.Case(_TXDATA):
with m.If(f.tx_bytes.w_rdy):
m.d.sync += [f.tx_bytes.w_data.eq(self.rx_data),
f.tx_bytes.w_en.eq(1),
t.txdata_cnt.eq(t.txdata_cnt + 1)]
with m.Case(_NCRB): m.d.sync += r.ncrb.eq(self.rx_data)
with m.Case(_BP_LO): m.d.sync += r.bp_lo.eq(self.rx_data)
with m.Case(_BP_HI): m.d.sync += r.bp_hi.eq(self.rx_data)
with m.Case(_TLBP_LO): m.d.sync += r.tlbp_lo.eq(self.rx_data)
with m.Case(_TLBP_HI): m.d.sync += r.tlbp_hi.eq(self.rx_data)
with m.Case(_RXINTT_LO): m.d.sync += r.rxintt_lo.eq(self.rx_data)
with m.Case(_RXINTT_HI): m.d.sync += r.rxintt_hi.eq(self.rx_data)
with m.Case(_RHBP_LO): m.d.sync += r.rhbp_lo.eq(self.rx_data)
with m.Case(_RHBP_HI): m.d.sync += r.rhbp_hi.eq(self.rx_data)
with m.Case(_NWAYC): m.d.sync += r.nwayc.eq(self.rx_data)
with m.Case(_GCA): m.d.sync += r.gca.eq(self.rx_data)
with m.Case(_MISC): m.d.sync += r.misc.eq(self.rx_data)
with m.Case(_MISC2): m.d.sync += r.misc2.eq(self.rx_data)
# TXDATA is a FIFO port, not a register: the address must not advance
# across it or a multi-byte frame write would spill into 0x49 upwards.
with m.If(t.rd_sel != _TXDATA):
m.d.sync += t.eff_addr.eq(t.eff_addr + 1)
def _transaction_fsm(self, m, r, f, t, rd_val):
"""Decode one EXI transaction.
States, in the order a transaction visits them:
HEADER0 byte 0: pick address space and direction
RHDR_HI register space, address high byte
RHDR_LO register space, address low byte. The address is now
complete, so an SPRAM read is issued here to use the
pad byte as its prefetch window.
RHDR_PAD register space, trailing pad byte. A register read
pushes its first response byte during it.
CHDR_1 command space, header byte 1
CMD_READ command space read data phase
CMD_WRITE command space write data phase
WRITE register space write data phase
REG_READ register space read data phase
SPRAM_STREAM register space read from the RX ring, one request in
flight, streaming until CS rises
SPRAM_END drain the in-flight SPRAM request after CS
DEVID device-ID query; the reply comes from SPIMode3Slave so
this only consumes the clocked bytes
TRAILER consume anything else until CS rises
Every data phase ends on `~cs_active`, never on a byte count.
"""
with m.FSM(domain="sync", name="exi_fsm"):
with m.State("HEADER0"):
with m.If(self.rx_rdy):
m.d.comb += self.rx_en.eq(1)
m.d.sync += [
t.is_write.eq(self.rx_data[6]),
t.prefetched.eq(0),
t.outstanding.eq(0),
t.sp_pending.eq(0),
t.cmd_have.eq(0),
t.cmd_pushes.eq(0),
]
with m.If(self.rx_data[7]):
m.d.sync += t.cmd_space.eq(0)
m.next = "RHDR_HI"
with m.Else():
m.d.sync += [
t.cmd_space.eq(1),
t.cmd_reg.eq(self.rx_data[0:6]),
t.addr.eq(self.rx_data[0:6]),
t.eff_addr.eq(self.rx_data[0:6]),
]
m.next = "CHDR_1"
with m.State("RHDR_HI"):
with m.If(self.rx_rdy):
m.d.comb += self.rx_en.eq(1)
m.d.sync += t.addr[8:16].eq(self.rx_data)
m.next = "RHDR_LO"
with m.State("RHDR_LO"):
with m.If(self.rx_rdy):
m.d.comb += self.rx_en.eq(1)
full = Cat(self.rx_data, t.addr[8:16])
m.d.sync += [t.addr.eq(full), t.eff_addr.eq(full),
t.sp_addr.eq(full)]
with m.If(~t.is_write & (full >= _SPRAM_BASE)
& f.spram_req.w_rdy):
# Combinational w_en: the pad byte is the whole budget
# for this round trip and a registered strobe does not
# fit in it.
m.d.comb += [f.spram_req.w_data.eq(full),
f.spram_req.w_en.eq(1)]
m.d.sync += [t.sp_addr.eq(full + 1),
t.outstanding.eq(1)]
m.next = "RHDR_PAD"
with m.State("RHDR_PAD"):
with m.If(self.rx_rdy):
m.d.comb += self.rx_en.eq(1)
with m.If(t.is_write):
m.next = "WRITE"
with m.Elif(t.addr >= _SPRAM_BASE):
m.next = "SPRAM_STREAM"
with m.Else():
m.next = "REG_READ"
with m.Elif(~t.is_write & (t.addr < _SPRAM_BASE)
& self.tx_rdy & ~t.prefetched):
# One-shot: this branch is re-evaluated every cycle the pad
# byte has not arrived, and repeating it would consume the
# register stream early.
m.d.comb += [self.tx_data.eq(rd_val), self.tx_en.eq(1)]
m.d.sync += [t.eff_addr.eq(t.eff_addr + 1),
t.prefetched.eq(1)]
with m.State("CHDR_1"):
with m.If(self.rx_rdy):
m.d.comb += self.rx_en.eq(1)
with m.If(~t.is_write & (t.cmd_reg == 0)):
m.next = "DEVID"
with m.Elif(t.is_write):
m.next = "CMD_WRITE"
with m.Else():
m.next = "CMD_READ"
with m.Elif(~t.is_write & self.tx_rdy & (t.cmd_pushes == 0)):
m.d.comb += [self.tx_data.eq(rd_val), self.tx_en.eq(1)]
m.d.sync += t.cmd_pushes.eq(1)
with m.State("CMD_READ"):
# The response is pushed TWICE so it appears in both of the
# byte positions hosts read it from.
with m.If(self.tx_rdy & (t.cmd_pushes < 2)):
m.d.comb += [self.tx_data.eq(rd_val), self.tx_en.eq(1)]
m.d.sync += t.cmd_pushes.eq(t.cmd_pushes + 1)
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("CMD_WRITE"):
# The LAST data byte before CS rise is the value: hosts send
# either one data byte or a pad plus the value, and taking the
# last one serves both without counting bytes.
with m.If(self.rx_rdy):
m.d.comb += self.rx_en.eq(1)
m.d.sync += [t.cmd_wdata.eq(self.rx_data),
t.cmd_wreg.eq(t.cmd_reg),
t.cmd_have.eq(1)]
with m.If(~self.cs_active):
with m.If(t.cmd_have):
m.d.sync += t.cmd_wr.eq(1)
m.next = "HEADER0"
with m.State("WRITE"):
with m.If(self.rx_rdy):
m.d.comb += self.rx_en.eq(1)
self._register_write(m, r, f, t)
with m.If(~self.cs_active):
m.next = "HEADER0"
with m.State("REG_READ"):
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 += t.eff_addr.eq(t.eff_addr + 1)
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("DEVID"):
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"):
# One request in flight: the response has nowhere to wait, so
# it must be consumed in the cycle it is valid.
with m.If(self.cs_active & f.spram_req.w_rdy & self.tx_rdy
& (t.outstanding == 0) & ~t.sp_pending):
m.d.comb += [f.spram_req.w_data.eq(t.sp_addr),
f.spram_req.w_en.eq(1)]
m.d.sync += [t.sp_addr.eq(t.sp_addr + 1),
t.outstanding.eq(t.outstanding + 1)]
with m.If(self.rx_rdy):
m.d.comb += self.rx_en.eq(1)
with m.If(~self.cs_active):
m.next = "SPRAM_END"
with m.State("SPRAM_END"):
with m.If(self.rx_rdy):
m.d.comb += self.rx_en.eq(1)
with m.If((t.outstanding == 0) & ~t.sp_pending & ~self.rx_rdy):
m.next = "HEADER0"
with m.State("TRAILER"):
with m.If(self.rx_rdy):
m.d.comb += self.rx_en.eq(1)
with m.If(~self.cs_active):
m.next = "HEADER0"
def _spram_responses(self, m, t):
"""Forward SPRAM read data to the EXI TX stream.
Runs outside the FSM because a response can arrive while the FSM is
still in a header state. `sp_hold` carries one byte for the case
where the TX FIFO is momentarily full.
"""
mine = self.spram_rsp_valid & (t.outstanding != 0)
with m.If(mine & self.tx_rdy & ~t.sp_pending):
m.d.comb += [self.tx_data.eq(self.spram_rsp_data),
self.tx_en.eq(1)]
m.d.sync += t.outstanding.eq(t.outstanding - 1)
with m.Elif(mine):
m.d.sync += [t.sp_hold.eq(self.spram_rsp_data),
t.sp_pending.eq(1),
t.outstanding.eq(t.outstanding - 1)]
with m.Elif(t.sp_pending & self.tx_rdy):
m.d.comb += [self.tx_data.eq(t.sp_hold), self.tx_en.eq(1)]
m.d.sync += t.sp_pending.eq(0)
def _command_write(self, m, r, t):
"""Apply a pending command-space write, one cycle after the data byte."""
with m.If(t.cmd_wr):
m.d.sync += t.cmd_wr.eq(0)
with m.Switch(t.cmd_wreg):
with m.Case(_CMD_REVID):
pass # read-only
with m.Case(_CMD_IMR):
m.d.sync += r.cmd_imr.eq(t.cmd_wdata)
with m.Case(_CMD_IR):
m.d.sync += r.cmd_ir.eq(r.cmd_ir & ~t.cmd_wdata)
with m.Default():
with m.If(t.cmd_wreg <= _CMD_LAST):
m.d.sync += r.cmd_misc[t.cmd_wreg[0:4]].eq(t.cmd_wdata)
def _interrupts(self, m, r, t):
"""The two interrupt levels.
MAC level: `IR & IMR`, the per-event bits a host clears by writing 1.
EXI level: command register `_CMD_IR`, whose bit 7 is a live level
meaning "a MAC interrupt is pending", masked by `_CMD_IMR`. INT_N
comes from the EXI level, so masking there silences the line while a
host works through its handler.
Bit 7 is a level rather than a latch because hosts clear the MAC IR
first and the EXI status afterwards; a latch would lose an event that
arrived in between.
"""
mac_int = Signal()
m.d.comb += [
mac_int.eq((r.ir & r.imr).any()),
t.cmd_ir_val.eq(Cat(r.cmd_ir[0:_CMD_IR_MAC], mac_int)),
]
m.d.sync += self.exi_int_n.eq(~(t.cmd_ir_val & r.cmd_imr).any())
def _ethernet_events(self, m, r, f, t):
"""Register-side effects of things the ethernet engines report."""
# The assembler's write pointer, so the host can see new frames.
with m.If(f.rx_wptr.r_rdy):
m.d.sync += [f.rx_wptr.r_en.eq(1),
r.rwp_lo.eq(f.rx_wptr.r_data)]
with m.If(self.rx_irq_i):
m.d.sync += r.ir[_IR_RI].eq(1)
with m.If(self.rx_overflow):
m.d.sync += r.ir[_IR_RBFI].eq(1)
# TX done: flag it, release the start bits, and allow the next frame
# to be queued.
with m.If(self.tx_irq_i):
m.d.sync += [r.ir[_IR_TI].eq(1),
r.ncra[1:3].eq(0),
t.tx_queued.eq(0)]
# Deferred TX queue, armed by a TWD_HI write. Held until tx_ctrl has
# room so a full control FIFO cannot drop the trigger.
with m.If(t.twd_armed & f.tx_ctrl.w_rdy):
m.d.sync += [f.tx_ctrl.w_data.eq(t.tx_frame_len),
f.tx_ctrl.w_en.eq(1),
t.twd_armed.eq(0),
t.tx_queued.eq(1),
t.txdata_cnt.eq(0)]
# ──────────────────────────── elaborate ─────────────────────────────
def elaborate(self, platform):
m = Module()
f = self._make_fifos(m)
r = self._make_registers(m)
# Transaction and TX-path state, shared across the blocks below.
t = SimpleNamespace(
# Current transaction.
addr=Signal(16), # register address from the header
eff_addr=Signal(16), # running address of the current data byte
is_write=Signal(),
cmd_space=Signal(),
cmd_reg=Signal(6),
prefetched=Signal(), # first response byte already pushed
# Command-space write, applied a cycle after the data byte.
cmd_wr=Signal(),
cmd_wreg=Signal(6),
cmd_wdata=Signal(8),
cmd_have=Signal(), # at least one data byte seen
cmd_pushes=Signal(2), # command-read responses pushed so far
cmd_ir_val=Signal(8), # command IR including the live MAC bit
# SPRAM streaming reads.
sp_addr=Signal(16), # next SPRAM byte address to request
outstanding=Signal(4), # requests issued but not yet answered
sp_hold=Signal(8), # response held while the TX FIFO was full
sp_pending=Signal(),
# TX frame length. `txdata_cnt` counts what the host actually
# streamed and is what the start-transmit path uses, because a
# host need not write TWD at all.
tx_frame_len=Signal(16),
txdata_cnt=Signal(16),
twd_armed=Signal(),
tx_queued=Signal(),
ncra_rst=Signal(),
)
t.rd_sel = t.eff_addr[0:8]
m.d.comb += self.ncra_rst_o.eq(t.ncra_rst)
self._register_taps(m, r)
# Strobes default low; the blocks below assert them for one cycle.
m.d.sync += [
f.tx_bytes.w_en.eq(0),
f.tx_ctrl.w_en .eq(0),
f.rx_wptr.r_en .eq(0),
t.ncra_rst .eq(0),
]
m.d.comb += [
f.spram_req.w_en .eq(0),
f.spram_req.w_data.eq(0),
self.rx_en .eq(0),
self.tx_en .eq(0),
self.tx_data.eq(0),
]
self._interrupts(m, r, t)
rd_val = self._read_mux(m, r, t)
self._transaction_fsm(m, r, f, t, rd_val)
self._spram_responses(m, t)
self._command_write(m, r, t)
self._ethernet_events(m, r, f, t)
return m
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
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, short form: 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
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
async def cmd_write_long(ctx, creg, val):
"""Command-space write, long form: ONE 4-byte transfer
[0x40|reg, 0x00, 0x00, val] -- exi_imm_read_write(..., 4)."""
ctx.set(reg.cs_active, 1)
await push_rx(ctx, 0x40 | (creg & 0x3F))
await push_rx(ctx, 0x00)
await push_rx(ctx, 0x00) # padding, NOT the value
await push_rx(ctx, val) # the value is the LAST byte
ctx.set(reg.cs_active, 0)
await ctx.tick("sync").repeat(6)
async def cmd_read_long(ctx, creg):
"""Command-space read, long form: a 4-byte read/write whose byte 3
is the response. Returns (byte2, byte3)."""
tx_log.clear()
ctx.set(reg.cs_active, 1)
await push_rx(ctx, creg & 0x3F)
for _ in range(4):
sample_tx(ctx)
await ctx.tick("sync")
await push_rx(ctx, 0x00)
got = await collect_tx(ctx, 2)
ctx.set(reg.cs_active, 0)
await ctx.tick("sync").repeat(6)
return got
# 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.
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
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
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.
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.
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
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 write, 2-byte header + a 1-byte data phase ──────
await cmd_write(ctx, 0x02, [0xF8]) # unmask
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: start transmit with NO TWD write; length must be counted ─
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)
# ── T15: reset defaults, and every stored register reads back ────
defaults = ((_NWAYC, _NWAYC_VAL), (_MISC, _MISC_VAL),
(_BP_LO, _BP_VAL), (_RHBP_LO, _RHBP_VAL))
for addr, want in defaults:
got = (await exi_read(ctx, addr, 1))[0]
if got != want:
errors.append(f"T15 reg 0x{addr:02X} default 0x{got:02X}, "
f"want 0x{want:02X}")
print(f"T15 defaults: NWAYC/MISC/BP/RHBP = "
f"{[hex((await exi_read(ctx, a, 1))[0]) for a, _ in defaults]}")
# A host's init sequence, in order.
init_writes = ((_NCRB, 0x52), (_RXINTT_LO, 0x00), (_RXINTT_HI, 0x06),
(_MISC2, 0x80), (_TLBP_LO, 0x00), (_TLBP_HI, 0x00),
(_BP_LO, 0x01), (_BP_HI, 0x00),
(_RHBP_LO, 0x0F), (_RHBP_HI, 0x00),
(_GCA, 0x08), (_NWAYC, 0x0C))
for addr, val in init_writes:
await exi_write(ctx, addr, [val])
for addr, val in init_writes:
got = (await exi_read(ctx, addr, 1))[0]
if got != val:
errors.append(f"T15 reg 0x{addr:02X}: wrote 0x{val:02X}, "
f"read 0x{got:02X}")
print(f"T15 all {len(init_writes)} init writes read back")
# NCRB is what the RX filter will be steered from, so the bits must
# survive: 0x52 = AB | CA | 2_PACKETS_PER_INT, with PR and PM CLEAR
# (so the real BBA delivers own-MAC + broadcast only).
ncrb = ctx.get(reg.ncrb)
pr = (ncrb >> NCRB_PR) & 1
pm = (ncrb >> _NCRB_PM) & 1
ab = (ncrb >> _NCRB_AB) & 1
print(f"T15 ncrb export = 0x{ncrb:02X} (PR={pr} PM={pm} AB={ab})")
if ncrb != 0x52 or pr or pm or not ab:
errors.append(f"T15 ncrb export 0x{ncrb:02X}, want 0x52 "
f"(PR=0 PM=0 AB=1)")
if ctx.get(reg.rhbp) != 0x00F or ctx.get(reg.bp) != 0x001:
errors.append(f"T15 ring geometry exports: bp="
f"0x{ctx.get(reg.bp):03X} rhbp="
f"0x{ctx.get(reg.rhbp):03X}, want 0x001/0x00F")
print(f"T15 geometry exports: bp=0x{ctx.get(reg.bp):03X} "
f"rhbp=0x{ctx.get(reg.rhbp):03X}")
# ── T14: the command-register map ────────────────────────────────
rev = (await cmd_read(ctx, _CMD_REVID, 1))[0]
print(f"T14 cmd 0x01 revision ID = 0x{rev:02X} (want 0x00, real HW)")
if rev != _CMD_REVID_VAL:
errors.append(f"T14 revision ID 0x{rev:02X}, want "
f"0x{_CMD_REVID_VAL:02X}")
# Read-only: a write must not change it.
await cmd_write(ctx, _CMD_REVID, [0xA5])
rev2 = (await cmd_read(ctx, _CMD_REVID, 1))[0]
if rev2 != _CMD_REVID_VAL:
errors.append(f"T14 revision ID writable: became 0x{rev2:02X}")
# The write-only ones must at least STICK, so software that writes
# unconditionally.
for creg, val in ((0x04, 0xD1), (0x05, 0x4E), (0x09, 0x3C),
(0x0B, 0x77), (0x0F, 0x11)):
await cmd_write(ctx, creg, [val])
rb = (await cmd_read(ctx, creg, 1))[0]
if rb != val:
errors.append(f"T14 cmd 0x{creg:02X}: wrote 0x{val:02X}, "
f"read 0x{rb:02X}")
print(f"T14 cmd 0x04/05/09/0B/0F all read back what was written")
# Out of range still reads 0, so a stray high register cannot alias
# onto a real one.
hi = (await cmd_read(ctx, 0x20, 1))[0]
if hi != 0x00:
errors.append(f"T14 cmd 0x20 reads 0x{hi:02X}, want 0x00")
# ── T13: the command-level interrupt and its mask ────────────────
await exi_write(ctx, _IR, [0xFF]) # clear MAC IR
await exi_write(ctx, _IMR, [0x86]) # enable RBFI|TI|RI
await cmd_write(ctx, 0x02, [0xF8]) # IRMASKNONE
before = (await cmd_read(ctx, 0x03, 1))[0]
int_idle = ctx.get(reg.exi_int_n)
ctx.set(reg.rx_irq_i, 1) # a frame arrives
await ctx.tick("sync")
ctx.set(reg.rx_irq_i, 0)
await ctx.tick("sync").repeat(3)
after = (await cmd_read(ctx, 0x03, 1))[0]
int_pend = ctx.get(reg.exi_int_n)
print(f"T13 cmd 0x03: idle=0x{before:02X} after RI=0x{after:02X}; "
f"INT_N {int_idle} -> {int_pend}")
if before & 0x80:
errors.append(f"T13 cmd 0x03 bit7 set while idle: 0x{before:02X}")
if not (after & 0x80):
errors.append(f"T13 cmd 0x03 bit7 NOT set after RI: 0x{after:02X} "
f"- both drivers would skip bba_interrupt()")
if int_pend:
errors.append("T13 INT_N not asserted with an unmasked MAC event")
# Masking at the COMMAND level must drop INT_N -- this is how both
# drivers stop the line re-asserting while their handler runs.
await cmd_write(ctx, 0x02, [0x00]) # IRMASKALL
await ctx.tick("sync").repeat(3)
masked = ctx.get(reg.exi_int_n)
still = (await cmd_read(ctx, 0x03, 1))[0]
print(f"T13b IRMASKALL: INT_N={masked} (want 1), "
f"status still 0x{still:02X}")
if not masked:
errors.append("T13b INT_N still asserted after IRMASKALL")
if not (still & 0x80):
errors.append("T13b masking must not clear the pending status")
await cmd_write(ctx, 0x02, [0xF8])
await exi_write(ctx, _IR, [0xFF]) # tidy up
# ── T12: command write as ONE 4-byte transfer, value in byte 3 ───
await cmd_write_long(ctx, 0x02, 0x5A)
both = await cmd_read_long(ctx, 0x02)
print(f"T12 4-byte cmd write 0x5A -> bytes 2,3 = "
f"{[hex(x) for x in both]}")
if len(both) < 2 or both[1] != 0x5A:
errors.append(f"T12 long-form cmd write: byte 3 = "
f"{[hex(x) for x in both]}, want 0x5A")
if len(both) >= 1 and both[0] != 0x5A:
errors.append(f"T12 byte-2 slot also wrong: {hex(both[0])}")
await cmd_write_long(ctx, 0x02, 0x00) # restore
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 TX without 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.
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.")
Reference in New Issue View Git Blame Copy Permalink