gram/test/test_core_multiplexer.py

   1 from nmigen import *
   2
   3 from gram.core.multiplexer import _AntiStarvation, _CommandChooser, _Steerer, STEER_NOP, STEER_CMD
   4 from gram.common import cmd_request_rw_layout
   5 from gram.phy.dfi import Interface
   6 import gram.stream as stream
   7 from gram.test.utils import *
   8
   9 class CommandChooserTestCase(FHDLTestCase):
  10     def prepare_testbench(self):
  11         a = 16
  12         ba = 3
  13         requests = []
  14         for i in range(10):
  15             requests += [stream.Endpoint(cmd_request_rw_layout(a, ba))]
  16         dut = _CommandChooser(requests)
  17
  18         return (requests, dut)
  19
  20     def test_wants(self):
  21         requests, dut = self.prepare_testbench()
  22
  23         def process():
  24             for i in range(10):
  25                 yield requests[i].a.eq(i)
  26
  27             # Fake requests, non valid requests shouldn't be picked
  28             yield requests[5].is_read.eq(1)
  29             yield requests[5].valid.eq(1)
  30             yield requests[6].is_read.eq(1)
  31             yield requests[6].valid.eq(0)
  32             yield requests[7].is_write.eq(1)
  33             yield requests[7].valid.eq(1)
  34             yield requests[8].is_write.eq(1)
  35             yield requests[8].valid.eq(0)
  36
  37             # want_writes
  38             yield dut.want_writes.eq(1)
  39             yield; yield Delay(1e-9)
  40             self.assertEqual((yield dut.cmd.a), 7)
  41
  42             # want_reads
  43             yield dut.want_writes.eq(0)
  44             yield dut.want_reads.eq(1)
  45             yield; yield Delay(1e-9)
  46             self.assertEqual((yield dut.cmd.a), 5)
  47
  48         runSimulation(dut, process, "test_core_multiplexer_commandchooser.vcd")
  49
  50     def test_helpers(self):
  51         requests, dut = self.prepare_testbench()
  52
  53         def process():
  54             for i in range(10):
  55                 yield requests[i].a.eq(i)
  56
  57             # Fake requests
  58             yield requests[5].is_read.eq(1)
  59             yield requests[5].valid.eq(1)
  60             yield requests[6].is_read.eq(1)
  61             yield requests[6].valid.eq(0)
  62             yield requests[7].is_write.eq(1)
  63             yield requests[7].valid.eq(1)
  64             yield requests[8].is_write.eq(1)
  65             yield requests[8].valid.eq(0)
  66
  67             # want_writes
  68             yield dut.want_writes.eq(1)
  69             yield; yield Delay(1e-9)
  70             self.assertTrue((yield dut.write()))
  71             self.assertFalse((yield dut.read()))
  72
  73             # want_reads
  74             yield dut.want_writes.eq(0)
  75             yield dut.want_reads.eq(1)
  76             yield; yield Delay(1e-9)
  77             self.assertTrue((yield dut.read()))
  78             self.assertFalse((yield dut.write()))
  79
  80         runSimulation(dut, process, "test_core_multiplexer_commandchooser.vcd")
  81
  82 class SteererTestCase(FHDLTestCase):
  83     def test_nop(self):
  84         a = 12
  85         ba = 3
  86
  87         commands = [
  88             stream.Endpoint(cmd_request_rw_layout(a, ba)), # NOP
  89             stream.Endpoint(cmd_request_rw_layout(a, ba)), # CMD
  90             stream.Endpoint(cmd_request_rw_layout(a, ba)), # REQ
  91             stream.Endpoint(cmd_request_rw_layout(a, ba)), # REFRESH
  92         ]
  93         dfi = Interface(a, ba, 1, 8, nphases=2)
  94         dut = _Steerer(commands, dfi)
  95
  96         def process():
  97             yield dut.sel[0].eq(STEER_NOP)
  98             yield dut.sel[1].eq(STEER_NOP)
  99             yield
 100
 101             # Check for NOP command on CAS/RAS/WE on phase #0
 102             self.assertFalse((yield dfi.phases[0].cas))
 103             self.assertFalse((yield dfi.phases[0].ras))
 104             self.assertFalse((yield dfi.phases[0].we))
 105
 106             # Check for NOP command on CAS/RAS/WE on phase #1
 107             self.assertFalse((yield dfi.phases[1].cas))
 108             self.assertFalse((yield dfi.phases[1].ras))
 109             self.assertFalse((yield dfi.phases[1].we))
 110
 111         runSimulation(dut, process, "test_core_multiplexer_steerer.vcd")
 112
 113     def test_cmd(self):
 114         a = 12
 115         ba = 3
 116
 117         commands = [
 118             stream.Endpoint(cmd_request_rw_layout(a, ba)), # NOP
 119             stream.Endpoint(cmd_request_rw_layout(a, ba)), # CMD
 120             stream.Endpoint(cmd_request_rw_layout(a, ba)), # REQ
 121             stream.Endpoint(cmd_request_rw_layout(a, ba)), # REFRESH
 122         ]
 123         dfi = Interface(a, ba, 1, 8, nphases=2)
 124         dut = _Steerer(commands, dfi)
 125
 126         def process():
 127             yield dut.sel[0].eq(STEER_CMD)
 128             yield dut.sel[1].eq(STEER_NOP)
 129             yield commands[STEER_CMD].cas.eq(1)
 130             yield commands[STEER_CMD].ras.eq(0)
 131             yield commands[STEER_CMD].we.eq(1)
 132             yield commands[STEER_CMD].ready.eq(0)
 133             yield
 134
 135             # Check for NOP command on CAS/RAS/WE on phase #0
 136             self.assertFalse((yield dfi.phases[0].cas))
 137             self.assertFalse((yield dfi.phases[0].ras))
 138             self.assertFalse((yield dfi.phases[0].we))
 139
 140             # Check for NOP command on CAS/RAS/WE on phase #1
 141             self.assertFalse((yield dfi.phases[1].cas))
 142             self.assertFalse((yield dfi.phases[1].ras))
 143             self.assertFalse((yield dfi.phases[1].we))
 144
 145             yield commands[STEER_CMD].valid.eq(1)
 146             yield; yield Delay(1e-9)
 147
 148             # Check for NOP command on CAS/RAS/WE on phase #0
 149             self.assertFalse((yield dfi.phases[0].cas))
 150             self.assertFalse((yield dfi.phases[0].ras))
 151             self.assertFalse((yield dfi.phases[0].we))
 152
 153             # Check for NOP command on CAS/RAS/WE on phase #1
 154             self.assertFalse((yield dfi.phases[1].cas))
 155             self.assertFalse((yield dfi.phases[1].ras))
 156             self.assertFalse((yield dfi.phases[1].we))
 157
 158             yield commands[STEER_CMD].ready.eq(1)
 159             yield; yield Delay(1e-9)
 160
 161             # Check for READ/WRITE command on CAS/RAS/WE on phase #0
 162             self.assertTrue((yield dfi.phases[0].cas))
 163             self.assertFalse((yield dfi.phases[0].ras))
 164             self.assertTrue((yield dfi.phases[0].we))
 165
 166             # Check for NOP command on CAS/RAS/WE on phase #1
 167             self.assertFalse((yield dfi.phases[1].cas))
 168             self.assertFalse((yield dfi.phases[1].ras))
 169             self.assertFalse((yield dfi.phases[1].we))
 170
 171         runSimulation(dut, process, "test_core_multiplexer_steerer.vcd")
 172
 173 class AntiStarvationTestCase(FHDLTestCase):
 174     def test_duration(self):
 175         def generic_test(timeout):
 176             dut = _AntiStarvation(timeout)
 177
 178             def process():
 179                 yield dut.en.eq(1)
 180                 yield
 181                 yield dut.en.eq(0)
 182                 yield
 183
 184                 for i in range(timeout):
 185                     self.assertFalse((yield dut.max_time))
 186                     yield
 187
 188                 self.assertTrue((yield dut.max_time))
 189
 190             runSimulation(dut, process, "test_core_multiplexer_antistarvation.vcd")
 191
 192     def test_formal(self):
 193         def generic_test(timeout):
 194             dut = _AntiStarvation(timeout)
 195             self.assertFormal(dut, mode="bmc", depth=timeout+1)
 196
 197         generic_test(5)
 198         generic_test(10)
 199         generic_test(0x20)