WIP AvalonMM interface and Avalon to Wishbone Bridge (#1674)

Add initial AvalonMM interface and AvalonMM2Wishbone.

litex/soc/interconnect/avalon.py

@@ -2,6 +2,7 @@
 # This file is part of LiteX.
 #
 # Copyright (c) 2019-2020 Florent Kermarrec <florent@enjoy-digital.fr>
+# Copyright (c) 2023 Hans Baier <hansfbaier@gmail.com>
 # 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 ----------------------------------------------------------
 

test/test_avalon.py

@@ -0,0 +1,72 @@
+#
+# This file is part of LiteX.
+#
+# Copyright (c) 2023 Hans Baier <hansfbaier@gmail.com>
+# 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")