from liteeth.mac import LiteEthMAC
from liteeth.core.arp import LiteEthARP
from liteeth.core.ip import LiteEthIP
+from liteeth.core.udp import LiteEthUDP
class LiteEthIPCore(Module, AutoCSR):
def __init__(self, phy, mac_address, ip_address):
self.phy = phy
- self.submodules.mac = mac = LiteEthMAC(phy, 8, interface="crossbar", with_hw_preamble_crc=True)
- self.submodules.arp = arp = LiteEthARP(mac, mac_address, ip_address)
- self.submodules.ip = ip = LiteEthIP(mac, mac_address, ip_address, arp.table)
+ self.submodules.mac = LiteEthMAC(phy, 8, interface="crossbar", with_hw_preamble_crc=True)
+ self.submodules.arp = LiteEthARP(self.mac, mac_address, ip_address)
+ self.submodules.ip = LiteEthIP(self.mac, mac_address, ip_address, self.arp.table)
self.sink, self.source = self.ip.sink, self.ip.source
+
+class LiteEthUDPIPCore(LiteEthIPCore):
+ def __init__(self, phy, mac_address, ip_address):
+ LiteEthIPCore.__init__(self, phy, mac_address, ip_address)
+ self.submodules.udp = LiteEthUDP(self.ip, ip_address)
+ self.sink, self.source = self.udp.sink, self.udp.source
self.sink = Sink(eth_udp_user_description(8))
self.source = Source(eth_ipv4_user_description(8))
###
- packetizer = LiteEthUDPV4Packetizer()
+ packetizer = LiteEthUDPPacketizer()
self.submodules += packetizer
self.comb += [
packetizer.sink.stb.eq(self.sink.stb),
)
fsm.act("SEND",
Record.connect(packetizer.source, self.source),
- self.source.length.eq(),
+ self.source.length.eq(packetizer.sink.length + ipv4_header_len),
self.source.protocol.eq(udp_protocol),
self.source.ip_address.eq(self.sink.ip_address),
If(self.source.stb & self.source.eop & self.source.ack,
self.sink = Sink(eth_ipv4_user_description(8))
self.source = source = Source(eth_udp_user_description(8))
###
- depacketizer = LiteEthUDPV4Depacketizer()
+ depacketizer = LiteEthUDPDepacketizer()
self.submodules += depacketizer
self.comb += Record.connect(self.sink, depacketizer.sink)
sink = depacketizer.source
valid = Signal()
self.comb += valid.eq(
sink.stb &
- (sink.protocol == udp_protocol) &
- (sink.ip_address == ip_address)
+ (self.sink.protocol == udp_protocol) &
+ (self.sink.ip_address == ip_address)
)
fsm.act("CHECK",
)
class LiteEthUDP(Module):
- def __init__(self, ip_address):
+ def __init__(self, ip, ip_address):
self.submodules.tx = LiteEthUDPTX(ip_address)
self.submodules.rx = LiteEthUDPRX(ip_address)
+ self.comb += [
+ Record.connect(self.tx.source, ip.sink),
+ Record.connect(ip.source, self.rx.sink)
+ ]
self.sink, self.source = self.tx.sink, self.rx.source
ip_tb:
$(CMD) ip_tb.py
-udp_tb:
- $(CMD) udp_tb.py
+udpip_tb:
+ $(CMD) udpip_tb.py
--- /dev/null
+from migen.fhdl.std import *
+from migen.bus import wishbone
+from migen.bus.transactions import *
+from migen.sim.generic import run_simulation
+
+from liteeth.common import *
+from liteeth.core import LiteEthUDPIPCore
+
+from liteeth.test.common import *
+from liteeth.test.model import phy, mac, arp, ip, udp
+
+ip_address = 0x12345678
+mac_address = 0x12345678abcd
+
+class TB(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=True)
+ self.submodules.udp_model = udp.UDP(self.ip_model, ip_address, debug=False, loopback=True)
+
+ self.submodules.udp_ip = LiteEthUDPIPCore(self.phy_model, mac_address, ip_address)
+
+ # use sys_clk for each clock_domain
+ self.clock_domains.cd_eth_rx = ClockDomain()
+ self.clock_domains.cd_eth_tx = ClockDomain()
+ self.comb += [
+ self.cd_eth_rx.clk.eq(ClockSignal()),
+ self.cd_eth_rx.rst.eq(ResetSignal()),
+ self.cd_eth_tx.clk.eq(ClockSignal()),
+ self.cd_eth_tx.rst.eq(ResetSignal()),
+ ]
+
+ def gen_simulation(self, selfp):
+ selfp.cd_eth_rx.rst = 1
+ selfp.cd_eth_tx.rst = 1
+ yield
+ selfp.cd_eth_rx.rst = 0
+ selfp.cd_eth_tx.rst = 0
+
+ for i in range(100):
+ yield
+
+ while True:
+ selfp.udp_ip.sink.stb = 1
+ selfp.udp_ip.sink.sop = 1
+ selfp.udp_ip.sink.eop = 1
+ selfp.udp_ip.sink.ip_address = 0x12345678
+ selfp.udp_ip.sink.source_port = 0x1234
+ selfp.udp_ip.sink.destination_port = 0x5678
+ selfp.udp_ip.sink.length = 64
+
+ selfp.udp_ip.source.ack = 1
+ if selfp.udp_ip.source.stb == 1 and selfp.udp_ip.source.sop == 1:
+ print("IP Packet / from ip_address %08x" %selfp.udp_ip.sink.source_port)
+
+ yield
+
+if __name__ == "__main__":
+ run_simulation(TB(), ncycles=2048, vcd_name="my.vcd", keep_files=True)