test/test_axi_lite.py

   1 # This file is Copyright (c) 2020 Antmicro <www.antmicro.com>
   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, csr_bus
  11
  12 # Helpers ------------------------------------------------------------------------------------------
  13
  14 def _int_or_call(int_or_func):
  15     if callable(int_or_func):
  16         return int_or_func()
  17     return int_or_func
  18
  19 @passive
  20 def timeout_generator(ticks):
  21     import os
  22     for i in range(ticks):
  23         if os.environ.get("TIMEOUT_DEBUG", "") == "1":
  24             print("tick {}".format(i))
  25         yield
  26     raise TimeoutError("Timeout after %d ticks" % ticks)
  27
  28 class AXILiteChecker:
  29     def __init__(self, ready_latency=0, response_latency=0, rdata_generator=None):
  30         self.ready_latency = ready_latency
  31         self.response_latency = response_latency
  32         self.rdata_generator = rdata_generator or (lambda adr: 0xbaadc0de)
  33         self.writes = []  # (addr, data, strb)
  34         self.reads = []  # (addr, data)
  35
  36     def delay(self, latency):
  37         for _ in range(_int_or_call(latency)):
  38             yield
  39
  40     def handle_write(self, axi_lite):
  41         # aw
  42         while not (yield axi_lite.aw.valid):
  43             yield
  44         yield from self.delay(self.ready_latency)
  45         addr = (yield axi_lite.aw.addr)
  46         yield axi_lite.aw.ready.eq(1)
  47         yield
  48         yield axi_lite.aw.ready.eq(0)
  49         while not (yield axi_lite.w.valid):
  50             yield
  51         yield from self.delay(self.ready_latency)
  52         # w
  53         data = (yield axi_lite.w.data)
  54         strb = (yield axi_lite.w.strb)
  55         yield axi_lite.w.ready.eq(1)
  56         yield
  57         yield axi_lite.w.ready.eq(0)
  58         yield from self.delay(self.response_latency)
  59         # b
  60         yield axi_lite.b.valid.eq(1)
  61         yield axi_lite.b.resp.eq(RESP_OKAY)
  62         yield
  63         while not (yield axi_lite.b.ready):
  64             yield
  65         yield axi_lite.b.valid.eq(0)
  66         self.writes.append((addr, data, strb))
  67
  68     def handle_read(self, axi_lite):
  69         # ar
  70         while not (yield axi_lite.ar.valid):
  71             yield
  72         yield from self.delay(self.ready_latency)
  73         addr = (yield axi_lite.ar.addr)
  74         yield axi_lite.ar.ready.eq(1)
  75         yield
  76         yield axi_lite.ar.ready.eq(0)
  77         yield from self.delay(self.response_latency)
  78         # r
  79         data = self.rdata_generator(addr)
  80         yield axi_lite.r.valid.eq(1)
  81         yield axi_lite.r.resp.eq(RESP_OKAY)
  82         yield axi_lite.r.data.eq(data)
  83         yield
  84         while not (yield axi_lite.r.ready):
  85             yield
  86         yield axi_lite.r.valid.eq(0)
  87         yield axi_lite.r.data.eq(0)
  88         self.reads.append((addr, data))
  89
  90     @passive
  91     def handler(self, axi_lite):
  92         while True:
  93             if (yield axi_lite.aw.valid):
  94                 yield from self.handle_write(axi_lite)
  95             if (yield axi_lite.ar.valid):
  96                 yield from self.handle_read(axi_lite)
  97             yield
  98
  99 class AXILitePatternGenerator:
 100     def __init__(self, axi_lite, pattern, delay=0):
 101         # patter: (rw, addr, data)
 102         self.axi_lite = axi_lite
 103         self.pattern = pattern
 104         self.delay = delay
 105         self.errors = 0
 106         self.read_errors = []
 107         self.resp_errors = {"w": 0, "r": 0}
 108
 109     def handler(self):
 110         for rw, addr, data in self.pattern:
 111             assert rw in ["w", "r"]
 112             if rw == "w":
 113                 strb = 2**len(self.axi_lite.w.strb) - 1
 114                 resp = (yield from self.axi_lite.write(addr, data, strb))
 115             else:
 116                 rdata, resp = (yield from self.axi_lite.read(addr))
 117                 if rdata != data:
 118                     self.read_errors.append((rdata, data))
 119                     self.errors += 1
 120             if resp != RESP_OKAY:
 121                 self.resp_errors[rw] += 1
 122                 self.errors += 1
 123             for _ in range(_int_or_call(self.delay)):
 124                 yield
 125         for _ in range(16):
 126             yield
 127
 128 # TestAXILite --------------------------------------------------------------------------------------
 129
 130 class TestAXILite(unittest.TestCase):
 131     def test_wishbone2axi2wishbone(self):
 132         class DUT(Module):
 133             def __init__(self):
 134                 self.wishbone = wishbone.Interface(data_width=32)
 135
 136                 # # #
 137
 138                 axi = AXILiteInterface(data_width=32, address_width=32)
 139                 wb  = wishbone.Interface(data_width=32)
 140
 141                 wishbone2axi = Wishbone2AXILite(self.wishbone, axi)
 142                 axi2wishbone = AXILite2Wishbone(axi, wb)
 143                 self.submodules += wishbone2axi, axi2wishbone
 144
 145                 sram = wishbone.SRAM(1024, init=[0x12345678, 0xa55aa55a])
 146                 self.submodules += sram
 147                 self.comb += wb.connect(sram.bus)
 148
 149         def generator(dut):
 150             dut.errors = 0
 151             if (yield from dut.wishbone.read(0)) != 0x12345678:
 152                 dut.errors += 1
 153             if (yield from dut.wishbone.read(1)) != 0xa55aa55a:
 154                 dut.errors += 1
 155             for i in range(32):
 156                 yield from dut.wishbone.write(i, i)
 157             for i in range(32):
 158                 if (yield from dut.wishbone.read(i)) != i:
 159                     dut.errors += 1
 160
 161         dut = DUT()
 162         run_simulation(dut, [generator(dut)])
 163         self.assertEqual(dut.errors, 0)
 164
 165     def test_axilite2csr(self):
 166         @passive
 167         def csr_mem_handler(csr, mem):
 168             while True:
 169                 adr = (yield csr.adr)
 170                 yield csr.dat_r.eq(mem[adr])
 171                 if (yield csr.we):
 172                     mem[adr] = (yield csr.dat_w)
 173                 yield
 174
 175         class DUT(Module):
 176             def __init__(self):
 177                 self.axi_lite = AXILiteInterface()
 178                 self.csr = csr_bus.Interface()
 179                 self.submodules.axilite2csr = AXILite2CSR(self.axi_lite, self.csr)
 180                 self.errors = 0
 181
 182         prng = random.Random(42)
 183         mem_ref = [prng.randrange(255) for i in range(100)]
 184
 185         def generator(dut):
 186             dut.errors = 0
 187
 188             for adr, ref in enumerate(mem_ref):
 189                 adr = adr << 2
 190                 data, resp = (yield from dut.axi_lite.read(adr))
 191                 self.assertEqual(resp, 0b00)
 192                 if data != ref:
 193                     dut.errors += 1
 194
 195             write_data = [prng.randrange(255) for _ in mem_ref]
 196
 197             for adr, wdata in enumerate(write_data):
 198                 adr = adr << 2
 199                 resp = (yield from dut.axi_lite.write(adr, wdata))
 200                 self.assertEqual(resp, 0b00)
 201                 rdata, resp = (yield from dut.axi_lite.read(adr))
 202                 self.assertEqual(resp, 0b00)
 203                 if rdata != wdata:
 204                     dut.errors += 1
 205
 206         dut = DUT()
 207         mem = [v for v in mem_ref]
 208         run_simulation(dut, [generator(dut), csr_mem_handler(dut.csr, mem)])
 209         self.assertEqual(dut.errors, 0)
 210
 211     def test_axilite_sram(self):
 212         class DUT(Module):
 213             def __init__(self, size, init):
 214                 self.axi_lite = AXILiteInterface()
 215                 self.submodules.sram = AXILiteSRAM(size, init=init, bus=self.axi_lite)
 216                 self.errors = 0
 217
 218         def generator(dut, ref_init):
 219             for adr, ref in enumerate(ref_init):
 220                 adr = adr << 2
 221                 data, resp = (yield from dut.axi_lite.read(adr))
 222                 self.assertEqual(resp, 0b00)
 223                 if data != ref:
 224                     dut.errors += 1
 225
 226             write_data = [prng.randrange(255) for _ in ref_init]
 227
 228             for adr, wdata in enumerate(write_data):
 229                 adr = adr << 2
 230                 resp = (yield from dut.axi_lite.write(adr, wdata))
 231                 self.assertEqual(resp, 0b00)
 232                 rdata, resp = (yield from dut.axi_lite.read(adr))
 233                 self.assertEqual(resp, 0b00)
 234                 if rdata != wdata:
 235                     dut.errors += 1
 236
 237         prng = random.Random(42)
 238         init = [prng.randrange(2**32) for i in range(100)]
 239
 240         dut = DUT(size=len(init)*4, init=[v for v in init])
 241         run_simulation(dut, [generator(dut, init)])
 242         self.assertEqual(dut.errors, 0)
 243
 244     def converter_test(self, width_from, width_to,
 245                        write_pattern=None, write_expected=None,
 246                        read_pattern=None, read_expected=None):
 247         assert not (write_pattern is None and read_pattern is None)
 248
 249         if write_pattern is None:
 250             write_pattern = []
 251             write_expected = []
 252         elif len(write_pattern[0]) == 2:
 253             # add w.strb
 254             write_pattern = [(adr, data, 2**(width_from//8)-1) for adr, data in write_pattern]
 255
 256         if read_pattern is None:
 257             read_pattern = []
 258             read_expected = []
 259
 260         class DUT(Module):
 261             def __init__(self, width_from, width_to):
 262                 self.master = AXILiteInterface(data_width=width_from)
 263                 self.slave = AXILiteInterface(data_width=width_to)
 264                 self.submodules.converter = AXILiteConverter(self.master, self.slave)
 265
 266         def generator(axi_lite):
 267             for addr, data, strb in write_pattern or []:
 268                 resp = (yield from axi_lite.write(addr, data, strb))
 269                 self.assertEqual(resp, RESP_OKAY)
 270             for _ in range(16):
 271                 yield
 272
 273             for addr, refdata in read_pattern or []:
 274                 data, resp = (yield from axi_lite.read(addr))
 275                 self.assertEqual(resp, RESP_OKAY)
 276                 self.assertEqual(data, refdata)
 277             for _ in range(4):
 278                 yield
 279
 280         def rdata_generator(adr):
 281             for a, v in read_expected:
 282                 if a == adr:
 283                     return v
 284             return 0xbaadc0de
 285
 286         _latency = 0
 287         def latency():
 288             nonlocal _latency
 289             _latency = (_latency + 1) % 3
 290             return _latency
 291
 292         dut = DUT(width_from=width_from, width_to=width_to)
 293         checker = AXILiteChecker(ready_latency=latency, rdata_generator=rdata_generator)
 294         run_simulation(dut, [generator(dut.master), checker.handler(dut.slave)])
 295         self.assertEqual(checker.writes, write_expected)
 296         self.assertEqual(checker.reads, read_expected)
 297
 298     def test_axilite_down_converter_32to16(self):
 299         write_pattern = [
 300             (0x00000000, 0x22221111),
 301             (0x00000004, 0x44443333),
 302             (0x00000008, 0x66665555),
 303             (0x00000100, 0x88887777),
 304         ]
 305         write_expected = [
 306             (0x00000000, 0x1111, 0b11),
 307             (0x00000002, 0x2222, 0b11),
 308             (0x00000004, 0x3333, 0b11),
 309             (0x00000006, 0x4444, 0b11),
 310             (0x00000008, 0x5555, 0b11),
 311             (0x0000000a, 0x6666, 0b11),
 312             (0x00000100, 0x7777, 0b11),
 313             (0x00000102, 0x8888, 0b11),
 314         ]
 315         read_pattern = write_pattern
 316         read_expected = [(adr, data) for (adr, data, _) in write_expected]
 317         self.converter_test(width_from=32, width_to=16,
 318                             write_pattern=write_pattern, write_expected=write_expected,
 319                             read_pattern=read_pattern, read_expected=read_expected)
 320
 321     def test_axilite_down_converter_32to8(self):
 322         write_pattern = [
 323             (0x00000000, 0x44332211),
 324             (0x00000004, 0x88776655),
 325         ]
 326         write_expected = [
 327             (0x00000000, 0x11, 0b1),
 328             (0x00000001, 0x22, 0b1),
 329             (0x00000002, 0x33, 0b1),
 330             (0x00000003, 0x44, 0b1),
 331             (0x00000004, 0x55, 0b1),
 332             (0x00000005, 0x66, 0b1),
 333             (0x00000006, 0x77, 0b1),
 334             (0x00000007, 0x88, 0b1),
 335         ]
 336         read_pattern = write_pattern
 337         read_expected = [(adr, data) for (adr, data, _) in write_expected]
 338         self.converter_test(width_from=32, width_to=8,
 339                             write_pattern=write_pattern, write_expected=write_expected,
 340                             read_pattern=read_pattern, read_expected=read_expected)
 341
 342     def test_axilite_down_converter_64to32(self):
 343         write_pattern = [
 344             (0x00000000, 0x2222222211111111),
 345             (0x00000008, 0x4444444433333333),
 346         ]
 347         write_expected = [
 348             (0x00000000, 0x11111111, 0b1111),
 349             (0x00000004, 0x22222222, 0b1111),
 350             (0x00000008, 0x33333333, 0b1111),
 351             (0x0000000c, 0x44444444, 0b1111),
 352         ]
 353         read_pattern = write_pattern
 354         read_expected = [(adr, data) for (adr, data, _) in write_expected]
 355         self.converter_test(width_from=64, width_to=32,
 356                             write_pattern=write_pattern, write_expected=write_expected,
 357                             read_pattern=read_pattern, read_expected=read_expected)
 358
 359     def test_axilite_down_converter_strb(self):
 360         write_pattern = [
 361             (0x00000000, 0x22221111, 0b1100),
 362             (0x00000004, 0x44443333, 0b1111),
 363             (0x00000008, 0x66665555, 0b1011),
 364             (0x00000100, 0x88887777, 0b0011),
 365         ]
 366         write_expected = [
 367             (0x00000002, 0x2222, 0b11),
 368             (0x00000004, 0x3333, 0b11),
 369             (0x00000006, 0x4444, 0b11),
 370             (0x00000008, 0x5555, 0b11),
 371             (0x0000000a, 0x6666, 0b10),
 372             (0x00000100, 0x7777, 0b11),
 373         ]
 374         self.converter_test(width_from=32, width_to=16,
 375                             write_pattern=write_pattern, write_expected=write_expected)
 376
 377 # TestAXILiteInterconnet ---------------------------------------------------------------------------
 378
 379 class TestAXILiteInterconnect(unittest.TestCase):
 380     def test_interconnect_p2p(self):
 381         class DUT(Module):
 382             def __init__(self):
 383                 self.master = master = AXILiteInterface()
 384                 self.slave  = slave  = AXILiteInterface()
 385                 self.submodules.interconnect = AXILiteInterconnectPointToPoint(master, slave)
 386
 387         pattern = [
 388             ("w", 0x00000004, 0x11111111),
 389             ("w", 0x0000000c, 0x22222222),
 390             ("r", 0x00000010, 0x33333333),
 391             ("r", 0x00000018, 0x44444444),
 392         ]
 393
 394         def rdata_generator(adr):
 395             for rw, a, v in pattern:
 396                 if rw == "r" and a == adr:
 397                     return v
 398             return 0xbaadc0de
 399
 400         dut = DUT()
 401         checker = AXILiteChecker(rdata_generator=rdata_generator)
 402         generators = [
 403             AXILitePatternGenerator(dut.master, pattern).handler(),
 404             checker.handler(dut.slave),
 405         ]
 406         run_simulation(dut, generators)
 407         self.assertEqual(checker.writes, [(addr, data, 0b1111) for rw, addr, data in pattern if rw == "w"])
 408         self.assertEqual(checker.reads, [(addr, data) for rw, addr, data in pattern if rw == "r"])
 409
 410     def test_timeout(self):
 411         class DUT(Module):
 412             def __init__(self):
 413                 self.master = master = AXILiteInterface()
 414                 self.slave  = slave  = AXILiteInterface()
 415                 self.submodules.interconnect = AXILiteInterconnectPointToPoint(master, slave)
 416                 self.submodules.timeout = AXILiteTimeout(master, 16)
 417
 418         def generator(axi_lite):
 419             resp = (yield from axi_lite.write(0x00001000, 0x11111111))
 420             self.assertEqual(resp, RESP_OKAY)
 421             resp = (yield from axi_lite.write(0x00002000, 0x22222222))
 422             self.assertEqual(resp, RESP_SLVERR)
 423             data, resp = (yield from axi_lite.read(0x00003000))
 424             self.assertEqual(resp, RESP_SLVERR)
 425             self.assertEqual(data, 0xffffffff)
 426             yield
 427
 428         def checker(axi_lite):
 429             for _ in range(16):
 430                 yield
 431             yield axi_lite.aw.ready.eq(1)
 432             yield axi_lite.w.ready.eq(1)
 433             yield
 434             yield axi_lite.aw.ready.eq(0)
 435             yield axi_lite.w.ready.eq(0)
 436             yield axi_lite.b.valid.eq(1)
 437             yield
 438             while not (yield axi_lite.b.ready):
 439                 yield
 440             yield axi_lite.b.valid.eq(0)
 441
 442         dut = DUT()
 443         generators = [
 444             generator(dut.master),
 445             checker(dut.slave),
 446             timeout_generator(300),
 447         ]
 448         run_simulation(dut, generators)
 449
 450     def test_arbiter_order(self):
 451         class DUT(Module):
 452             def __init__(self, n_masters):
 453                 self.masters = [AXILiteInterface() for _ in range(n_masters)]
 454                 self.slave   = AXILiteInterface()
 455                 self.submodules.arbiter = AXILiteArbiter(self.masters, self.slave)
 456
 457         def generator(n, axi_lite, delay=0):
 458             def gen(i):
 459                 return 100*n + i
 460
 461             for i in range(4):
 462                 resp = (yield from axi_lite.write(gen(i), gen(i)))
 463                 self.assertEqual(resp, RESP_OKAY)
 464                 for _ in range(delay):
 465                     yield
 466             for i in range(4):
 467                 data, resp = (yield from axi_lite.read(gen(i)))
 468                 self.assertEqual(resp, RESP_OKAY)
 469                 for _ in range(delay):
 470                     yield
 471             for _ in range(8):
 472                 yield
 473
 474         n_masters = 3
 475
 476         # with no delay each master will do all transfers at once
 477         with self.subTest(delay=0):
 478             dut = DUT(n_masters)
 479             checker = AXILiteChecker()
 480             generators = [generator(i, master, delay=0) for i, master in enumerate(dut.masters)]
 481             generators += [timeout_generator(300), checker.handler(dut.slave)]
 482             run_simulation(dut, generators)
 483             order = [0, 1, 2, 3, 100, 101, 102, 103, 200, 201, 202, 203]
 484             self.assertEqual([addr for addr, data, strb in checker.writes], order)
 485             self.assertEqual([addr for addr, data in checker.reads], order)
 486
 487         # with some delay, the round-robin arbiter will iterate over masters
 488         with self.subTest(delay=1):
 489             dut = DUT(n_masters)
 490             checker = AXILiteChecker()
 491             generators = [generator(i, master, delay=1) for i, master in enumerate(dut.masters)]
 492             generators += [timeout_generator(300), checker.handler(dut.slave)]
 493             run_simulation(dut, generators)
 494             order = [0, 100, 200, 1, 101, 201, 2, 102, 202, 3, 103, 203]
 495             self.assertEqual([addr for addr, data, strb in checker.writes], order)
 496             self.assertEqual([addr for addr, data in checker.reads], order)
 497
 498     def test_arbiter_holds_grant_until_response(self):
 499         class DUT(Module):
 500             def __init__(self, n_masters):
 501                 self.masters = [AXILiteInterface() for _ in range(n_masters)]
 502                 self.slave   = AXILiteInterface()
 503                 self.submodules.arbiter = AXILiteArbiter(self.masters, self.slave)
 504
 505         def generator(n, axi_lite, delay=0):
 506             def gen(i):
 507                 return 100*n + i
 508
 509             for i in range(4):
 510                 resp = (yield from axi_lite.write(gen(i), gen(i)))
 511                 self.assertEqual(resp, RESP_OKAY)
 512                 for _ in range(delay):
 513                     yield
 514             for i in range(4):
 515                 data, resp = (yield from axi_lite.read(gen(i)))
 516                 self.assertEqual(resp, RESP_OKAY)
 517                 for _ in range(delay):
 518                     yield
 519             for _ in range(8):
 520                 yield
 521
 522         n_masters = 3
 523
 524         # with no delay each master will do all transfers at once
 525         with self.subTest(delay=0):
 526             dut = DUT(n_masters)
 527             checker = AXILiteChecker(response_latency=lambda: 3)
 528             generators = [generator(i, master, delay=0) for i, master in enumerate(dut.masters)]
 529             generators += [timeout_generator(300), checker.handler(dut.slave)]
 530             run_simulation(dut, generators)
 531             order = [0, 1, 2, 3, 100, 101, 102, 103, 200, 201, 202, 203]
 532             self.assertEqual([addr for addr, data, strb in checker.writes], order)
 533             self.assertEqual([addr for addr, data in checker.reads], order)
 534
 535         # with some delay, the round-robin arbiter will iterate over masters
 536         with self.subTest(delay=1):
 537             dut = DUT(n_masters)
 538             checker = AXILiteChecker(response_latency=lambda: 3)
 539             generators = [generator(i, master, delay=1) for i, master in enumerate(dut.masters)]
 540             generators += [timeout_generator(300), checker.handler(dut.slave)]
 541             run_simulation(dut, generators)
 542             order = [0, 100, 200, 1, 101, 201, 2, 102, 202, 3, 103, 203]
 543             self.assertEqual([addr for addr, data, strb in checker.writes], order)
 544             self.assertEqual([addr for addr, data in checker.reads], order)
 545
 546     def address_decoder(self, i, size=0x100, python=False):
 547         # bytes to 32-bit words aligned
 548         _size   = (size) >> 2
 549         _origin = (size * i) >> 2
 550         if python:  # for python integers
 551             shift = log2_int(_size)
 552             return lambda a: ((a >> shift) == (_origin >> shift))
 553         # for migen signals
 554         return lambda a: (a[log2_int(_size):] == (_origin >> log2_int(_size)))
 555
 556     def decoder_test(self, n_slaves, pattern, generator_delay=0):
 557         class DUT(Module):
 558             def __init__(self, decoders):
 559                 self.master = AXILiteInterface()
 560                 self.slaves = [AXILiteInterface() for _ in range(len(decoders))]
 561                 slaves = list(zip(decoders, self.slaves))
 562                 self.submodules.decoder = AXILiteDecoder(self.master, slaves)
 563
 564         def rdata_generator(adr):
 565             for rw, a, v in pattern:
 566                 if rw == "r" and a == adr:
 567                     return v
 568             return 0xbaadc0de
 569
 570         dut = DUT([self.address_decoder(i) for i in range(n_slaves)])
 571         checkers = [AXILiteChecker(rdata_generator=rdata_generator) for _ in dut.slaves]
 572
 573         generators = [AXILitePatternGenerator(dut.master, pattern, delay=generator_delay).handler()]
 574         generators += [checker.handler(slave) for (slave, checker) in zip(dut.slaves, checkers)]
 575         generators += [timeout_generator(300)]
 576         run_simulation(dut, generators)
 577
 578         return checkers
 579
 580     def test_decoder_write(self):
 581         for delay in [0, 1, 0]:
 582             with self.subTest(delay=delay):
 583                 slaves = self.decoder_test(n_slaves=3, pattern=[
 584                     ("w", 0x010, 1),
 585                     ("w", 0x110, 2),
 586                     ("w", 0x210, 3),
 587                     ("w", 0x011, 1),
 588                     ("w", 0x012, 1),
 589                     ("w", 0x111, 2),
 590                     ("w", 0x112, 2),
 591                     ("w", 0x211, 3),
 592                     ("w", 0x212, 3),
 593                 ], generator_delay=delay)
 594
 595                 def addr(checker_list):
 596                     return [entry[0] for entry in checker_list]
 597
 598                 self.assertEqual(addr(slaves[0].writes), [0x010, 0x011, 0x012])
 599                 self.assertEqual(addr(slaves[1].writes), [0x110, 0x111, 0x112])
 600                 self.assertEqual(addr(slaves[2].writes), [0x210, 0x211, 0x212])
 601                 for slave in slaves:
 602                     self.assertEqual(slave.reads, [])
 603
 604     def test_decoder_read(self):
 605         for delay in [0, 1]:
 606             with self.subTest(delay=delay):
 607                 slaves = self.decoder_test(n_slaves=3, pattern=[
 608                     ("r", 0x010, 1),
 609                     ("r", 0x110, 2),
 610                     ("r", 0x210, 3),
 611                     ("r", 0x011, 1),
 612                     ("r", 0x012, 1),
 613                     ("r", 0x111, 2),
 614                     ("r", 0x112, 2),
 615                     ("r", 0x211, 3),
 616                     ("r", 0x212, 3),
 617                 ], generator_delay=delay)
 618
 619                 def addr(checker_list):
 620                     return [entry[0] for entry in checker_list]
 621
 622                 self.assertEqual(addr(slaves[0].reads), [0x010, 0x011, 0x012])
 623                 self.assertEqual(addr(slaves[1].reads), [0x110, 0x111, 0x112])
 624                 self.assertEqual(addr(slaves[2].reads), [0x210, 0x211, 0x212])
 625                 for slave in slaves:
 626                     self.assertEqual(slave.writes, [])
 627
 628     def test_decoder_read_write(self):
 629         for delay in [0, 1]:
 630             with self.subTest(delay=delay):
 631                 slaves = self.decoder_test(n_slaves=3, pattern=[
 632                     ("w", 0x010, 1),
 633                     ("w", 0x110, 2),
 634                     ("r", 0x111, 2),
 635                     ("r", 0x011, 1),
 636                     ("r", 0x211, 3),
 637                     ("w", 0x210, 3),
 638                 ], generator_delay=delay)
 639
 640                 def addr(checker_list):
 641                     return [entry[0] for entry in checker_list]
 642
 643                 self.assertEqual(addr(slaves[0].writes), [0x010])
 644                 self.assertEqual(addr(slaves[0].reads),  [0x011])
 645                 self.assertEqual(addr(slaves[1].writes), [0x110])
 646                 self.assertEqual(addr(slaves[1].reads),  [0x111])
 647                 self.assertEqual(addr(slaves[2].writes), [0x210])
 648                 self.assertEqual(addr(slaves[2].reads),  [0x211])
 649
 650     def test_decoder_stall(self):
 651         with self.assertRaises(TimeoutError):
 652             self.decoder_test(n_slaves=3, pattern=[
 653                 ("w", 0x300, 1),
 654             ])
 655         with self.assertRaises(TimeoutError):
 656             self.decoder_test(n_slaves=3, pattern=[
 657                 ("r", 0x300, 1),
 658             ])
 659
 660     def interconnect_test(self, master_patterns, slave_decoders,
 661                                  master_delay=0, slave_ready_latency=0, slave_response_latency=0,
 662                                  disconnected_slaves=None, timeout=300, interconnect=AXILiteInterconnectShared,
 663                                  **kwargs):
 664         # number of masters/slaves is defined by the number of patterns/decoders
 665         # master_patterns: list of patterns per master, pattern = list(tuple(rw, addr, data))
 666         # slave_decoders: list of address decoders per slave
 667         # delay/latency: control the speed of masters/slaves
 668         # disconnected_slaves: list of slave numbers that shouldn't respond to any transactions
 669         class DUT(Module):
 670             def __init__(self, n_masters, decoders, **kwargs):
 671                 self.masters = [AXILiteInterface(name="master") for _ in range(n_masters)]
 672                 self.slaves  = [AXILiteInterface(name="slave") for _ in range(len(decoders))]
 673                 slaves = list(zip(decoders, self.slaves))
 674                 self.submodules.interconnect = interconnect(self.masters, slaves, **kwargs)
 675
 676         class ReadDataGenerator:
 677             # Generates data based on decoded addresses and data defined in master_patterns
 678             def __init__(self, patterns):
 679                 self.mem = {}
 680                 for pattern in patterns:
 681                     for rw, addr, val in pattern:
 682                         if rw == "r":
 683                             assert addr not in self.mem
 684                             self.mem[addr] = val
 685
 686             def getter(self, n):
 687                 # on miss will give default data depending on slave n
 688                 return lambda addr: self.mem.get(addr, 0xbaad0000 + n)
 689
 690         def new_checker(rdata_generator):
 691             return AXILiteChecker(ready_latency=slave_ready_latency,
 692                                   response_latency=slave_response_latency,
 693                                   rdata_generator=rdata_generator)
 694
 695         # perpare test
 696         dut = DUT(len(master_patterns), slave_decoders, **kwargs)
 697         rdata_generator = ReadDataGenerator(master_patterns)
 698         checkers = [new_checker(rdata_generator.getter(i)) for i, _ in enumerate(master_patterns)]
 699         pattern_generators = [AXILitePatternGenerator(dut.masters[i], pattern, delay=master_delay)
 700                               for i, pattern in enumerate(master_patterns)]
 701
 702         # run simulator
 703         generators = [gen.handler() for gen in pattern_generators]
 704         generators += [checker.handler(slave)
 705                        for i, (slave, checker) in enumerate(zip(dut.slaves, checkers))
 706                        if i not in (disconnected_slaves or [])]
 707         generators += [timeout_generator(timeout)]
 708         run_simulation(dut, generators, vcd_name='sim.vcd')
 709
 710         return pattern_generators, checkers
 711
 712     def test_interconnect_shared_basic(self):
 713         master_patterns = [
 714             [("w", 0x000, 0), ("w", 0x101, 0), ("w", 0x202, 0)],
 715             [("w", 0x010, 0), ("w", 0x111, 0), ("w", 0x112, 0)],
 716             [("w", 0x220, 0), ("w", 0x221, 0), ("w", 0x222, 0)],
 717         ]
 718         slave_decoders = [self.address_decoder(i) for i in range(3)]
 719
 720         generators, checkers = self.interconnect_test(master_patterns, slave_decoders,
 721                                                       master_delay=1)
 722
 723         for gen in generators:
 724             self.assertEqual(gen.errors, 0)
 725
 726         def addr(checker_list):
 727             return [entry[0] for entry in checker_list]
 728
 729         self.assertEqual(addr(checkers[0].writes), [0x000, 0x010])
 730         self.assertEqual(addr(checkers[1].writes), [0x101, 0x111, 0x112])
 731         self.assertEqual(addr(checkers[2].writes), [0x220, 0x221, 0x202, 0x222])
 732         self.assertEqual(addr(checkers[0].reads), [])
 733         self.assertEqual(addr(checkers[1].reads), [])
 734         self.assertEqual(addr(checkers[2].reads), [])
 735
 736     def interconnect_stress_test(self, timeout=1000, **kwargs):
 737         prng = random.Random(42)
 738
 739         n_masters = 3
 740         n_slaves = 3
 741         pattern_length = 64
 742         slave_region_size = 0x10000000
 743         # for testing purpose each master will access only its own region of a slave
 744         master_region_size = 0x1000
 745         assert n_masters*master_region_size < slave_region_size
 746
 747         def gen_pattern(n, length):
 748             assert length < master_region_size
 749             for i_access in range(length):
 750                 rw = "w" if prng.randint(0, 1) == 0 else "r"
 751                 i_slave = prng.randrange(n_slaves)
 752                 addr = i_slave*slave_region_size + n*master_region_size + i_access
 753                 data = addr
 754                 yield rw, addr, data
 755
 756         master_patterns   = [list(gen_pattern(i, pattern_length)) for i in range(n_masters)]
 757         slave_decoders    = [self.address_decoder(i, size=slave_region_size) for i in range(n_slaves)]
 758         slave_decoders_py = [self.address_decoder(i, size=slave_region_size, python=True)
 759                              for i in range(n_slaves)]
 760
 761         generators, checkers = self.interconnect_test(master_patterns, slave_decoders,
 762                                                       timeout=timeout, **kwargs)
 763
 764         for gen in generators:
 765             read_errors = ["  0x{:08x} vs 0x{:08x}".format(v, ref) for v, ref in gen.read_errors]
 766             msg = "\ngen.resp_errors = {}\ngen.read_errors = \n{}".format(
 767                 gen.resp_errors, "\n".join(read_errors))
 768             if not kwargs.get("disconnected_slaves", None):
 769                 self.assertEqual(gen.errors, 0, msg=msg)
 770             else:  # when some slaves are disconnected we should have some errors
 771                 self.assertNotEqual(gen.errors, 0, msg=msg)
 772
 773         # make sure all the accesses at slave side are in correct address region
 774         for i_slave, (checker, decoder) in enumerate(zip(checkers, slave_decoders_py)):
 775             for addr in (entry[0] for entry in checker.writes + checker.reads):
 776                 # compensate for the fact that decoders work on word-aligned addresses
 777                 self.assertNotEqual(decoder(addr >> 2), 0)
 778
 779     def test_interconnect_shared_stress_no_delay(self):
 780         self.interconnect_stress_test(timeout=1000,
 781                                       master_delay=0,
 782                                       slave_ready_latency=0,
 783                                       slave_response_latency=0)
 784
 785     def test_interconnect_shared_stress_rand_short(self):
 786         prng = random.Random(42)
 787         rand = lambda: prng.randrange(4)
 788         self.interconnect_stress_test(timeout=2000,
 789                                       master_delay=rand,
 790                                       slave_ready_latency=rand,
 791                                       slave_response_latency=rand)
 792
 793     def test_interconnect_shared_stress_rand_long(self):
 794         prng = random.Random(42)
 795         rand = lambda: prng.randrange(16)
 796         self.interconnect_stress_test(timeout=4000,
 797                                       master_delay=rand,
 798                                       slave_ready_latency=rand,
 799                                       slave_response_latency=rand)
 800
 801     def test_interconnect_shared_stress_timeout(self):
 802         self.interconnect_stress_test(timeout=4000,
 803                                       disconnected_slaves=[1],
 804                                       timeout_cycles=50)
 805
 806     def test_crossbar_stress_no_delay(self):
 807         self.interconnect_stress_test(timeout=1000,
 808                                       master_delay=0,
 809                                       slave_ready_latency=0,
 810                                       slave_response_latency=0,
 811                                       interconnect=AXILiteCrossbar)
 812
 813     def test_crossbar_stress_rand(self):
 814         prng = random.Random(42)
 815         rand = lambda: prng.randrange(4)
 816         self.interconnect_stress_test(timeout=2000,
 817                                       master_delay=rand,
 818                                       slave_ready_latency=rand,
 819                                       slave_response_latency=rand,
 820                                       interconnect=AXILiteCrossbar)