lambdasoc/test/test_periph_sdram.py

   1 # nmigen: UnusedElaboratable=no
   2
   3 import unittest
   4
   5 from nmigen import *
   6 from nmigen.test.utils import *
   7 from nmigen.asserts import *
   8 from nmigen.utils import log2_int
   9
  10 from nmigen_soc import wishbone
  11
  12 from ..periph.sdram import *
  13 from ..cores import litedram
  14
  15 from .utils.wishbone import WishboneSubordinateSpec
  16 from .utils.formal import FormalTestCase
  17
  18
  19 class WritebackCacheSpec(Elaboratable):
  20     def __init__(self, dut):
  21         self.dut = dut
  22
  23     def elaborate(self, platform):
  24         m = Module()
  25
  26         m.submodules.dut = dut = self.dut
  27
  28         # Wishbone interface
  29
  30         m.submodules.wb_sub_spec = WishboneSubordinateSpec(self.dut.intr_bus)
  31
  32         ratio        = dut.dram_port.data_width // dut.intr_bus.data_width
  33         nb_lines     = (dut.size * dut.intr_bus.granularity) // dut.dram_port.data_width
  34         intr_bus_adr = Record([
  35             ("offset", log2_int(ratio)),
  36             ("line",   log2_int(nb_lines)),
  37             ("tag",    dut.intr_bus.addr_width - log2_int(nb_lines) - log2_int(ratio)),
  38         ])
  39         m.d.comb += [
  40             Assert(len(intr_bus_adr) == len(dut.intr_bus.adr)),
  41             intr_bus_adr.eq(dut.intr_bus.adr),
  42         ]
  43
  44         dram_spec_addr = AnyConst(dut.dram_port.addr_width)
  45         dram_spec_data = Signal(dut.dram_port.data_width)
  46         dram_spec_line = Array(
  47             Signal(dut.intr_bus.data_width, name=f"dram_spec_word_{i}")
  48             for i in range(ratio)
  49         )
  50
  51         m.d.comb += dram_spec_data.eq(Cat(dram_spec_line))
  52
  53         with m.If(Initial()):
  54             m.d.comb += Assume(dram_spec_data == 0)
  55
  56         intr_dram_addr = Signal.like(dram_spec_addr)
  57         m.d.comb += intr_dram_addr.eq(Cat(intr_bus_adr.line, intr_bus_adr.tag))
  58
  59         with m.If(dut.intr_bus.cyc & dut.intr_bus.stb & dut.intr_bus.ack
  60                 & (intr_dram_addr == dram_spec_addr)
  61                 & dut.intr_bus.we):
  62             dram_spec_word = dram_spec_line[intr_bus_adr.offset]
  63             for i, sel_bit in enumerate(dut.intr_bus.sel):
  64                 dram_spec_bits = dram_spec_word    .word_select(i, dut.intr_bus.granularity)
  65                 intr_data_bits = dut.intr_bus.dat_w.word_select(i, dut.intr_bus.granularity)
  66                 with m.If(sel_bit):
  67                     m.d.sync += dram_spec_bits.eq(intr_data_bits)
  68
  69         # * A cache hit at `dram_spec_addr` must be coherent with `dram_spec_data`.
  70
  71         with m.If(dut.intr_bus.cyc & dut.intr_bus.stb & dut.intr_bus.ack
  72                 & (intr_dram_addr == dram_spec_addr)):
  73             dram_spec_word = dram_spec_line[intr_bus_adr.offset]
  74             m.d.comb += Assert(dut.intr_bus.dat_r == dram_spec_word)
  75
  76         # DRAM interface
  77
  78         dram_cmd  = Record([
  79             ("addr", dut.dram_port.addr_width),
  80             ("we",   1),
  81         ])
  82         dram_data = Record([
  83             ("r", dut.dram_port.data_width),
  84             ("w", dut.dram_port.data_width),
  85         ])
  86         dram_done = Record([
  87             ("cmd", 1),
  88             ("r",   1),
  89             ("w",   1),
  90         ])
  91
  92         with m.If(Initial()):
  93             m.d.comb += Assume(~dram_done.any())
  94
  95         with m.If(dut.dram_port.cmd.valid & dut.dram_port.cmd.ready):
  96             m.d.sync += [
  97                 dram_done.cmd.eq(1),
  98                 dram_cmd.addr.eq(dut.dram_port.cmd.addr),
  99                 dram_cmd.we  .eq(dut.dram_port.cmd.we),
 100             ]
 101         with m.If(dut.dram_port.w.valid & dut.dram_port.w.ready):
 102             m.d.sync += [
 103                 dram_done.w.eq(1),
 104                 dram_data.w.eq(dut.dram_port.w.data),
 105             ]
 106         with m.If(dut.dram_port.r.ready & dut.dram_port.r.valid):
 107             m.d.sync += [
 108                 dram_done.r.eq(1),
 109                 dram_data.r.eq(dut.dram_port.r.data),
 110             ]
 111
 112         with m.If(dram_done.cmd & dram_done.r):
 113             with m.If(dram_cmd.addr == dram_spec_addr):
 114                 m.d.comb += Assume(dram_data.r == dram_spec_data)
 115
 116         # Some of the following constraints are tighter than what the LiteDRAM native interface
 117         # actually allows. We may relax them in the future to improve cache performance.
 118
 119         # * A new command must not be initiated before the previous one has completed.
 120
 121         with m.If(dut.dram_port.cmd.valid):
 122             m.d.comb += Assert(~dram_done.cmd)
 123         with m.If(dut.dram_port.w.valid):
 124             m.d.comb += Assert(~dram_done.w)
 125         with m.If(dut.dram_port.r.ready):
 126             m.d.comb += Assert(~dram_done.r)
 127
 128         # * A command may either be a read or a write, but not both.
 129
 130         with m.If(dram_done.cmd):
 131             with m.If(dram_cmd.we):
 132                 m.d.comb += Assert(~dram_done.r)
 133             with m.Else():
 134                 m.d.comb += Assert(~dram_done.w)
 135
 136         m.d.comb += [
 137             Assert(dram_done.r.implies(~dram_done.w)),
 138             Assert(dram_done.w.implies(~dram_done.r)),
 139         ]
 140
 141         # * A read command consists of:
 142         #   - a transaction on the `dram.cmd` stream with `dram.cmd.we` de-asserted;
 143         #   - a transaction on the `dram.r` stream.
 144
 145         with m.If(dram_done.cmd & dram_done.r):
 146             m.d.comb += Assert(~dram_cmd.we)
 147             m.d.sync += [
 148                 dram_done.cmd.eq(0),
 149                 dram_done.r  .eq(0),
 150             ]
 151
 152         # * A write command consists of:
 153         #   - a transaction on the `dram.cmd` stream with `dram.cmd.we` asserted;
 154         #   - a transaction on the `dram.w` stream.
 155
 156         with m.If(dram_done.cmd & dram_done.w):
 157             m.d.comb += Assert(dram_cmd.we)
 158             m.d.sync += [
 159                 dram_done.cmd.eq(0),
 160                 dram_done.w  .eq(0),
 161             ]
 162
 163         # * A DRAM write at `dram_spec_addr` must be coherent with `dram_spec_data`.
 164
 165         with m.If(dram_done.cmd & dram_done.w):
 166             with m.If(dram_cmd.addr == dram_spec_addr):
 167                 m.d.comb += Assert(dram_data.w == dram_spec_data)
 168
 169         # For fairness, assume that any stream transaction completes in at most 3 clock cycles.
 170
 171         dram_wait = Record([
 172             ("cmd", range(2)),
 173             ("r",   range(2)),
 174             ("w",   range(2)),
 175         ])
 176
 177         with m.If(dut.dram_port.cmd.valid & ~dut.dram_port.cmd.ready):
 178             m.d.sync += dram_wait.cmd.eq(dram_wait.cmd + 1)
 179         with m.Else():
 180             m.d.sync += dram_wait.cmd.eq(0)
 181
 182         with m.If(dut.dram_port.w.valid & ~dut.dram_port.w.ready):
 183             m.d.sync += dram_wait.w.eq(dram_wait.w + 1)
 184         with m.Else():
 185             m.d.sync += dram_wait.w.eq(0)
 186
 187         with m.If(dut.dram_port.r.ready & ~dut.dram_port.r.valid):
 188             m.d.sync += dram_wait.r.eq(dram_wait.r + 1)
 189         with m.Else():
 190             m.d.sync += dram_wait.r.eq(0)
 191
 192         m.d.comb += [
 193             Assume(dram_wait.cmd < 2),
 194             Assume(dram_wait.r   < 2),
 195             Assume(dram_wait.w   < 2),
 196         ]
 197
 198         return m
 199
 200
 201 class WritebackCacheTestCase(FormalTestCase):
 202     def test_wrong_dram_port(self):
 203         with self.assertRaisesRegex(TypeError,
 204                 r"DRAM port must be an instance of lambdasoc\.cores\.litedram\.NativePort, "
 205                 r"not 'foo'"):
 206             WritebackCache(dram_port="foo", size=8, data_width=16)
 207
 208     def test_wrong_size(self):
 209         dram_port = litedram.NativePort(addr_width=23, data_width=32)
 210         with self.assertRaisesRegex(ValueError,
 211                 r"Cache size must be a positive power of two integer, not 'foo'"):
 212             WritebackCache(dram_port, size="foo", data_width=16)
 213         with self.assertRaisesRegex(ValueError,
 214                 r"Cache size must be a positive power of two integer, not -1"):
 215             WritebackCache(dram_port, size=-1, data_width=16)
 216         with self.assertRaisesRegex(ValueError,
 217                 r"Cache size must be a positive power of two integer, not 42"):
 218             WritebackCache(dram_port, size=42, data_width=16)
 219
 220     def test_wrong_data_width(self):
 221         dram_port = litedram.NativePort(addr_width=23, data_width=32)
 222         with self.assertRaisesRegex(ValueError,
 223                 r"Data width must be a positive power of two integer, not 'foo'"):
 224             WritebackCache(dram_port, size=8, data_width="foo")
 225         with self.assertRaisesRegex(ValueError,
 226                 r"Data width must be a positive power of two integer, not -1"):
 227             WritebackCache(dram_port, size=8, data_width=-1)
 228         with self.assertRaisesRegex(ValueError,
 229                 r"Data width must be a positive power of two integer, not 42"):
 230             WritebackCache(dram_port, size=8, data_width=42)
 231
 232     def test_wrong_ratio(self):
 233         dram_port = litedram.NativePort(addr_width=23, data_width=32)
 234         with self.assertRaisesRegex(ValueError,
 235                 r"DRAM port data width must be a multiple of data width, but 32 is not a multiple "
 236                 r"of 64"):
 237             WritebackCache(dram_port, size=8, data_width=64)
 238
 239     def check(self, dut):
 240         m = Module()
 241         m.domains.sync = ClockDomain(reset_less=True)
 242         m.submodules.spec = WritebackCacheSpec(dut)
 243         self.assertFormal(m, mode="hybrid", depth=10)
 244
 245     def test_spec_simple(self):
 246         dram_port = litedram.NativePort(addr_width=23, data_width=32)
 247         dut = WritebackCache(dram_port, size=8, data_width=16, granularity=8)
 248         self.check(dut)
 249
 250
 251 class SDRAMPeripheralTestCase(unittest.TestCase):
 252     def test_wrong_core(self):
 253         with self.assertRaisesRegex(TypeError,
 254                 r"LiteDRAM core must be an instance of lambdasoc\.cores\.litedram\.Core, "
 255                 r"not 'foo'"):
 256             sdram = SDRAMPeripheral(core="foo", cache_size=8)