baulin.fa/rtl_libs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: baulin.fa:cocotb_proxy
Branches Tags
baulin.fa:master baulin.fa:dev baulin.fa:cocotb_proxy
..
compare: baulin.fa:dev
Branches Tags
baulin.fa:master baulin.fa:dev baulin.fa:cocotb_proxy

2 Commits
cocotb_pro ... dev

Author SHA1 Message Date
Phil
a0a7137064 fix: axi dma wrapper 2026-07-10 18:59:50 +03:00
Phil
3ddf9130d3 fix: reg_pulse support 2026-07-10 17:30:51 +03:00
10 changed files with 741 additions and 739 deletions
Expand all files Collapse all files

38
axi/axi_reg/axi4l_reg_map.sv
View File

@ -12,7 +12,8 @@ module axi4l_reg_map #(
axi4l_if.slave s_axil,
input logic [N_REGS-1:0][31:0] reg_i,
output logic [N_REGS-1:0][31:0] reg_o
output logic [N_REGS-1:0][31:0] reg_o,
output logic [N_REGS-1:0][31:0] reg_pulse
);
import axi_pkg::*;
@ -105,6 +106,7 @@ module axi4l_reg_map #(
rdata_q <= '0;
reg_o <= REG_RST;
end else begin
reg_pulse <= '0;
for (int r = 0; r < N_REGS; r++) begin
for (int bit_idx = 0; bit_idx < 32; bit_idx++) begin
if (reg_bit_mode_t'(REG_MODE[r][bit_idx]) == REG_BIT_W1S)
@ -136,12 +138,33 @@ module axi4l_reg_map #(
for (b = 0; b < 32; b = b + 1) begin
if (wr_mask[b]) begin
unique case (reg_bit_mode_t'(REG_MODE[wr_idx][b]))
REG_BIT_RSVD: begin end
REG_BIT_RO : begin bresp_q <= 2'b10; end
REG_BIT_RW : rw_new[b] = wr_data32[b];
REG_BIT_W1S : if (wr_data32[b]) rw_new[b] = 1'b1;
REG_BIT_W1C : if (wr_data32[b]) rw_new[b] = 1'b0;
default : begin end
REG_BIT_RSVD: begin
end
REG_BIT_RO: begin
bresp_q <= 2'b10;
end
REG_BIT_RW: begin
rw_new[b] = wr_data32[b];
end
REG_BIT_W1S: begin
if (wr_data32[b]) begin
rw_new[b] = 1'b1;
reg_pulse[wr_idx][b] <= 1'b1;
end
end
REG_BIT_W1C: begin
if (wr_data32[b]) begin
rw_new[b] = 1'b0;
reg_pulse[wr_idx][b] <= 1'b1;
end
end
default: begin
end
endcase
end
end
@ -174,6 +197,7 @@ module axi4l_reg_map #(
REG_BIT_W1C : rd_word[b] = reg_o[rd_idx][b];
default : rd_word[b] = 1'b0;
endcase
end
end

24
axi/rtl/axi_dma_wrapper.sv
View File

@ -271,7 +271,7 @@ module axi_dma_wrapper #(
// local read descriptor interface -> DMA flat input
axis_if_to_flat #(
.DATA_W (AXI_ADDR_WIDTH + LEN_WIDTH + TAG_WIDTH),
.KEEP_W ('0),
.KEEP_W (AXIS_KEEP_WIDTH),
.ID_W (AXIS_ID_WIDTH),
.DEST_W (AXIS_DEST_WIDTH),
.USER_W (AXIS_USER_WIDTH)
@ -282,7 +282,7 @@ module axi_dma_wrapper #(
.m_axis_tstrb (),
.m_axis_tlast (),
.m_axis_tid (dma_s_axis_read_desc_id),
.m_axis_tdest (dma_s_axis_read_desc_tdest),
.m_axis_tdest (dma_s_axis_read_desc_dest),
.m_axis_tuser (dma_s_axis_read_desc_user),
.m_axis_tvalid (dma_s_axis_read_desc_valid),
.m_axis_tready (dma_s_axis_read_desc_ready)
@ -291,10 +291,10 @@ module axi_dma_wrapper #(
// DMA read descriptor status flat output -> local status interface
axis_flat_to_if #(
.DATA_W (TAG_WIDTH + 4),
.KEEP_W ('0),
.ID_W ('0),
.DEST_W ('0),
.USER_W ('0)
.KEEP_W (AXIS_KEEP_WIDTH),
.ID_W (AXIS_ID_WIDTH),
.DEST_W (AXIS_DEST_WIDTH),
.USER_W (AXIS_USER_WIDTH)
) i_read_desc_status_f2i(
.s_axis_tdata ({dma_m_axis_read_desc_status_tag, dma_m_axis_read_desc_status_error}),
.s_axis_tkeep (),
@ -311,7 +311,7 @@ module axi_dma_wrapper #(
// local write descriptor interface -> DMA flat input
axis_if_to_flat #(
.DATA_W (AXI_ADDR_WIDTH + LEN_WIDTH + TAG_WIDTH),
.KEEP_W ('0),
.KEEP_W (AXIS_KEEP_WIDTH),
.ID_W (AXIS_ID_WIDTH),
.DEST_W (AXIS_DEST_WIDTH),
.USER_W (AXIS_USER_WIDTH)
@ -321,9 +321,9 @@ module axi_dma_wrapper #(
.m_axis_tkeep (),
.m_axis_tstrb (),
.m_axis_tlast (),
.m_axis_tid (dma_s_axis_write_desc_id),
.m_axis_tdest (dma_s_axis_write_desc_tdest),
.m_axis_tuser (dma_s_axis_write_desc_user),
.m_axis_tid (),
.m_axis_tdest (),
.m_axis_tuser (),
.m_axis_tvalid (dma_s_axis_write_desc_valid),
.m_axis_tready (dma_s_axis_write_desc_ready)
);
@ -332,7 +332,7 @@ module axi_dma_wrapper #(
// DMA write descriptor status flat output -> local status interface
axis_flat_to_if #(
.DATA_W (TAG_WIDTH + LEN_WIDTH + 4),
.KEEP_W ('0),
.KEEP_W (AXIS_KEEP_WIDTH),
.ID_W (AXIS_ID_WIDTH),
.DEST_W (AXIS_DEST_WIDTH),
.USER_W (AXIS_USER_WIDTH)
@ -448,6 +448,6 @@ module axi_dma_wrapper #(
.m_axi (m_axi)
);
endmodule : axi_dma_if_wrapper
endmodule : axi_dma_wrapper
`default_nettype wire

430
axi/tb/axi_cocotb_loopback_test/axi4_proxy.py
View File

@ -1,430 +0,0 @@
# proxy to match cocotb and custom axi4_if
import logging
from cocotb.types import LogicArray
from cocotbext.axi.axi_channels import (
AxiAWBus,
AxiWBus,
AxiBBus,
AxiARBus,
AxiRBus,
AxiBus,
)
def install_cocotbext_axi_slice_compat():
"""
cocotbext-axi StreamMonitor uses RisingEdge(self.valid/self.ready).
That works only for real cocotb LogicObject handles.
Our SliceSignal is a Python proxy over packed req/resp vector, so
RisingEdge(SliceSignal) fails in cocotb 2.x.
This patch makes StreamMonitor wake on signal.value_change instead.
For SliceSignal this should return parent packed vector value_change.
"""
# this sucks...
from cocotbext.axi import stream as axi_stream
async def _run_valid_monitor_value_change(self):
while True:
await self.valid.value_change
self.wake_event.set()
async def _run_ready_monitor_value_change(self):
while True:
await self.ready.value_change
self.wake_event.set()
axi_stream.StreamMonitor._run_valid_monitor = _run_valid_monitor_value_change
axi_stream.StreamMonitor._run_ready_monitor = _run_ready_monitor_value_change
def _safe_int(value) -> int:
try:
return int(value)
except Exception:
s = str(value).upper()
bits = []
for ch in s:
if ch in "01":
bits.append(ch)
elif ch in "XZUW-":
bits.append("0")
if not bits:
return 0
return int("".join(bits), 2)
def _logic_array(value: int, width: int):
value &= (1 << width) - 1
return LogicArray.from_unsigned(value, width)
def _param_int(obj, name: str, override):
if override is not None:
return int(override)
p = getattr(obj, name)
try:
return int(p)
except Exception:
pass
try:
return int(p.value)
except Exception:
pass
raise RuntimeError(f"Cannot read parameter {name} from {obj!r}")
def _packed_offsets(fields):
"""
de-construct SV signal order.
"""
total = sum(width for _, width in fields)
pos = total
offsets = {}
for name, width in fields:
pos -= width
offsets[name] = (pos, width)
return total, offsets
class PackedVector:
def __init__(self, handle, name=None):
self.handle = handle
self.name = name or getattr(handle, "_path", repr(handle))
self.width = len(handle)
self.mask = (1 << self.width) - 1
self.shadow = _safe_int(handle.value) & self.mask
self.dirty = False
def read(self, lo: int, width: int) -> int:
src = self.shadow if self.dirty else (
_safe_int(self.handle.value) & self.mask)
return (src >> lo) & ((1 << width) - 1)
def write(self, lo: int, width: int, value, immediate=False):
value = _safe_int(value)
field_mask = ((1 << width) - 1) << lo
self.shadow &= ~field_mask
self.shadow |= (value << lo) & field_mask
self.shadow &= self.mask
self.dirty = True
v = _logic_array(self.shadow, self.width)
if immediate:
self.handle.setimmediatevalue(v)
else:
self.handle.value = v
class SliceSignal:
"""
cocotbext-axi proxy
looks like:
sig.value
sig.value = ...
sig.setimmediatevalue(...)
len(sig)
"""
def __init__(self, name: str, parent: PackedVector, lo: int, width: int):
self._name = name
self._path = name
self._log = logging.getLogger(f"cocotb.{name}")
self.parent = parent
self.lo = lo
self.width = width
def __len__(self):
return self.width
def __repr__(self):
return f"<SliceSignal {self._path}[{self.lo + self.width - 1}:{self.lo}]>"
@property
def value(self):
return _logic_array(self.parent.read(self.lo, self.width), self.width)
@value.setter
def value(self, value):
self.parent.write(self.lo, self.width, value, immediate=False)
def setimmediatevalue(self, value):
self.parent.write(self.lo, self.width, value, immediate=True)
def set(self, value):
self.value = value
def get(self):
return self.value
@property
def value_change(self):
# dirty
return self.parent.handle.value_change
class ProxyEntity:
def __init__(self, name: str, **signals):
self._name = name
self._path = name
self._log = logging.getLogger(f"cocotb.{name}")
self.__dict__.update(signals)
def __dir__(self):
return list(self.__dict__.keys())
def _sig(name, parent, base_lo, layout, field):
lo, width = layout[field]
return SliceSignal(name, parent, base_lo + lo, width)
def axi4_bus_from_packed_if(
iface,
name="axi",
*,
addr_width=None,
data_width=None,
id_width=None,
user_width=None,
):
"""
Build cocotbext-axi AxiBus from custom packed axi4_if.
iface should be instance axi4_if:
dut.axi_in
dut.axi_out
example:
master = AxiMaster(
axi4_bus_from_packed_if(dut.axi_in, "s_axi"),
dut.clk,
dut.rst,
)
ram = AxiRam(
axi4_bus_from_packed_if(dut.axi_out, "m_axi"),
dut.clk,
dut.rst,
size=2**20,
)
"""
ADDR_W = _param_int(iface, "ADDR_W", addr_width)
DATA_W = _param_int(iface, "DATA_W", data_width)
ID_W = _param_int(iface, "ID_W", id_width)
USER_W = _param_int(iface, "USER_W", user_width)
if DATA_W % 8 != 0:
raise ValueError(f"AXI DATA_W must be divisible by 8, got {DATA_W}")
STRB_W = DATA_W // 8
aw_total, aw_layout = _packed_offsets([
("id", ID_W),
("addr", ADDR_W),
("len", 8),
("size", 3),
("burst", 2),
("lock", 1),
("cache", 4),
("prot", 3),
("qos", 4),
("region", 4),
("user", USER_W),
("valid", 1),
])
w_total, w_layout = _packed_offsets([
("data", DATA_W),
("strb", STRB_W),
("last", 1),
("user", USER_W),
("valid", 1),
])
b_total, b_layout = _packed_offsets([
("id", ID_W),
("resp", 2),
("user", USER_W),
("valid", 1),
])
ar_total, ar_layout = _packed_offsets([
("id", ID_W),
("addr", ADDR_W),
("len", 8),
("size", 3),
("burst", 2),
("lock", 1),
("cache", 4),
("prot", 3),
("qos", 4),
("region", 4),
("user", USER_W),
("valid", 1),
])
r_total, r_layout = _packed_offsets([
("id", ID_W),
("data", DATA_W),
("resp", 2),
("last", 1),
("user", USER_W),
("valid", 1),
])
req_total, req_layout = _packed_offsets([
("aw", aw_total),
("w", w_total),
("b_ready", 1),
("ar", ar_total),
("r_ready", 1),
])
resp_total, resp_layout = _packed_offsets([
("aw_ready", 1),
("w_ready", 1),
("b", b_total),
("ar_ready", 1),
("r", r_total),
])
req = PackedVector(iface.req, f"{name}.req")
resp = PackedVector(iface.resp, f"{name}.resp")
if req_total != req.width:
raise RuntimeError(
f"{name}.req layout mismatch: calculated {req_total} bits, "
f"simulator has {req.width} bits"
)
if resp_total != resp.width:
raise RuntimeError(
f"{name}.resp layout mismatch: calculated {resp_total} bits, "
f"simulator has {resp.width} bits"
)
aw_lo, _ = req_layout["aw"]
w_lo, _ = req_layout["w"]
ar_lo, _ = req_layout["ar"]
b_lo, _ = resp_layout["b"]
r_lo, _ = resp_layout["r"]
aw = AxiAWBus.from_entity(ProxyEntity(
f"{name}_aw",
awid=_sig(f"{name}_awid", req, aw_lo, aw_layout, "id"),
awaddr=_sig(f"{name}_awaddr", req, aw_lo, aw_layout, "addr"),
awlen=_sig(f"{name}_awlen", req, aw_lo, aw_layout, "len"),
awsize=_sig(f"{name}_awsize", req, aw_lo, aw_layout, "size"),
awburst=_sig(f"{name}_awburst", req, aw_lo, aw_layout, "burst"),
awlock=_sig(f"{name}_awlock", req, aw_lo, aw_layout, "lock"),
awcache=_sig(f"{name}_awcache", req, aw_lo, aw_layout, "cache"),
awprot=_sig(f"{name}_awprot", req, aw_lo, aw_layout, "prot"),
awqos=_sig(f"{name}_awqos", req, aw_lo, aw_layout, "qos"),
awregion=_sig(f"{name}_awregion", req, aw_lo, aw_layout, "region"),
awuser=_sig(f"{name}_awuser", req, aw_lo, aw_layout, "user"),
awvalid=_sig(f"{name}_awvalid", req, aw_lo, aw_layout, "valid"),
awready=SliceSignal(
f"{name}_awready",
resp,
*resp_layout["aw_ready"],
),
))
w = AxiWBus.from_entity(ProxyEntity(
f"{name}_w",
wdata=_sig(f"{name}_wdata", req, w_lo, w_layout, "data"),
wstrb=_sig(f"{name}_wstrb", req, w_lo, w_layout, "strb"),
wlast=_sig(f"{name}_wlast", req, w_lo, w_layout, "last"),
wuser=_sig(f"{name}_wuser", req, w_lo, w_layout, "user"),
wvalid=_sig(f"{name}_wvalid", req, w_lo, w_layout, "valid"),
wready=SliceSignal(
f"{name}_wready",
resp,
*resp_layout["w_ready"],
),
))
b = AxiBBus.from_entity(ProxyEntity(
f"{name}_b",
bid=_sig(f"{name}_bid", resp, b_lo, b_layout, "id"),
bresp=_sig(f"{name}_bresp", resp, b_lo, b_layout, "resp"),
buser=_sig(f"{name}_buser", resp, b_lo, b_layout, "user"),
bvalid=_sig(f"{name}_bvalid", resp, b_lo, b_layout, "valid"),
bready=SliceSignal(
f"{name}_bready",
req,
*req_layout["b_ready"],
),
))
ar = AxiARBus.from_entity(ProxyEntity(
f"{name}_ar",
arid=_sig(f"{name}_arid", req, ar_lo, ar_layout, "id"),
araddr=_sig(f"{name}_araddr", req, ar_lo, ar_layout, "addr"),
arlen=_sig(f"{name}_arlen", req, ar_lo, ar_layout, "len"),
arsize=_sig(f"{name}_arsize", req, ar_lo, ar_layout, "size"),
arburst=_sig(f"{name}_arburst", req, ar_lo, ar_layout, "burst"),
arlock=_sig(f"{name}_arlock", req, ar_lo, ar_layout, "lock"),
arcache=_sig(f"{name}_arcache", req, ar_lo, ar_layout, "cache"),
arprot=_sig(f"{name}_arprot", req, ar_lo, ar_layout, "prot"),
arqos=_sig(f"{name}_arqos", req, ar_lo, ar_layout, "qos"),
arregion=_sig(f"{name}_arregion", req, ar_lo, ar_layout, "region"),
aruser=_sig(f"{name}_aruser", req, ar_lo, ar_layout, "user"),
arvalid=_sig(f"{name}_arvalid", req, ar_lo, ar_layout, "valid"),
arready=SliceSignal(
f"{name}_arready",
resp,
*resp_layout["ar_ready"],
),
))
r = AxiRBus.from_entity(ProxyEntity(
f"{name}_r",
rid=_sig(f"{name}_rid", resp, r_lo, r_layout, "id"),
rdata=_sig(f"{name}_rdata", resp, r_lo, r_layout, "data"),
rresp=_sig(f"{name}_rresp", resp, r_lo, r_layout, "resp"),
rlast=_sig(f"{name}_rlast", resp, r_lo, r_layout, "last"),
ruser=_sig(f"{name}_ruser", resp, r_lo, r_layout, "user"),
rvalid=_sig(f"{name}_rvalid", resp, r_lo, r_layout, "valid"),
rready=SliceSignal(
f"{name}_rready",
req,
*req_layout["r_ready"],
),
))
return AxiBus.from_channels(aw, w, b, ar, r)

221
axi/tb/axi_cocotb_loopback_test/tb_axi4_loopback.sv
View File

@ -1,11 +1,113 @@
`timescale 1ns/1ps
module tb_axi4_loopback #(
parameter int unsigned ADDR_W = 32,
parameter int unsigned DATA_W = 64,
parameter int unsigned ID_W = 4,
parameter int unsigned USER_W = 1
)(
input logic clk,
input logic rst
input logic clk,
input logic rst,
// slave-side flat AXI port (cocotb driven)
input logic [ID_W-1:0] s_axi_awid,
input logic [ADDR_W-1:0] s_axi_awaddr,
input logic [7:0] s_axi_awlen,
input logic [2:0] s_axi_awsize,
input logic [1:0] s_axi_awburst,
input logic s_axi_awlock,
input logic [3:0] s_axi_awcache,
input logic [2:0] s_axi_awprot,
input logic [3:0] s_axi_awqos,
input logic [3:0] s_axi_awregion,
input logic [USER_W-1:0] s_axi_awuser,
input logic s_axi_awvalid,
output logic s_axi_awready,
input logic [DATA_W-1:0] s_axi_wdata,
input logic [DATA_W/8-1:0] s_axi_wstrb,
input logic s_axi_wlast,
input logic [USER_W-1:0] s_axi_wuser,
input logic s_axi_wvalid,
output logic s_axi_wready,
output logic [ID_W-1:0] s_axi_bid,
output logic [1:0] s_axi_bresp,
output logic [USER_W-1:0] s_axi_buser,
output logic s_axi_bvalid,
input logic s_axi_bready,
input logic [ID_W-1:0] s_axi_arid,
input logic [ADDR_W-1:0] s_axi_araddr,
input logic [7:0] s_axi_arlen,
input logic [2:0] s_axi_arsize,
input logic [1:0] s_axi_arburst,
input logic s_axi_arlock,
input logic [3:0] s_axi_arcache,
input logic [2:0] s_axi_arprot,
input logic [3:0] s_axi_arqos,
input logic [3:0] s_axi_arregion,
input logic [USER_W-1:0] s_axi_aruser,
input logic s_axi_arvalid,
output logic s_axi_arready,
output logic [ID_W-1:0] s_axi_rid,
output logic [DATA_W-1:0] s_axi_rdata,
output logic [1:0] s_axi_rresp,
output logic s_axi_rlast,
output logic [USER_W-1:0] s_axi_ruser,
output logic s_axi_rvalid,
input logic s_axi_rready,
// master-side flat AXI port for coco-tb
output logic [ID_W-1:0] m_axi_awid,
output logic [ADDR_W-1:0] m_axi_awaddr,
output logic [7:0] m_axi_awlen,
output logic [2:0] m_axi_awsize,
output logic [1:0] m_axi_awburst,
output logic m_axi_awlock,
output logic [3:0] m_axi_awcache,
output logic [2:0] m_axi_awprot,
output logic [3:0] m_axi_awqos,
output logic [3:0] m_axi_awregion,
output logic [USER_W-1:0] m_axi_awuser,
output logic m_axi_awvalid,
input logic m_axi_awready,
output logic [DATA_W-1:0] m_axi_wdata,
output logic [DATA_W/8-1:0] m_axi_wstrb,
output logic m_axi_wlast,
output logic [USER_W-1:0] m_axi_wuser,
output logic m_axi_wvalid,
input logic m_axi_wready,
input logic [ID_W-1:0] m_axi_bid,
input logic [1:0] m_axi_bresp,
input logic [USER_W-1:0] m_axi_buser,
input logic m_axi_bvalid,
output logic m_axi_bready,
output logic [ID_W-1:0] m_axi_arid,
output logic [ADDR_W-1:0] m_axi_araddr,
output logic [7:0] m_axi_arlen,
output logic [2:0] m_axi_arsize,
output logic [1:0] m_axi_arburst,
output logic m_axi_arlock,
output logic [3:0] m_axi_arcache,
output logic [2:0] m_axi_arprot,
output logic [3:0] m_axi_arqos,
output logic [3:0] m_axi_arregion,
output logic [USER_W-1:0] m_axi_aruser,
output logic m_axi_arvalid,
input logic m_axi_arready,
input logic [ID_W-1:0] m_axi_rid,
input logic [DATA_W-1:0] m_axi_rdata,
input logic [1:0] m_axi_rresp,
input logic m_axi_rlast,
input logic [USER_W-1:0] m_axi_ruser,
input logic m_axi_rvalid,
output logic m_axi_rready
);
logic aresetn;
@ -31,9 +133,120 @@ module tb_axi4_loopback #(
.aresetn(aresetn)
);
axi4_loopback dut (
axi4_flat_to_if #(
.ADDR_W(ADDR_W),
.DATA_W(DATA_W),
.ID_W(ID_W),
.USER_W(USER_W)
) u_flat_to_if (
.s_axi_awid (s_axi_awid),
.s_axi_awaddr (s_axi_awaddr),
.s_axi_awlen (s_axi_awlen),
.s_axi_awsize (s_axi_awsize),
.s_axi_awburst (s_axi_awburst),
.s_axi_awlock (s_axi_awlock),
.s_axi_awcache (s_axi_awcache),
.s_axi_awprot (s_axi_awprot),
.s_axi_awqos (s_axi_awqos),
.s_axi_awregion (s_axi_awregion),
.s_axi_awuser (s_axi_awuser),
.s_axi_awvalid (s_axi_awvalid),
.s_axi_awready (s_axi_awready),
.s_axi_wdata (s_axi_wdata),
.s_axi_wstrb (s_axi_wstrb),
.s_axi_wlast (s_axi_wlast),
.s_axi_wuser (s_axi_wuser),
.s_axi_wvalid (s_axi_wvalid),
.s_axi_wready (s_axi_wready),
.s_axi_bid (s_axi_bid),
.s_axi_bresp (s_axi_bresp),
.s_axi_buser (s_axi_buser),
.s_axi_bvalid (s_axi_bvalid),
.s_axi_bready (s_axi_bready),
.s_axi_arid (s_axi_arid),
.s_axi_araddr (s_axi_araddr),
.s_axi_arlen (s_axi_arlen),
.s_axi_arsize (s_axi_arsize),
.s_axi_arburst (s_axi_arburst),
.s_axi_arlock (s_axi_arlock),
.s_axi_arcache (s_axi_arcache),
.s_axi_arprot (s_axi_arprot),
.s_axi_arqos (s_axi_arqos),
.s_axi_arregion (s_axi_arregion),
.s_axi_aruser (s_axi_aruser),
.s_axi_arvalid (s_axi_arvalid),
.s_axi_arready (s_axi_arready),
.s_axi_rid (s_axi_rid),
.s_axi_rdata (s_axi_rdata),
.s_axi_rresp (s_axi_rresp),
.s_axi_rlast (s_axi_rlast),
.s_axi_ruser (s_axi_ruser),
.s_axi_rvalid (s_axi_rvalid),
.s_axi_rready (s_axi_rready),
.m_axi (axi_in)
);
axi4_loopback #(
.ADDR_W(ADDR_W),
.DATA_W(DATA_W),
.ID_W(ID_W),
.USER_W(USER_W)
) dut (
.s_axi(axi_in),
.m_axi(axi_out)
);
endmodule
axi4_if_to_flat #(
.ADDR_W(ADDR_W),
.DATA_W(DATA_W),
.ID_W(ID_W),
.USER_W(USER_W)
) u_if_to_flat (
.s_axi (axi_out),
.m_axi_awid (m_axi_awid),
.m_axi_awaddr (m_axi_awaddr),
.m_axi_awlen (m_axi_awlen),
.m_axi_awsize (m_axi_awsize),
.m_axi_awburst (m_axi_awburst),
.m_axi_awlock (m_axi_awlock),
.m_axi_awcache (m_axi_awcache),
.m_axi_awprot (m_axi_awprot),
.m_axi_awqos (m_axi_awqos),
.m_axi_awregion (m_axi_awregion),
.m_axi_awuser (m_axi_awuser),
.m_axi_awvalid (m_axi_awvalid),
.m_axi_awready (m_axi_awready),
.m_axi_wdata (m_axi_wdata),
.m_axi_wstrb (m_axi_wstrb),
.m_axi_wlast (m_axi_wlast),
.m_axi_wuser (m_axi_wuser),
.m_axi_wvalid (m_axi_wvalid),
.m_axi_wready (m_axi_wready),
.m_axi_bid (m_axi_bid),
.m_axi_bresp (m_axi_bresp),
.m_axi_buser (m_axi_buser),
.m_axi_bvalid (m_axi_bvalid),
.m_axi_bready (m_axi_bready),
.m_axi_arid (m_axi_arid),
.m_axi_araddr (m_axi_araddr),
.m_axi_arlen (m_axi_arlen),
.m_axi_arsize (m_axi_arsize),
.m_axi_arburst (m_axi_arburst),
.m_axi_arlock (m_axi_arlock),
.m_axi_arcache (m_axi_arcache),
.m_axi_arprot (m_axi_arprot),
.m_axi_arqos (m_axi_arqos),
.m_axi_arregion (m_axi_arregion),
.m_axi_aruser (m_axi_aruser),
.m_axi_arvalid (m_axi_arvalid),
.m_axi_arready (m_axi_arready),
.m_axi_rid (m_axi_rid),
.m_axi_rdata (m_axi_rdata),
.m_axi_rresp (m_axi_rresp),
.m_axi_rlast (m_axi_rlast),
.m_axi_ruser (m_axi_ruser),
.m_axi_rvalid (m_axi_rvalid),
.m_axi_rready (m_axi_rready)
);
endmodule

35
axi/tb/axi_cocotb_loopback_test/test_axi4_loopback.py
View File

@ -3,12 +3,7 @@ import itertools
import cocotb
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge, Timer
from cocotbext.axi import AxiMaster, AxiRam
from cocotbext.axi import AxiMaster, AxiRam
from axi4_proxy import axi4_bus_from_packed_if, install_cocotbext_axi_slice_compat
install_cocotbext_axi_slice_compat()
from cocotbext.axi import AxiBus, AxiMaster, AxiRam
class TB:
@ -17,28 +12,21 @@ class TB:
cocotb.start_soon(Clock(dut.clk, 10, units="ns").start())
self.master = AxiMaster(
axi4_bus_from_packed_if(dut.axi_in, "s_axi"),
dut.clk,
dut.rst,
)
self.ram = AxiRam(
axi4_bus_from_packed_if(dut.axi_out, "m_axi"),
dut.clk,
dut.rst,
size=2**20,
)
# Forencich-style connection:
# s_axi: cocotb AXI master -> flat_to_if -> compact AXI interface
# m_axi: compact AXI interface -> if_to_flat -> cocotb AXI RAM
self.master = AxiMaster(AxiBus.from_prefix(
dut, "s_axi"), dut.clk, dut.rst)
self.ram = AxiRam(AxiBus.from_prefix(dut, "m_axi"),
dut.clk, dut.rst, size=2**20)
async def reset(self):
self.dut.rst.setimmediatevalue(0)
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
self.dut.rst.value = 1
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
self.dut.rst.value = 0
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
@ -48,13 +36,6 @@ def cycle_pause():
return itertools.cycle([1, 1, 1, 0])
@cocotb.test()
async def inspect_axi_if(dut):
dut._log.info("axi_in fields: %s", dir(dut.axi_in))
dut._log.info("axi_in.req fields: %s", dir(dut.axi_in.req))
dut._log.info("axi_in.resp fields: %s", dir(dut.axi_in.resp))
@cocotb.test()
async def run_basic_write_read_test(dut):
# simple loopback test..

4
axi/tb/axi_dma_axis_compat/Makefile
View File

@ -44,8 +44,8 @@ VERILOG_SOURCES += $(FORENCICH_AXI_RTL_DIR)/axi_dma_rd.v
VERILOG_SOURCES += $(FORENCICH_AXI_RTL_DIR)/axi_dma_wr.v
VERILOG_SOURCES += $(AXI_IF_RTL_DIR)/axi_dma_if_wrapper.sv
VERILOG_SOURCES += $(AXI_IF_RTL_DIR)/axi_dma_compat.sv
VERILOG_SOURCES += $(AXI_IF_RTL_DIR)/axi_dma_wrapper.sv
VERILOG_SOURCES += $(TB_DIR)/tb_axi_dma_axis_compat.sv
COMPILE_ARGS += -I$(AXI_IF_RTL_DIR)

237
axi/tb/axi_dma_axis_compat/tb_axi_dma_axis_compat.sv
View File

@ -26,18 +26,14 @@ module tb_axi_dma_axis_compat #(
parameter int unsigned ENABLE_SG = 0,
parameter int unsigned ENABLE_UNALIGNED = 0,
parameter int unsigned READ_DESC_WIDTH = AXI_ADDR_WIDTH + LEN_WIDTH + TAG_WIDTH + AXIS_ID_WIDTH + AXIS_DEST_WIDTH + AXIS_USER_WIDTH,
parameter int unsigned WRITE_DESC_WIDTH = AXI_ADDR_WIDTH + LEN_WIDTH + TAG_WIDTH,
parameter int unsigned DESC_CMD_RAW_WIDTH = READ_DESC_WIDTH + WRITE_DESC_WIDTH,
parameter int unsigned DESC_CMD_DATA_WIDTH = ((DESC_CMD_RAW_WIDTH + 7) / 8) * 8,
parameter int unsigned DESC_CMD_KEEP_WIDTH = DESC_CMD_DATA_WIDTH / 8,
parameter int unsigned READ_STATUS_WIDTH = TAG_WIDTH + 4,
parameter int unsigned WRITE_STATUS_WIDTH = LEN_WIDTH + TAG_WIDTH + AXIS_ID_WIDTH + AXIS_DEST_WIDTH + AXIS_USER_WIDTH + 4,
parameter int unsigned DESC_STATUS_RAW_WIDTH = READ_STATUS_WIDTH + WRITE_STATUS_WIDTH,
parameter int unsigned DESC_STATUS_DATA_WIDTH = ((DESC_STATUS_RAW_WIDTH + 7) / 8) * 8,
parameter int unsigned DESC_STATUS_KEEP_WIDTH = DESC_STATUS_DATA_WIDTH / 8,
parameter int unsigned DESC_STATUS_USER_WIDTH = 2
parameter int unsigned READ_DESC_RAW_WIDTH = AXI_ADDR_WIDTH + LEN_WIDTH + TAG_WIDTH,
parameter int unsigned READ_DESC_DATA_WIDTH = ((READ_DESC_RAW_WIDTH + 7) / 8) * 8,
parameter int unsigned WRITE_DESC_RAW_WIDTH = AXI_ADDR_WIDTH + LEN_WIDTH + TAG_WIDTH,
parameter int unsigned WRITE_DESC_DATA_WIDTH = ((WRITE_DESC_RAW_WIDTH + 7) / 8) * 8,
parameter int unsigned READ_STATUS_RAW_WIDTH = TAG_WIDTH + 4,
parameter int unsigned READ_STATUS_DATA_WIDTH = ((READ_STATUS_RAW_WIDTH + 7) / 8) * 8,
parameter int unsigned WRITE_STATUS_RAW_WIDTH = LEN_WIDTH + TAG_WIDTH + 4,
parameter int unsigned WRITE_STATUS_DATA_WIDTH = ((WRITE_STATUS_RAW_WIDTH + 7) / 8) * 8
)(
input logic clk,
input logic rst,
@ -46,27 +42,35 @@ module tb_axi_dma_axis_compat #(
input logic write_enable,
input logic write_abort,
// Unified descriptor command AXIS, flat for cocotb
input logic [DESC_CMD_DATA_WIDTH-1:0] s_axis_desc_cmd_tdata,
input logic [DESC_CMD_KEEP_WIDTH-1:0] s_axis_desc_cmd_tkeep,
input logic [DESC_CMD_KEEP_WIDTH-1:0] s_axis_desc_cmd_tstrb,
input logic s_axis_desc_cmd_tlast,
input logic s_axis_desc_cmd_tid,
input logic s_axis_desc_cmd_tdest,
input logic s_axis_desc_cmd_tuser,
input logic s_axis_desc_cmd_tvalid,
output logic s_axis_desc_cmd_tready,
// Read descriptor AXIS input, flat for cocotb.
// tdata layout: {addr, len, tag}; id/dest/user use AXIS sidebands.
input logic [READ_DESC_DATA_WIDTH-1:0] s_axis_read_desc_tdata,
input logic [AXIS_ID_WIDTH-1:0] s_axis_read_desc_tid,
input logic [AXIS_DEST_WIDTH-1:0] s_axis_read_desc_tdest,
input logic [AXIS_USER_WIDTH-1:0] s_axis_read_desc_tuser,
input logic s_axis_read_desc_tvalid,
output logic s_axis_read_desc_tready,
// Unified descriptor status AXIS, flat for cocotb
output logic [DESC_STATUS_DATA_WIDTH-1:0] m_axis_desc_status_tdata,
output logic [DESC_STATUS_KEEP_WIDTH-1:0] m_axis_desc_status_tkeep,
output logic [DESC_STATUS_KEEP_WIDTH-1:0] m_axis_desc_status_tstrb,
output logic m_axis_desc_status_tlast,
output logic m_axis_desc_status_tid,
output logic m_axis_desc_status_tdest,
output logic [DESC_STATUS_USER_WIDTH-1:0] m_axis_desc_status_tuser,
output logic m_axis_desc_status_tvalid,
input logic m_axis_desc_status_tready,
// Read descriptor status AXIS output.
// tdata layout: {tag, error}.
output logic [READ_STATUS_DATA_WIDTH-1:0] m_axis_read_desc_status_tdata,
output logic m_axis_read_desc_status_tvalid,
input logic m_axis_read_desc_status_tready,
// Write descriptor AXIS input.
// tdata layout: {addr, len, tag}.
input logic [WRITE_DESC_DATA_WIDTH-1:0] s_axis_write_desc_tdata,
input logic s_axis_write_desc_tvalid,
output logic s_axis_write_desc_tready,
// Write descriptor status AXIS output.
// tdata layout: {len, tag, error}; id/dest/user use AXIS sidebands.
output logic [WRITE_STATUS_DATA_WIDTH-1:0] m_axis_write_desc_status_tdata,
output logic [AXIS_ID_WIDTH-1:0] m_axis_write_desc_status_tid,
output logic [AXIS_DEST_WIDTH-1:0] m_axis_write_desc_status_tdest,
output logic [AXIS_USER_WIDTH-1:0] m_axis_write_desc_status_tuser,
output logic m_axis_write_desc_status_tvalid,
input logic m_axis_write_desc_status_tready,
// Read data AXIS output
output logic [AXIS_DATA_WIDTH-1:0] m_axis_read_data_tdata,
@ -144,23 +148,45 @@ module tb_axi_dma_axis_compat #(
wire rstn = ~rst;
axis_if #(
.DATA_W (DESC_CMD_DATA_WIDTH),
.KEEP_W (DESC_CMD_KEEP_WIDTH),
.ID_W (1),
.DEST_W (1),
.USER_W (1)
) desc_cmd_if (
.DATA_W (READ_DESC_DATA_WIDTH),
.KEEP_W (READ_DESC_DATA_WIDTH / 8),
.ID_W (AXIS_ID_WIDTH),
.DEST_W (AXIS_DEST_WIDTH),
.USER_W (AXIS_USER_WIDTH)
) read_desc_if (
.aclk (clk),
.aresetn (rstn)
);
axis_if #(
.DATA_W (DESC_STATUS_DATA_WIDTH),
.KEEP_W (DESC_STATUS_KEEP_WIDTH),
.DATA_W (READ_STATUS_DATA_WIDTH),
.KEEP_W (READ_STATUS_DATA_WIDTH / 8),
.ID_W (1),
.DEST_W (1),
.USER_W (DESC_STATUS_USER_WIDTH)
) desc_status_if (
.USER_W (1)
) read_desc_status_if (
.aclk (clk),
.aresetn (rstn)
);
axis_if #(
.DATA_W (WRITE_DESC_DATA_WIDTH),
.KEEP_W (WRITE_DESC_DATA_WIDTH / 8),
.ID_W (1),
.DEST_W (1),
.USER_W (1)
) write_desc_if (
.aclk (clk),
.aresetn (rstn)
);
axis_if #(
.DATA_W (WRITE_STATUS_DATA_WIDTH),
.KEEP_W (WRITE_STATUS_DATA_WIDTH / 8),
.ID_W (AXIS_ID_WIDTH),
.DEST_W (AXIS_DEST_WIDTH),
.USER_W (AXIS_USER_WIDTH)
) write_desc_status_if (
.aclk (clk),
.aresetn (rstn)
);
@ -198,41 +224,79 @@ module tb_axi_dma_axis_compat #(
);
axis_flat_to_if #(
.DATA_W (DESC_CMD_DATA_WIDTH),
.KEEP_W (DESC_CMD_KEEP_WIDTH),
.ID_W (1),
.DEST_W (1),
.USER_W (1)
) u_desc_cmd_flat_to_if (
.s_axis_tdata (s_axis_desc_cmd_tdata),
.s_axis_tkeep (s_axis_desc_cmd_tkeep),
.s_axis_tstrb (s_axis_desc_cmd_tstrb),
.s_axis_tlast (s_axis_desc_cmd_tlast),
.s_axis_tid (s_axis_desc_cmd_tid),
.s_axis_tdest (s_axis_desc_cmd_tdest),
.s_axis_tuser (s_axis_desc_cmd_tuser),
.s_axis_tvalid (s_axis_desc_cmd_tvalid),
.s_axis_tready (s_axis_desc_cmd_tready),
.m_axis (desc_cmd_if)
.DATA_W (READ_DESC_DATA_WIDTH),
.KEEP_W (READ_DESC_DATA_WIDTH / 8),
.ID_W (AXIS_ID_WIDTH),
.DEST_W (AXIS_DEST_WIDTH),
.USER_W (AXIS_USER_WIDTH)
) u_read_desc_flat_to_if (
.s_axis_tdata (s_axis_read_desc_tdata),
.s_axis_tkeep ({(READ_DESC_DATA_WIDTH/8){1'b1}}),
.s_axis_tstrb ({(READ_DESC_DATA_WIDTH/8){1'b1}}),
.s_axis_tlast (1'b1),
.s_axis_tid (s_axis_read_desc_tid),
.s_axis_tdest (s_axis_read_desc_tdest),
.s_axis_tuser (s_axis_read_desc_tuser),
.s_axis_tvalid (s_axis_read_desc_tvalid),
.s_axis_tready (s_axis_read_desc_tready),
.m_axis (read_desc_if)
);
axis_if_to_flat #(
.DATA_W (DESC_STATUS_DATA_WIDTH),
.KEEP_W (DESC_STATUS_KEEP_WIDTH),
.DATA_W (READ_STATUS_DATA_WIDTH),
.KEEP_W (READ_STATUS_DATA_WIDTH / 8),
.ID_W (1),
.DEST_W (1),
.USER_W (DESC_STATUS_USER_WIDTH)
) u_desc_status_if_to_flat (
.s_axis (desc_status_if),
.m_axis_tdata (m_axis_desc_status_tdata),
.m_axis_tkeep (m_axis_desc_status_tkeep),
.m_axis_tstrb (m_axis_desc_status_tstrb),
.m_axis_tlast (m_axis_desc_status_tlast),
.m_axis_tid (m_axis_desc_status_tid),
.m_axis_tdest (m_axis_desc_status_tdest),
.m_axis_tuser (m_axis_desc_status_tuser),
.m_axis_tvalid (m_axis_desc_status_tvalid),
.m_axis_tready (m_axis_desc_status_tready)
.USER_W (1)
) u_read_desc_status_if_to_flat (
.s_axis (read_desc_status_if),
.m_axis_tdata (m_axis_read_desc_status_tdata),
.m_axis_tkeep (),
.m_axis_tstrb (),
.m_axis_tlast (),
.m_axis_tid (),
.m_axis_tdest (),
.m_axis_tuser (),
.m_axis_tvalid (m_axis_read_desc_status_tvalid),
.m_axis_tready (m_axis_read_desc_status_tready)
);
axis_flat_to_if #(
.DATA_W (WRITE_DESC_DATA_WIDTH),
.KEEP_W (WRITE_DESC_DATA_WIDTH / 8),
.ID_W (1),
.DEST_W (1),
.USER_W (1)
) u_write_desc_flat_to_if (
.s_axis_tdata (s_axis_write_desc_tdata),
.s_axis_tkeep ({(WRITE_DESC_DATA_WIDTH/8){1'b1}}),
.s_axis_tstrb ({(WRITE_DESC_DATA_WIDTH/8){1'b1}}),
.s_axis_tlast (1'b1),
.s_axis_tid (1'b0),
.s_axis_tdest (1'b0),
.s_axis_tuser (1'b0),
.s_axis_tvalid (s_axis_write_desc_tvalid),
.s_axis_tready (s_axis_write_desc_tready),
.m_axis (write_desc_if)
);
axis_if_to_flat #(
.DATA_W (WRITE_STATUS_DATA_WIDTH),
.KEEP_W (WRITE_STATUS_DATA_WIDTH / 8),
.ID_W (AXIS_ID_WIDTH),
.DEST_W (AXIS_DEST_WIDTH),
.USER_W (AXIS_USER_WIDTH)
) u_write_desc_status_if_to_flat (
.s_axis (write_desc_status_if),
.m_axis_tdata (m_axis_write_desc_status_tdata),
.m_axis_tkeep (),
.m_axis_tstrb (),
.m_axis_tlast (),
.m_axis_tid (m_axis_write_desc_status_tid),
.m_axis_tdest (m_axis_write_desc_status_tdest),
.m_axis_tuser (m_axis_write_desc_status_tuser),
.m_axis_tvalid (m_axis_write_desc_status_tvalid),
.m_axis_tready (m_axis_write_desc_status_tready)
);
axis_if_to_flat #(
@ -273,7 +337,7 @@ module tb_axi_dma_axis_compat #(
.m_axis (write_data_if)
);
axi_dma_axis_compat #(
axi_dma_wrapper #(
.AXI_DATA_WIDTH (AXI_DATA_WIDTH),
.AXI_ADDR_WIDTH (AXI_ADDR_WIDTH),
.AXI_STRB_WIDTH (AXI_STRB_WIDTH),
@ -293,21 +357,20 @@ module tb_axi_dma_axis_compat #(
.LEN_WIDTH (LEN_WIDTH),
.TAG_WIDTH (TAG_WIDTH),
.ENABLE_SG (ENABLE_SG),
.ENABLE_UNALIGNED (ENABLE_UNALIGNED),
.DESC_CMD_DATA_WIDTH (DESC_CMD_DATA_WIDTH),
.DESC_STATUS_DATA_WIDTH (DESC_STATUS_DATA_WIDTH),
.DESC_STATUS_USER_WIDTH (DESC_STATUS_USER_WIDTH)
.ENABLE_UNALIGNED (ENABLE_UNALIGNED)
) u_dut (
.clk (clk),
.rst (rst),
.s_axis_desc_cmd (desc_cmd_if),
.m_axis_desc_status (desc_status_if),
.m_axis_read_data (read_data_if),
.s_axis_write_data (write_data_if),
.m_axi (m_axi_if),
.read_enable (read_enable),
.write_enable (write_enable),
.write_abort (write_abort)
.clk (clk),
.rst (rst),
.s_axis_read_desc (read_desc_if),
.m_axis_read_desc_status (read_desc_status_if),
.m_axis_read_data (read_data_if),
.s_axis_write_desc (write_desc_if),
.m_axis_write_desc_status (write_desc_status_if),
.s_axis_write_data (write_data_if),
.m_axi (m_axi_if),
.read_enable (read_enable),
.write_enable (write_enable),
.write_abort (write_abort)
);
axi4_if_to_flat #(

305
axi/tb/axi_dma_axis_compat/test_axi_dma_axis_compat.py
View File

@ -3,7 +3,7 @@ import os
import cocotb
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge, Combine
from cocotb.triggers import RisingEdge
from cocotbext.axi import AxiBus, AxiRam
from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink
@ -16,42 +16,33 @@ AXIS_ID_WIDTH = int(os.getenv("PARAM_AXIS_ID_WIDTH", "8"))
AXIS_DEST_WIDTH = int(os.getenv("PARAM_AXIS_DEST_WIDTH", "8"))
AXIS_USER_WIDTH = int(os.getenv("PARAM_AXIS_USER_WIDTH", "1"))
READ_DESC_WIDTH = AXI_ADDR_WIDTH + LEN_WIDTH + TAG_WIDTH + \
AXIS_ID_WIDTH + AXIS_DEST_WIDTH + AXIS_USER_WIDTH
WRITE_DESC_WIDTH = AXI_ADDR_WIDTH + LEN_WIDTH + TAG_WIDTH
DESC_CMD_RAW_WIDTH = READ_DESC_WIDTH + WRITE_DESC_WIDTH
DESC_CMD_DATA_WIDTH = ((DESC_CMD_RAW_WIDTH + 7) // 8) * 8
DESC_CMD_BYTES = DESC_CMD_DATA_WIDTH // 8
READ_DESC_RAW_WIDTH = AXI_ADDR_WIDTH + LEN_WIDTH + TAG_WIDTH
READ_DESC_DATA_WIDTH = ((READ_DESC_RAW_WIDTH + 7) // 8) * 8
READ_DESC_BYTES = READ_DESC_DATA_WIDTH // 8
READ_STATUS_WIDTH = TAG_WIDTH + 4
WRITE_STATUS_WIDTH = LEN_WIDTH + TAG_WIDTH + \
AXIS_ID_WIDTH + AXIS_DEST_WIDTH + AXIS_USER_WIDTH + 4
DESC_STATUS_RAW_WIDTH = READ_STATUS_WIDTH + WRITE_STATUS_WIDTH
DESC_STATUS_DATA_WIDTH = ((DESC_STATUS_RAW_WIDTH + 7) // 8) * 8
DESC_STATUS_BYTES = DESC_STATUS_DATA_WIDTH // 8
WRITE_DESC_RAW_WIDTH = AXI_ADDR_WIDTH + LEN_WIDTH + TAG_WIDTH
WRITE_DESC_DATA_WIDTH = ((WRITE_DESC_RAW_WIDTH + 7) // 8) * 8
WRITE_DESC_BYTES = WRITE_DESC_DATA_WIDTH // 8
RD_ADDR_LSB = 0
RD_LEN_LSB = RD_ADDR_LSB + AXI_ADDR_WIDTH
RD_TAG_LSB = RD_LEN_LSB + LEN_WIDTH
RD_ID_LSB = RD_TAG_LSB + TAG_WIDTH
RD_DEST_LSB = RD_ID_LSB + AXIS_ID_WIDTH
RD_USER_LSB = RD_DEST_LSB + AXIS_DEST_WIDTH
READ_STATUS_RAW_WIDTH = TAG_WIDTH + 4
READ_STATUS_DATA_WIDTH = ((READ_STATUS_RAW_WIDTH + 7) // 8) * 8
WR_BASE_LSB = READ_DESC_WIDTH
WR_ADDR_LSB = WR_BASE_LSB
WR_LEN_LSB = WR_ADDR_LSB + AXI_ADDR_WIDTH
WR_TAG_LSB = WR_LEN_LSB + LEN_WIDTH
WRITE_STATUS_RAW_WIDTH = LEN_WIDTH + TAG_WIDTH + 4
WRITE_STATUS_DATA_WIDTH = ((WRITE_STATUS_RAW_WIDTH + 7) // 8) * 8
RD_STS_TAG_LSB = 0
RD_STS_ERROR_LSB = RD_STS_TAG_LSB + TAG_WIDTH
# axi_dma_wrapper maps descriptor tdata as {addr, len, tag}.
DESC_TAG_LSB = 0
DESC_LEN_LSB = DESC_TAG_LSB + TAG_WIDTH
DESC_ADDR_LSB = DESC_LEN_LSB + LEN_WIDTH
WR_STS_BASE_LSB = READ_STATUS_WIDTH
WR_STS_LEN_LSB = WR_STS_BASE_LSB
WR_STS_TAG_LSB = WR_STS_LEN_LSB + LEN_WIDTH
WR_STS_ID_LSB = WR_STS_TAG_LSB + TAG_WIDTH
WR_STS_DEST_LSB = WR_STS_ID_LSB + AXIS_ID_WIDTH
WR_STS_USER_LSB = WR_STS_DEST_LSB + AXIS_DEST_WIDTH
WR_STS_ERROR_LSB = WR_STS_USER_LSB + AXIS_USER_WIDTH
# Read status tdata is {tag, error}.
READ_STATUS_ERROR_LSB = 0
READ_STATUS_TAG_LSB = READ_STATUS_ERROR_LSB + 4
# Write status tdata is {len, tag, error}.
WRITE_STATUS_ERROR_LSB = 0
WRITE_STATUS_TAG_LSB = WRITE_STATUS_ERROR_LSB + 4
WRITE_STATUS_LEN_LSB = WRITE_STATUS_TAG_LSB + TAG_WIDTH
def mask(width):
@ -68,59 +59,60 @@ def get(value, lsb, width):
return (int(value) >> lsb) & mask(width)
def pack_desc_cmd(
*,
read_addr=0,
read_len=0,
read_tag=0,
read_id=0,
read_dest=0,
read_user=0,
write_addr=0,
write_len=0,
write_tag=0,
):
def pack_read_desc(*, addr, length, tag, axis_id=0, dest=0, user=0):
value = 0
value = put(value, RD_ADDR_LSB, AXI_ADDR_WIDTH, read_addr)
value = put(value, RD_LEN_LSB, LEN_WIDTH, read_len)
value = put(value, RD_TAG_LSB, TAG_WIDTH, read_tag)
value = put(value, RD_ID_LSB, AXIS_ID_WIDTH, read_id)
value = put(value, RD_DEST_LSB, AXIS_DEST_WIDTH, read_dest)
value = put(value, RD_USER_LSB, AXIS_USER_WIDTH, read_user)
value = put(value, DESC_TAG_LSB, TAG_WIDTH, tag)
value = put(value, DESC_LEN_LSB, LEN_WIDTH, length)
value = put(value, DESC_ADDR_LSB, AXI_ADDR_WIDTH, addr)
value = put(value, WR_ADDR_LSB, AXI_ADDR_WIDTH, write_addr)
value = put(value, WR_LEN_LSB, LEN_WIDTH, write_len)
value = put(value, WR_TAG_LSB, TAG_WIDTH, write_tag)
return AxiStreamFrame(
value.to_bytes(READ_DESC_BYTES, byteorder="little"),
tid=axis_id,
tdest=dest,
tuser=user,
)
return value.to_bytes(DESC_CMD_BYTES, byteorder="little")
def pack_write_desc(*, addr, length, tag):
value = 0
value = put(value, DESC_TAG_LSB, TAG_WIDTH, tag)
value = put(value, DESC_LEN_LSB, LEN_WIDTH, length)
value = put(value, DESC_ADDR_LSB, AXI_ADDR_WIDTH, addr)
return AxiStreamFrame(
value.to_bytes(WRITE_DESC_BYTES, byteorder="little")
)
def frame_to_int(frame):
return int.from_bytes(bytes(frame.tdata), byteorder="little")
def frame_tuser(frame):
def frame_sideband(frame, name):
value = getattr(frame, name)
try:
return int(frame.tuser)
return int(value)
except TypeError:
return int(frame.tuser[0]) if frame.tuser else 0
return int(value[0]) if value else 0
def unpack_status(frame):
def unpack_read_status(frame):
value = frame_to_int(frame)
tuser = frame_tuser(frame)
return {
"read_valid": bool(tuser & 0x1),
"write_valid": bool(tuser & 0x2),
"read_tag": get(value, RD_STS_TAG_LSB, TAG_WIDTH),
"read_error": get(value, RD_STS_ERROR_LSB, 4),
"write_len": get(value, WR_STS_LEN_LSB, LEN_WIDTH),
"write_tag": get(value, WR_STS_TAG_LSB, TAG_WIDTH),
"write_id": get(value, WR_STS_ID_LSB, AXIS_ID_WIDTH),
"write_dest": get(value, WR_STS_DEST_LSB, AXIS_DEST_WIDTH),
"write_user": get(value, WR_STS_USER_LSB, AXIS_USER_WIDTH),
"write_error": get(value, WR_STS_ERROR_LSB, 4),
"tag": get(value, READ_STATUS_TAG_LSB, TAG_WIDTH),
"error": get(value, READ_STATUS_ERROR_LSB, 4),
}
def unpack_write_status(frame):
value = frame_to_int(frame)
return {
"len": get(value, WRITE_STATUS_LEN_LSB, LEN_WIDTH),
"tag": get(value, WRITE_STATUS_TAG_LSB, TAG_WIDTH),
"error": get(value, WRITE_STATUS_ERROR_LSB, 4),
"id": frame_sideband(frame, "tid"),
"dest": frame_sideband(frame, "tdest"),
"user": frame_sideband(frame, "tuser"),
}
@ -132,24 +124,45 @@ class TB:
cocotb.start_soon(Clock(dut.clk, 10, units="ns").start())
self.desc_cmd_source = AxiStreamSource(
AxiStreamBus.from_prefix(dut, "s_axis_desc_cmd"), dut.clk, dut.rst
self.read_desc_source = AxiStreamSource(
AxiStreamBus.from_prefix(dut, "s_axis_read_desc"),
dut.clk,
dut.rst,
)
self.desc_status_sink = AxiStreamSink(
AxiStreamBus.from_prefix(
dut, "m_axis_desc_status"), dut.clk, dut.rst
self.read_desc_status_sink = AxiStreamSink(
AxiStreamBus.from_prefix(dut, "m_axis_read_desc_status"),
dut.clk,
dut.rst,
)
self.write_desc_source = AxiStreamSource(
AxiStreamBus.from_prefix(dut, "s_axis_write_desc"),
dut.clk,
dut.rst,
)
self.write_desc_status_sink = AxiStreamSink(
AxiStreamBus.from_prefix(dut, "m_axis_write_desc_status"),
dut.clk,
dut.rst,
)
self.read_data_sink = AxiStreamSink(
AxiStreamBus.from_prefix(dut, "m_axis_read_data"), dut.clk, dut.rst
AxiStreamBus.from_prefix(dut, "m_axis_read_data"),
dut.clk,
dut.rst,
)
self.write_data_source = AxiStreamSource(
AxiStreamBus.from_prefix(
dut, "s_axis_write_data"), dut.clk, dut.rst
AxiStreamBus.from_prefix(dut, "s_axis_write_data"),
dut.clk,
dut.rst,
)
self.axi_ram = AxiRam(AxiBus.from_prefix(
dut, "m_axi"), dut.clk, dut.rst, size=2**16)
self.axi_ram = AxiRam(
AxiBus.from_prefix(dut, "m_axi"),
dut.clk,
dut.rst,
size=2**AXI_ADDR_WIDTH,
)
dut.read_enable.setimmediatevalue(0)
dut.write_enable.setimmediatevalue(0)
@ -166,8 +179,11 @@ class TB:
await RisingEdge(self.dut.clk)
await RisingEdge(self.dut.clk)
async def send_desc_cmd(self, payload):
await self.desc_cmd_source.send(AxiStreamFrame(payload))
async def send_read_desc(self, **kwargs):
await self.read_desc_source.send(pack_read_desc(**kwargs))
async def send_write_desc(self, **kwargs):
await self.write_desc_source.send(pack_write_desc(**kwargs))
@cocotb.test()
@ -184,28 +200,26 @@ async def test_axis_compat_write_path(dut):
tb.axi_ram.write(addr - 16, b"\xaa" * (len(data) + 32))
cmd = pack_desc_cmd(write_addr=addr, write_len=len(data), write_tag=tag)
cmd_task = cocotb.start_soon(tb.send_desc_cmd(cmd))
desc_task = cocotb.start_soon(
tb.send_write_desc(addr=addr, length=len(data), tag=tag)
)
data_task = cocotb.start_soon(
tb.write_data_source.send(AxiStreamFrame(data, tid=tag))
)
await cmd_task
await desc_task
await data_task
status_frame = await tb.desc_status_sink.recv()
status = unpack_status(status_frame)
status = unpack_write_status(await tb.write_desc_status_sink.recv())
assert not status["read_valid"]
assert status["write_valid"]
assert status["write_tag"] == tag
assert status["write_len"] == len(data)
assert status["write_id"] == tag
assert status["write_error"] == 0
assert status["tag"] == tag
assert status["len"] == len(data)
assert status["id"] == tag
assert status["error"] == 0
assert tb.axi_ram.read(addr - 8, len(data) +
16) == b"\xaa" * 8 + data + b"\xaa" * 8
assert tb.axi_ram.read(addr - 8, len(data) + 16) == (
b"\xaa" * 8 + data + b"\xaa" * 8
)
@cocotb.test()
@ -222,32 +236,26 @@ async def test_axis_compat_read_path(dut):
tb.axi_ram.write(addr, data)
cmd = pack_desc_cmd(
read_addr=addr,
read_len=len(data),
read_tag=tag,
read_id=tag,
read_dest=0,
read_user=0,
await tb.send_read_desc(
addr=addr,
length=len(data),
tag=tag,
axis_id=tag,
dest=0,
user=0,
)
await tb.send_desc_cmd(cmd)
data_frame = await tb.read_data_sink.recv()
assert bytes(data_frame.tdata) == data
assert int(data_frame.tid) == tag
assert frame_sideband(data_frame, "tid") == tag
status_frame = await tb.desc_status_sink.recv()
status = unpack_status(status_frame)
assert status["read_valid"]
assert not status["write_valid"]
assert status["read_tag"] == tag
assert status["read_error"] == 0
status = unpack_read_status(await tb.read_desc_status_sink.recv())
assert status["tag"] == tag
assert status["error"] == 0
@cocotb.test()
async def test_axis_compat_read_and_write_same_command(dut):
async def test_axis_compat_read_and_write_independent_desc_ports(dut):
tb = TB(dut)
await tb.reset()
@ -264,51 +272,46 @@ async def test_axis_compat_read_and_write_same_command(dut):
tb.axi_ram.write(read_addr, read_data)
tb.axi_ram.write(write_addr - 8, b"\xaa" * (len(write_data) + 16))
cmd = pack_desc_cmd(
read_addr=read_addr,
read_len=len(read_data),
read_tag=read_tag,
read_id=read_tag,
write_addr=write_addr,
write_len=len(write_data),
write_tag=write_tag,
read_desc_task = cocotb.start_soon(
tb.send_read_desc(
addr=read_addr,
length=len(read_data),
tag=read_tag,
axis_id=read_tag,
dest=0,
user=0,
)
)
cmd_task = cocotb.start_soon(tb.send_desc_cmd(cmd))
data_task = cocotb.start_soon(
write_desc_task = cocotb.start_soon(
tb.send_write_desc(
addr=write_addr,
length=len(write_data),
tag=write_tag,
)
)
write_data_task = cocotb.start_soon(
tb.write_data_source.send(AxiStreamFrame(write_data, tid=write_tag))
)
await cmd_task
await data_task
await read_desc_task
await write_desc_task
await write_data_task
data_frame = await tb.read_data_sink.recv()
assert bytes(data_frame.tdata) == read_data
assert int(data_frame.tid) == read_tag
assert frame_sideband(data_frame, "tid") == read_tag
# status might be in 2 beats, depending on timings
seen_read = False
seen_write = False
read_status = unpack_read_status(await tb.read_desc_status_sink.recv())
write_status = unpack_write_status(await tb.write_desc_status_sink.recv())
for _ in range(2):
status_frame = await tb.desc_status_sink.recv()
status = unpack_status(status_frame)
assert read_status["tag"] == read_tag
assert read_status["error"] == 0
if status["read_valid"]:
assert status["read_tag"] == read_tag
assert status["read_error"] == 0
seen_read = True
assert write_status["tag"] == write_tag
assert write_status["len"] == len(write_data)
assert write_status["id"] == write_tag
assert write_status["error"] == 0
if status["write_valid"]:
assert status["write_tag"] == write_tag
assert status["write_len"] == len(write_data)
assert status["write_error"] == 0
seen_write = True
if seen_read and seen_write:
break
assert seen_read
assert seen_write
assert tb.axi_ram.read(write_addr - 8, len(write_data) +
16) == b"\xaa" * 8 + write_data + b"\xaa" * 8
assert tb.axi_ram.read(write_addr - 8, len(write_data) + 16) == (
b"\xaa" * 8 + write_data + b"\xaa" * 8
)

4
axi/tb/axil_reg_cocotb/tb_axi4l_reg_map.sv
View File

@ -92,6 +92,7 @@ module tb_axi4l_reg_map;
logic [N_REGS-1:0][31:0] reg_i;
logic [N_REGS-1:0][31:0] reg_o;
logic [N_REGS-1:0][31:0] reg_pulse;
// debug probes visible from cocotb. they are useful for checking W1S pulse-like behavior
logic [31:0] reg0_o;
@ -157,7 +158,8 @@ module tb_axi4l_reg_map;
.rst_n (rst_n),
.s_axil (s_axil_if.slave),
.reg_i (reg_i),
.reg_o (reg_o)
.reg_o (reg_o),
.reg_pulse(reg_pulse)
);
endmodule

182
axi/tb/axil_reg_cocotb/test_axi4l_reg_map.py
View File

@ -1,14 +1,12 @@
# SPDX-License-Identifier: MIT
import cocotb
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge
from cocotb.triggers import ReadOnly, RisingEdge
from cocotbext.axi import AxiLiteBus, AxiLiteMaster
OKAY = 0
SLVERR = 2
REG0_CTRL = 0x00
REG0_CTRL = 0x00
REG1_STATUS = 0x04
REG2_CONFIG = 0x08
REG_INVALID = 0x0C # N_REGS=3, so index 3 is bad
@ -35,7 +33,8 @@ def _read_data(resp):
class TB:
def __init__(self, dut):
self.dut = dut
self.axil = AxiLiteMaster(AxiLiteBus.from_prefix(dut, "s_axil"), dut.clk, dut.rst)
self.axil = AxiLiteMaster(AxiLiteBus.from_prefix(
dut, "s_axil"), dut.clk, dut.rst)
async def reset(self):
self.dut.rst.value = 1
@ -114,28 +113,175 @@ async def test_rw_and_w1c_bits(dut):
assert await tb.read32(REG2_CONFIG) == 0xDEADBEEF
def _reg0_pulse_value(dut):
"""Return REG0 pulse bits regardless of how the test wrapper exposes them."""
if hasattr(dut, "reg0_pulse"):
return int(dut.reg0_pulse.value)
if hasattr(dut, "reg_pulse"):
# Packed SystemVerilog array [N_REGS-1:0][31:0]: REG0 occupies bits 31:0.
return int(dut.reg_pulse.value) & 0xFFFF_FFFF
raise AssertionError(
"DUT must expose either reg0_pulse[31:0] or packed reg_pulse"
)
async def _sample_reg0_pulse_cycles(dut, cycles):
"""Sample REG0 pulse output after each active clock edge."""
samples = []
for _ in range(cycles):
await RisingEdge(dut.clk)
await ReadOnly()
samples.append(_reg0_pulse_value(dut))
return samples
def _assert_single_cycle_bit_pulse(samples, bit, name):
"""Check that one selected bit was high for exactly one sampled cycle."""
high_cycles = [
index for index, value in enumerate(samples)
if value & (1 << bit)
]
assert len(high_cycles) == 1, (
f"{name}: expected one high cycle, got {len(high_cycles)}; "
f"samples={[f'0x{value:08x}' for value in samples]}"
)
return high_cycles[0]
@cocotb.test()
async def test_w1s_seen_on_reg_o_probe(dut):
"""Check that a W1S write creates a short-lived internal reg_o bit."""
async def test_reg_pulse_is_zero_after_reset(dut):
"""No W1 event may be reported after reset without a write."""
cocotb.start_soon(Clock(dut.clk, 10, units="ns").start())
tb = TB(dut)
await tb.reset()
seen = False
samples = await _sample_reg0_pulse_cycles(dut, 4)
assert not any(samples), (
"reg_pulse was asserted without a W1 write: "
f"{[f'0x{value:08x}' for value in samples]}"
)
async def monitor_w1s_bit():
nonlocal seen
for _ in range(20):
await RisingEdge(dut.clk)
if int(dut.reg0_o.value) & 0x1:
seen = True
mon = cocotb.start_soon(monitor_w1s_bit())
@cocotb.test()
async def test_w1s_generates_single_cycle_reg_pulse(dut):
"""Writing 1 to REG0[0] W1S must pulse REG0 pulse bit 0 once."""
cocotb.start_soon(Clock(dut.clk, 10, units="ns").start())
tb = TB(dut)
await tb.reset()
monitor = cocotb.start_soon(_sample_reg0_pulse_cycles(dut, 20))
await tb.write32(REG0_CTRL, 0x00000001)
await mon
samples = await monitor
assert seen, "W1S bit was never observed on reg0_o[0]"
assert await tb.read32(REG0_CTRL) == 0x00000004 # W1S reads as 0, W1C reset bit remains set
_assert_single_cycle_bit_pulse(samples, bit=0, name="reg_pulse[0][0]")
# No unrelated W1 bit may pulse.
assert not any(value & (1 << 2) for value in samples)
# W1S is a command bit and is not returned by reads.
assert await tb.read32(REG0_CTRL) == 0x00000004
@cocotb.test()
async def test_w1c_generates_single_cycle_reg_pulse(dut):
"""Writing 1 to REG0[2] W1C must pulse bit 2 once and clear state."""
cocotb.start_soon(Clock(dut.clk, 10, units="ns").start())
tb = TB(dut)
await tb.reset()
assert await tb.read32(REG0_CTRL) == 0x00000004
monitor = cocotb.start_soon(_sample_reg0_pulse_cycles(dut, 20))
await tb.write32(REG0_CTRL, 0x00000004)
samples = await monitor
_assert_single_cycle_bit_pulse(samples, bit=2, name="reg_pulse[0][2]")
# No unrelated W1 bit may pulse.
assert not any(value & (1 << 0) for value in samples)
assert await tb.read32(REG0_CTRL) == 0x00000000
@cocotb.test()
async def test_zero_to_w1_and_rw_write_do_not_generate_reg_pulse(dut):
"""Only written ones in W1 fields may create reg_pulse."""
cocotb.start_soon(Clock(dut.clk, 10, units="ns").start())
tb = TB(dut)
await tb.reset()
monitor = cocotb.start_soon(_sample_reg0_pulse_cycles(dut, 20))
# bit1 is RW; both W1 fields receive zero.
await tb.write32(REG0_CTRL, 0x00000002)
samples = await monitor
assert not any(samples), (
"RW write or zero written to W1 fields generated reg_pulse: "
f"{[f'0x{value:08x}' for value in samples]}"
)
assert await tb.read32(REG0_CTRL) == 0x00000006
@cocotb.test()
async def test_w1s_and_w1c_pulse_together(dut):
"""W1S and W1C ones in one AXI write must pulse on the same cycle."""
cocotb.start_soon(Clock(dut.clk, 10, units="ns").start())
tb = TB(dut)
await tb.reset()
monitor = cocotb.start_soon(_sample_reg0_pulse_cycles(dut, 20))
await tb.write32(REG0_CTRL, 0x00000005)
samples = await monitor
w1s_cycle = _assert_single_cycle_bit_pulse(
samples, bit=0, name="reg_pulse[0][0]"
)
w1c_cycle = _assert_single_cycle_bit_pulse(
samples, bit=2, name="reg_pulse[0][2]"
)
assert w1s_cycle == w1c_cycle, (
"W1 bits written by one transaction pulsed on different cycles: "
f"W1S={w1s_cycle}, W1C={w1c_cycle}"
)
# Only bits 0 and 2 are allowed to pulse.
assert samples[w1s_cycle] == 0x00000005
assert await tb.read32(REG0_CTRL) == 0x00000000
@cocotb.test()
async def test_each_w1_write_creates_a_new_pulse(dut):
"""Two separate W1 writes must create two separate one-cycle pulses."""
cocotb.start_soon(Clock(dut.clk, 10, units="ns").start())
tb = TB(dut)
await tb.reset()
monitor = cocotb.start_soon(_sample_reg0_pulse_cycles(dut, 40))
await tb.write32(REG0_CTRL, 0x00000001)
for _ in range(3):
await RisingEdge(dut.clk)
await tb.write32(REG0_CTRL, 0x00000001)
samples = await monitor
high_cycles = [
index for index, value in enumerate(samples)
if value & 0x1
]
assert len(high_cycles) == 2, (
f"expected two W1S pulses, got cycles {high_cycles}; "
f"samples={[f'0x{value:08x}' for value in samples]}"
)
assert high_cycles[1] > high_cycles[0] + 1, (
f"separate writes did not produce separate pulses: {high_cycles}"
)
@cocotb.test()