import enum
from nmigen import *
+from nmigen.utils import log2_int
from ..memory import MemoryMap
-__all__ = ["Element", "Interface", "Decoder"]
+__all__ = ["Element", "Interface", "Decoder", "Multiplexer"]
class Element(Record):
Address width. At most ``(2 ** addr_width) * data_width`` register bits will be available.
data_width : int
Data width. Registers are accessed in ``data_width`` sized chunks.
+ alignment : int
+ Alignment. See :class:`MemoryMap`.
name : str
Name of the underlying record.
Attributes
----------
+ memory_map : MemoryMap
+ Map of the bus.
addr : Signal(addr_width)
Address for reads and writes.
r_data : Signal(data_width)
- Read data. Valid on the next cycle after ``r_stb`` is asserted.
+ Read data. Valid on the next cycle after ``r_stb`` is asserted. Otherwise, zero. (Keeping
+ read data of an unused interface at zero simplifies multiplexers.)
r_stb : Signal()
Read strobe. If ``addr`` points to the first chunk of a register, captures register value
and causes read side effects to be performed (if any). If ``addr`` points to any chunk
nothing.
"""
- def __init__(self, *, addr_width, data_width, name=None):
+ def __init__(self, *, addr_width, data_width, alignment=0, name=None):
if not isinstance(addr_width, int) or addr_width <= 0:
raise ValueError("Address width must be a positive integer, not {!r}"
.format(addr_width))
.format(data_width))
self.addr_width = addr_width
self.data_width = data_width
+ self.memory_map = MemoryMap(addr_width=addr_width, data_width=data_width,
+ alignment=alignment)
super().__init__([
("addr", addr_width),
data_width : int
Data width. See :class:`Interface`.
alignment : int
- Register alignment. The address assigned to each register will be a multiple of
- ``2 ** alignment``.
+ Register alignment. See :class:`Interface`.
Attributes
----------
CSR bus providing access to registers.
"""
def __init__(self, *, addr_width, data_width, alignment=0):
- self.bus = Interface(addr_width=addr_width, data_width=data_width)
- self._map = MemoryMap(addr_width=addr_width, data_width=data_width, alignment=alignment)
+ self.bus = Interface(addr_width=addr_width, data_width=data_width, alignment=alignment)
+ self._map = self.bus.memory_map
def align_to(self, alignment):
"""Align the implicit address of the next register.
m.d.comb += self.bus.r_data.eq(r_data_fanin)
return m
+
+
+class Multiplexer(Elaboratable):
+ """CSR bus multiplexer.
+
+ An address-based multiplexer for subordinate CSR buses.
+
+ Usage
+ -----
+
+ Although there is no functional difference between adding a set of registers directly to
+ a :class:`Decoder` and adding a set of reigsters to multiple :class:`Decoder`s that are
+ aggregated with a :class:`Multiplexer`, hierarchical CSR buses are useful for organizing
+ a hierarchical design. If many peripherals are directly served by a single :class:`Decoder`,
+ a very large amount of ports will connect the peripheral registers with the decoder, and
+ the cost of decoding logic would not be attributed to specific peripherals. With a multiplexer,
+ only five signals per peripheral will be used, and the logic could be kept together with
+ the peripheral.
+
+ Parameters
+ ----------
+ addr_width : int
+ Address width. See :class:`Interface`.
+ data_width : int
+ Data width. See :class:`Interface`.
+ alignment : int
+ Window alignment. See :class:`Interface`.
+
+ Attributes
+ ----------
+ bus : :class:`Interface`
+ CSR bus providing access to subordinate buses.
+ """
+ def __init__(self, *, addr_width, data_width, alignment=0):
+ self.bus = Interface(addr_width=addr_width, data_width=data_width, alignment=alignment)
+ self._map = self.bus.memory_map
+ self._subs = dict()
+
+ def align_to(self, alignment):
+ """Align the implicit address of the next window.
+
+ See :meth:`MemoryMap.align_to` for details.
+ """
+ return self._map.align_to(alignment)
+
+ def add(self, sub_bus, *, addr=None):
+ """Add a window to a subordinate bus.
+
+ See :meth:`MemoryMap.add_resource` for details.
+ """
+ if not isinstance(sub_bus, Interface):
+ raise TypeError("Subordinate bus must be an instance of csr.Interface, not {!r}"
+ .format(sub_bus))
+ if sub_bus.data_width != self.bus.data_width:
+ raise ValueError("Subordinate bus has data width {}, which is not the same as "
+ "multiplexer data width {}"
+ .format(sub_bus.data_width, self.bus.data_width))
+
+ start, end, ratio = window_range = self._map.add_window(sub_bus.memory_map, addr=addr)
+ assert ratio == 1
+ pattern = "{:0{}b}{}".format(start >> sub_bus.addr_width,
+ self.bus.addr_width - sub_bus.addr_width,
+ "-" * sub_bus.addr_width)
+ self._subs[pattern] = sub_bus
+ return window_range
+
+ def elaborate(self, platform):
+ m = Module()
+
+ # See Decoder.elaborate above.
+ r_data_fanin = 0
+
+ with m.Switch(self.bus.addr):
+ for sub_pat, sub_bus in self._subs.items():
+ m.d.comb += sub_bus.addr.eq(self.bus.addr[:sub_bus.addr_width])
+
+ # The CSR bus interface is defined to output zero when idle, allowing us to avoid
+ # adding a multiplexer here.
+ r_data_fanin |= sub_bus.r_data
+ m.d.comb += sub_bus.w_data.eq(self.bus.w_data)
+
+ with m.Case(sub_pat):
+ m.d.comb += sub_bus.r_stb.eq(self.bus.r_stb)
+ m.d.comb += sub_bus.w_stb.eq(self.bus.w_stb)
+
+ m.d.comb += self.bus.r_data.eq(r_data_fanin)
+
+ return m
sim.add_clock(1e-6)
sim.add_sync_process(sim_test())
sim.run()
+
+
+class MultiplexerTestCase(unittest.TestCase):
+ def setUp(self):
+ self.dut = Multiplexer(addr_width=16, data_width=8)
+ Fragment.get(self.dut, platform=None) # silence UnusedElaboratable
+
+ def test_add_wrong_sub_bus(self):
+ with self.assertRaisesRegex(TypeError,
+ r"Subordinate bus must be an instance of csr\.Interface, not 1"):
+ self.dut.add(1)
+
+ def test_add_wrong_data_width(self):
+ decoder = Decoder(addr_width=10, data_width=16)
+ Fragment.get(decoder, platform=None) # silence UnusedElaboratable
+
+ with self.assertRaisesRegex(ValueError,
+ r"Subordinate bus has data width 16, which is not the same as "
+ r"multiplexer data width 8"):
+ self.dut.add(decoder.bus)
+
+ def test_sim(self):
+ dec_1 = Decoder(addr_width=10, data_width=8)
+ self.dut.add(dec_1.bus)
+ elem_1 = Element(8, "rw")
+ dec_1.add(elem_1)
+
+ dec_2 = Decoder(addr_width=10, data_width=8)
+ self.dut.add(dec_2.bus)
+ elem_2 = Element(8, "rw")
+ dec_2.add(elem_2, addr=2)
+
+ elem_1_addr, _, _ = self.dut.bus.memory_map.find_resource(elem_1)
+ elem_2_addr, _, _ = self.dut.bus.memory_map.find_resource(elem_2)
+ self.assertEqual(elem_1_addr, 0x0000)
+ self.assertEqual(elem_2_addr, 0x0402)
+
+ bus = self.dut.bus
+
+ def sim_test():
+ yield bus.addr.eq(elem_1_addr)
+ yield bus.w_stb.eq(1)
+ yield bus.w_data.eq(0x55)
+ yield
+ yield bus.w_stb.eq(0)
+ yield
+ self.assertEqual((yield elem_1.w_data), 0x55)
+
+ yield bus.addr.eq(elem_2_addr)
+ yield bus.w_stb.eq(1)
+ yield bus.w_data.eq(0xaa)
+ yield
+ yield bus.w_stb.eq(0)
+ yield
+ self.assertEqual((yield elem_2.w_data), 0xaa)
+
+ yield elem_1.r_data.eq(0x55)
+ yield elem_2.r_data.eq(0xaa)
+
+ yield bus.addr.eq(elem_1_addr)
+ yield bus.r_stb.eq(1)
+ yield
+ yield bus.addr.eq(elem_2_addr)
+ yield
+ self.assertEqual((yield bus.r_data), 0x55)
+ yield
+ self.assertEqual((yield bus.r_data), 0xaa)
+
+ m = Module()
+ m.submodules += self.dut, dec_1, dec_2
+ with Simulator(m, vcd_file=open("test.vcd", "w")) as sim:
+ sim.add_clock(1e-6)
+ sim.add_sync_process(sim_test())
+ sim.run()