test/test_axi.py

   1 # This file is Copyright (c) 2019 Florent Kermarrec <florent@enjoy-digital.fr>
   2 # License: BSD
   3
   4 import unittest
   5 import random
   6
   7 from migen import *
   8
   9 from litex.soc.interconnect.axi import *
  10 from litex.soc.interconnect import wishbone
  11
  12 # Software Models ----------------------------------------------------------------------------------
  13
  14 class Burst:
  15     def __init__(self, addr, type=BURST_FIXED, len=0, size=0):
  16         self.addr = addr
  17         self.type = type
  18         self.len  = len
  19         self.size = size
  20
  21     def to_beats(self):
  22         r = []
  23         for i in range(self.len + 1):
  24             if self.type == BURST_INCR:
  25                 offset = i*2**(self.size)
  26                 r += [Beat(self.addr + offset)]
  27             elif self.type == BURST_WRAP:
  28                 offset = (i*2**(self.size))%((2**self.size)*(self.len + 1))
  29                 r += [Beat(self.addr + offset)]
  30             else:
  31                 r += [Beat(self.addr)]
  32         return r
  33
  34
  35 class Beat:
  36     def __init__(self, addr):
  37         self.addr = addr
  38
  39
  40 class Access(Burst):
  41     def __init__(self, addr, data, id, **kwargs):
  42         Burst.__init__(self, addr, **kwargs)
  43         self.data = data
  44         self.id   = id
  45
  46
  47 class Write(Access):
  48     pass
  49
  50
  51 class Read(Access):
  52     pass
  53
  54 # TestAXI ------------------------------------------------------------------------------------------
  55
  56 class TestAXI(unittest.TestCase):
  57     def test_burst2beat(self):
  58         def bursts_generator(ax, bursts, valid_rand=50):
  59             prng = random.Random(42)
  60             for burst in bursts:
  61                 yield ax.valid.eq(1)
  62                 yield ax.addr.eq(burst.addr)
  63                 yield ax.burst.eq(burst.type)
  64                 yield ax.len.eq(burst.len)
  65                 yield ax.size.eq(burst.size)
  66                 while (yield ax.ready) == 0:
  67                     yield
  68                 yield ax.valid.eq(0)
  69                 while prng.randrange(100) < valid_rand:
  70                     yield
  71                 yield
  72
  73         @passive
  74         def beats_checker(ax, beats, ready_rand=50):
  75             self.errors = 0
  76             yield ax.ready.eq(0)
  77             prng = random.Random(42)
  78             for beat in beats:
  79                 while ((yield ax.valid) and (yield ax.ready)) == 0:
  80                     if prng.randrange(100) > ready_rand:
  81                         yield ax.ready.eq(1)
  82                     else:
  83                         yield ax.ready.eq(0)
  84                     yield
  85                 ax_addr = (yield ax.addr)
  86                 if ax_addr != beat.addr:
  87                     self.errors += 1
  88                 yield
  89
  90         # dut
  91         ax_burst = stream.Endpoint(ax_description(32, 32))
  92         ax_beat = stream.Endpoint(ax_description(32, 32))
  93         dut =  AXIBurst2Beat(ax_burst, ax_beat)
  94
  95         # generate dut input (bursts)
  96         prng = random.Random(42)
  97         bursts = []
  98         for i in range(32):
  99             bursts.append(Burst(prng.randrange(2**32), BURST_FIXED, prng.randrange(255), log2_int(32//8)))
 100             bursts.append(Burst(prng.randrange(2**32), BURST_INCR, prng.randrange(255), log2_int(32//8)))
 101         bursts.append(Burst(4, BURST_WRAP, 4-1, log2_int(2)))
 102
 103         # generate expected dut output (beats for reference)
 104         beats = []
 105         for burst in bursts:
 106             beats += burst.to_beats()
 107
 108         # simulation
 109         generators = [
 110             bursts_generator(ax_burst, bursts),
 111             beats_checker(ax_beat, beats)
 112         ]
 113         run_simulation(dut, generators)
 114         self.assertEqual(self.errors, 0)
 115
 116
 117     def _test_axi2wishbone(self,
 118         naccesses=16, simultaneous_writes_reads=False,
 119         # random: 0: min (no random), 100: max.
 120         # burst randomness
 121         id_rand_enable   = False,
 122         len_rand_enable  = False,
 123         data_rand_enable = False,
 124         # flow valid randomness
 125         aw_valid_random  = 0,
 126         w_valid_random   = 0,
 127         ar_valid_random  = 0,
 128         r_valid_random   = 0,
 129         # flow ready randomness
 130         w_ready_random   = 0,
 131         b_ready_random   = 0,
 132         r_ready_random   = 0
 133         ):
 134
 135         def writes_cmd_generator(axi_port, writes):
 136             prng = random.Random(42)
 137             for write in writes:
 138                 while prng.randrange(100) < aw_valid_random:
 139                     yield
 140                 # send command
 141                 yield axi_port.aw.valid.eq(1)
 142                 yield axi_port.aw.addr.eq(write.addr<<2)
 143                 yield axi_port.aw.burst.eq(write.type)
 144                 yield axi_port.aw.len.eq(write.len)
 145                 yield axi_port.aw.size.eq(write.size)
 146                 yield axi_port.aw.id.eq(write.id)
 147                 yield
 148                 while (yield axi_port.aw.ready) == 0:
 149                     yield
 150                 yield axi_port.aw.valid.eq(0)
 151
 152         def writes_data_generator(axi_port, writes):
 153             prng = random.Random(42)
 154             yield axi_port.w.strb.eq(2**(len(axi_port.w.data)//8) - 1)
 155             for write in writes:
 156                 for i, data in enumerate(write.data):
 157                     while prng.randrange(100) < w_valid_random:
 158                         yield
 159                     # send data
 160                     yield axi_port.w.valid.eq(1)
 161                     if (i == (len(write.data) - 1)):
 162                         yield axi_port.w.last.eq(1)
 163                     else:
 164                         yield axi_port.w.last.eq(0)
 165                     yield axi_port.w.data.eq(data)
 166                     yield
 167                     while (yield axi_port.w.ready) == 0:
 168                         yield
 169                     yield axi_port.w.valid.eq(0)
 170             axi_port.reads_enable = True
 171
 172         def writes_response_generator(axi_port, writes):
 173             prng = random.Random(42)
 174             self.writes_id_errors = 0
 175             for write in writes:
 176                 # wait response
 177                 yield axi_port.b.ready.eq(0)
 178                 yield
 179                 while (yield axi_port.b.valid) == 0:
 180                     yield
 181                 while prng.randrange(100) < b_ready_random:
 182                     yield
 183                 yield axi_port.b.ready.eq(1)
 184                 yield
 185                 if (yield axi_port.b.id) != write.id:
 186                     self.writes_id_errors += 1
 187
 188         def reads_cmd_generator(axi_port, reads):
 189             prng = random.Random(42)
 190             while not axi_port.reads_enable:
 191                 yield
 192             for read in reads:
 193                 while prng.randrange(100) < ar_valid_random:
 194                     yield
 195                 # send command
 196                 yield axi_port.ar.valid.eq(1)
 197                 yield axi_port.ar.addr.eq(read.addr<<2)
 198                 yield axi_port.ar.burst.eq(read.type)
 199                 yield axi_port.ar.len.eq(read.len)
 200                 yield axi_port.ar.size.eq(read.size)
 201                 yield axi_port.ar.id.eq(read.id)
 202                 yield
 203                 while (yield axi_port.ar.ready) == 0:
 204                     yield
 205                 yield axi_port.ar.valid.eq(0)
 206
 207         def reads_response_data_generator(axi_port, reads):
 208             prng = random.Random(42)
 209             self.reads_data_errors = 0
 210             self.reads_id_errors   = 0
 211             self.reads_last_errors = 0
 212             while not axi_port.reads_enable:
 213                 yield
 214             for read in reads:
 215                 for i, data in enumerate(read.data):
 216                     # wait data / response
 217                     yield axi_port.r.ready.eq(0)
 218                     yield
 219                     while (yield axi_port.r.valid) == 0:
 220                         yield
 221                     while prng.randrange(100) < r_ready_random:
 222                         yield
 223                     yield axi_port.r.ready.eq(1)
 224                     yield
 225                     if (yield axi_port.r.data) != data:
 226                         self.reads_data_errors += 1
 227                     if (yield axi_port.r.id) != read.id:
 228                         self.reads_id_errors += 1
 229                     if i == (len(read.data) - 1):
 230                         if (yield axi_port.r.last) != 1:
 231                             self.reads_last_errors += 1
 232                     else:
 233                         if (yield axi_port.r.last) != 0:
 234                             self.reads_last_errors += 1
 235
 236         # dut
 237         class DUT(Module):
 238             def __init__(self):
 239                 self.axi      = AXIInterface(data_width=32, address_width=32, id_width=8)
 240                 self.wishbone = wishbone.Interface(data_width=32)
 241
 242                 axi2wishbone = AXI2Wishbone(self.axi, self.wishbone)
 243                 self.submodules += axi2wishbone
 244
 245                 wishbone_mem = wishbone.SRAM(1024, bus=self.wishbone)
 246                 self.submodules += wishbone_mem
 247
 248         dut = DUT()
 249
 250         # generate writes/reads
 251         prng   = random.Random(42)
 252         writes = []
 253         offset = 1
 254         for i in range(naccesses):
 255             _id = prng.randrange(2**8) if id_rand_enable else i
 256             _len = prng.randrange(32) if len_rand_enable else i
 257             _data = [prng.randrange(2**32) if data_rand_enable else j for j in range(_len + 1)]
 258             writes.append(Write(offset, _data, _id, type=BURST_INCR, len=_len, size=log2_int(32//8)))
 259             offset += _len + 1
 260         # dummy reads to ensure datas have been written before the effective reads start.
 261         dummy_reads = [Read(1023, [0], 0, type=BURST_FIXED, len=0, size=log2_int(32//8)) for _ in range(32)]
 262         reads = writes
 263
 264         # simulation
 265         if simultaneous_writes_reads:
 266             dut.axi.reads_enable = True
 267         else:
 268             dut.axi.reads_enable = False # will be set by writes_data_generator
 269         generators = [
 270             writes_cmd_generator(dut.axi, writes),
 271             writes_data_generator(dut.axi, writes),
 272             writes_response_generator(dut.axi, writes),
 273             reads_cmd_generator(dut.axi, reads),
 274             reads_response_data_generator(dut.axi, reads)
 275         ]
 276         run_simulation(dut, generators)
 277         self.assertEqual(self.writes_id_errors, 0)
 278         self.assertEqual(self.reads_data_errors, 0)
 279         self.assertEqual(self.reads_id_errors, 0)
 280         self.assertEqual(self.reads_last_errors, 0)
 281
 282     # test with no randomness
 283     def test_axi2wishbone_writes_then_reads_no_random(self):
 284         self._test_axi2wishbone(simultaneous_writes_reads=False)
 285
 286     # test randomness one parameter at a time
 287     def test_axi2wishbone_writes_then_reads_random_bursts(self):
 288         self._test_axi2wishbone(
 289             simultaneous_writes_reads = False,
 290             id_rand_enable   = True,
 291             len_rand_enable  = True,
 292             data_rand_enable = True)
 293
 294     def test_axi2wishbone_random_w_ready(self):
 295         self._test_axi2wishbone(w_ready_random=90)
 296
 297     def test_axi2wishbone_random_b_ready(self):
 298         self._test_axi2wishbone(b_ready_random=90)
 299
 300     def test_axi2wishbone_random_r_ready(self):
 301         self._test_axi2wishbone(r_ready_random=90)
 302
 303     def test_axi2wishbone_random_aw_valid(self):
 304         self._test_axi2wishbone(aw_valid_random=90)
 305
 306     def test_axi2wishbone_random_w_valid(self):
 307         self._test_axi2wishbone(w_valid_random=90)
 308
 309     def test_axi2wishbone_random_ar_valid(self):
 310         self._test_axi2wishbone(ar_valid_random=90)
 311
 312     def test_axi2wishbone_random_r_valid(self):
 313         self._test_axi2wishbone(r_valid_random=90)
 314
 315     # now let's stress things a bit... :)
 316     def test_axi2wishbone_random_all(self):
 317         self._test_axi2wishbone(
 318             simultaneous_writes_reads = False,
 319             id_rand_enable  = True,
 320             len_rand_enable = True,
 321             aw_valid_random = 50,
 322             w_ready_random  = 50,
 323             b_ready_random  = 50,
 324             w_valid_random  = 50,
 325             ar_valid_random = 90,
 326             r_valid_random  = 90,
 327             r_ready_random  = 90
 328         )