diff --git a/liteeth/common.py b/liteeth/common.py index 2c4f88fe..e110aa7b 100644 --- a/liteeth/common.py +++ b/liteeth/common.py @@ -96,7 +96,8 @@ "msgtype": HeaderField(0, 0, 8), "code": HeaderField(1, 0, 8), "checksum": HeaderField(2, 0, 16), - "quench": HeaderField(4, 0, 32) + "ident": HeaderField(4, 0, 16), + "sequence": HeaderField(6, 0, 16) } icmp_header = Header(icmp_header_fields, icmp_header_length, swap_field_bytes=True) diff --git a/liteeth/core/arp.py b/liteeth/core/arp.py index 18ccfbc2..a74d4222 100644 --- a/liteeth/core/arp.py +++ b/liteeth/core/arp.py @@ -202,9 +202,9 @@ def __init__(self, clk_freq, max_requests=8): # table in the future to improve performance when packets are # targeting multiple destinations. update = Signal() - cached_valid = Signal() - cached_ip_address = Signal(32, reset_less=True) - cached_mac_address = Signal(48, reset_less=True) + self.cached_valid = Signal() + self.cached_ip_address = Signal(32, reset_less=True) + self.cached_mac_address = Signal(48, reset_less=True) cached_timer = WaitTimer(clk_freq*10) self.submodules += cached_timer @@ -238,21 +238,21 @@ def __init__(self, clk_freq, max_requests=8): ) self.sync += \ If(update, - cached_valid.eq(1), - cached_ip_address.eq(sink.ip_address), - cached_mac_address.eq(sink.mac_address), + self.cached_valid.eq(1), + self.cached_ip_address.eq(sink.ip_address), + self.cached_mac_address.eq(sink.mac_address), ).Else( If(cached_timer.done, - cached_valid.eq(0) + self.cached_valid.eq(0) ) ) self.comb += cached_timer.wait.eq(~update) fsm.act("CHECK_TABLE", - If(cached_valid, - If(request_ip_address == cached_ip_address, + If(self.cached_valid, + If(request_ip_address == self.cached_ip_address, request_ip_address_reset.eq(1), NextState("PRESENT_RESPONSE"), - ).Elif(request.ip_address == cached_ip_address, + ).Elif(request.ip_address == self.cached_ip_address, request.ready.eq(request.valid), NextState("PRESENT_RESPONSE"), ).Else( @@ -282,7 +282,7 @@ def __init__(self, clk_freq, max_requests=8): request_counter_reset.eq(1), request_pending_clr.eq(1) ), - response.mac_address.eq(cached_mac_address) + response.mac_address.eq(self.cached_mac_address) ] fsm.act("PRESENT_RESPONSE", response.valid.eq(1), diff --git a/liteeth/core/icmp.py b/liteeth/core/icmp.py index ee1f7d49..2f70984d 100644 --- a/liteeth/core/icmp.py +++ b/liteeth/core/icmp.py @@ -35,7 +35,8 @@ def __init__(self, ip_address, dw=8): "msgtype", "code", "checksum", - "quench", + "ident", + "sequence", "data", "last_be"}) @@ -98,7 +99,8 @@ def __init__(self, ip_address, dw=8): "msgtype", "code", "checksum", - "quench", + "ident", + "sequence", "data", "error", "last_be"}), diff --git a/test/model/dumps.py b/test/model/dumps.py index e9ea8f26..01952603 100644 --- a/test/model/dumps.py +++ b/test/model/dumps.py @@ -86,7 +86,7 @@ def verify_packet(packet, infos): ping_request = format_dump(""" 00 50 56 e0 14 49 00 0c 29 34 0b de 08 00 45 00 00 3c d7 43 00 00 80 01 2b 73 c0 a8 9e 8b ae 89 -2a 4d 08 00 2a 5c 02 00 21 00 61 62 63 64 65 66 +2a 4d 08 00 20 b4 02 00 21 00 61 62 63 64 65 66 67 68 69 6a 6b 6c 6d 6e 6f 70 71 72 73 74 75 76 77 61 62 63 64 65 66 67 68 69""") @@ -104,3 +104,13 @@ def verify_packet(packet, infos): 77 61 62 63 64 65 66 67 68 69""") ping_reply_infos = {} + +# ICMP: Destination unreachable (Port unreachable) +icmp_unreachable_reply = format_dump(""" +00 00 00 00 00 00 00 00 00 00 00 00 08 00 45 c0 +00 3e 11 a7 00 00 40 01 6a 56 7f 00 00 01 7f 00 +00 01 03 03 d8 18 00 00 00 00 45 00 00 22 a0 52 +40 00 40 11 9c 76 7f 00 00 01 7f 00 00 01 de 05 +04 d2 00 0e fe 21 68 65 6c 6c 6f 0a +""") + diff --git a/test/model/etherbone.py b/test/model/etherbone.py index 4fbb7fbf..34cfaa65 100644 --- a/test/model/etherbone.py +++ b/test/model/etherbone.py @@ -31,23 +31,26 @@ def __init__(self, udp, debug=False): udp.set_etherbone_callback(self.callback) - def send(self, packet): + def send(self, packet, target_ip=0x12345678): packet.encode() if self.debug: print_etherbone(">>>>>>>>") print_etherbone(packet) - udp_packet = udp.UDPPacket(packet) + udp_packet = udp.UDPPacket(packet.bytes) udp_packet.src_port = 0x1234 # FIXME udp_packet.dst_port = 0x1234 # FIXME - udp_packet.length = len(packet) + udp_packet.length = len(packet.bytes) + udp_header_length udp_packet.checksum = 0 - self.udp.send(udp_packet) + self.udp.send(udp_packet, target_ip) - def receive(self): + def receive(self, timeout=None): self.rx_packet = EtherbonePacket() - while not self.rx_packet.done: + i = 0 + while not self.rx_packet.done and ((timeout is None) or (timeout >= i)): + i += 1 yield + def callback(self, packet): packet = EtherbonePacket(packet) packet.decode() diff --git a/test/model/icmp.py b/test/model/icmp.py index faf5c90d..19235d95 100644 --- a/test/model/icmp.py +++ b/test/model/icmp.py @@ -32,6 +32,9 @@ def decode(self): setattr(self, k, get_field_data(v, header)) def encode(self): + ''' + insert header bytes in front of the payload bytes + ''' header = 0 for k, v in sorted(icmp_header.fields.items()): value = merge_bytes(split_bytes(getattr(self, k), @@ -41,6 +44,13 @@ def encode(self): for d in split_bytes(header, icmp_header.length): self.insert(0, d) + def insert_checksum(self): + self[2] = 0 + self[3] = 0 + c = ip.checksum(self) + self[2] = c & 0xff + self[3] = (c >> 8) & 0xff + def __repr__(self): r = "--------\n" for k in sorted(icmp_header.fields.keys()): @@ -63,21 +73,22 @@ def __init__(self, ip, ip_address, debug=False): self.ip.set_icmp_callback(self.callback) - def send(self, packet): + def send(self, packet, target_ip=0xFFFFFFFF): packet.encode() + packet.insert_checksum() if self.debug: print_icmp(">>>>>>>>") print_icmp(packet) ip_packet = ip.IPPacket(packet) ip_packet.version = 0x4 ip_packet.ihl = 0x5 - ip_packet.total_length = len(packet) + ip_packet.ihl + ip_packet.total_length = len(packet) + 20 ip_packet.identification = 0 ip_packet.flags = 0 ip_packet.fragment_offset = 0 ip_packet.ttl = 0x80 ip_packet.sender_ip = self.ip_address - ip_packet.target_ip = 0x12345678 # XXX + ip_packet.target_ip = target_ip ip_packet.checksum = 0 ip_packet.protocol = icmp_protocol self.ip.send(ip_packet) diff --git a/test/model/ip.py b/test/model/ip.py index ea4e2c67..5183bf00 100644 --- a/test/model/ip.py +++ b/test/model/ip.py @@ -24,10 +24,17 @@ def carry_around_add(a, b): def checksum(msg): + ''' + http://www.faqs.org/rfcs/rfc1071.html + ''' s = 0 - for i in range(0, len(msg), 2): - w = msg[i] + (msg[i+1] << 8) + for i in range(len(msg) // 2): + w = msg[2 * i] + (msg[2 * i + 1] << 8) s = carry_around_add(s, w) + # print(hex(w), hex(s)) + if len(msg) % 2: # Add the single remaining byte + s = carry_around_add(s, msg[-1]) + # print(hex(msg[-1]), hex(s)) return ~s & 0xffff diff --git a/test/model/phy.py b/test/model/phy.py index c11904a9..e07349db 100644 --- a/test/model/phy.py +++ b/test/model/phy.py @@ -4,6 +4,7 @@ # Copyright (c) 2015-2019 Florent Kermarrec # SPDX-License-Identifier: BSD-2-Clause +from struct import pack from litex.soc.interconnect.stream_sim import * from liteeth.common import * @@ -13,56 +14,89 @@ def print_phy(s): print_with_prefix(s, "[PHY]") - # PHY Source --------------------------------------------------------------------------------------- class PHYSource(PacketStreamer): - def __init__(self, dw): - PacketStreamer.__init__(self, eth_phy_description(dw)) + def __init__(self, dw, assertStall): + PacketStreamer.__init__(self, eth_phy_description(dw), dw=dw, assertStall=assertStall) # PHY Sink ----------------------------------------------------------------------------------------- class PHYSink(PacketLogger): - def __init__(self, dw): - PacketLogger.__init__(self, eth_phy_description(dw)) + def __init__(self, dw, assertStall): + PacketLogger.__init__(self, eth_phy_description(dw), dw=dw, assertStall=assertStall) # PHY ---------------------------------------------------------------------------------------------- +LINKTYPE_ETHERNET = 1 +LINKTYPE_RAW = 101 +LINKTYPE_ETHERNET_MPACKET = 274 + class PHY(Module): - def __init__(self, dw, debug=False): + def __init__(self, dw, debug=False, pcap_file=None, assertStall=False): self.dw = dw self.debug = debug - self.submodules.phy_source = PHYSource(dw) - self.submodules.phy_sink = PHYSink(dw) + self.submodules.phy_source = PHYSource(dw, assertStall) + self.submodules.phy_sink = PHYSink(dw, assertStall) self.source = self.phy_source.source self.sink = self.phy_sink.sink self.mac_callback = None + self.cc = 0 + self.pcap_file = pcap_file + if pcap_file is not None: + file_header = pack( + 'IHHiIII', + 0xa1b2c3d4, + 2, + 4, + 0, + 0, + 65535, + LINKTYPE_ETHERNET_MPACKET + ) + with open(pcap_file, 'wb') as f: + f.write(file_header) + def set_mac_callback(self, callback): self.mac_callback = callback def send(self, datas): - packet = Packet(datas) + n_bytes = len(datas) if self.debug: r = ">>>>>>>>\n" - r += "length " + str(len(datas)) + "\n" + r += "length " + str(n_bytes) + "\n" for d in datas: - r += "{:02x}".format(d) + r += f'{d:02x} ' print_phy(r) - self.phy_source.send(packet) + + if self.pcap_file is not None and n_bytes > 0: + with open(self.pcap_file, 'ab') as f: + f.write(pack('IIII', self.cc, 0, n_bytes, n_bytes)) + f.write(bytes(datas)) + + self.phy_source.send(Packet(datas)) def receive(self): yield from self.phy_sink.receive() + self.packet = p = self.phy_sink.packet # Each item is a byte if self.debug: r = "<<<<<<<<\n" - r += "length " + str(len(self.phy_sink.packet)) + "\n" - for d in self.phy_sink.packet: - r += "{:02x}".format(d) + r += "length " + str(len(p)) + "\n" + for d in p: + r += f'{d:02x} ' print_phy(r) - self.packet = self.phy_sink.packet + + if self.pcap_file is not None: + ll = len(self.packet) # - 8 + if ll > 0: + with open(self.pcap_file, 'ab') as f: + f.write(pack('IIII', self.cc, 0, ll, ll)) + f.write(bytes(self.packet)) + @passive def generator(self): @@ -70,3 +104,4 @@ def generator(self): yield from self.receive() if self.mac_callback is not None: self.mac_callback(self.packet) + self.cc += 1 diff --git a/test/model/udp.py b/test/model/udp.py index 05aaabc5..ec163723 100644 --- a/test/model/udp.py +++ b/test/model/udp.py @@ -69,7 +69,7 @@ def __init__(self, ip, ip_address, debug=False, loopback=False): def set_etherbone_callback(self, callback): self.etherbone_callback = callback - def send(self, packet): + def send(self, packet, target_ip=0x12345678): packet.encode() if self.debug: print_udp(">>>>>>>>") @@ -77,13 +77,13 @@ def send(self, packet): ip_packet = ip.IPPacket(packet) ip_packet.version = 0x4 ip_packet.ihl = 0x5 - ip_packet.total_length = len(packet) + ip_packet.ihl + ip_packet.total_length = len(packet) + ip_packet.ihl * 4 ip_packet.identification = 0 ip_packet.flags = 0 ip_packet.fragment_offset = 0 ip_packet.ttl = 0x80 ip_packet.sender_ip = self.ip_address - ip_packet.target_ip = 0x12345678 # FIXME + ip_packet.target_ip = target_ip ip_packet.checksum = 0 ip_packet.protocol = udp_protocol self.ip.send(ip_packet) diff --git a/test/test_etherbone.py b/test/test_etherbone.py index e5cb2bcb..47df48ae 100644 --- a/test/test_etherbone.py +++ b/test/test_etherbone.py @@ -14,102 +14,137 @@ from liteeth.common import * from liteeth.core import LiteEthUDPIPCore from liteeth.frontend.etherbone import LiteEthEtherbone - from test.model import phy, mac, arp, ip, udp, etherbone - from litex.gen.sim import * + ip_address = 0x12345678 mac_address = 0x12345678abcd class DUT(Module): - def __init__(self): - self.submodules.phy_model = phy.PHY(8, debug=False) - self.submodules.mac_model = mac.MAC(self.phy_model, debug=False, loopback=False) - self.submodules.arp_model = arp.ARP(self.mac_model, mac_address, ip_address, debug=False) - self.submodules.ip_model = ip.IP(self.mac_model, mac_address, ip_address, debug=False, loopback=False) - self.submodules.udp_model = udp.UDP(self.ip_model, ip_address, debug=False, loopback=False) + def __init__(self, dw): + self.submodules.phy_model = phy.PHY(dw, assertStall=dw < 64, pcap_file='dump_wishbone.pcap') + self.submodules.mac_model = mac.MAC(self.phy_model) + self.submodules.arp_model = arp.ARP(self.mac_model, mac_address, ip_address) + self.submodules.ip_model = ip.IP(self.mac_model, mac_address, ip_address) + self.submodules.udp_model = udp.UDP(self.ip_model, ip_address) self.submodules.etherbone_model = etherbone.Etherbone(self.udp_model, debug=False) - self.submodules.core = LiteEthUDPIPCore(self.phy_model, mac_address, ip_address, 100000) - self.submodules.etherbone = LiteEthEtherbone(self.core.udp, 0x1234) + self.submodules.core = LiteEthUDPIPCore( + self.phy_model, + mac_address + 1, + ip_address + 1, + 100000, + with_icmp=False, + dw=dw + ) + self.submodules.etherbone = LiteEthEtherbone(self.core.udp, 0x1234, buffer_depth=8) self.submodules.sram = wishbone.SRAM(1024) self.submodules.interconnect = wishbone.InterconnectPointToPoint(self.etherbone.wishbone.bus, self.sram.bus) -def main_generator(dut): - test_probe = True - test_writes = True - test_reads = True - - # test probe - if test_probe: +class TestEtherbone(unittest.TestCase): + def do_probe(self, dut): packet = etherbone.EtherbonePacket() packet.pf = 1 + packet.encode() + packet.bytes += bytes([0x00] * 4) # Add padding + + dut.etherbone_model.send(packet, ip_address + 1) + yield from dut.etherbone_model.receive(400) + self.assertEqual(dut.etherbone_model.rx_packet.pr, 1) + + def do_writes(self, dut, writes_datas): + for i, wd in enumerate(writes_datas): + yield dut.sram.mem[i].eq(0) + + writes = etherbone.EtherboneWrites(base_addr=0x1000, datas=writes_datas) + record = etherbone.EtherboneRecord() + record.writes = writes + record.reads = None + record.bca = 0 + record.rca = 0 + record.rff = 0 + record.cyc = 0 + record.wca = 0 + record.wff = 0 + record.byte_enable = 0xf + record.wcount = len(writes_datas) + record.rcount = 0 + + packet = etherbone.EtherbonePacket() + packet.records = [record] dut.etherbone_model.send(packet) - yield from dut.etherbone_model.receive() - print("probe: " + str(bool(dut.etherbone_model.rx_packet.pr))) - - for i in range(2): - # test writes - if test_writes: - writes_datas = [j for j in range(4)] - writes = etherbone.EtherboneWrites(base_addr=0x1000, datas=writes_datas) - record = etherbone.EtherboneRecord() - record.writes = writes - record.reads = None - record.bca = 0 - record.rca = 0 - record.rff = 0 - record.cyc = 0 - record.wca = 0 - record.wff = 0 - record.byte_enable = 0xf - record.wcount = len(writes_datas) - record.rcount = 0 - - packet = etherbone.EtherbonePacket() - packet.records = [record] - dut.etherbone_model.send(packet) - for i in range(256): - yield - - # test reads - if test_reads: - reads_addrs = [0x1000 + 4*j for j in range(4)] - reads = etherbone.EtherboneReads(base_ret_addr=0x1000, addrs=reads_addrs) - record = etherbone.EtherboneRecord() - record.writes = None - record.reads = reads - record.bca = 0 - record.rca = 0 - record.rff = 0 - record.cyc = 0 - record.wca = 0 - record.wff = 0 - record.byte_enable = 0xf - record.wcount = 0 - record.rcount = len(reads_addrs) - - packet = etherbone.EtherbonePacket() - packet.records = [record] - dut.etherbone_model.send(packet) - yield from dut.etherbone_model.receive() - loopback_writes_datas = [] - loopback_writes_datas = dut.etherbone_model.rx_packet.records.pop().writes.get_datas() - - # check resultss - s, l, e = check(writes_datas, loopback_writes_datas) - print("shift " + str(s) + " / length " + str(l) + " / errors " + str(e)) + for i in range(256): + yield + if (yield dut.sram.bus.cyc): + break + for i in range(32): + yield + + for i, wd in enumerate(writes_datas): + val = (yield dut.sram.mem[i]) + self.assertEqual(val, wd) + + # Check for infinite packet send loop (last_be bug in StrideConverter) + self.assertEqual((yield dut.etherbone.record.receiver.fsm.state), 0) + + def do_reads(self, dut, writes_datas): + reads_addrs = [] + for i, wd in enumerate(writes_datas): + yield dut.sram.mem[i].eq(wd) + reads_addrs.append(0x1000 + i * 4) + + reads = etherbone.EtherboneReads(base_ret_addr=0x1000, addrs=reads_addrs) + record = etherbone.EtherboneRecord() + record.writes = None + record.reads = reads + record.bca = 0 + record.rca = 0 + record.rff = 0 + record.cyc = 0 + record.wca = 0 + record.wff = 0 + record.byte_enable = 0xf + record.wcount = 0 + record.rcount = len(reads_addrs) -class TestEtherbone(unittest.TestCase): - def test_etherbone(self): - dut = DUT() + packet = etherbone.EtherbonePacket() + packet.records = [record] + dut.etherbone_model.send(packet) + yield from dut.etherbone_model.receive(400) + + # Check for infinite packet send loop (crossbar bug) + self.assertEqual((yield dut.etherbone.record.receiver.fsm.state), 0) + + reads_datas = dut.etherbone_model.rx_packet.records.pop().writes.get_datas() + + # check results + self.assertEqual(writes_datas, reads_datas) + + def main_generator(self, dut): + writes_datas = [((0xA + j) << 24) + j for j in range(4)] + + # push IP address into ARP table to speed up sim. + yield dut.core.arp.table.cached_valid.eq(1) + yield dut.core.arp.table.cached_ip_address.eq(ip_address) + yield dut.core.arp.table.cached_mac_address.eq(mac_address) + + with self.subTest("do_probe"): + yield from self.do_probe(dut) + with self.subTest("do_writes"): + yield from self.do_writes(dut, [writes_datas[0]]) + yield from self.do_writes(dut, writes_datas) + with self.subTest("do_reads"): + yield from self.do_reads(dut, [writes_datas[0]]) + yield from self.do_reads(dut, writes_datas) + + def do_test(self, dut): generators = { - "sys" : [main_generator(dut)], + "sys" : [self.main_generator(dut)], "eth_tx": [dut.phy_model.phy_sink.generator(), dut.phy_model.generator()], "eth_rx": dut.phy_model.phy_source.generator() @@ -117,4 +152,13 @@ def test_etherbone(self): clocks = {"sys": 10, "eth_rx": 10, "eth_tx": 10} - #run_simulation(dut, generators, clocks, vcd_name="sim.vcd") # FIXME: hanging + run_simulation(dut, generators, clocks, vcd_name=f'test_etherbone.vcd') + + def test_etherbone_dw_8(self): + self.do_test(DUT(8)) + + def test_etherbone_dw_32(self): + self.do_test(DUT(32)) + + def test_etherbone_dw_64(self): + self.do_test(DUT(64)) diff --git a/test/test_icmp.py b/test/test_icmp.py index 9bb3af18..051e133c 100644 --- a/test/test_icmp.py +++ b/test/test_icmp.py @@ -8,7 +8,6 @@ from migen import * -from litex.soc.interconnect import wishbone from litex.soc.interconnect.stream_sim import * from liteeth.common import * @@ -22,44 +21,100 @@ from litex.gen.sim import * -ip_address = 0x12345678 -mac_address = 0x12345678abcd +model_ip = convert_ip("192.168.10.1") +model_mac = 0x12345678abcd + +dut_ip = convert_ip("192.168.10.50") +dut_mac = 0x12345678ffff + +got_ping_reply = False + + +class ICMP(icmp.ICMP): + def process(self, p): + global got_ping_reply + print("Received ping reply", p) + tc = unittest.TestCase() + tc.assertEqual(p.code, 0) + tc.assertEqual(p.ident, 0x69b3) + tc.assertEqual(p.checksum, 0xaac0) + tc.assertEqual(p.sequence, 1) + got_ping_reply = True class DUT(Module): - def __init__(self): - self.submodules.phy_model = phy.PHY(8, debug=True) - self.submodules.mac_model = mac.MAC(self.phy_model, debug=True, loopback=False) - self.submodules.arp_model = arp.ARP(self.mac_model, mac_address, ip_address, debug=True) - self.submodules.ip_model = ip.IP(self.mac_model, mac_address, ip_address, debug=True, loopback=False) - self.submodules.icmp_model = icmp.ICMP(self.ip_model, ip_address, debug=True) + def __init__(self, dw=8): + self.clock_domains.cd_sys = ClockDomain() + self.clock_domains.eth_tx = ClockDomain() + self.clock_domains.eth_rx = ClockDomain() + self.dw = dw + + self.submodules.phy_model = phy.PHY(self.dw, pcap_file='dump_icmp.pcap', assertStall=True) + self.submodules.mac_model = mac.MAC(self.phy_model) + self.submodules.arp_model = arp.ARP(self.mac_model, model_mac, model_ip) + self.submodules.ip_model = ip.IP(self.mac_model, model_mac, model_ip) + self.submodules.icmp_model = ICMP(self.ip_model, model_ip, debug=True) + + self.submodules.ip = LiteEthIPCore(self.phy_model, dut_mac, dut_ip, 100000, dw=dw) + - self.submodules.ip = LiteEthIPCore(self.phy_model, mac_address, ip_address, 100000) +def send_icmp(dut, msgtype=icmp_type_ping_request, code=0): + p = icmp.ICMPPacket(b"Hello World 123456") + p.code = code + p.checksum = 0 + p.msgtype = msgtype + p.ident = 0x69b3 + p.sequence = 0x1 + dut.icmp_model.send(p, target_ip=dut_ip) def main_generator(dut): - packet = MACPacket(ping_request) - packet.decode_remove_header() - packet = IPPacket(packet) - packet.decode() - packet = ICMPPacket(packet) - packet.decode() - dut.icmp_model.send(packet) - - for i in range(256): + global got_ping_reply + tc = unittest.TestCase() + + # push IP address into ARP table to speed up sim. + yield dut.ip.arp.table.cached_valid.eq(1) + yield dut.ip.arp.table.cached_ip_address.eq(model_ip) + yield dut.ip.arp.table.cached_mac_address.eq(model_mac) + + # We expect a ping reply to this (after ARP query) + send_icmp(dut) + for i in range(512): yield + if got_ping_reply: + break + tc.assertTrue(got_ping_reply, "Missing ping reply") + + # We expect no ping reply to this + # got_ping_reply = False + # send_icmp(dut, 3, 3) + # for i in range(256): + # tc.assertFalse(got_ping_reply, "Inappropriate ping reply") + # yield class TestICMP(unittest.TestCase): - def test(self): - dut = DUT() + def work(self, dw): + global got_ping_reply + got_ping_reply = False + dut = DUT(dw) generators = { "sys" : [main_generator(dut)], "eth_tx": [dut.phy_model.phy_sink.generator(), dut.phy_model.generator()], "eth_rx": dut.phy_model.phy_source.generator() } - clocks = {"sys": 10, + # f_sys must be >= f_eth_* + clocks = {"sys": 9, "eth_rx": 10, "eth_tx": 10} - run_simulation(dut, generators, clocks, vcd_name="sim.vcd") + run_simulation(dut, generators, clocks, vcd_name=f'test_icmp_{dw}.vcd') + + def test_8(self): + self.work(8) + + def test_32(self): + self.work(32) + + def test_64(self): + self.work(64) diff --git a/test/test_mac_core.py b/test/test_mac_core.py index 3a1a6480..f3ef0ed0 100644 --- a/test/test_mac_core.py +++ b/test/test_mac_core.py @@ -25,7 +25,7 @@ def __init__(self): self.submodules.mac_model = mac.MAC(self.phy_model, debug=False, loopback=True) self.submodules.core = LiteEthMACCore(phy=self.phy_model, dw=8, with_preamble_crc=True) - self.submodules.streamer = PacketStreamer(eth_phy_description(8), last_be=1) + self.submodules.streamer = PacketStreamer(eth_phy_description(8)) self.submodules.streamer_randomizer = Randomizer(eth_phy_description(8), level=50) self.submodules.logger_randomizer = Randomizer(eth_phy_description(8), level=50) diff --git a/test/test_packet_streamer.py b/test/test_packet_streamer.py new file mode 100644 index 00000000..ccd1ac64 --- /dev/null +++ b/test/test_packet_streamer.py @@ -0,0 +1,45 @@ +''' +TODO add checking of output stream +''' +import unittest + +from migen import * +from litex.soc.interconnect.stream_sim import * +from liteeth.common import * + + +class DUT(Module): + def __init__(self, dw=8): + self.source = stream.Endpoint(eth_phy_description(dw)) + + ### + + self.dw = dw + self.submodules.phy_source = PacketStreamer( + eth_phy_description(dw), dw=dw + ) + self.comb += [ + self.phy_source.source.connect(self.source) + ] + + +def main_generator(dut): + print() + p = Packet(range(10)) + dut.phy_source.send(p) + # dut.phy_source.send(p) + for i in range(64): + yield (dut.source.ready.eq(i % 2)) + yield + + +class TestPacketStreamer(unittest.TestCase): + def test(self): + dut = DUT(64) + generators = { + "sys": [ + main_generator(dut), + dut.phy_source.generator() + ], + } + run_simulation(dut, generators, vcd_name="sim.vcd")