Roflin/rebbarb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
41ba716759d8d4cb5ce0baa9e9927f8abf950f19
rebbarb/exi_bba/w5100_parallel_master.py
T
Roflin 41d487e1a7 Cut dead signals and the unused W5100 debug peek 
An audit for signals that are written but never read turned up three things no
longer connected to anything.  All verified unreferenced across every module,
testbench and script before removal.

  exi_capture      cs_n_prev          registered on sync, read by nothing.
  w5100            udp_rx_sof         driven in the UDP rx stream mux; no
                                      consumer uses it.  uart_shell needs only
                                      data/valid/eof, because the request
                                      frames the transfer.  The MACRAW rx
                                      interface keeps its sof -- the frame
                                      assembler does need it.
  w5100            the debug peek      4 ports, 2 FSM states (DBG_RD/DBG_RD_W)
                                      and a branch in the IDLE arbiter's
                                      priority chain.  dbg_rd_req was driven by
                                      nothing -- not even its own testbench --
                                      so this was untested logic sitting in the
                                      path that decides what the bus master
                                      does next.

Kept deliberately, against the audit: the unused register-address constants in
both ethernet masters (_IR, _TMSR, _S0_SR, _S0_TX_FSR, _S0_TX_RD and the three
W5500 buffer ones).  They cost no logic and they document the chip's register
map in one place; that map was the source of three hardware bugs, and a partial
map invites re-guessing the missing half.  Same for status_panel's eth_rst_n,
which is a documented panel feature synth.py has not wired yet -- a gap, not
dead weight.  And rx_frame_assembler's geom_honoured, which is one comb line
asserting a property the tests otherwise could not observe.

Also fixed the register file's opening docstring, which claimed the module did
"three jobs" and invited splitting it into three.  It does not: the interrupt
level is IR & IMR, a frame arriving is a register side effect, and the read mux
feeds the TX stream combinationally in the cycle the FSM pushes it.  Those are
facets of one job that share the register bank, not separable parts.  _read_mux
now records why that path must stay unregistered -- one pipeline register there
is 42 ns against a 296 ns byte budget, which is the tx FIFO underrun that the
+1 descriptor hunt ended in.

Full suite passes; bba_top T9 6/6 at EXI_HALF=1, 2 and 8; W5100 elaborates with
enable_udp_test both ways.  Sweep: 8/8 seeds PASS, best seed 3 at clk 29.08 MHz
(1.21x the 24 MHz target), and area down again:

    ICESTORM_LC   4350 original -> 4328 after the rx_rptr cut -> 4305 now

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

1491 lines
70 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""W5100 parallel-bus master — sync domain.
A drop-in alternative to `W5500SPIMaster` that talks to a WIZnet **W5100** over
its **indirect parallel bus** instead of SPI. The external streaming interface
(init_req/init_done/par, tx_*, rx_*) is identical, so BBATop wiring is unchanged;
only the physical pins differ (a parallel bus instead of 4 SPI wires).
Why parallel
------------
SPI serialises 8 bits per byte, so on this UP5K (whose W5500-operating logic
closes only ~40 MHz) the SPI byte rate caps at ~12 Mbit/s. A parallel bus moves
a whole byte per access, so the same ~24 MHz sync logic clears the 27 Mbit/s EXI
ceiling — the real hard limit — with margin. See CLAUDE.md.
W5100 indirect bus interface (IDM)
----------------------------------
Only two address lines A[1:0] are wired (the upper address lines are tied to 0
on the board, so a power-up *direct*-mode access at A=00 still lands on MR):
A[1:0] register
00 MR (Mode Register — also reachable directly at power-up)
01 IDM_AR0 (indirect address, high byte)
10 IDM_AR1 (indirect address, low byte)
11 IDM_DR (indirect data — accesses mem[IDM_AR]; auto-increments
IDM_AR when MR.AI is set)
So a register/buffer access is: write IDM_AR0/AR1 with the 16-bit address, then
read/write IDM_DR. With MR.AI=1 a multi-byte block is one address-set followed
by a burst of IDM_DR accesses (the chip auto-increments) — used for SHAR and for
streaming frame data.
A bus cycle drives A + (for writes) D with /CS and /RD or /WR asserted for
`strobe_cycles` sync clocks (≥ the W5100's ~80 ns access time at 24 MHz).
Phase status
------------
Phase 1 (this file): bus access engine + transaction engine + init sequence,
verified against a W5100 bus model. TX/RX MACRAW (with socket-buffer ring
wraparound) land in phases 2–3.
"""
from amaranth import *
__all__ = ["W5100ParallelMaster"]
# ── W5100 register addresses (indirect 16-bit address space) ────────────────
_MR = 0x0000 # Mode register (common)
_GAR0 = 0x0001 # Gateway IP, 4 bytes
_SUBR0 = 0x0005 # Subnet mask, 4 bytes
_SHAR0 = 0x0009 # Source MAC, 6 bytes
_SIPR0 = 0x000F # Source IP, 4 bytes
_NETLCKR = 0x0071 # W5100S Network Lock Register (WO) — GWR/SUBR/SHAR/SIPR are
# Default RSTn low time before release (sync cycles). 24_000 @ 24 MHz = 1 ms;
# the W5100S datasheet minimum is ~2 us, so this is generous on purpose.
_RST_LOW_DEFAULT = 24_000
_NETLCK_UNLOCK = 0x3A # writable ONLY while NETLCKR==0x3A (default 0x00 = locked)
_IR = 0x0015 # Interrupt register
_IMR = 0x0016 # Interrupt mask
_RMSR = 0x001A # RX memory size (2 bits/socket)
_TMSR = 0x001B # TX memory size
_S0_MR = 0x0400 # Socket 0 mode
_S0_CR = 0x0401 # Socket 0 command
_S0_IR = 0x0402 # Socket 0 interrupt
_S0_SR = 0x0403 # Socket 0 status
_S0_TX_FSR = 0x0420 # Socket 0 TX free size (2 bytes)
_S0_TX_RD = 0x0422 # Socket 0 TX read pointer
_S0_TX_WR = 0x0424 # Socket 0 TX write pointer
_S0_RX_RSR = 0x0426 # Socket 0 RX received size (2 bytes)
_S0_RX_RD = 0x0428 # Socket 0 RX read pointer
# Per-socket register/buffer geometry (2 KB per socket, RMSR/TMSR=0x55).
# The UDP-test socket number is configurable; addresses are computed from it in
# __init__ (see _socket_addrs). Socket-n register block base = 0x0400+n*0x100,
# TX buffer base = 0x4000+n*0x800, RX buffer base = 0x6000+n*0x800.
# NOTE: the W5100S datasheet's pseudo-code quotes 0x8000/0xC000 for the TX/RX
# memory blocks, but the PARALLEL/indirect (IDM) interface only decodes a 15-bit
# offset (mask 0x7FFF) -- bench-proven 2026-09-20 by an address-aliasing probe:
# 0xC000 and 0x4000 read back the SAME cell, while 0x6000 is distinct. So over
# IDM the buffers live at the original W5100 addresses. Using 0x8000 makes
# payload writes alias down onto 0x0000+, silently CLOBBERING SHAR/SIPR (frames
# then go out with a zero source MAC/IP and get dropped). Do not "fix" these to
# 0x8000/0xC000 again.
_TX_BASE = 0x4000 # Socket 0 TX buffer base
_RX_BASE = 0x6000 # Socket 0 RX buffer base
_S0_TX_MASK = 0x07FF # 2 KB ring mask
_S0_RX_MASK = 0x07FF
_SN_MASK = 0x07FF # 2 KB ring mask (any socket)
def _socket_addrs(n):
"""Return the register/buffer addresses for socket `n` (0..3)."""
base = 0x0400 + n * 0x0100
return dict(
MR=base + 0x00, CR=base + 0x01, IR=base + 0x02, SR=base + 0x03,
PORT=base + 0x04, DIPR=base + 0x0C, DPORT=base + 0x10,
TX_WR=base + 0x24, RX_RSR=base + 0x26, RX_RD=base + 0x28,
TX_BASE=0x4000 + n * 0x0800, RX_BASE=0x6000 + n * 0x0800,
)
# MR bits / command / mode values
_MR_RST = 0x80
_MR_AI = 0x02 # address auto-increment (indirect mode)
_MR_IND = 0x01 # indirect bus interface mode
_S0_MR_MACRAW = 0x04
_S0_MR_MF = 0x40 # MACRAW MAC filter: own-MAC (SHAR) + broadcast only
_S1_MR_UDP = 0x02 # socket UDP mode
_CR_OPEN = 0x01
_CR_SEND = 0x20
_CR_RECV = 0x40
def _ip_bytes(dotted):
"""'192.168.1.100' → [192, 168, 1, 100] (big-endian, network order)."""
parts = [int(x) for x in dotted.split(".")]
if len(parts) != 4 or any(not 0 <= p <= 255 for p in parts):
raise ValueError(f"invalid IPv4 address: {dotted!r}")
return parts
def _port_bytes(port):
"""6464 → [0x19, 0x40] (big-endian)."""
return [(port >> 8) & 0xFF, port & 0xFF]
# Indirect-mode address selects (A[1:0])
_A_MR = 0b00
_A_AR0 = 0b01 # IDM_AR high byte
_A_AR1 = 0b10 # IDM_AR low byte
_A_DR = 0b11 # IDM_DR (data)
class W5100ParallelMaster(Elaboratable):
"""W5100 master over the indirect parallel bus, sync clock domain.
Physical bus pins
-----------------
bus_addr : A[1:0] output
bus_data_o : D[7:0] output value (drive when bus_data_oe=1)
bus_data_oe: data-bus output enable (1=FPGA drives D, 0=W5100 drives D)
bus_data_i : D[7:0] input value (sampled during reads)
cs_n / rd_n / wr_n : bus control (active low)
w5100_int_n : W5100 INT_N input (active low)
w5100_rst_n : W5100 hardware reset (active low)
Init / TX / RX interfaces are identical to W5500SPIMaster.
"""
def __init__(self, strobe_cycles=3, reset_cycles=2_000_000,
enable_udp_test=False, udp_socket=3,
src_ip="192.168.1.123", subnet="255.255.255.0",
gateway="192.168.1.1", dst_ip="192.168.1.100",
src_port=6464, dst_port=6464,
udp_payload=b"REBBARB-UDP-TEST\r\n"):
# /RD//WR strobe width in sync cycles (≥ W5100 access time).
self._strobe = strobe_cycles
# MR-reset settle wait; testbench overrides with a small value.
self._reset_cycles = reset_cycles
# RSTn low time. Clamped to reset_cycles so the testbenches (which
# pass a tiny reset_cycles to stay fast) keep wait_ctr wide enough to
# actually reach it -- an unreachable target would hang the FSM.
self._rst_low = min(_RST_LOW_DEFAULT, reset_cycles)
# ── Optional socket-1 UDP bring-up test ───────────────────────────────
# When enabled, init also configures the W5100's IP stack (GAR/SUBR/
# SIPR) and opens socket 1 in UDP mode; pulsing `udp_send_req` then
# emits one UDP datagram to `udp_dst_ip`:dst_port with the payload
# supplied on the `udp_pl_*` stream. MACRAW socket 0 is untouched.
# Everything below is elaborated only when enabled, so a plain BBA
# build (and the existing MACRAW testbench) is bit-for-bit unchanged.
#
# Network config is build-time (the board's own address); the
# destination and payload are RUNTIME so the UART shell can drive them
# from a typed command. `dst_ip` here is only a power-on default for
# the udp_dst_ip input.
if not 1 <= udp_socket <= 3:
raise ValueError("udp_socket must be 1..3 (socket 0 is MACRAW)")
self._enable_udp = enable_udp_test
self._sn = _socket_addrs(udp_socket)
self._src_ip = _ip_bytes(src_ip)
self._subnet = _ip_bytes(subnet)
self._gateway = _ip_bytes(gateway)
self._src_port = _port_bytes(src_port)
self._dst_port = _port_bytes(dst_port)
_dst_default = _ip_bytes(dst_ip)
# Runtime UDP-send control / datapath.
self.udp_send_req = Signal() # pulse: emit one datagram
self.udp_test_busy = Signal() # level: send in progress
self.udp_dst_ip = Signal(32, # destination IP (big-endian:
init=int.from_bytes(bytes(_dst_default), "big")) # [0]=MSB octet)
# Payload byte stream (driven by the shell; consumed during the send).
self.udp_pl_data = Signal(8)
self.udp_pl_valid = Signal()
self.udp_pl_last = Signal()
self.udp_pl_ready = Signal()
# UDP receive (poll-driven): pulse udp_rx_req to check the socket's RX
# buffer for a datagram. Either udp_rx_none pulses (nothing waiting) or
# the payload streams out on udp_rx_* (sof/eof frame the datagram);
# udp_rx_busy is high while a check/read is in progress.
self.udp_rx_req = Signal()
self.udp_rx_busy = Signal()
self.udp_rx_none = Signal()
self.udp_rx_data = Signal(8)
self.udp_rx_valid = Signal()
self.udp_rx_eof = Signal()
self.udp_rx_ready = Signal()
# Physical parallel bus
self.bus_addr = Signal(2)
self.bus_data_o = Signal(8)
self.bus_data_oe = Signal()
self.bus_data_i = Signal(8)
self.cs_n = Signal(init=1)
self.rd_n = Signal(init=1)
self.wr_n = Signal(init=1)
self.w5100_int_n = Signal(init=1)
self.w5100_rst_n = Signal(init=1)
# Init control
self.init_req = Signal()
self.init_done = Signal()
self.par = Signal(48) # MAC address (PAR0..5 packed)
# NCRB.PR from the GC. Low (the normal case) means MACRAW filters to
# own-MAC + broadcast, which is what the real BBA does -- see the
# comment on the S0_MODE write. Sampled when the socket is OPENed,
# i.e. at init; a later change takes effect on the next NCRA reset.
self.promisc = Signal()
# TX stream
self.tx_data = Signal(8)
self.tx_valid = Signal()
self.tx_ready = Signal()
self.tx_sof = Signal()
self.tx_eof = Signal()
# RX stream
self.rx_data = Signal(8)
self.rx_valid = Signal()
self.rx_ready = Signal()
self.rx_sof = Signal()
self.rx_eof = Signal()
def elaborate(self, platform):
m = Module()
STROBE = self._strobe
sn = self._sn # UDP-test socket register/buffer addresses
# ── Bus access engine: one indirect-bus read or write cycle ──────────
bus_go = Signal()
bus_rw = Signal() # 1 = write, 0 = read
bus_a = Signal(2)
bus_wdata = Signal(8)
bus_rdata = Signal(8)
bus_done = Signal()
bus_ctr = Signal(range(STROBE + 2))
rw_r = Signal()
# registered physical outputs
a_o = Signal(2)
d_o = Signal(8)
d_oe = Signal()
cs_r = Signal(init=1)
rd_r = Signal(init=1)
wr_r = Signal(init=1)
m.d.comb += [
self.bus_addr .eq(a_o),
self.bus_data_o .eq(d_o),
self.bus_data_oe.eq(d_oe),
self.cs_n .eq(cs_r),
self.rd_n .eq(rd_r),
self.wr_n .eq(wr_r),
]
m.d.sync += bus_done.eq(0)
with m.FSM(domain="sync", name="bus_fsm"):
with m.State("IDLE"):
m.d.sync += [cs_r.eq(1), rd_r.eq(1), wr_r.eq(1), d_oe.eq(0)]
with m.If(bus_go):
m.d.sync += [a_o.eq(bus_a), rw_r.eq(bus_rw),
cs_r.eq(0), bus_ctr.eq(0)]
with m.If(bus_rw):
m.d.sync += [d_o.eq(bus_wdata), d_oe.eq(1), wr_r.eq(0)]
with m.Else():
m.d.sync += rd_r.eq(0)
m.next = "STROBE"
with m.State("STROBE"):
m.d.sync += bus_ctr.eq(bus_ctr + 1)
with m.If(bus_ctr == STROBE - 1):
with m.If(~rw_r):
m.d.sync += bus_rdata.eq(self.bus_data_i) # sample read
m.d.sync += [rd_r.eq(1), wr_r.eq(1)]
m.next = "FINISH"
with m.State("FINISH"):
m.d.sync += [cs_r.eq(1), d_oe.eq(0), bus_done.eq(1)]
m.next = "IDLE"
# ── Transaction engine: address-set + payload over the bus engine ────
WBUF = 8
xfer_start = Signal()
xfer_direct = Signal() # 1 = single A=00 access (MR), addr ignored
xfer_addr = Signal(16)
xfer_rw = Signal() # payload direction: 1=write, 0=read
xfer_len = Signal(range(WBUF + 1))
xfer_stream = Signal() # stream-write payload from s_*
xfer_sread = Signal() # stream-read payload to r_*
xfer_rcount = Signal(16)
xfer_done = Signal()
wbuf = Array([Signal(8, name=f"wbuf{i}") for i in range(WBUF)])
rbuf = Array([Signal(8, name=f"rbuf{i}") for i in range(WBUF)])
s_count = Signal(16) # bytes streamed-written (advances pointers)
xfer_idx = Signal(range(WBUF + 1))
s_last_r = Signal()
r_idx = Signal(16)
# Streaming payload interfaces.
s_data, s_valid, s_last, s_consume = Signal(8), Signal(), Signal(), Signal()
r_data, r_valid, r_first, r_last, r_ready = (
Signal(8), Signal(), Signal(), Signal(), Signal())
# TX stream-write source mux: during a UDP-test send the payload comes
# from the external `udp_pl_*` stream (the shell); otherwise the normal
# MACRAW TX interface feeds it. `udp_streaming` is raised only while the
# UDP payload is being written to the socket-1 TX buffer.
udp_streaming = Signal()
if self._enable_udp:
with m.If(udp_streaming):
m.d.comb += [s_data.eq(self.udp_pl_data),
s_valid.eq(self.udp_pl_valid),
s_last.eq(self.udp_pl_last),
self.udp_pl_ready.eq(s_consume)]
with m.Else():
m.d.comb += [s_data.eq(self.tx_data), s_valid.eq(self.tx_valid),
s_last.eq(self.tx_eof), self.tx_ready.eq(s_consume)]
else:
m.d.comb += [s_data.eq(self.tx_data), s_valid.eq(self.tx_valid),
s_last.eq(self.tx_eof), self.tx_ready.eq(s_consume)]
# RX stream-read sink mux: during a socket-N UDP receive the payload
# streams out on `udp_rx_*` (to the shell); otherwise it goes to the
# MACRAW rx interface (to the frame assembler). `udp_rx_streaming` is
# raised only while the UDP payload is being read out.
udp_rx_streaming = Signal()
if self._enable_udp:
with m.If(udp_rx_streaming):
m.d.comb += [self.udp_rx_data.eq(r_data),
self.udp_rx_valid.eq(r_valid),
self.udp_rx_eof.eq(r_last),
r_ready.eq(self.udp_rx_ready)]
with m.Else():
m.d.comb += [self.rx_data.eq(r_data), self.rx_valid.eq(r_valid),
self.rx_sof.eq(r_first), self.rx_eof.eq(r_last),
r_ready.eq(self.rx_ready)]
else:
m.d.comb += [self.rx_data.eq(r_data), self.rx_valid.eq(r_valid),
self.rx_sof.eq(r_first), self.rx_eof.eq(r_last),
r_ready.eq(self.rx_ready)]
# Socket-buffer ring wraparound. Unlike the W5500, the W5100's IDM
# address does NOT auto-wrap at the socket-buffer boundary — it just
# increments linearly into the next region. So when a streamed access
# reaches `xfer_wend`, the engine re-sets IDM_AR back to `xfer_wbase`.
xfer_wrap = Signal()
xfer_wbase = Signal(16)
xfer_wend = Signal(16)
cur_addr = Signal(16)
m.d.comb += [bus_go.eq(0), bus_rw.eq(0), bus_a.eq(0), bus_wdata.eq(0)]
m.d.comb += [s_consume.eq(0), r_valid.eq(0), r_data.eq(0),
r_first.eq(0), r_last.eq(0)]
m.d.sync += xfer_done.eq(0)
def bus_write(a, data):
m.d.comb += [bus_go.eq(1), bus_rw.eq(1), bus_a.eq(a), bus_wdata.eq(data)]
def bus_read(a):
m.d.comb += [bus_go.eq(1), bus_rw.eq(0), bus_a.eq(a)]
with m.FSM(domain="sync", name="xfer_fsm"):
with m.State("IDLE"):
with m.If(xfer_start):
m.d.sync += [xfer_idx.eq(0), s_count.eq(0), r_idx.eq(0),
cur_addr.eq(xfer_addr)]
with m.If(xfer_direct):
m.next = "DIRECT"
with m.Else():
m.next = "AR_HI"
# Direct MR write (A=00)
with m.State("DIRECT"):
bus_write(_A_MR, wbuf[0])
m.next = "DIRECT_W"
with m.State("DIRECT_W"):
with m.If(bus_done):
m.next = "FINISH"
# Set indirect address IDM_AR (high then low)
with m.State("AR_HI"):
bus_write(_A_AR0, xfer_addr[8:16])
m.next = "AR_HI_W"
with m.State("AR_HI_W"):
with m.If(bus_done):
m.next = "AR_LO"
with m.State("AR_LO"):
bus_write(_A_AR1, xfer_addr[0:8])
m.next = "AR_LO_W"
with m.State("AR_LO_W"):
with m.If(bus_done):
with m.If(xfer_stream):
m.next = "SW_LOAD"
with m.Elif(xfer_sread):
m.next = "SR_LOAD"
with m.Elif(xfer_rw):
m.next = "WB_ISSUE"
with m.Else():
m.next = "RB_ISSUE"
# Fixed-length write from wbuf (IDM_DR burst, auto-increment)
with m.State("WB_ISSUE"):
bus_write(_A_DR, wbuf[xfer_idx])
m.next = "WB_WAIT"
with m.State("WB_WAIT"):
with m.If(bus_done):
m.d.sync += xfer_idx.eq(xfer_idx + 1)
with m.If(xfer_idx + 1 == xfer_len):
m.next = "FINISH"
with m.Else():
m.next = "WB_ISSUE"
# Fixed-length read into rbuf (with ring wrap, for the length header)
with m.State("RB_ISSUE"):
with m.If(xfer_wrap & (cur_addr == xfer_wend)):
m.next = "RB_WRAP_HI"
with m.Else():
bus_read(_A_DR)
m.next = "RB_WAIT"
with m.State("RB_WAIT"):
with m.If(bus_done):
m.d.sync += rbuf[xfer_idx].eq(bus_rdata)
m.d.sync += [xfer_idx.eq(xfer_idx + 1), cur_addr.eq(cur_addr + 1)]
with m.If(xfer_idx + 1 == xfer_len):
m.next = "FINISH"
with m.Else():
m.next = "RB_ISSUE"
with m.State("RB_WRAP_HI"):
bus_write(_A_AR0, xfer_wbase[8:16])
m.next = "RB_WRAP_HI_W"
with m.State("RB_WRAP_HI_W"):
with m.If(bus_done):
m.next = "RB_WRAP_LO"
with m.State("RB_WRAP_LO"):
bus_write(_A_AR1, xfer_wbase[0:8])
m.next = "RB_WRAP_LO_W"
with m.State("RB_WRAP_LO_W"):
with m.If(bus_done):
m.d.sync += cur_addr.eq(xfer_wbase)
m.next = "RB_ISSUE"
# Stream-write payload from s_* until s_last (with ring wrap)
with m.State("SW_LOAD"):
with m.If(xfer_wrap & (cur_addr == xfer_wend)):
m.next = "SW_WRAP_HI"
with m.Elif(s_valid):
bus_write(_A_DR, s_data)
m.d.sync += s_last_r.eq(s_last)
m.next = "SW_WAIT"
with m.State("SW_WAIT"):
with m.If(bus_done):
m.d.comb += s_consume.eq(1)
m.d.sync += [s_count.eq(s_count + 1), cur_addr.eq(cur_addr + 1)]
with m.If(s_last_r):
m.next = "FINISH"
with m.Else():
m.next = "SW_LOAD"
with m.State("SW_WRAP_HI"):
bus_write(_A_AR0, xfer_wbase[8:16])
m.next = "SW_WRAP_HI_W"
with m.State("SW_WRAP_HI_W"):
with m.If(bus_done):
m.next = "SW_WRAP_LO"
with m.State("SW_WRAP_LO"):
bus_write(_A_AR1, xfer_wbase[0:8])
m.next = "SW_WRAP_LO_W"
with m.State("SW_WRAP_LO_W"):
with m.If(bus_done):
m.d.sync += cur_addr.eq(xfer_wbase)
m.next = "SW_LOAD"
# Stream-read payload to r_* for rcount bytes (with ring wrap)
with m.State("SR_LOAD"):
with m.If(r_idx == xfer_rcount):
m.next = "FINISH"
with m.Elif(xfer_wrap & (cur_addr == xfer_wend)):
m.next = "SR_WRAP_HI"
with m.Else():
bus_read(_A_DR)
m.next = "SR_WAIT"
with m.State("SR_WAIT"):
with m.If(bus_done):
m.next = "SR_PUSH"
with m.State("SR_PUSH"):
m.d.comb += [r_data.eq(bus_rdata), r_valid.eq(1),
r_first.eq(r_idx == 0),
r_last.eq(r_idx + 1 == xfer_rcount)]
with m.If(r_ready):
m.d.sync += [r_idx.eq(r_idx + 1), cur_addr.eq(cur_addr + 1)]
m.next = "SR_LOAD"
with m.State("SR_WRAP_HI"):
bus_write(_A_AR0, xfer_wbase[8:16])
m.next = "SR_WRAP_HI_W"
with m.State("SR_WRAP_HI_W"):
with m.If(bus_done):
m.next = "SR_WRAP_LO"
with m.State("SR_WRAP_LO"):
bus_write(_A_AR1, xfer_wbase[0:8])
m.next = "SR_WRAP_LO_W"
with m.State("SR_WRAP_LO_W"):
with m.If(bus_done):
m.d.sync += cur_addr.eq(xfer_wbase)
m.next = "SR_LOAD"
with m.State("FINISH"):
m.d.sync += xfer_done.eq(1)
m.next = "IDLE"
# ── Control regs ─────────────────────────────────────────────────────
mac_shadow = Array([Signal(8, name=f"mac{i}") for i in range(6)])
wait_ctr = Signal(range(self._reset_cycles + 2))
tx_wr = Signal(16)
rx_rsr = Signal(16)
rx_rd = Signal(16)
pkt_len = Signal(16)
s1_tx_wr = Signal(16) # UDP-socket TX write pointer (send)
sn_rx_rsr = Signal(16) # UDP-socket RX received size
sn_rx_rd = Signal(16) # UDP-socket RX read pointer
sn_pkt_len = Signal(16) # UDP datagram payload length (from header)
def write_reg(name, addr, payload, nxt, direct=False):
"""Emit a 2-state block that writes `payload` (a list) to `addr`."""
with m.State(name):
m.d.sync += [xfer_addr.eq(addr), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0),
xfer_direct.eq(1 if direct else 0),
xfer_len.eq(len(payload))]
for i, b in enumerate(payload):
m.d.sync += wbuf[i].eq(b)
m.d.sync += xfer_start.eq(1)
m.next = name + "_W"
with m.State(name + "_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = nxt
# ── Main control FSM (Phase 1: init only) ────────────────────────────
if self._enable_udp:
m.d.sync += self.udp_rx_none.eq(0) # pulse: default low
with m.FSM(domain="sync", name="main_fsm"):
with m.State("IDLE"):
m.d.sync += self.init_done.eq(0)
with m.If(self.init_req):
for i in range(6):
m.d.sync += mac_shadow[i].eq(self.par[i*8:(i+1)*8])
m.d.sync += [self.w5100_rst_n.eq(0), wait_ctr.eq(0)]
m.next = "HW_RST"
with m.Elif(~self.w5100_int_n):
m.next = "RX_CHECK"
with m.Elif(self.tx_valid & self.tx_sof):
m.next = "TX_START"
if self._enable_udp:
with m.Elif(self.udp_send_req):
m.next = "UDP_DIPR"
with m.Elif(self.udp_rx_req):
m.next = "UDP_RX_RSR"
# Pulse the W5100's RSTn pin, then let it boot, BEFORE touching the
# bus. This used to be missing entirely: w5100_rst_n was declared
# and never driven, so the only reset the chip ever got was the MR
# software reset -- which a wedged chip does not answer, because it
# arrives over the very bus that is stuck. The result was a W5100
# that could only be recovered by a 12 V power cycle; even
# reconfiguring the FPGA did not help, since RSTn just floats to its
# pull-up (R36) while the FPGA is unconfigured. Driving it here
# makes an NCRA reset able to recover the chip on its own.
# RSTn low: datasheet minimum is ~2 us; _RST_LOW is far longer and
# costs nothing on a path that runs once per init.
with m.State("HW_RST"):
with m.If(wait_ctr == self._rst_low):
m.d.sync += [self.w5100_rst_n.eq(1), wait_ctr.eq(0)]
m.next = "HW_RST_WAIT"
with m.Else():
m.d.sync += wait_ctr.eq(wait_ctr + 1)
# RSTn released: the W5100S needs 60.3 ms before it answers at all.
with m.State("HW_RST_WAIT"):
with m.If(wait_ctr == self._reset_cycles):
m.d.sync += wait_ctr.eq(0)
m.next = "MR_RST"
with m.Else():
m.d.sync += wait_ctr.eq(wait_ctr + 1)
# MR = 0x80 software reset (direct A=00), then settle.
#
# reset_cycles MUST cover the W5100S's post-reset dead time (the
# datasheet's 60.3 ms; our own bus self-test needed ~85 ms before
# the chip answered). The default 2_000_000 = 83 ms at the 24 MHz
# sync clock. It was 24_000 = 1 ms, which is MARGINAL: the chip
# sometimes answered and sometimes did not, so every init write
# after this point (NETLCKR, SHAR, SIPR, socket regs) could land on
# a still-resetting chip and be silently dropped. The symptom was
# a W5100 that wedged after an NCRA reset -- TX never completing
# and the UART shell's own `udp unicast` hanging -- recoverable
# only by a 12 V power cycle. Do NOT shrink this for the synthesis
# build; the testbenches pass a small value explicitly.
write_reg("MR_RST", _MR, [_MR_RST], "MR_WAIT", direct=True)
with m.State("MR_WAIT"):
with m.If(wait_ctr == self._reset_cycles):
m.d.sync += wait_ctr.eq(0)
m.next = "MR_MODE"
with m.Else():
m.d.sync += wait_ctr.eq(wait_ctr + 1)
# MR = indirect + auto-increment (direct A=00).
write_reg("MR_MODE", _MR, [_MR_IND | _MR_AI], "NETLCK", direct=True)
# W5100S-only: GWR/SUBR/SHAR/SIPR are LOCKED until NETLCKR (0x0071)
# is written 0x3A (default 0x00 = locked). Without this, every
# SHAR/SIPR/GAR/SUBR write below is silently DROPPED -> no source
# MAC/IP -> the UDP/MACRAW send builds nothing and the chip transmits
# nothing on the wire despite reporting SEND_OK (bench-confirmed
# 2026-09-20; the original W5100 had no such lock). Indirect mode is
# already on from MR_MODE, so this reaches 0x0071.
write_reg("NETLCK", _NETLCKR, [_NETLCK_UNLOCK], "SHAR")
# SHAR = source MAC (6-byte auto-increment burst).
with m.State("SHAR"):
m.d.sync += [xfer_addr.eq(_SHAR0), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_direct.eq(0), xfer_len.eq(6)]
for i in range(6):
m.d.sync += wbuf[i].eq(mac_shadow[i])
m.d.sync += xfer_start.eq(1)
m.next = "SHAR_W"
with m.State("SHAR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = "MEMSZ"
# RMSR/TMSR = 0x55 (2 KB per socket — default; socket 0 used).
write_reg("MEMSZ", _RMSR, [0x55, 0x55], "S0_MODE") # RMSR then TMSR
# Socket 0: MACRAW mode, OPEN, enable interrupt.
#
# ⚠️ MF (MAC filter) IS SET unless the GC asked for promiscuous.
# Without it MACRAW delivers EVERY frame on the segment, which is
# not what the real BBA does: libogc2 configures
# NCRB = AB|CA|2_PACKETS_PER_INT (0x52) -- PR=0, PM=0, AB=1 --
# i.e. own-MAC + broadcast only.
#
# Unfiltered, this HUNG SWISS. libogc2's bba_interrupt() loops
# while (status) { clear IR; start_rx; status = IR & IMR; }
# which only exits when IR & IMR reaches zero. On a live segment
# RI re-asserted faster than the loop could clear it, so it never
# exited and Swiss sat on "Initialising Network" forever. With
# the cable unplugged it got through, which is what identified it.
#
# Filtering against SHAR is safe here because libogc2 READS the
# MAC out of PAR0-5 rather than writing it, so SHAR (written from
# reg.par during this same init) cannot go stale underneath it.
# A driver that WRITES PAR0-5 after init would need SHAR
# refreshed -- see docs/bba_compat_todo.md item 1.
write_reg("S0_MODE", _S0_MR,
[Mux(self.promisc, _S0_MR_MACRAW,
_S0_MR_MACRAW | _S0_MR_MF)], "S0_OPEN")
write_reg("S0_OPEN", _S0_CR, [_CR_OPEN], "S0_IMR")
# After S0 IMR, either finish (plain BBA) or configure the IP stack
# and open socket 1 in UDP mode (bring-up test build).
write_reg("S0_IMR", _IMR, [0x01],
"UDP_CFG_GAR" if self._enable_udp else "INIT_DONE")
if self._enable_udp:
# IP stack: gateway, subnet, source IP (SHAR already set above).
write_reg("UDP_CFG_GAR", _GAR0, self._gateway, "UDP_CFG_SUBR")
write_reg("UDP_CFG_SUBR", _SUBR0, self._subnet, "UDP_CFG_SIPR")
write_reg("UDP_CFG_SIPR", _SIPR0, self._src_ip, "UDP_CFG_MR")
# Socket 1: UDP mode, source port, OPEN, fixed dest port.
write_reg("UDP_CFG_MR", sn["MR"], [_S1_MR_UDP], "UDP_CFG_PORT")
write_reg("UDP_CFG_PORT", sn["PORT"], self._src_port, "UDP_CFG_OPEN")
write_reg("UDP_CFG_OPEN", sn["CR"], [_CR_OPEN], "UDP_CFG_DPORT")
write_reg("UDP_CFG_DPORT", sn["DPORT"], self._dst_port, "INIT_DONE")
with m.State("INIT_DONE"):
m.d.sync += self.init_done.eq(1)
m.next = "IDLE"
# ── TX MACRAW ────────────────────────────────────────────────────
# read S0_TX_WR → stream frame into the TX buffer at that offset
# (ring-wrapping at the 2 KB boundary) → advance S0_TX_WR → SEND.
with m.State("TX_START"): # read S0_TX_WR (2 bytes)
m.d.sync += [xfer_addr.eq(_S0_TX_WR), xfer_rw.eq(0),
xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0),
xfer_direct.eq(0), xfer_len.eq(2)]
m.d.sync += xfer_start.eq(1)
m.next = "TX_RDPTR_W"
with m.State("TX_RDPTR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += tx_wr.eq(Cat(rbuf[1], rbuf[0])) # big-endian
m.next = "TX_DATA"
with m.State("TX_DATA"): # stream frame → TX buffer
m.d.sync += [xfer_addr.eq(_TX_BASE + (tx_wr & _S0_TX_MASK)),
xfer_rw.eq(1), xfer_stream.eq(1), xfer_sread.eq(0),
xfer_direct.eq(0), xfer_wrap.eq(1),
xfer_wbase.eq(_TX_BASE),
xfer_wend.eq(_TX_BASE + _S0_TX_MASK + 1)]
m.d.sync += xfer_start.eq(1)
m.next = "TX_DATA_W"
with m.State("TX_DATA_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += [xfer_stream.eq(0), xfer_wrap.eq(0),
tx_wr.eq(tx_wr + s_count)] # advanced pointer
m.next = "TX_UPDPTR"
with m.State("TX_UPDPTR"): # write back S0_TX_WR
m.d.sync += [xfer_addr.eq(_S0_TX_WR), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0),
xfer_direct.eq(0), xfer_len.eq(2)]
m.d.sync += [wbuf[0].eq(tx_wr[8:16]), wbuf[1].eq(tx_wr[0:8])]
m.d.sync += xfer_start.eq(1)
m.next = "TX_UPDPTR_W"
with m.State("TX_UPDPTR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = "TX_SEND"
# S0_CR = SEND
write_reg("TX_SEND", _S0_CR, [_CR_SEND], "IDLE")
# ── RX MACRAW ────────────────────────────────────────────────────
# On W5100 INT: read RX_RSR; if non-zero read RX_RD, read the 2-byte
# MACRAW length, stream (length−2) frame bytes out (ring-wrapping),
# advance RX_RD by the length, issue RECV, clear the RECV interrupt.
with m.State("RX_CHECK"): # read S0_RX_RSR (2 bytes)
m.d.sync += [xfer_addr.eq(_S0_RX_RSR), xfer_rw.eq(0),
xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0),
xfer_direct.eq(0), xfer_len.eq(2)]
m.d.sync += xfer_start.eq(1)
m.next = "RX_RSR_W"
with m.State("RX_RSR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += rx_rsr.eq(Cat(rbuf[1], rbuf[0]))
m.next = "RX_RSR_CHK"
with m.State("RX_RSR_CHK"):
with m.If(rx_rsr == 0):
# INT_N is asserted but there is no data. Something other
# than RECV set Sn_IR -- in practice SEND_OK (bit 4), which
# fires after every successful transmit. Returning to IDLE
# without clearing it leaves INT_N low forever, and since
# IDLE tests ~w5100_int_n BEFORE tx_valid, the FSM spins
# IDLE->RX_CHECK->here and TX is starved permanently. That
# showed up on hardware as: first frame transmits, every
# later one never sets TI, while RX keeps working.
m.next = "RX_IR_CLR"
with m.Else():
m.next = "RX_RDPTR"
# Clear every Sn_IR bit EXCEPT RECV (0x04). Keeping RECV means a
# frame that lands between the RSR read and this write still raises
# INT_N and gets served on the next pass instead of being lost.
write_reg("RX_IR_CLR", _S0_IR, [0x1B], "IDLE")
with m.State("RX_RDPTR"): # read S0_RX_RD (2 bytes)
m.d.sync += [xfer_addr.eq(_S0_RX_RD), xfer_rw.eq(0),
xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0),
xfer_direct.eq(0), xfer_len.eq(2)]
m.d.sync += xfer_start.eq(1)
m.next = "RX_RDPTR_W"
with m.State("RX_RDPTR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += rx_rd.eq(Cat(rbuf[1], rbuf[0]))
m.next = "RX_LEN"
with m.State("RX_LEN"): # read 2-byte MACRAW length (wrap)
m.d.sync += [xfer_addr.eq(_RX_BASE + (rx_rd & _S0_RX_MASK)),
xfer_rw.eq(0), xfer_stream.eq(0), xfer_sread.eq(0),
xfer_direct.eq(0), xfer_len.eq(2), xfer_wrap.eq(1),
xfer_wbase.eq(_RX_BASE),
xfer_wend.eq(_RX_BASE + _S0_RX_MASK + 1)]
m.d.sync += xfer_start.eq(1)
m.next = "RX_LEN_W"
with m.State("RX_LEN_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += pkt_len.eq(Cat(rbuf[1], rbuf[0]))
m.next = "RX_FRAME"
with m.State("RX_FRAME"): # stream (pkt_len−2) frame bytes
m.d.sync += [xfer_addr.eq(_RX_BASE + ((rx_rd + 2) & _S0_RX_MASK)),
xfer_rw.eq(0), xfer_stream.eq(0), xfer_sread.eq(1),
xfer_direct.eq(0), xfer_rcount.eq(pkt_len - 2),
xfer_wrap.eq(1), xfer_wbase.eq(_RX_BASE),
xfer_wend.eq(_RX_BASE + _S0_RX_MASK + 1)]
m.d.sync += xfer_start.eq(1)
m.next = "RX_FRAME_W"
with m.State("RX_FRAME_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += [xfer_sread.eq(0), xfer_wrap.eq(0)]
m.next = "RX_UPDRD"
with m.State("RX_UPDRD"): # S0_RX_RD += pkt_len, write back
m.d.sync += [xfer_addr.eq(_S0_RX_RD), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0), xfer_wrap.eq(0),
xfer_direct.eq(0), xfer_len.eq(2)]
m.d.sync += [wbuf[0].eq((rx_rd + pkt_len)[8:16]),
wbuf[1].eq((rx_rd + pkt_len)[0:8])]
m.d.sync += xfer_start.eq(1)
m.next = "RX_UPDRD_W"
with m.State("RX_UPDRD_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = "RX_RECV"
# S0_CR = RECV, then clear the RECV interrupt bit (S0_IR[2]).
write_reg("RX_RECV", _S0_CR, [_CR_RECV], "RX_CLR_IR")
write_reg("RX_CLR_IR", _S0_IR, [0x04], "IDLE")
# ── UDP test send (configurable socket) ──────────────────────────
# Runtime dest IP → Sn_DIPR, read Sn_TX_WR, stream the shell-
# supplied payload into the socket TX buffer (ring-wrapping),
# advance Sn_TX_WR, SEND. DPORT + socket open were done at init.
if self._enable_udp:
with m.State("UDP_DIPR"): # write Sn_DIPR (runtime, 4 B)
m.d.sync += self.udp_test_busy.eq(1)
m.d.sync += [xfer_addr.eq(sn["DIPR"]), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_wrap.eq(0), xfer_direct.eq(0),
xfer_len.eq(4)]
m.d.sync += [wbuf[0].eq(self.udp_dst_ip[24:32]),
wbuf[1].eq(self.udp_dst_ip[16:24]),
wbuf[2].eq(self.udp_dst_ip[8:16]),
wbuf[3].eq(self.udp_dst_ip[0:8])]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_DIPR_W"
with m.State("UDP_DIPR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = "UDP_TXWR"
with m.State("UDP_TXWR"): # read Sn_TX_WR (2 B)
m.d.sync += [xfer_addr.eq(sn["TX_WR"]), xfer_rw.eq(0),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_wrap.eq(0), xfer_direct.eq(0),
xfer_len.eq(2)]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_TXWR_W"
with m.State("UDP_TXWR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += s1_tx_wr.eq(Cat(rbuf[1], rbuf[0]))
m.next = "UDP_DATA"
with m.State("UDP_DATA"): # stream payload → Sn TX buffer
m.d.sync += [xfer_addr.eq(sn["TX_BASE"] + (s1_tx_wr & _SN_MASK)),
xfer_rw.eq(1), xfer_stream.eq(1),
xfer_sread.eq(0), xfer_direct.eq(0),
xfer_wrap.eq(1), xfer_wbase.eq(sn["TX_BASE"]),
xfer_wend.eq(sn["TX_BASE"] + _SN_MASK + 1),
udp_streaming.eq(1)]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_DATA_W"
with m.State("UDP_DATA_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += [xfer_stream.eq(0), xfer_wrap.eq(0),
udp_streaming.eq(0),
s1_tx_wr.eq(s1_tx_wr + s_count)]
m.next = "UDP_UPDPTR"
with m.State("UDP_UPDPTR"): # write back Sn_TX_WR (2 B)
m.d.sync += [xfer_addr.eq(sn["TX_WR"]), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_wrap.eq(0), xfer_direct.eq(0),
xfer_len.eq(2)]
m.d.sync += [wbuf[0].eq(s1_tx_wr[8:16]),
wbuf[1].eq(s1_tx_wr[0:8])]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_UPDPTR_W"
with m.State("UDP_UPDPTR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = "UDP_SEND"
write_reg("UDP_SEND", sn["CR"], [_CR_SEND], "UDP_DONE")
with m.State("UDP_DONE"):
m.d.sync += self.udp_test_busy.eq(0)
m.next = "IDLE"
# ── UDP test receive (configurable socket) ───────────────────────
# Poll-driven (the shell pulses udp_rx_req). Read Sn_RX_RSR; if 0,
# pulse udp_rx_none. Otherwise read Sn_RX_RD, read the 8-byte WIZnet
# UDP header ([srcIP 4][srcPort 2][len 2]), stream `len` payload
# bytes out on udp_rx_* (ring-wrapping), advance Sn_RX_RD by 8+len,
# RECV, clear the socket IR. MACRAW socket-0 RX is untouched.
if self._enable_udp:
with m.State("UDP_RX_RSR"): # read Sn_RX_RSR (2 B)
m.d.sync += self.udp_rx_busy.eq(1)
m.d.sync += [xfer_addr.eq(sn["RX_RSR"]), xfer_rw.eq(0),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_wrap.eq(0), xfer_direct.eq(0),
xfer_len.eq(2)]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_RX_RSR_W"
with m.State("UDP_RX_RSR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += sn_rx_rsr.eq(Cat(rbuf[1], rbuf[0]))
m.next = "UDP_RX_CHK"
with m.State("UDP_RX_CHK"):
with m.If(sn_rx_rsr == 0):
m.d.sync += [self.udp_rx_none.eq(1),
self.udp_rx_busy.eq(0)]
m.next = "IDLE"
with m.Else():
m.next = "UDP_RX_RD"
with m.State("UDP_RX_RD"): # read Sn_RX_RD (2 B)
m.d.sync += [xfer_addr.eq(sn["RX_RD"]), xfer_rw.eq(0),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_wrap.eq(0), xfer_direct.eq(0),
xfer_len.eq(2)]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_RX_RD_W"
with m.State("UDP_RX_RD_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += sn_rx_rd.eq(Cat(rbuf[1], rbuf[0]))
m.next = "UDP_RX_HDR"
with m.State("UDP_RX_HDR"): # read 8-byte UDP header (wrap)
m.d.sync += [xfer_addr.eq(sn["RX_BASE"] + (sn_rx_rd & _SN_MASK)),
xfer_rw.eq(0), xfer_stream.eq(0), xfer_sread.eq(0),
xfer_direct.eq(0), xfer_len.eq(8), xfer_wrap.eq(1),
xfer_wbase.eq(sn["RX_BASE"]),
xfer_wend.eq(sn["RX_BASE"] + _SN_MASK + 1)]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_RX_HDR_W"
with m.State("UDP_RX_HDR_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
# header[6:8] = payload length (big-endian)
m.d.sync += sn_pkt_len.eq(Cat(rbuf[7], rbuf[6]))
m.next = "UDP_RX_FRAME"
with m.State("UDP_RX_FRAME"): # stream `len` payload bytes out
m.d.sync += [xfer_addr.eq(sn["RX_BASE"] + ((sn_rx_rd + 8) & _SN_MASK)),
xfer_rw.eq(0), xfer_stream.eq(0), xfer_sread.eq(1),
xfer_direct.eq(0), xfer_rcount.eq(sn_pkt_len),
xfer_wrap.eq(1), xfer_wbase.eq(sn["RX_BASE"]),
xfer_wend.eq(sn["RX_BASE"] + _SN_MASK + 1),
udp_rx_streaming.eq(1)]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_RX_FRAME_W"
with m.State("UDP_RX_FRAME_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.d.sync += [xfer_sread.eq(0), xfer_wrap.eq(0),
udp_rx_streaming.eq(0)]
m.next = "UDP_RX_UPDRD"
with m.State("UDP_RX_UPDRD"): # Sn_RX_RD += 8 + len, write back
m.d.sync += [xfer_addr.eq(sn["RX_RD"]), xfer_rw.eq(1),
xfer_stream.eq(0), xfer_sread.eq(0),
xfer_wrap.eq(0), xfer_direct.eq(0), xfer_len.eq(2)]
m.d.sync += [wbuf[0].eq((sn_rx_rd + 8 + sn_pkt_len)[8:16]),
wbuf[1].eq((sn_rx_rd + 8 + sn_pkt_len)[0:8])]
m.d.sync += xfer_start.eq(1)
m.next = "UDP_RX_UPDRD_W"
with m.State("UDP_RX_UPDRD_W"):
m.d.sync += xfer_start.eq(0)
with m.If(xfer_done):
m.next = "UDP_RX_RECV"
write_reg("UDP_RX_RECV", sn["CR"], [_CR_RECV], "UDP_RX_CLR_IR")
write_reg("UDP_RX_CLR_IR", sn["IR"], [0x04], "UDP_RX_DONE")
with m.State("UDP_RX_DONE"):
m.d.sync += self.udp_rx_busy.eq(0)
m.next = "IDLE"
return m
# ── Testbench ─────────────────────────────────────────────────────────────
if __name__ == "__main__":
import sys
from amaranth.sim import Simulator, Period
dut = W5100ParallelMaster(strobe_cycles=3, reset_cycles=10)
errors = []
MAC = [0x11, 0x22, 0x33, 0x44, 0x55, 0x66]
PAR = sum(b << (8 * i) for i, b in enumerate(MAC))
# Expected indirect-address writes captured by the model (addr, value).
# MR is written directly (A=00) → captured as ('MR', value).
EXPECTED = [
("MR", _MR_RST),
("MR", _MR_IND | _MR_AI),
(_NETLCKR, _NETLCK_UNLOCK), # W5100S: unlock GWR/SUBR/SHAR/SIPR
(_SHAR0 + 0, MAC[0]), (_SHAR0 + 1, MAC[1]), (_SHAR0 + 2, MAC[2]),
(_SHAR0 + 3, MAC[3]), (_SHAR0 + 4, MAC[4]), (_SHAR0 + 5, MAC[5]),
(_RMSR + 0, 0x55), (_RMSR + 1, 0x55),
(_S0_MR, _S0_MR_MACRAW | _S0_MR_MF), # MF set: own-MAC + broadcast
(_S0_CR, _CR_OPEN),
(_IMR, 0x01),
]
writes = [] # captured (addr-or-'MR', value) — IDM_DR + MR writes
model_mem = {} # W5100 memory image (registers + TX/RX buffers)
async def w5100_model(ctx):
"""W5100 indirect-bus slave model: tracks MR/IDM_AR, records IDM_DR and
MR writes, and drives bus_data_i for reads. Mode-0 timing: a write is
latched on /WR rising while /CS low; reads driven while /RD low."""
idm_ar = 0
mr = 0
prev_cs = prev_rd = prev_wr = 1
async for vals in ctx.tick("sync").sample(
dut.cs_n, dut.rd_n, dut.wr_n,
dut.bus_addr, dut.bus_data_o, dut.bus_data_oe):
cs, rd, wr, a, do, doe = vals[-6:]
ai = (mr >> 1) & 1 # MR.AI
# Drive read data while /RD asserted (combinational, before sample).
if cs == 0 and rd == 0:
if a == _A_MR:
val = mr
elif a == _A_AR0:
val = (idm_ar >> 8) & 0xFF
elif a == _A_AR1:
val = idm_ar & 0xFF
else:
val = model_mem.get(idm_ar, 0)
ctx.set(dut.bus_data_i, val)
# Latch write on /WR rising edge.
if cs == 0 and prev_wr == 0 and wr == 1:
if a == _A_MR:
mr = do
writes.append(("MR", do))
elif a == _A_AR0:
idm_ar = (idm_ar & 0x00FF) | (do << 8)
elif a == _A_AR1:
idm_ar = (idm_ar & 0xFF00) | do
else: # IDM_DR
model_mem[idm_ar] = do
writes.append((idm_ar, do))
# RECV command consumes the RX data: clear RSR (mirrors HW).
if idm_ar == _S0_CR and do == _CR_RECV:
model_mem[_S0_RX_RSR] = 0
model_mem[_S0_RX_RSR + 1] = 0
if ai:
idm_ar = (idm_ar + 1) & 0xFFFF
# Auto-increment after a data read (/RD rising, A=DR).
if cs == 0 and prev_rd == 0 and rd == 1 and a == _A_DR and ai:
idm_ar = (idm_ar + 1) & 0xFFFF
prev_cs, prev_rd, prev_wr = cs, rd, wr
async def testbench(ctx):
ctx.set(dut.par, PAR)
await ctx.tick("sync").repeat(2)
# T1: trigger init, wait for init_done.
ctx.set(dut.init_req, 1)
await ctx.tick("sync").repeat(1)
ctx.set(dut.init_req, 0)
done = False
for _ in range(4000):
await ctx.tick("sync").repeat(1)
if ctx.get(dut.init_done):
done = True
break
if not done:
errors.append("init_done never asserted")
print(f"T1 init captured {len(writes)} writes")
if writes != EXPECTED:
errors.append("init write sequence mismatch")
for i in range(max(len(writes), len(EXPECTED))):
g = writes[i] if i < len(writes) else None
e = EXPECTED[i] if i < len(EXPECTED) else None
mark = "" if g == e else " <-- MISMATCH"
gs = f"({g[0]:#06x},{g[1]:#04x})" if g and isinstance(g[0], int) else str(g)
es = f"({e[0]:#06x},{e[1]:#04x})" if e and isinstance(e[0], int) else str(e)
print(f" [{i:2}] got {gs:20} exp {es:20}{mark}")
else:
print("T1 init sequence matches expected (MR, SHAR, mem sizes, "
"S0 MACRAW/OPEN, IMR)")
# T1b: NCRB.PR -> promiscuous, i.e. MF CLEARED.
#
# The default (MF set) filters to own-MAC + broadcast, matching the
# real BBA. Running unfiltered hung Swiss: libogc2's bba_interrupt()
# loops until IR & IMR reaches zero, and on a live segment RI
# re-asserted faster than the loop could clear it. But the GC can ask
# for promiscuous via NCRB bit 0, and then it must actually get it.
s0_default = [v for a, v in writes if a == _S0_MR]
writes.clear()
ctx.set(dut.promisc, 1)
ctx.set(dut.init_req, 1)
await ctx.tick("sync").repeat(1)
ctx.set(dut.init_req, 0)
for _ in range(4000):
await ctx.tick("sync").repeat(1)
if ctx.get(dut.init_done):
break
s0_promisc = [v for a, v in writes if a == _S0_MR]
print(f"T1b S0_MR: default={[hex(v) for v in s0_default]} "
f"promisc={[hex(v) for v in s0_promisc]}")
if s0_default != [_S0_MR_MACRAW | _S0_MR_MF]:
errors.append(f"T1b default S0_MR {s0_default}, want MACRAW|MF "
f"- unfiltered MACRAW hangs Swiss")
if s0_promisc != [_S0_MR_MACRAW]:
errors.append(f"T1b promiscuous S0_MR {s0_promisc}, want MACRAW "
f"with MF clear")
ctx.set(dut.promisc, 0)
writes.clear()
writes.extend(EXPECTED) # restore for any later comparison
# ── helper: stream one TX frame through the external tx interface ─────
async def feed_frame(ctx, frame):
for i, b in enumerate(frame):
ctx.set(dut.tx_data, b)
ctx.set(dut.tx_valid, 1)
ctx.set(dut.tx_sof, 1 if i == 0 else 0)
ctx.set(dut.tx_eof, 1 if i == len(frame) - 1 else 0)
got = False
for _ in range(400):
await ctx.tick("sync").repeat(1)
if ctx.get(dut.tx_ready):
got = True
break
if not got:
errors.append(f"feed_frame: byte {i} never consumed")
return
ctx.set(dut.tx_valid, 0)
ctx.set(dut.tx_sof, 0)
ctx.set(dut.tx_eof, 0)
# let TX_UPDPTR + SEND complete
for _ in range(200):
await ctx.tick("sync").repeat(1)
if model_mem.get(_S0_CR) == _CR_SEND:
break
# ── T2: TX MACRAW frame (TX_WR=0, no wrap) ───────────────────────────
FRAME = [0xAA, 0xBB, 0xCC, 0xDD, 0xEE, 0xFF, 0x10, 0x20]
await feed_frame(ctx, FRAME)
buf = [model_mem.get(_TX_BASE + i, None) for i in range(len(FRAME))]
if buf != FRAME:
errors.append(f"T2 TX buffer mismatch: {buf} != {FRAME}")
tx_wr_hi = model_mem.get(_S0_TX_WR, 0)
tx_wr_lo = model_mem.get(_S0_TX_WR + 1, 0)
adv = (tx_wr_hi << 8) | tx_wr_lo
if adv != len(FRAME):
errors.append(f"T2 S0_TX_WR advance: got {adv}, want {len(FRAME)}")
if model_mem.get(_S0_CR) != _CR_SEND:
errors.append("T2 SEND command not issued")
print(f"T2 TX: buffer={['0x%02X' % b for b in buf]} "
f"TX_WR={adv} SEND={model_mem.get(_S0_CR)==_CR_SEND}")
# ── T3: TX MACRAW with ring wraparound (TX_WR near 2 KB boundary) ─────
# Pre-load S0_TX_WR = 0x07FE so a 6-byte frame straddles the boundary:
# offsets 0x7FE,0x7FF then wraps to 0x000,0x001,0x002,0x003.
model_mem[_S0_TX_WR] = 0x07
model_mem[_S0_TX_WR + 1] = 0xFE
model_mem[_S0_CR] = 0x00 # clear so we can detect the new SEND
WFRAME = [0x41, 0x42, 0x43, 0x44, 0x45, 0x46]
await feed_frame(ctx, WFRAME)
# expected physical layout
exp = {
_TX_BASE + 0x7FE: WFRAME[0],
_TX_BASE + 0x7FF: WFRAME[1],
_TX_BASE + 0x000: WFRAME[2],
_TX_BASE + 0x001: WFRAME[3],
_TX_BASE + 0x002: WFRAME[4],
_TX_BASE + 0x003: WFRAME[5],
}
for addr, want in exp.items():
got = model_mem.get(addr)
if got != want:
errors.append(f"T3 wrap byte @0x{addr:04X}: got {got}, want 0x{want:02X}")
adv2 = (model_mem.get(_S0_TX_WR, 0) << 8) | model_mem.get(_S0_TX_WR + 1, 0)
want_wr = (0x07FE + len(WFRAME)) & 0xFFFF
if adv2 != want_wr:
errors.append(f"T3 wrap S0_TX_WR: got 0x{adv2:04X}, want 0x{want_wr:04X}")
ok = all(model_mem.get(a) == v for a, v in exp.items())
print(f"T3 TX wrap: bytes_placed_ok={ok} TX_WR=0x{adv2:04X} (want 0x{want_wr:04X})")
# ── helper: drive an RX event and collect the streamed-out frame ─────
def load_rx(rx_rd_off, frame):
"""Place a MACRAW packet [len_hi,len_lo,frame...] in the RX buffer at
offset rx_rd_off (ring), set RX_RSR/RX_RD, return the 16-bit length."""
plen = len(frame) + 2
payload = [(plen >> 8) & 0xFF, plen & 0xFF] + list(frame)
for i, b in enumerate(payload):
off = (rx_rd_off + i) & _S0_RX_MASK
model_mem[_RX_BASE + off] = b
model_mem[_S0_RX_RSR] = (plen >> 8) & 0xFF
model_mem[_S0_RX_RSR + 1] = plen & 0xFF
model_mem[_S0_RX_RD] = (rx_rd_off >> 8) & 0xFF
model_mem[_S0_RX_RD + 1] = rx_rd_off & 0xFF
return plen
async def do_rx(ctx, rx_rd_off, frame):
plen = load_rx(rx_rd_off, frame)
ctx.set(dut.rx_ready, 1)
collected = []
ctx.set(dut.w5100_int_n, 0) # assert RX interrupt
for _ in range(1500):
await ctx.tick("sync").repeat(1)
if ctx.get(dut.rx_valid) and ctx.get(dut.rx_ready):
collected.append(ctx.get(dut.rx_data))
if model_mem.get(_S0_CR) == _CR_RECV:
break
ctx.set(dut.w5100_int_n, 1) # deassert; let it finish + idle
for _ in range(300):
await ctx.tick("sync").repeat(1)
ctx.set(dut.rx_ready, 0)
return collected, plen
# ── T4: RX MACRAW frame (RX_RD=0, no wrap) ───────────────────────────
model_mem[_S0_CR] = 0x00
RX_FRAME = [0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02, 0x03]
got, plen = await do_rx(ctx, 0x0000, RX_FRAME)
if got != RX_FRAME:
errors.append(f"T4 RX frame mismatch: {['0x%02X'%b for b in got]} != "
f"{['0x%02X'%b for b in RX_FRAME]}")
new_rd = (model_mem.get(_S0_RX_RD, 0) << 8) | model_mem.get(_S0_RX_RD + 1, 0)
if new_rd != plen:
errors.append(f"T4 RX_RD advance: got 0x{new_rd:04X}, want 0x{plen:04X}")
print(f"T4 RX: frame={['0x%02X'%b for b in got]} RX_RD=0x{new_rd:04X} "
f"RECV={model_mem.get(_S0_CR)==_CR_RECV}")
# ── T5: RX MACRAW with ring wraparound (RX_RD near 2 KB boundary) ─────
model_mem[_S0_CR] = 0x00
RX_FRAME2 = [0x51, 0x52, 0x53, 0x54, 0x55]
# rx_rd = 0x07FD: [len_hi@7FD][len_lo@7FE][f0@7FF][f1@000][f2@001]...
got2, plen2 = await do_rx(ctx, 0x07FD, RX_FRAME2)
if got2 != RX_FRAME2:
errors.append(f"T5 RX wrap frame mismatch: {['0x%02X'%b for b in got2]} != "
f"{['0x%02X'%b for b in RX_FRAME2]}")
new_rd2 = (model_mem.get(_S0_RX_RD, 0) << 8) | model_mem.get(_S0_RX_RD + 1, 0)
want_rd2 = (0x07FD + plen2) & 0xFFFF
if new_rd2 != want_rd2:
errors.append(f"T5 RX wrap RX_RD: got 0x{new_rd2:04X}, want 0x{want_rd2:04X}")
print(f"T5 RX wrap: frame={['0x%02X'%b for b in got2]} "
f"RX_RD=0x{new_rd2:04X} (want 0x{want_rd2:04X})")
sim = Simulator(dut)
sim.add_clock(Period(MHz=24), domain="sync")
sim.add_testbench(testbench)
sim.add_process(w5100_model)
sim.run()
# ── UDP-test path (configurable socket, enable_udp_test=True) ───────────
# Reuses the same address-agnostic bus model (it reads/writes model_mem at
# whatever IDM_AR is set), so it transparently covers the IP-config and
# socket registers. Verifies: init programs GAR/SUBR/SIPR + opens the
# socket in UDP mode; a udp_send_req writes Sn_DIPR to the runtime dest,
# streams the payload into the socket TX buffer, advances Sn_TX_WR, issues
# SEND; and a udp_rx_req reads a datagram (header + payload) back out.
SRC_IP, SUBNET, GATEWAY = "10.0.0.7", "255.255.255.0", "10.0.0.1"
DEF_DST, DST_PORT, SRC_PORT = "192.168.1.100", 6464, 40000
UDP_SOCK = 3
S = _socket_addrs(UDP_SOCK)
dut2 = W5100ParallelMaster(
strobe_cycles=3, reset_cycles=10, enable_udp_test=True,
udp_socket=UDP_SOCK, src_ip=SRC_IP, subnet=SUBNET, gateway=GATEWAY,
dst_ip=DEF_DST, src_port=SRC_PORT, dst_port=DST_PORT)
writes2, model_mem2 = [], {}
async def w5100_model2(ctx):
idm_ar = 0
mr = 0
prev_wr = prev_rd = 1
async for vals in ctx.tick("sync").sample(
dut2.cs_n, dut2.rd_n, dut2.wr_n,
dut2.bus_addr, dut2.bus_data_o):
cs, rd, wr, a, do = vals[-5:]
ai = (mr >> 1) & 1
if cs == 0 and rd == 0:
if a == _A_MR: val = mr
elif a == _A_AR0: val = (idm_ar >> 8) & 0xFF
elif a == _A_AR1: val = idm_ar & 0xFF
else: val = model_mem2.get(idm_ar, 0)
ctx.set(dut2.bus_data_i, val)
if cs == 0 and prev_wr == 0 and wr == 1:
if a == _A_MR:
mr = do; writes2.append(("MR", do))
elif a == _A_AR0: idm_ar = (idm_ar & 0x00FF) | (do << 8)
elif a == _A_AR1: idm_ar = (idm_ar & 0xFF00) | do
else:
model_mem2[idm_ar] = do; writes2.append((idm_ar, do))
if ai: idm_ar = (idm_ar + 1) & 0xFFFF
if cs == 0 and prev_rd == 0 and rd == 1 and a == _A_DR and ai:
idm_ar = (idm_ar + 1) & 0xFFFF
prev_wr, prev_rd = wr, rd
def _u16(addr):
return (model_mem2.get(addr, 0) << 8) | model_mem2.get(addr + 1, 0)
async def testbench2(ctx):
ctx.set(dut2.par, PAR)
ctx.set(dut2.udp_pl_valid, 0)
await ctx.tick("sync").repeat(2)
# U1: init → IP stack + socket-1 UDP open.
ctx.set(dut2.init_req, 1)
await ctx.tick("sync").repeat(1)
ctx.set(dut2.init_req, 0)
done = False
for _ in range(6000):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.init_done):
done = True; break
if not done:
errors.append("UDP: init_done never asserted")
checks = {
_GAR0: _ip_bytes(GATEWAY), _SUBR0: _ip_bytes(SUBNET),
_SIPR0: _ip_bytes(SRC_IP),
}
for base, octets in checks.items():
got = [model_mem2.get(base + i) for i in range(4)]
if got != octets:
errors.append(f"UDP init {base:#06x}: got {got}, want {octets}")
if model_mem2.get(S["MR"]) != _S1_MR_UDP:
errors.append(f"UDP init Sn_MR: got {model_mem2.get(S['MR'])}, want UDP")
if _u16(S["PORT"]) != SRC_PORT:
errors.append(f"UDP init Sn_PORT: {_u16(S['PORT'])} != {SRC_PORT}")
if _u16(S["DPORT"]) != DST_PORT:
errors.append(f"UDP init Sn_DPORT: {_u16(S['DPORT'])} != {DST_PORT}")
if (S["CR"], _CR_OPEN) not in writes2:
errors.append("UDP init: socket OPEN not issued")
print(f"U1 init (socket {UDP_SOCK}): GAR/SUBR/SIPR set, Sn_MR=UDP, "
f"Sn_PORT={_u16(S['PORT'])} Sn_DPORT={_u16(S['DPORT'])} OPEN issued")
# U2: send one datagram to a runtime dest IP with a payload.
DST = "192.168.1.55"
PAYLOAD = list(b"HELLO-ETH")
ctx.set(dut2.udp_dst_ip, int.from_bytes(bytes(_ip_bytes(DST)), "big"))
idx = 0
ctx.set(dut2.udp_pl_data, PAYLOAD[0])
ctx.set(dut2.udp_pl_last, 1 if len(PAYLOAD) == 1 else 0)
ctx.set(dut2.udp_pl_valid, 1)
ctx.set(dut2.udp_send_req, 1)
await ctx.tick("sync").repeat(1)
ctx.set(dut2.udp_send_req, 0)
sent = False
for _ in range(4000):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.udp_pl_ready) and ctx.get(dut2.udp_pl_valid):
idx += 1
if idx < len(PAYLOAD):
ctx.set(dut2.udp_pl_data, PAYLOAD[idx])
ctx.set(dut2.udp_pl_last, 1 if idx == len(PAYLOAD) - 1 else 0)
else:
ctx.set(dut2.udp_pl_valid, 0)
if model_mem2.get(S["CR"]) == _CR_SEND:
sent = True; break
ctx.set(dut2.udp_pl_valid, 0)
if not sent:
errors.append("UDP send: SEND command never issued")
for _ in range(50):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.udp_test_busy) == 0:
break
dip = [model_mem2.get(S["DIPR"] + i) for i in range(4)]
if dip != _ip_bytes(DST):
errors.append(f"UDP send DIPR: got {dip}, want {_ip_bytes(DST)}")
buf = [model_mem2.get(S["TX_BASE"] + i) for i in range(len(PAYLOAD))]
if buf != PAYLOAD:
errors.append(f"UDP send payload: {buf} != {PAYLOAD}")
if _u16(S["TX_WR"]) != len(PAYLOAD):
errors.append(f"UDP send Sn_TX_WR: {_u16(S['TX_WR'])} != {len(PAYLOAD)}")
print(f"U2 send: DIPR={dip} payload={bytes(buf)!r} "
f"Sn_TX_WR={_u16(S['TX_WR'])} SEND={sent}")
# ── U3: receive a UDP datagram from the socket RX buffer ─────────────
# WIZnet UDP RX format: [srcIP(4)][srcPort(2)][len(2)][payload].
model_mem2[S["CR"]] = 0x00
SRC = _ip_bytes("192.168.1.9")
SPORT = 6464
RPAYLOAD = list(b"WORLD!")
rlen = len(RPAYLOAD)
rx_rd0 = 0x0000
hdr = SRC + [SPORT >> 8, SPORT & 0xFF, (rlen >> 8) & 0xFF, rlen & 0xFF]
packet = hdr + RPAYLOAD
for i, b in enumerate(packet):
model_mem2[S["RX_BASE"] + ((rx_rd0 + i) & _SN_MASK)] = b
total = len(packet)
model_mem2[S["RX_RSR"]] = (total >> 8) & 0xFF
model_mem2[S["RX_RSR"] + 1] = total & 0xFF
model_mem2[S["RX_RD"]] = (rx_rd0 >> 8) & 0xFF
model_mem2[S["RX_RD"] + 1] = rx_rd0 & 0xFF
ctx.set(dut2.udp_rx_ready, 1)
rx_got = []
got_none = [False]
ctx.set(dut2.udp_rx_req, 1)
# hold req until the master goes busy, then drop it
for _ in range(200):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.udp_rx_busy):
ctx.set(dut2.udp_rx_req, 0); break
recvd = False
for _ in range(3000):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.udp_rx_valid) and ctx.get(dut2.udp_rx_ready):
rx_got.append(ctx.get(dut2.udp_rx_data))
if ctx.get(dut2.udp_rx_none):
got_none[0] = True
if model_mem2.get(S["CR"]) == _CR_RECV:
recvd = True; break
for _ in range(30):
await ctx.tick("sync").repeat(1)
ctx.set(dut2.udp_rx_ready, 0)
if rx_got != RPAYLOAD:
errors.append(f"U3 RX payload: {bytes(rx_got)!r} != {bytes(RPAYLOAD)!r}")
new_rd = _u16(S["RX_RD"])
if new_rd != rx_rd0 + total:
errors.append(f"U3 RX_RD advance: got {new_rd}, want {rx_rd0 + total}")
if model_mem2.get(S["CR"]) != _CR_RECV:
errors.append("U3 RX: RECV command not issued")
if got_none[0]:
errors.append("U3 RX: udp_rx_none pulsed despite data present")
print(f"U3 recv: payload={bytes(rx_got)!r} RX_RD={new_rd} "
f"RECV={model_mem2.get(S['CR'])==_CR_RECV}")
# ── U4: RX poll with an empty buffer → udp_rx_none, no stream ────────
model_mem2[S["RX_RSR"]] = 0
model_mem2[S["RX_RSR"] + 1] = 0
model_mem2[S["CR"]] = 0x00
none_seen = False
streamed = False
ctx.set(dut2.udp_rx_req, 1)
for _ in range(200):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.udp_rx_busy):
ctx.set(dut2.udp_rx_req, 0); break
for _ in range(200):
await ctx.tick("sync").repeat(1)
if ctx.get(dut2.udp_rx_none):
none_seen = True
if ctx.get(dut2.udp_rx_valid):
streamed = True
if none_seen and not ctx.get(dut2.udp_rx_busy):
break
if not none_seen:
errors.append("U4 RX-empty: udp_rx_none never pulsed")
if streamed:
errors.append("U4 RX-empty: streamed data despite empty buffer")
if model_mem2.get(S["CR"]) == _CR_RECV:
errors.append("U4 RX-empty: RECV issued on empty buffer")
print(f"U4 recv-empty: none={none_seen} streamed={streamed}")
sim2 = Simulator(dut2)
sim2.add_clock(Period(MHz=24), domain="sync")
sim2.add_testbench(testbench2)
sim2.add_process(w5100_model2)
sim2.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