bslathi19/taxi
1
0
Fork 0
mirror of https://github.com/fpganinja/taxi.git synced 2026-05-17 21:10:55 -07:00
Code Issues Packages Projects Releases Wiki Activity

axis: Add AXI stream padding module and testbench

Browse Source 
Signed-off-by: Alex Forencich <alex@alexforencich.com>
This commit is contained in:
Alex Forencich
2026-05-11 15:22:56 -07:00
parent 2063e47a4e
commit e608e8b73e
5 changed files with 1033 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@@ -115,6 +115,7 @@ The Taxi transport library contains many smaller components that can be composed
* Demultiplexer * Demultiplexer
* Broadcaster * Broadcaster
* Concatenator * Concatenator
* Pading inserter
* Switch * Switch
* COBS encoder * COBS encoder
* COBS decoder * COBS decoder

302
src/axis/rtl/taxi_axis_pad.sv Normal file
View File

@@ -0,0 +1,302 @@
// SPDX-License-Identifier: CERN-OHL-S-2.0
/*
Copyright (c) 2025-2026 FPGA Ninja, LLC
Authors:
- Alex Forencich
*/
`resetall
`timescale 1ns / 1ps
`default_nettype none
/*
* AXI4-Stream padding logic
*/
module taxi_axis_pad #(
parameter logic ID_PAD_REG_EN = 1'b1,
parameter logic DEST_PAD_REG_EN = 1'b1,
parameter logic USER_PAD_REG_EN = 1'b1,
parameter MIN_LEN_W = 8,
parameter logic UNDERFLOW_DROP_EN = 1'b0
)
(
input wire logic clk,
input wire logic rst,
/*
* AXI4-Stream input (sink)
*/
taxi_axis_if.snk s_axis,
/*
* AXI4-Stream output (source)
*/
taxi_axis_if.src m_axis,
/*
* Configuration
*/
input wire logic cfg_pad_en = 1'b1,
input wire logic [MIN_LEN_W-1:0] cfg_min_pkt_len,
/*
* Status
*/
output wire logic stat_pad_frame,
output wire logic stat_err_user,
output wire logic stat_err_underflow
);
// extract parameters
localparam DATA_W = s_axis.DATA_W;
localparam logic KEEP_EN = s_axis.KEEP_EN && m_axis.KEEP_EN;
localparam KEEP_W = s_axis.KEEP_W;
localparam logic STRB_EN = s_axis.STRB_EN && m_axis.STRB_EN;
localparam logic LAST_EN = s_axis.LAST_EN && m_axis.LAST_EN;
localparam logic ID_EN = s_axis.ID_EN && m_axis.ID_EN;
localparam ID_W = s_axis.ID_W;
localparam logic DEST_EN = s_axis.DEST_EN && m_axis.DEST_EN;
localparam DEST_W = s_axis.DEST_W;
localparam logic USER_EN = s_axis.USER_EN && m_axis.USER_EN;
localparam USER_W = s_axis.USER_W;
localparam KEEP_W_INT = KEEP_EN ? KEEP_W : 1;
localparam BYTE_LANES = KEEP_W_INT;
localparam BYTE_SIZE = DATA_W/BYTE_LANES;
localparam PAD_LANE_W = KEEP_EN && KEEP_W > 1 ? $clog2(BYTE_LANES) : 1;
localparam PAD_CYC_W = KEEP_EN && KEEP_W > 1 ? MIN_LEN_W-PAD_LANE_W : MIN_LEN_W;
// check configuration
if (m_axis.DATA_W != DATA_W)
$fatal(0, "Error: Interface DATA_W parameter mismatch (instance %m)");
if (KEEP_EN && m_axis.KEEP_W != KEEP_W)
$fatal(0, "Error: Interface KEEP_W parameter mismatch (instance %m)");
if (BYTE_SIZE * BYTE_LANES != DATA_W)
$fatal(0, "Error: input data width not evenly divisible (instance %m)");
if (2**$clog2(BYTE_LANES) != BYTE_LANES)
$fatal(0, "Error: byte lane count be even power of two (instance %m)");
if (!LAST_EN)
$fatal(0, "Error: LAST_EN must be set (instance %m)");
wire [PAD_LANE_W-1:0] min_pkt_len_lanes;
wire [PAD_CYC_W-1:0] min_pkt_len_cycles;
if (KEEP_EN && KEEP_W > 1) begin
assign {min_pkt_len_cycles, min_pkt_len_lanes} = cfg_min_pkt_len;
end else begin
assign min_pkt_len_lanes = '0;
assign min_pkt_len_cycles = cfg_min_pkt_len;
end
logic frame_in_reg = 1'b0, frame_in_next;
logic frame_out_reg = 1'b0, frame_out_next;
logic drop_frame_reg = 1'b0, drop_frame_next;
logic drop_pad_reg = 1'b0, drop_pad_next;
logic [PAD_CYC_W-1:0] pad_cyc_reg = '0, pad_cyc_next;
logic pad_reg = 1'b0, pad_next;
logic extend_reg = 1'b0, extend_next;
logic pad_done_reg = 1'b0, pad_done_next;
logic [KEEP_W-1:0] pad_mask_reg = '1, pad_mask_next;
logic [ID_W-1:0] pad_id_reg = '0, pad_id_next;
logic [DEST_W-1:0] pad_dest_reg = '0, pad_dest_next;
logic [USER_W-1:0] pad_user_reg = '0, pad_user_next;
logic stat_pad_frame_reg = 1'b0, stat_pad_frame_next;
logic stat_err_user_reg = 1'b0, stat_err_user_next;
logic stat_err_underflow_reg = 1'b0, stat_err_underflow_next;
// Mask input data
for (genvar n = 0; n < BYTE_LANES; n = n + 1) begin
assign m_axis.tdata[n*BYTE_SIZE +: BYTE_SIZE] = (s_axis.tkeep[n] && !extend_reg) ? s_axis.tdata[n*BYTE_SIZE +: BYTE_SIZE] : '0;
end
if (KEEP_EN) begin
assign m_axis.tkeep = (extend_reg ? '0 : (UNDERFLOW_DROP_EN && !s_axis.tvalid ? '1 : s_axis.tkeep)) | pad_mask_reg;
end else begin
assign m_axis.tkeep = '1;
end
assign m_axis.tstrb = m_axis.tkeep;
assign m_axis.tvalid = UNDERFLOW_DROP_EN ? ((s_axis.tvalid && !drop_frame_reg) || frame_out_reg) : s_axis.tvalid || extend_reg;
assign m_axis.tlast = (pad_reg || (UNDERFLOW_DROP_EN && !s_axis.tvalid)) ? pad_done_reg : (s_axis.tlast || (UNDERFLOW_DROP_EN && drop_pad_reg));
if (ID_EN) begin
assign m_axis.tid = ((extend_reg || (UNDERFLOW_DROP_EN && !s_axis.tvalid)) && ID_PAD_REG_EN) ? pad_id_reg : s_axis.tid;
end else begin
assign m_axis.tid = '0;
end
if (DEST_EN) begin
assign m_axis.tdest = ((extend_reg || (UNDERFLOW_DROP_EN && !s_axis.tvalid)) && DEST_PAD_REG_EN) ? pad_dest_reg : s_axis.tdest;
end else begin
assign m_axis.tdest = '0;
end
if (USER_EN) begin
assign m_axis.tuser = (((extend_reg || (UNDERFLOW_DROP_EN && !s_axis.tvalid)) && USER_PAD_REG_EN) ? pad_user_reg : s_axis.tuser) | USER_W'(UNDERFLOW_DROP_EN && (drop_pad_reg || (!s_axis.tvalid && !extend_reg)));
end else begin
assign m_axis.tuser = '0;
end
assign s_axis.tready = (m_axis.tready && !extend_reg) || (UNDERFLOW_DROP_EN && drop_frame_reg);
assign stat_pad_frame = stat_pad_frame_reg;
assign stat_err_user = stat_err_user_reg;
assign stat_err_underflow = UNDERFLOW_DROP_EN ? stat_err_underflow_reg : 1'b0;
always_comb begin
frame_in_next = frame_in_reg;
frame_out_next = frame_out_reg;
drop_frame_next = drop_frame_reg;
drop_pad_next = drop_pad_reg;
pad_cyc_next = pad_cyc_reg;
pad_next = pad_reg;
extend_next = extend_reg;
pad_done_next = pad_done_reg;
pad_mask_next = pad_mask_reg;
pad_id_next = pad_id_reg;
pad_dest_next = pad_dest_reg;
pad_user_next = pad_user_reg;
stat_pad_frame_next = 1'b0;
stat_err_user_next = s_axis.tvalid && s_axis.tready && s_axis.tlast && s_axis.tuser[0];
stat_err_underflow_next = 1'b0;
if (UNDERFLOW_DROP_EN && frame_in_reg && s_axis.tready && !s_axis.tvalid && !drop_frame_reg) begin
// underflow
drop_frame_next = 1'b1;
drop_pad_next = 1'b1;
if (pad_reg && !pad_done_reg) begin
// inside minimum frame length
extend_next = 1'b1;
end
end
if (s_axis.tvalid && s_axis.tready) begin
frame_in_next = !s_axis.tlast;
if (s_axis.tlast) begin
drop_frame_next = 1'b0;
stat_err_underflow_next = drop_frame_reg;
end
end
if (m_axis.tvalid && m_axis.tready) begin
frame_out_next = !m_axis.tlast;
if (m_axis.tlast) begin
drop_pad_next = 1'b0;
end
end
if (s_axis.tvalid && s_axis.tready) begin
if (!extend_reg && !drop_frame_reg && !drop_pad_reg) begin
pad_id_next = s_axis.tid;
pad_dest_next = s_axis.tdest;
pad_user_next = s_axis.tuser;
if (s_axis.tlast) begin
if (pad_reg && !pad_done_reg) begin
// inside minimum frame length
extend_next = 1'b1;
end
end
end
end
if (m_axis.tvalid && m_axis.tready) begin
if ((pad_cyc_reg >> 1) != 0) begin
pad_cyc_next = pad_cyc_reg - 1;
pad_next = 1'b1;
pad_done_next = 1'b0;
pad_mask_next = '1;
end else begin
pad_next = extend_reg || s_axis.tlast;
pad_done_next = 1'b1;
pad_mask_next = {KEEP_W{1'b1}} >> PAD_LANE_W'(KEEP_W-1-min_pkt_len_lanes);
end
if (pad_done_reg) begin
pad_next = 1'b0;
pad_mask_next = '0;
extend_next = 1'b0;
drop_pad_next = 1'b0;
if (s_axis.tlast) begin
// padding within final cycle
stat_pad_frame_next = (~s_axis.tkeep & pad_mask_reg) != 0;
end
if (extend_reg || drop_pad_reg) begin
// padding started in a previous cycle
stat_pad_frame_next = 1'b1;
end
if (extend_reg || drop_pad_reg || s_axis.tlast) begin
if (cfg_pad_en) begin
pad_cyc_next = min_pkt_len_cycles;
pad_next = min_pkt_len_cycles != 0;
pad_done_next = min_pkt_len_cycles == 0;
pad_mask_next = min_pkt_len_cycles == 0 ? {KEEP_W{1'b1}} >> PAD_LANE_W'(KEEP_W-1-min_pkt_len_lanes) : '1;
end else begin
pad_cyc_next = '0;
pad_next = 1'b0;
pad_done_next = 1'b1;
pad_mask_next = '0;
end
pad_user_next = '0;
end
end
end else if (!frame_in_reg && !extend_reg && !drop_pad_reg) begin
if (cfg_pad_en) begin
pad_cyc_next = min_pkt_len_cycles;
pad_next = min_pkt_len_cycles != 0;
pad_done_next = min_pkt_len_cycles == 0;
pad_mask_next = min_pkt_len_cycles == 0 ? {KEEP_W{1'b1}} >> PAD_LANE_W'(KEEP_W-1-min_pkt_len_lanes) : '1;
end else begin
pad_cyc_next = '0;
pad_next = 1'b0;
pad_done_next = 1'b1;
pad_mask_next = '0;
end
pad_user_next = '0;
end
end
always_ff @(posedge clk) begin
frame_in_reg <= frame_in_next;
frame_out_reg <= frame_out_next;
drop_frame_reg <= drop_frame_next;
drop_pad_reg <= drop_pad_next;
pad_cyc_reg <= pad_cyc_next;
pad_reg <= pad_next;
extend_reg <= extend_next;
pad_done_reg <= pad_done_next;
pad_mask_reg <= pad_mask_next;
pad_id_reg <= pad_id_next;
pad_dest_reg <= pad_dest_next;
pad_user_reg <= pad_user_next;
stat_pad_frame_reg <= stat_pad_frame_next;
stat_err_user_reg <= stat_err_user_next;
stat_err_underflow_reg <= stat_err_underflow_next;
if (rst) begin
frame_in_reg <= 1'b0;
frame_out_reg <= 1'b0;
drop_frame_reg <= 1'b0;
pad_cyc_reg <= min_pkt_len_cycles;
pad_reg <= min_pkt_len_cycles != 0;
extend_reg <= 1'b0;
pad_done_reg <= min_pkt_len_cycles == 0;
pad_mask_reg <= min_pkt_len_cycles == 0 ? {KEEP_W{1'b1}} >> PAD_LANE_W'(KEEP_W-1-min_pkt_len_lanes) : '1;
pad_user_reg <= '0;
stat_pad_frame_reg <= 1'b0;
stat_err_user_reg <= 1'b0;
stat_err_underflow_reg <= 1'b0;
end
end
endmodule
`resetall

66
src/axis/tb/taxi_axis_pad/Makefile Normal file
View File

@@ -0,0 +1,66 @@
# SPDX-License-Identifier: CERN-OHL-S-2.0
#
# Copyright (c) 2021-2026 FPGA Ninja, LLC
#
# Authors:
# - Alex Forencich
TOPLEVEL_LANG = verilog
SIM ?= verilator
WAVES ?= 0
COCOTB_HDL_TIMEUNIT = 1ns
COCOTB_HDL_TIMEPRECISION = 1ps
RTL_DIR = ../../rtl
LIB_DIR = ../../lib
TAXI_SRC_DIR = $(LIB_DIR)/taxi/src
DUT = taxi_axis_pad
COCOTB_TEST_MODULES = test_$(DUT)
COCOTB_TOPLEVEL = test_$(DUT)
MODULE = $(COCOTB_TEST_MODULES)
TOPLEVEL = $(COCOTB_TOPLEVEL)
VERILOG_SOURCES += $(COCOTB_TOPLEVEL).sv
VERILOG_SOURCES += $(RTL_DIR)/$(DUT).sv
VERILOG_SOURCES += $(RTL_DIR)/taxi_axis_if.sv
# handle file list files
process_f_file = $(call process_f_files,$(addprefix $(dir $1),$(shell cat $1)))
process_f_files = $(foreach f,$1,$(if $(filter %.f,$f),$(call process_f_file,$f),$f))
uniq_base = $(if $1,$(call uniq_base,$(foreach f,$1,$(if $(filter-out $(notdir $(lastword $1)),$(notdir $f)),$f,))) $(lastword $1))
VERILOG_SOURCES := $(call uniq_base,$(call process_f_files,$(VERILOG_SOURCES)))
# module parameters
export PARAM_DATA_W := 8
export PARAM_KEEP_EN := $(shell echo $$(( $(PARAM_DATA_W) > 8 )))
export PARAM_KEEP_W := $(shell echo $$(( ( $(PARAM_DATA_W) + 7 ) / 8 )))
export PARAM_STRB_EN := 0
export PARAM_LAST_EN := 1
export PARAM_ID_EN := 1
export PARAM_ID_W := 8
export PARAM_DEST_EN := 1
export PARAM_DEST_W := 8
export PARAM_USER_EN := 1
export PARAM_USER_W := 1
export PARAM_ID_PAD_REG_EN := 1
export PARAM_DEST_PAD_REG_EN := 1
export PARAM_USER_PAD_REG_EN := 1
export PARAM_MIN_LEN_W := 16
export PARAM_UNDERFLOW_DROP_EN := 0
ifeq ($(SIM), icarus)
PLUSARGS += -fst
COMPILE_ARGS += $(foreach v,$(filter PARAM_%,$(.VARIABLES)),-P $(COCOTB_TOPLEVEL).$(subst PARAM_,,$(v))=$($(v)))
else ifeq ($(SIM), verilator)
COMPILE_ARGS += $(foreach v,$(filter PARAM_%,$(.VARIABLES)),-G$(subst PARAM_,,$(v))=$($(v)))
ifeq ($(WAVES), 1)
COMPILE_ARGS += --trace-fst
VERILATOR_TRACE = 1
endif
endif
include $(shell cocotb-config --makefiles)/Makefile.sim

562
src/axis/tb/taxi_axis_pad/test_taxi_axis_pad.py Normal file
View File

@@ -0,0 +1,562 @@
#!/usr/bin/env python
# SPDX-License-Identifier: CERN-OHL-S-2.0
"""
Copyright (c) 2021-2026 FPGA Ninja, LLC
Authors:
- Alex Forencich
"""
import itertools
import logging
import os
import random
import cocotb_test.simulator
import pytest
import cocotb
from cocotb.clock import Clock
from cocotb.triggers import RisingEdge, FallingEdge
from cocotb.regression import TestFactory
from cocotbext.axi import AxiStreamBus, AxiStreamFrame, AxiStreamSource, AxiStreamSink
class TB(object):
def __init__(self, dut):
self.dut = dut
self.log = logging.getLogger("cocotb.tb")
self.log.setLevel(logging.DEBUG)
cocotb.start_soon(Clock(dut.clk, 10, units="ns").start())
self.source = AxiStreamSource(AxiStreamBus.from_entity(dut.s_axis), dut.clk, dut.rst)
self.sink = AxiStreamSink(AxiStreamBus.from_entity(dut.m_axis), dut.clk, dut.rst)
dut.cfg_pad_en.setimmediatevalue(1)
dut.cfg_min_pkt_len.setimmediatevalue(60-1)
self.stats = {}
self.stats["stat_pad_frame"] = 0
self.stats["stat_err_user"] = 0
self.stats["stat_err_underflow"] = 0
cocotb.start_soon(self._run_stats_counters())
def set_idle_generator(self, generator=None):
if generator:
self.source.set_pause_generator(generator())
def set_backpressure_generator(self, generator=None):
if generator:
self.sink.set_pause_generator(generator())
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)
self.stats_reset()
def stats_reset(self):
for stat in self.stats:
self.stats[stat] = 0
async def _run_stats_counters(self):
while True:
await RisingEdge(self.dut.clk)
for stat in self.stats:
self.stats[stat] += int(getattr(self.dut, stat).value)
async def run_test(dut, pad_en=True, payload_lengths=None, payload_data=None, idle_inserter=None, backpressure_inserter=None):
tb = TB(dut)
id_count = 2**len(tb.source.bus.tid)
cur_id = 1
min_pkt_len = 60
dut.cfg_pad_en.value = pad_en
dut.cfg_min_pkt_len.value = min_pkt_len-1
await tb.reset()
tb.set_idle_generator(idle_inserter)
tb.set_backpressure_generator(backpressure_inserter)
test_frames = []
for test_data in [payload_data(x) for x in payload_lengths()]:
test_frame = AxiStreamFrame(test_data)
test_frame.tid = cur_id
test_frame.tdest = cur_id
test_frames.append(test_frame)
await tb.source.send(test_frame)
cur_id = (cur_id + 1) % id_count
for test_frame in test_frames:
rx_frame = await tb.sink.recv()
tb.log.info("Packet len: %d", len(rx_frame))
if pad_en:
assert len(rx_frame) >= min_pkt_len
assert rx_frame.tdata == test_frame.tdata.ljust(min_pkt_len, b'\x00')
else:
assert rx_frame.tdata == test_frame.tdata
assert rx_frame.tid == test_frame.tid
assert rx_frame.tdest == test_frame.tdest
assert not rx_frame.tuser
assert tb.sink.empty()
await RisingEdge(dut.clk)
await RisingEdge(dut.clk)
async def run_test_pad(dut, pad_en=True, idle_inserter=None, backpressure_inserter=None):
tb = TB(dut)
byte_lanes = tb.source.byte_lanes
id_count = 2**len(tb.source.bus.tid)
cur_id = 1
min_pkt_len = 60
dut.cfg_pad_en.value = pad_en
dut.cfg_min_pkt_len.value = min_pkt_len-1
await tb.reset()
tb.set_idle_generator(idle_inserter)
tb.set_backpressure_generator(backpressure_inserter)
for min_pkt_len in range(1, byte_lanes*3+1):
tb.dut.cfg_min_pkt_len.value = min_pkt_len-1
await RisingEdge(dut.clk)
for test_pkt_len in range(max(min_pkt_len-byte_lanes*2, 1), min_pkt_len+byte_lanes*2+1):
short = test_pkt_len < min_pkt_len
tb.log.info("min len %d, test len %d", min_pkt_len, test_pkt_len)
tb.stats_reset()
test_data_1 = bytearray(itertools.islice(itertools.cycle(range(256)), min_pkt_len))
test_data_2 = bytearray(itertools.islice(itertools.cycle(range(256)), test_pkt_len))
test_frames = []
for k in range(3):
if k == 1:
test_data = test_data_2
else:
test_data = test_data_1
test_frame = AxiStreamFrame(test_data)
test_frame.tid = cur_id
test_frame.tdest = cur_id
test_frames.append(test_frame)
await tb.source.send(test_frame)
cur_id = (cur_id + 1) % id_count
for test_frame in test_frames:
rx_frame = await tb.sink.recv()
tb.log.info("Packet len: %d", len(rx_frame))
if pad_en:
assert len(rx_frame) >= min_pkt_len
assert rx_frame.tdata == test_frame.tdata.ljust(min_pkt_len, b'\x00')
else:
assert rx_frame.tdata == test_frame.tdata
assert rx_frame.tid == test_frame.tid
assert rx_frame.tdest == test_frame.tdest
assert not rx_frame.tuser
assert tb.stats["stat_pad_frame"] == (1 if short and pad_en else 0)
assert tb.stats["stat_err_user"] == 0
assert tb.stats["stat_err_underflow"] == 0
assert tb.sink.empty()
await RisingEdge(dut.clk)
await RisingEdge(dut.clk)
async def run_test_tuser_assert(dut, pad_en=True, idle_inserter=None, backpressure_inserter=None):
tb = TB(dut)
byte_lanes = tb.source.byte_lanes
id_count = 2**len(tb.source.bus.tid)
cur_id = 1
min_pkt_len = 60
dut.cfg_pad_en.value = pad_en
dut.cfg_min_pkt_len.value = min_pkt_len-1
await tb.reset()
tb.set_idle_generator(idle_inserter)
tb.set_backpressure_generator(backpressure_inserter)
for min_pkt_len in range(1, byte_lanes*3+1):
tb.dut.cfg_min_pkt_len.value = min_pkt_len-1
await RisingEdge(dut.clk)
for test_pkt_len in range(max(min_pkt_len-byte_lanes*2, 1), min_pkt_len+byte_lanes*2+1):
short = test_pkt_len < min_pkt_len
tb.log.info("min len %d, test len %d", min_pkt_len, test_pkt_len)
tb.stats_reset()
test_data_1 = bytearray(itertools.islice(itertools.cycle(range(256)), min_pkt_len))
test_data_2 = bytearray(itertools.islice(itertools.cycle(range(256)), test_pkt_len))
test_frames = []
for k in range(3):
if k == 1:
test_data = test_data_2
else:
test_data = test_data_1
test_frame = AxiStreamFrame(test_data)
test_frame.tid = cur_id
test_frame.tdest = cur_id
if k == 1:
test_frame.tuser = 1
test_frames.append(test_frame)
await tb.source.send(test_frame)
cur_id = (cur_id + 1) % id_count
for k, test_frame in enumerate(test_frames):
rx_frame = await tb.sink.recv()
tb.log.info("Packet len: %d", len(rx_frame))
if pad_en:
assert len(rx_frame) >= min_pkt_len
assert rx_frame.tdata == test_frame.tdata.ljust(min_pkt_len, b'\x00')
else:
assert rx_frame.tdata == test_frame.tdata
assert rx_frame.tid == test_frame.tid
assert rx_frame.tdest == test_frame.tdest
if k == 1:
if isinstance(rx_frame.tuser, list):
assert rx_frame.tuser[-1]
else:
assert rx_frame.tuser
else:
assert not rx_frame.tuser
assert tb.stats["stat_pad_frame"] == (1 if short and pad_en else 0)
assert tb.stats["stat_err_user"] == 1
assert tb.stats["stat_err_underflow"] == 0
assert tb.sink.empty()
await RisingEdge(dut.clk)
await RisingEdge(dut.clk)
async def run_test_underflow(dut, pad_en=True, backpressure_inserter=None):
tb = TB(dut)
byte_lanes = tb.source.byte_lanes
id_count = 2**len(tb.source.bus.tid)
cur_id = 1
min_pkt_len = 60
dut.cfg_pad_en.value = pad_en
dut.cfg_min_pkt_len.value = min_pkt_len-1
await tb.reset()
tb.set_backpressure_generator(backpressure_inserter)
for min_pkt_len in range(byte_lanes, byte_lanes*4+1, byte_lanes):
tb.dut.cfg_min_pkt_len.value = min_pkt_len-1
await RisingEdge(dut.clk)
for test_pkt_len in range(byte_lanes, byte_lanes*8+1, byte_lanes):
for delay_cyc in range(0, (test_pkt_len // byte_lanes) + 1):
short = test_pkt_len < min_pkt_len
drop = delay_cyc > 0 and delay_cyc < (test_pkt_len // byte_lanes)
short_drop = drop and delay_cyc < (min_pkt_len // byte_lanes)-1
tb.log.info("min len %d, test len %d, delay %d, short %d, drop %d, short_drop %d", min_pkt_len, test_pkt_len, delay_cyc, short, drop, short_drop)
await FallingEdge(dut.clk)
for k in range(10):
await FallingEdge(dut.clk)
tb.stats_reset()
test_data_1 = bytearray(itertools.islice(itertools.cycle(range(256)), min_pkt_len))
test_data_2 = bytearray(itertools.islice(itertools.cycle(range(256)), test_pkt_len))
test_frames = []
for k in range(3):
if k == 1:
test_data = test_data_2
else:
test_data = test_data_1
test_frame = AxiStreamFrame(test_data)
test_frame.tid = cur_id
test_frame.tdest = cur_id
test_frames.append(test_frame)
await tb.source.send(test_frame)
cur_id = (cur_id + 1) % id_count
cycle_delay = (min_pkt_len // byte_lanes) + delay_cyc
while cycle_delay > 0:
await FallingEdge(dut.clk)
if not int(dut.m_axis.tvalid.value) or not int(dut.m_axis.tready.value):
continue
cycle_delay -= 1
tb.source.pause = True
await FallingEdge(dut.clk)
while not int(dut.m_axis.tready.value):
await FallingEdge(dut.clk)
tb.source.pause = False
for k, test_frame in enumerate(test_frames):
rx_frame = await tb.sink.recv()
tb.log.info("Packet len: %d", len(rx_frame))
if pad_en:
assert len(rx_frame) >= min_pkt_len
if k == 1 and drop:
if isinstance(rx_frame.tuser, list):
assert rx_frame.tuser[-1]
else:
assert rx_frame.tuser
else:
if pad_en:
assert rx_frame.tdata == test_frame.tdata.ljust(min_pkt_len, b'\x00')
else:
assert rx_frame.tdata == test_frame.tdata
assert rx_frame.tid == test_frame.tid
assert rx_frame.tdest == test_frame.tdest
assert not rx_frame.tuser
assert tb.stats["stat_pad_frame"] == (1 if (short or short_drop) and pad_en else 0)
assert tb.stats["stat_err_user"] == 0
assert tb.stats["stat_err_underflow"] == (1 if drop else 0)
assert tb.sink.empty()
await RisingEdge(dut.clk)
await RisingEdge(dut.clk)
async def run_stress_test(dut, pad_en=True, idle_inserter=None, backpressure_inserter=None):
tb = TB(dut)
byte_lanes = tb.source.byte_lanes
id_count = 2**len(tb.source.bus.tid)
cur_id = 1
min_pkt_len = 60
dut.cfg_pad_en.value = pad_en
dut.cfg_min_pkt_len.value = min_pkt_len-1
await tb.reset()
tb.set_idle_generator(idle_inserter)
tb.set_backpressure_generator(backpressure_inserter)
test_frames = []
for k in range(128):
length = random.randint(1, max(byte_lanes*16, min_pkt_len*4))
test_data = bytearray(itertools.islice(itertools.cycle(range(256)), length))
test_frame = AxiStreamFrame(test_data)
test_frame.tid = cur_id
test_frame.tdest = cur_id
test_frames.append(test_frame)
await tb.source.send(test_frame)
cur_id = (cur_id + 1) % id_count
for test_frame in test_frames:
rx_frame = await tb.sink.recv()
tb.log.info("Packet len: %d", len(rx_frame))
if pad_en:
assert len(rx_frame) >= min_pkt_len
assert rx_frame.tdata == test_frame.tdata.ljust(min_pkt_len, b'\x00')
else:
assert rx_frame.tdata == test_frame.tdata
assert rx_frame.tid == test_frame.tid
assert rx_frame.tdest == test_frame.tdest
assert not rx_frame.tuser
assert tb.sink.empty()
await RisingEdge(dut.clk)
await RisingEdge(dut.clk)
def cycle_pause():
return itertools.cycle([1, 1, 1, 0])
def size_list():
data_width = len(cocotb.top.m_axis.tdata)
byte_lanes = data_width // 8
return list(range(1, max(byte_lanes*4, 128))) + [512, 1514, 9214] + [1]*10
def incrementing_payload(length):
return bytearray(itertools.islice(itertools.cycle(range(256)), length))
if getattr(cocotb, 'top', None) is not None:
factory = TestFactory(run_test)
factory.add_option("payload_lengths", [size_list])
factory.add_option("pad_en", [True, False])
factory.add_option("payload_data", [incrementing_payload])
if not cocotb.top.UNDERFLOW_DROP_EN.value:
factory.add_option("idle_inserter", [None, cycle_pause])
factory.add_option("backpressure_inserter", [None, cycle_pause])
factory.generate_tests()
for test in [
run_test_pad,
run_test_tuser_assert,
]:
factory = TestFactory(test)
factory.add_option("pad_en", [True, False])
if not cocotb.top.UNDERFLOW_DROP_EN.value:
factory.add_option("idle_inserter", [None, cycle_pause])
factory.add_option("backpressure_inserter", [None, cycle_pause])
factory.generate_tests()
if cocotb.top.UNDERFLOW_DROP_EN.value:
for test in [run_test_underflow]:
factory = TestFactory(test)
factory.add_option("pad_en", [True, False])
factory.add_option("backpressure_inserter", [None, cycle_pause])
factory.generate_tests()
factory = TestFactory(run_stress_test)
factory.add_option("pad_en", [True, False])
if not cocotb.top.UNDERFLOW_DROP_EN.value:
factory.add_option("idle_inserter", [None, cycle_pause])
factory.add_option("backpressure_inserter", [None, cycle_pause])
factory.generate_tests()
# cocotb-test
tests_dir = os.path.dirname(__file__)
rtl_dir = os.path.abspath(os.path.join(tests_dir, '..', '..', 'rtl'))
lib_dir = os.path.abspath(os.path.join(tests_dir, '..', '..', 'lib'))
taxi_src_dir = os.path.abspath(os.path.join(lib_dir, 'taxi', 'src'))
def process_f_files(files):
lst = {}
for f in files:
if f[-2:].lower() == '.f':
with open(f, 'r') as fp:
l = fp.read().split()
for f in process_f_files([os.path.join(os.path.dirname(f), x) for x in l]):
lst[os.path.basename(f)] = f
else:
lst[os.path.basename(f)] = f
return list(lst.values())
@pytest.mark.parametrize("data_w", [8, 16, 32, 64])
@pytest.mark.parametrize("underflow_drop", [0, 1])
def test_taxi_axis_pad(request, data_w, underflow_drop):
dut = "taxi_axis_pad"
module = os.path.splitext(os.path.basename(__file__))[0]
toplevel = module
verilog_sources = [
os.path.join(tests_dir, f"{toplevel}.sv"),
os.path.join(rtl_dir, f"{dut}.sv"),
os.path.join(rtl_dir, "taxi_axis_if.sv"),
]
verilog_sources = process_f_files(verilog_sources)
parameters = {}
parameters['DATA_W'] = data_w
parameters['KEEP_EN'] = int(parameters['DATA_W'] > 8)
parameters['KEEP_W'] = (parameters['DATA_W'] + 7) // 8
parameters['STRB_EN'] = 0
parameters['LAST_EN'] = 1
parameters['ID_EN'] = 1
parameters['ID_W'] = 8
parameters['DEST_EN'] = 1
parameters['DEST_W'] = 8
parameters['USER_EN'] = 1
parameters['USER_W'] = 1
parameters['ID_PAD_REG_EN'] = 1
parameters['DEST_PAD_REG_EN'] = 1
parameters['USER_PAD_REG_EN'] = 1
parameters['MIN_LEN_W'] = 16
parameters['UNDERFLOW_DROP_EN'] = underflow_drop
extra_env = {f'PARAM_{k}': str(v) for k, v in parameters.items()}
sim_build = os.path.join(tests_dir, "sim_build",
request.node.name.replace('[', '-').replace(']', ''))
cocotb_test.simulator.run(
simulator="verilator",
python_search=[tests_dir],
verilog_sources=verilog_sources,
toplevel=toplevel,
module=module,
parameters=parameters,
sim_build=sim_build,
extra_env=extra_env,
)

102
src/axis/tb/taxi_axis_pad/test_taxi_axis_pad.sv Normal file
View File

@@ -0,0 +1,102 @@
// SPDX-License-Identifier: CERN-OHL-S-2.0
/*
Copyright (c) 2025-2026 FPGA Ninja, LLC
Authors:
- Alex Forencich
*/
`resetall
`timescale 1ns / 1ps
`default_nettype none
/*
* AXI4-Stream padding logic testbench
*/
module test_taxi_axis_pad #
(
/* verilator lint_off WIDTHTRUNC */
parameter DATA_W = 8,
parameter logic KEEP_EN = (DATA_W>8),
parameter KEEP_W = ((DATA_W+7)/8),
parameter logic STRB_EN = 1'b0,
parameter logic LAST_EN = 1'b1,
parameter logic ID_EN = 1'b0,
parameter ID_W = 8,
parameter logic DEST_EN = 1'b0,
parameter DEST_W = 8,
parameter logic USER_EN = 1'b1,
parameter USER_W = 1,
parameter logic ID_PAD_REG_EN = 1'b1,
parameter logic DEST_PAD_REG_EN = 1'b1,
parameter logic USER_PAD_REG_EN = 1'b1,
parameter MIN_LEN_W = 8,
parameter logic UNDERFLOW_DROP_EN = 1'b0
/* verilator lint_on WIDTHTRUNC */
)
();
logic clk;
logic rst;
taxi_axis_if #(
.DATA_W(DATA_W),
.KEEP_EN(KEEP_EN),
.KEEP_W(KEEP_W),
.STRB_EN(STRB_EN),
.LAST_EN(LAST_EN),
.ID_EN(ID_EN),
.ID_W(ID_W),
.DEST_EN(DEST_EN),
.DEST_W(DEST_W),
.USER_EN(USER_EN),
.USER_W(USER_W)
) s_axis(), m_axis();
logic cfg_pad_en = '0;
logic [MIN_LEN_W-1:0] cfg_min_pkt_len = 'd60-1;
logic stat_pad_frame;
logic stat_err_user;
logic stat_err_underflow;
taxi_axis_pad #(
.ID_PAD_REG_EN(ID_PAD_REG_EN),
.DEST_PAD_REG_EN(DEST_PAD_REG_EN),
.USER_PAD_REG_EN(USER_PAD_REG_EN),
.MIN_LEN_W(MIN_LEN_W),
.UNDERFLOW_DROP_EN(UNDERFLOW_DROP_EN)
)
uut (
.clk(clk),
.rst(rst),
/*
* AXI4-Stream input (sink)
*/
.s_axis(s_axis),
/*
* AXI4-Stream output (source)
*/
.m_axis(m_axis),
/*
* Configuration
*/
.cfg_pad_en(cfg_pad_en),
.cfg_min_pkt_len(cfg_min_pkt_len),
/*
* Status
*/
.stat_pad_frame(stat_pad_frame),
.stat_err_user(stat_err_user),
.stat_err_underflow(stat_err_underflow)
);
endmodule
`resetall