FPGA/corundum
1
0
Fork 0
mirror of https://github.com/corundum/corundum.git synced 2025-01-30 08:32:52 +08:00
Code Issues Projects Releases Wiki Activity

Test all interfaces

Browse Source
This commit is contained in:
Alex Forencich 2021-11-30 00:57:41 -08:00
parent 639117e53f
commit 9d817af8d1
4 changed files with 197 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
fpga/common/tb/mqnic_core_axi/test_mqnic_core_axi.py
View File

@ -187,13 +187,15 @@ async def run_test_nic(dut):
tb.log.info("Init driver")
await tb.driver.init_axi_dev(tb.pool, tb.hw_regs, irq=dut.irq)
await tb.driver.interfaces[0].open()
for interface in tb.driver.interfaces:
await interface.open()
# enable queues
tb.log.info("Enable queues")
await tb.driver.interfaces[0].ports[0].hw_regs.write_dword(mqnic.MQNIC_PORT_REG_SCHED_ENABLE, 0x00000001)
for k in range(tb.driver.interfaces[0].tx_queue_count):
await tb.driver.interfaces[0].ports[0].schedulers[0].hw_regs.write_dword(4*k, 0x00000003)
for interface in tb.driver.interfaces:
await interface.ports[0].hw_regs.write_dword(mqnic.MQNIC_PORT_REG_SCHED_ENABLE, 0x00000001)
for k in range(interface.tx_queue_count):
await interface.ports[0].schedulers[0].hw_regs.write_dword(4*k, 0x00000003)
# wait for all writes to complete
await tb.driver.hw_regs.read_dword(0)
@ -201,19 +203,20 @@ async def run_test_nic(dut):
tb.log.info("Send and receive single packet")
data = bytearray([x % 256 for x in range(1024)])
for interface in tb.driver.interfaces:
data = bytearray([x % 256 for x in range(1024)])
await tb.driver.interfaces[0].start_xmit(data, 0)
await interface.start_xmit(data, 0)
pkt = await tb.port_mac[0].tx.recv()
tb.log.info("Packet: %s", pkt)
pkt = await tb.port_mac[interface.index*interface.port_count].tx.recv()
tb.log.info("Packet: %s", pkt)
await tb.port_mac[0].rx.send(pkt)
await tb.port_mac[interface.index*interface.port_count].rx.send(pkt)
pkt = await tb.driver.interfaces[0].recv()
pkt = await interface.recv()
tb.log.info("Packet: %s", pkt)
assert pkt.rx_checksum == ~scapy.utils.checksum(bytes(pkt.data[14:])) & 0xffff
tb.log.info("Packet: %s", pkt)
assert pkt.rx_checksum == ~scapy.utils.checksum(bytes(pkt.data[14:])) & 0xffff
tb.log.info("RX and TX checksum tests")
@ -299,6 +302,27 @@ async def run_test_nic(dut):
tb.loopback_enable = False
if len(tb.driver.interfaces) > 1:
tb.log.info("All interfaces")
count = 64
pkts = [bytearray([(x+k) % 256 for x in range(1514)]) for k in range(count)]
tb.loopback_enable = True
for k, p in enumerate(pkts):
await tb.driver.interfaces[k % len(tb.driver.interfaces)].start_xmit(p, 0)
for k in range(count):
pkt = await tb.driver.interfaces[k % len(tb.driver.interfaces)].recv()
tb.log.info("Packet: %s", pkt)
assert pkt.data == pkts[k]
assert pkt.rx_checksum == ~scapy.utils.checksum(bytes(pkt.data[14:])) & 0xffff
tb.loopback_enable = False
tb.log.info("Read statistics counters")
await Timer(2000, 'ns')
@ -325,9 +349,15 @@ eth_rtl_dir = os.path.abspath(os.path.join(lib_dir, 'eth', 'rtl'))
pcie_rtl_dir = os.path.abspath(os.path.join(lib_dir, 'pcie', 'rtl'))
@pytest.mark.parametrize(("axi_data_width", "axis_data_width", "axis_sync_data_width"),
[(128, 64, 64), (128, 64, 128)])
def test_mqnic_core_axi(request, axi_data_width, axis_data_width, axis_sync_data_width):
@pytest.mark.parametrize(("if_count", "ports_per_if", "axi_data_width",
"axis_data_width", "axis_sync_data_width"), [
(1, 1, 128, 64, 64),
(2, 1, 256, 64, 64),
(1, 2, 256, 64, 64),
(1, 1, 128, 64, 128),
])
def test_mqnic_core_pcie_axi(request, if_count, ports_per_if, axi_data_width,
axis_data_width, axis_sync_data_width):
dut = "mqnic_core_axi"
module = os.path.splitext(os.path.basename(__file__))[0]
toplevel = dut
@ -400,8 +430,8 @@ def test_mqnic_core_axi(request, axi_data_width, axis_data_width, axis_sync_data
parameters = {}
# Structural configuration
parameters['IF_COUNT'] = 1
parameters['PORTS_PER_IF'] = 1
parameters['IF_COUNT'] = if_count
parameters['PORTS_PER_IF'] = ports_per_if
# PTP configuration
parameters['PTP_USE_SAMPLE_CLOCK'] = 0

67
fpga/common/tb/mqnic_core_pcie_s10/test_mqnic_core_pcie_s10.py
View File

@ -298,13 +298,15 @@ async def run_test_nic(dut):
tb.log.info("Init driver")
await tb.driver.init_pcie_dev(tb.rc, tb.dev.functions[0].pcie_id)
await tb.driver.interfaces[0].open()
for interface in tb.driver.interfaces:
await interface.open()
# enable queues
tb.log.info("Enable queues")
await tb.driver.interfaces[0].ports[0].hw_regs.write_dword(mqnic.MQNIC_PORT_REG_SCHED_ENABLE, 0x00000001)
for k in range(tb.driver.interfaces[0].tx_queue_count):
await tb.driver.interfaces[0].ports[0].schedulers[0].hw_regs.write_dword(4*k, 0x00000003)
for interface in tb.driver.interfaces:
await interface.ports[0].hw_regs.write_dword(mqnic.MQNIC_PORT_REG_SCHED_ENABLE, 0x00000001)
for k in range(interface.tx_queue_count):
await interface.ports[0].schedulers[0].hw_regs.write_dword(4*k, 0x00000003)
# wait for all writes to complete
await tb.driver.hw_regs.read_dword(0)
@ -312,19 +314,20 @@ async def run_test_nic(dut):
tb.log.info("Send and receive single packet")
data = bytearray([x % 256 for x in range(1024)])
for interface in tb.driver.interfaces:
data = bytearray([x % 256 for x in range(1024)])
await tb.driver.interfaces[0].start_xmit(data, 0)
await interface.start_xmit(data, 0)
pkt = await tb.port_mac[0].tx.recv()
tb.log.info("Packet: %s", pkt)
pkt = await tb.port_mac[interface.index*interface.port_count].tx.recv()
tb.log.info("Packet: %s", pkt)
await tb.port_mac[0].rx.send(pkt)
await tb.port_mac[interface.index*interface.port_count].rx.send(pkt)
pkt = await tb.driver.interfaces[0].recv()
pkt = await interface.recv()
tb.log.info("Packet: %s", pkt)
assert pkt.rx_checksum == ~scapy.utils.checksum(bytes(pkt.data[14:])) & 0xffff
tb.log.info("Packet: %s", pkt)
assert pkt.rx_checksum == ~scapy.utils.checksum(bytes(pkt.data[14:])) & 0xffff
tb.log.info("RX and TX checksum tests")
@ -410,6 +413,27 @@ async def run_test_nic(dut):
tb.loopback_enable = False
if len(tb.driver.interfaces) > 1:
tb.log.info("All interfaces")
count = 64
pkts = [bytearray([(x+k) % 256 for x in range(1514)]) for k in range(count)]
tb.loopback_enable = True
for k, p in enumerate(pkts):
await tb.driver.interfaces[k % len(tb.driver.interfaces)].start_xmit(p, 0)
for k in range(count):
pkt = await tb.driver.interfaces[k % len(tb.driver.interfaces)].recv()
tb.log.info("Packet: %s", pkt)
assert pkt.data == pkts[k]
assert pkt.rx_checksum == ~scapy.utils.checksum(bytes(pkt.data[14:])) & 0xffff
tb.loopback_enable = False
tb.log.info("Read statistics counters")
await Timer(2000, 'ns')
@ -436,9 +460,18 @@ eth_rtl_dir = os.path.abspath(os.path.join(lib_dir, 'eth', 'rtl'))
pcie_rtl_dir = os.path.abspath(os.path.join(lib_dir, 'pcie', 'rtl'))
@pytest.mark.parametrize(("pcie_data_width", "axis_eth_data_width", "axis_eth_sync_data_width"),
[(256, 64, 64), (256, 64, 128)]) #, (512, 64, 64), (512, 64, 128), (512, 512, 512)])
def test_mqnic_core_pcie_s10(request, pcie_data_width, axis_eth_data_width, axis_eth_sync_data_width):
@pytest.mark.parametrize(("if_count", "ports_per_if", "pcie_data_width",
"axis_eth_data_width", "axis_eth_sync_data_width"), [
(1, 1, 256, 64, 64),
(2, 1, 256, 64, 64),
(1, 2, 256, 64, 64),
(1, 1, 256, 64, 128),
# (1, 1, 512, 64, 64),
# (1, 1, 512, 64, 128),
# (1, 1, 512, 512, 512),
])
def test_mqnic_core_pcie_s10(request, if_count, ports_per_if, pcie_data_width,
axis_eth_data_width, axis_eth_sync_data_width):
dut = "mqnic_core_pcie_s10"
module = os.path.splitext(os.path.basename(__file__))[0]
toplevel = dut
@ -521,8 +554,8 @@ def test_mqnic_core_pcie_s10(request, pcie_data_width, axis_eth_data_width, axis
parameters = {}
# Structural configuration
parameters['IF_COUNT'] = 1
parameters['PORTS_PER_IF'] = 1
parameters['IF_COUNT'] = if_count
parameters['PORTS_PER_IF'] = ports_per_if
# PTP configuration
parameters['PTP_USE_SAMPLE_CLOCK'] = 0

67
fpga/common/tb/mqnic_core_pcie_us/test_mqnic_core_pcie_us.py
View File

@ -373,13 +373,15 @@ async def run_test_nic(dut):
tb.log.info("Init driver")
await tb.driver.init_pcie_dev(tb.rc, tb.dev.functions[0].pcie_id)
await tb.driver.interfaces[0].open()
for interface in tb.driver.interfaces:
await interface.open()
# enable queues
tb.log.info("Enable queues")
await tb.driver.interfaces[0].ports[0].hw_regs.write_dword(mqnic.MQNIC_PORT_REG_SCHED_ENABLE, 0x00000001)
for k in range(tb.driver.interfaces[0].tx_queue_count):
await tb.driver.interfaces[0].ports[0].schedulers[0].hw_regs.write_dword(4*k, 0x00000003)
for interface in tb.driver.interfaces:
await interface.ports[0].hw_regs.write_dword(mqnic.MQNIC_PORT_REG_SCHED_ENABLE, 0x00000001)
for k in range(interface.tx_queue_count):
await interface.ports[0].schedulers[0].hw_regs.write_dword(4*k, 0x00000003)
# wait for all writes to complete
await tb.driver.hw_regs.read_dword(0)
@ -387,19 +389,20 @@ async def run_test_nic(dut):
tb.log.info("Send and receive single packet")
data = bytearray([x % 256 for x in range(1024)])
for interface in tb.driver.interfaces:
data = bytearray([x % 256 for x in range(1024)])
await tb.driver.interfaces[0].start_xmit(data, 0)
await interface.start_xmit(data, 0)
pkt = await tb.port_mac[0].tx.recv()
tb.log.info("Packet: %s", pkt)
pkt = await tb.port_mac[interface.index*interface.port_count].tx.recv()
tb.log.info("Packet: %s", pkt)
await tb.port_mac[0].rx.send(pkt)
await tb.port_mac[interface.index*interface.port_count].rx.send(pkt)
pkt = await tb.driver.interfaces[0].recv()
pkt = await interface.recv()
tb.log.info("Packet: %s", pkt)
assert pkt.rx_checksum == ~scapy.utils.checksum(bytes(pkt.data[14:])) & 0xffff
tb.log.info("Packet: %s", pkt)
assert pkt.rx_checksum == ~scapy.utils.checksum(bytes(pkt.data[14:])) & 0xffff
tb.log.info("RX and TX checksum tests")
@ -485,6 +488,27 @@ async def run_test_nic(dut):
tb.loopback_enable = False
if len(tb.driver.interfaces) > 1:
tb.log.info("All interfaces")
count = 64
pkts = [bytearray([(x+k) % 256 for x in range(1514)]) for k in range(count)]
tb.loopback_enable = True
for k, p in enumerate(pkts):
await tb.driver.interfaces[k % len(tb.driver.interfaces)].start_xmit(p, 0)
for k in range(count):
pkt = await tb.driver.interfaces[k % len(tb.driver.interfaces)].recv()
tb.log.info("Packet: %s", pkt)
assert pkt.data == pkts[k]
assert pkt.rx_checksum == ~scapy.utils.checksum(bytes(pkt.data[14:])) & 0xffff
tb.loopback_enable = False
tb.log.info("Read statistics counters")
await Timer(2000, 'ns')
@ -511,9 +535,18 @@ eth_rtl_dir = os.path.abspath(os.path.join(lib_dir, 'eth', 'rtl'))
pcie_rtl_dir = os.path.abspath(os.path.join(lib_dir, 'pcie', 'rtl'))
@pytest.mark.parametrize(("axis_pcie_data_width", "axis_eth_data_width", "axis_eth_sync_data_width"),
[(256, 64, 64), (256, 64, 128), (512, 64, 64), (512, 64, 128), (512, 512, 512)])
def test_mqnic_core_pcie_us(request, axis_pcie_data_width, axis_eth_data_width, axis_eth_sync_data_width):
@pytest.mark.parametrize(("if_count", "ports_per_if", "axis_pcie_data_width",
"axis_eth_data_width", "axis_eth_sync_data_width"), [
(1, 1, 256, 64, 64),
(2, 1, 256, 64, 64),
(1, 2, 256, 64, 64),
(1, 1, 256, 64, 128),
(1, 1, 512, 64, 64),
(1, 1, 512, 64, 128),
(1, 1, 512, 512, 512),
])
def test_mqnic_core_pcie_us(request, if_count, ports_per_if, axis_pcie_data_width,
axis_eth_data_width, axis_eth_sync_data_width):
dut = "mqnic_core_pcie_us"
module = os.path.splitext(os.path.basename(__file__))[0]
toplevel = dut
@ -598,8 +631,8 @@ def test_mqnic_core_pcie_us(request, axis_pcie_data_width, axis_eth_data_width,
parameters = {}
# Structural configuration
parameters['IF_COUNT'] = 1
parameters['PORTS_PER_IF'] = 1
parameters['IF_COUNT'] = if_count
parameters['PORTS_PER_IF'] = ports_per_if
# PTP configuration
parameters['PTP_USE_SAMPLE_CLOCK'] = 0

67
fpga/common/tb/mqnic_core_pcie_us_tdma/test_mqnic_core_pcie_us.py
View File

@ -373,13 +373,15 @@ async def run_test_nic(dut):
tb.log.info("Init driver")
await tb.driver.init_pcie_dev(tb.rc, tb.dev.functions[0].pcie_id)
await tb.driver.interfaces[0].open()
for interface in tb.driver.interfaces:
await interface.open()
# enable queues
tb.log.info("Enable queues")
await tb.driver.interfaces[0].ports[0].hw_regs.write_dword(mqnic.MQNIC_PORT_REG_SCHED_ENABLE, 0x00000001)
for k in range(tb.driver.interfaces[0].tx_queue_count):
await tb.driver.interfaces[0].ports[0].schedulers[0].hw_regs.write_dword(4*k, 0x00000003)
for interface in tb.driver.interfaces:
await interface.ports[0].hw_regs.write_dword(mqnic.MQNIC_PORT_REG_SCHED_ENABLE, 0x00000001)
for k in range(interface.tx_queue_count):
await interface.ports[0].schedulers[0].hw_regs.write_dword(4*k, 0x00000003)
# wait for all writes to complete
await tb.driver.hw_regs.read_dword(0)
@ -387,19 +389,20 @@ async def run_test_nic(dut):
tb.log.info("Send and receive single packet")
data = bytearray([x % 256 for x in range(1024)])
for interface in tb.driver.interfaces:
data = bytearray([x % 256 for x in range(1024)])
await tb.driver.interfaces[0].start_xmit(data, 0)
await interface.start_xmit(data, 0)
pkt = await tb.port_mac[0].tx.recv()
tb.log.info("Packet: %s", pkt)
pkt = await tb.port_mac[interface.index*interface.port_count].tx.recv()
tb.log.info("Packet: %s", pkt)
await tb.port_mac[0].rx.send(pkt)
await tb.port_mac[interface.index*interface.port_count].rx.send(pkt)
pkt = await tb.driver.interfaces[0].recv()
pkt = await interface.recv()
tb.log.info("Packet: %s", pkt)
assert pkt.rx_checksum == ~scapy.utils.checksum(bytes(pkt.data[14:])) & 0xffff
tb.log.info("Packet: %s", pkt)
assert pkt.rx_checksum == ~scapy.utils.checksum(bytes(pkt.data[14:])) & 0xffff
tb.log.info("RX and TX checksum tests")
@ -485,6 +488,27 @@ async def run_test_nic(dut):
tb.loopback_enable = False
if len(tb.driver.interfaces) > 1:
tb.log.info("All interfaces")
count = 64
pkts = [bytearray([(x+k) % 256 for x in range(1514)]) for k in range(count)]
tb.loopback_enable = True
for k, p in enumerate(pkts):
await tb.driver.interfaces[k % len(tb.driver.interfaces)].start_xmit(p, 0)
for k in range(count):
pkt = await tb.driver.interfaces[k % len(tb.driver.interfaces)].recv()
tb.log.info("Packet: %s", pkt)
assert pkt.data == pkts[k]
assert pkt.rx_checksum == ~scapy.utils.checksum(bytes(pkt.data[14:])) & 0xffff
tb.loopback_enable = False
await Timer(1000, 'ns')
tb.log.info("TDMA")
@ -563,9 +587,18 @@ eth_rtl_dir = os.path.abspath(os.path.join(lib_dir, 'eth', 'rtl'))
pcie_rtl_dir = os.path.abspath(os.path.join(lib_dir, 'pcie', 'rtl'))
@pytest.mark.parametrize(("axis_pcie_data_width", "axis_eth_data_width", "axis_eth_sync_data_width"),
[(256, 64, 64), (256, 64, 128), (512, 64, 64), (512, 64, 128), (512, 512, 512)])
def test_mqnic_core_pcie_us(request, axis_pcie_data_width, axis_eth_data_width, axis_eth_sync_data_width):
@pytest.mark.parametrize(("if_count", "ports_per_if", "axis_pcie_data_width",
"axis_eth_data_width", "axis_eth_sync_data_width"), [
(1, 1, 256, 64, 64),
(2, 1, 256, 64, 64),
(1, 2, 256, 64, 64),
(1, 1, 256, 64, 128),
(1, 1, 512, 64, 64),
(1, 1, 512, 64, 128),
(1, 1, 512, 512, 512),
])
def test_mqnic_core_pcie_us(request, if_count, ports_per_if, axis_pcie_data_width,
axis_eth_data_width, axis_eth_sync_data_width):
dut = "mqnic_core_pcie_us"
module = os.path.splitext(os.path.basename(__file__))[0]
toplevel = dut
@ -652,8 +685,8 @@ def test_mqnic_core_pcie_us(request, axis_pcie_data_width, axis_eth_data_width,
parameters = {}
# Structural configuration
parameters['IF_COUNT'] = 1
parameters['PORTS_PER_IF'] = 1
parameters['IF_COUNT'] = if_count
parameters['PORTS_PER_IF'] = ports_per_if
# PTP configuration
parameters['PTP_USE_SAMPLE_CLOCK'] = 0