diff --git a/litex/soc/interconnect/avalon.py b/litex/soc/interconnect/avalon.py index fbea428b1..307b4b253 100644 --- a/litex/soc/interconnect/avalon.py +++ b/litex/soc/interconnect/avalon.py @@ -2,6 +2,7 @@ # This file is part of LiteX. # # Copyright (c) 2019-2020 Florent Kermarrec +# Copyright (c) 2023 Hans Baier # SPDX-License-Identifier: BSD-2-Clause """Avalon support for LiteX""" @@ -9,6 +10,235 @@ from migen import * from litex.soc.interconnect import stream +from litex.soc.interconnect import wishbone + +_layout = [ + ("address", "adr_width", DIR_M_TO_S), + ("writedata", "data_width", DIR_M_TO_S), + ("readdata", "data_width", DIR_S_TO_M), + ("readdatavalid", 1, DIR_S_TO_M), + ("byteenable", "sel_width", DIR_M_TO_S), + ("read", 1, DIR_M_TO_S), + ("write", 1, DIR_M_TO_S), + ("waitrequest", 1, DIR_S_TO_M), + ("burstbegin", 1, DIR_M_TO_S), # this is optional + ("burstcount", 8, DIR_M_TO_S), + ("chipselect", 1, DIR_M_TO_S), # this is optional +] + +class AvalonMMInterface(Record): + def __init__(self, data_width=32, adr_width=30, **kwargs): + self.data_width = data_width + if kwargs.get("adr_width", False): + adr_width = kwargs["adr_width"] - int(log2(data_width//8)) + self.adr_width = adr_width + Record.__init__(self, set_layout_parameters(_layout, + adr_width = adr_width, + data_width = data_width, + sel_width = data_width//8)) + self.address.reset_less = True + self.writedata.reset_less = True + self.readdata.reset_less = True + self.byteenable.reset_less = True + + @staticmethod + def like(other): + return AvalonMMInterface(len(other.writedata)) + + def get_ios(self, bus_name="avl"): + subsignals = [] + for name, width, direction in self.layout: + subsignals.append(Subsignal(name, Pins(width))) + ios = [(bus_name , 0) + tuple(subsignals)] + return ios + + def connect_to_pads(self, pads, mode="master"): + assert mode in ["slave", "master"] + r = [] + for name, width, direction in self.layout: + sig = getattr(self, name) + pad = getattr(pads, name) + if mode == "master": + if direction == DIR_M_TO_S: + r.append(pad.eq(sig)) + else: + r.append(sig.eq(pad)) + else: + if direction == DIR_S_TO_M: + r.append(pad.eq(sig)) + else: + r.append(sig.eq(pad)) + return r + + def bus_read(self, address, byteenable=None, burstcount=1, chipselect=None): + if byteenable is None: + byteenable = 2**len(self.byteenable) - 1 + yield self.address.eq(address) + yield self.write.eq(0) + yield self.read.eq(1) + yield self.byteenable.eq(byteenable) + if burstcount != 1: + yield self.burstcount.eq(burstcount) + if chipselect is not None: + yield self.chipselect.eq(chipselect) + yield + while (yield self.waitrequest): + yield + yield self.read.eq(0) + # actually don't care outside of a transaction + # this makes the traces look neater + yield self.byteenable.eq(0) + if burstcount != 1: + yield self.burstcount.eq(0) + if chipselect is not None: + yield self.chipselect.eq(0) + + while not (yield self.readdatavalid): + yield + return (yield self.readdata) + + def continue_read_burst(self): + yield + return (yield self.readdata) + + def bus_write(self, address, writedata, byteenable=None, chipselect=None): + if not isinstance(writedata, list): + writedata = [ writedata ] + burstcount = len(writedata) + if byteenable is None: + byteenable = 2**len(self.byteenable) - 1 + yield self.address.eq(address) + yield self.write.eq(1) + yield self.read.eq(0) + yield self.byteenable.eq(byteenable) + if burstcount is not None: + yield self.burstcount.eq(burstcount) + if chipselect is not None: + yield self.chipselect.eq(chipselect) + for data in writedata: + yield self.writedata.eq(data) + yield + while (yield self.waitrequest): + yield + yield self.burstcount.eq(0) + yield self.writedata.eq(0) + yield self.write.eq(0) + # actually don't care outside of a transaction + # this makes the traces look neater + yield self.byteenable.eq(0) + if chipselect is not None: + yield self.chipselect.eq(0) + +class AvalonMM2Wishbone(Module): + def __init__(self, data_width=32, address_width=32, wishbone_base_address=0x0, wishbone_extend_address_bits=0, avoid_combinatorial_loop=True): + word_width = data_width // 8 + word_width_bits = log2_int(word_width) + wishbone_address_width = address_width - word_width_bits + wishbone_extend_address_bits + + self.a2w_wb = wb = wishbone.Interface(data_width=data_width, adr_width=wishbone_address_width, bursting=True) + self.a2w_avl = avl = AvalonMMInterface (data_width=data_width, adr_width=address_width) + + read_access = Signal() + readdatavalid = Signal() + readdata = Signal(data_width) + + last_burst_cycle = Signal() + burst_cycle = Signal() + burst_counter = Signal.like(avl.burstcount) + burst_address = Signal(address_width) + burst_read = Signal() + burst_sel = Signal.like(avl.byteenable) + + self.sync += last_burst_cycle.eq(burst_cycle) + + # Some designs might have trouble with the combinatorial loop created + # by wb.ack, so cut it, incurring one clock cycle of overhead on each + # bus transaction + if avoid_combinatorial_loop: + self.sync += [ + If (wb.ack | wb.err, read_access.eq(0)) \ + .Elif(avl.read, read_access.eq(1)), + readdata.eq(wb.dat_r), + readdatavalid.eq((wb.ack | wb.err) & read_access), + ] + else: + self.comb += [ + read_access.eq(avl.read), + readdata.eq(wb.dat_r), + readdatavalid.eq((wb.ack | wb.err) & read_access), + ] + + # Wishbone -> Avalon + self.comb += [ + avl.waitrequest.eq(~(wb.ack | wb.err) | burst_read), + avl.readdata.eq(readdata), + avl.readdatavalid.eq(readdatavalid), + ] + + # Avalon -> Wishbone + self.comb += [ + # avalon is byte addresses, wishbone word addressed + wb.adr.eq(Mux(burst_cycle & last_burst_cycle, + burst_address, avl.address)[word_width_bits:] + + Constant(wishbone_base_address, (wishbone_address_width, 0))), + wb.dat_w.eq(avl.writedata), + wb.we.eq(avl.write), + wb.cyc.eq(read_access | avl.write | burst_cycle), + wb.stb.eq(read_access | avl.write), + wb.bte.eq(Constant(0, 2)), + ] + + self.submodules.fsm = fsm = FSM(reset_state="NORMAL") + fsm.act("NORMAL", + burst_cycle.eq(0), + wb.sel.eq(avl.byteenable), + wb.cti.eq(Mux(avl.burstcount > 1, + wishbone.CTI_BURST_INCREMENTING, + wishbone.CTI_BURST_NONE)), + If(~avl.waitrequest & (avl.burstcount > 1), + burst_cycle.eq(1), + NextValue(burst_counter, avl.burstcount - 1), + NextValue(burst_address, avl.address + word_width), + NextValue(burst_sel, avl.byteenable), + If(avl.write, NextState("BURST_WRITE")), + If(avl.read, + NextValue(burst_read, 1), + NextState("BURST_READ"))) + ) + fsm.act("BURST_WRITE", + burst_cycle.eq(1), + wb.sel.eq(burst_sel), + wb.cti.eq(Mux(burst_counter > 1, + wishbone.CTI_BURST_INCREMENTING, + Mux(burst_counter == 1, wishbone.CTI_BURST_END, wishbone.CTI_BURST_NONE))), + If(~avl.waitrequest, + NextValue(burst_address, burst_address + word_width), + NextValue(burst_counter, burst_counter - 1)), + If(burst_counter == 0, + burst_cycle.eq(0), + wb.sel.eq(avl.byteenable), + NextValue(burst_sel, 0), + NextState("NORMAL")) + ) + fsm.act("BURST_READ", # TODO + burst_cycle.eq(1), + wb.stb.eq(1), + wb.sel.eq(burst_sel), + wb.cti.eq(Mux(burst_counter > 1, + wishbone.CTI_BURST_INCREMENTING, + Mux(burst_counter == 1, wishbone.CTI_BURST_END, wishbone.CTI_BURST_NONE))), + If (wb.ack, + avl.readdatavalid.eq(1), + NextValue(burst_address, burst_address + word_width), + NextValue(burst_counter, burst_counter - 1)), + If (burst_counter == 0, + wb.cyc.eq(0), + wb.stb.eq(0), + wb.sel.eq(avl.byteenable), + NextValue(burst_sel, 0), + NextValue(burst_read, 0), + NextState("NORMAL")) + ) # Avalon-ST to/from native LiteX's stream ---------------------------------------------------------- diff --git a/test/test_avalon.py b/test/test_avalon.py new file mode 100644 index 000000000..4e7dbbd81 --- /dev/null +++ b/test/test_avalon.py @@ -0,0 +1,72 @@ +# +# This file is part of LiteX. +# +# Copyright (c) 2023 Hans Baier +# SPDX-License-Identifier: BSD-2-Clause + +import unittest + +from migen import * + +from litex.soc.interconnect import wishbone, avalon + +# TestWishbone ------------------------------------------------------------------------------------- + +class TestAvalon2Wishbone(unittest.TestCase): + + def test_sram(self): + def generator(dut): + yield from dut.avl.bus_write(0x0000, 0x01234567) + yield from dut.avl.bus_write(0x0004, 0x89abcdef) + yield from dut.avl.bus_write(0x0008, 0xdeadbeef) + yield from dut.avl.bus_write(0x000c, 0xc0ffee00) + yield from dut.avl.bus_write(0x0010, 0x76543210) + yield + self.assertEqual((yield from dut.avl.bus_read(0x0000)), 0x01234567) + self.assertEqual((yield from dut.avl.bus_read(0x0004)), 0x89abcdef) + self.assertEqual((yield from dut.avl.bus_read(0x0008)), 0xdeadbeef) + self.assertEqual((yield from dut.avl.bus_read(0x000c)), 0xc0ffee00) + self.assertEqual((yield from dut.avl.bus_read(0x0010)), 0x76543210) + + class DUT(Module): + def __init__(self): + a2w = avalon.AvalonMM2Wishbone() + self.avl = a2w.a2w_avl + wishbone_mem = wishbone.SRAM(32, bus=a2w.a2w_wb) + self.submodules += a2w + self.submodules += wishbone_mem + + dut = DUT() + run_simulation(dut, generator(dut)) #, vcd_name="avalon.vcd") + + def test_sram_burst(self): + def generator(dut): + yield from dut.avl.bus_write(0x0, [0x01234567, 0x89abcdef, 0xdeadbeef, 0xc0ffee00, 0x76543210]) + yield + self.assertEqual((yield from dut.avl.bus_read(0x0000, burstcount=5)), 0x01234567) + self.assertEqual((yield from dut.avl.continue_read_burst()), 0x89abcdef) + self.assertEqual((yield from dut.avl.continue_read_burst()), 0xdeadbeef) + self.assertEqual((yield from dut.avl.continue_read_burst()), 0xc0ffee00) + self.assertEqual((yield from dut.avl.continue_read_burst()), 0x76543210) + yield + yield + yield + yield + self.assertEqual((yield from dut.avl.bus_read(0x0000)), 0x01234567) + self.assertEqual((yield from dut.avl.bus_read(0x0004)), 0x89abcdef) + self.assertEqual((yield from dut.avl.bus_read(0x0008)), 0xdeadbeef) + self.assertEqual((yield from dut.avl.bus_read(0x000c)), 0xc0ffee00) + self.assertEqual((yield from dut.avl.bus_read(0x0010)), 0x76543210) + yield + yield + + class DUT(Module): + def __init__(self): + a2w = avalon.AvalonMM2Wishbone() + self.avl = a2w.a2w_avl + wishbone_mem = wishbone.SRAM(32, bus=a2w.a2w_wb) + self.submodules += a2w + self.submodules += wishbone_mem + + dut = DUT() + run_simulation(dut, generator(dut)) #, vcd_name="avalon_burst.vcd")