diff --git a/test/test_fifo.py b/test/test_fifo.py index bc4a6e5..e924d08 100644 --- a/test/test_fifo.py +++ b/test/test_fifo.py @@ -10,37 +10,238 @@ from litex.soc.interconnect.stream import * from litedram.common import LiteDRAMNativeWritePort from litedram.common import LiteDRAMNativeReadPort -from litedram.frontend.fifo import LiteDRAMFIFO +from litedram.frontend.fifo import LiteDRAMFIFO, \ + _LiteDRAMFIFOCtrl, _LiteDRAMFIFOWriter, _LiteDRAMFIFOReader from test.common import * -class FIFODUT(Module): - def __init__(self): - # ports - self.write_port = LiteDRAMNativeWritePort(address_width=32, data_width=32) - self.read_port = LiteDRAMNativeReadPort(address_width=32, data_width=32) - - # fifo - self.submodules.fifo = LiteDRAMFIFO( - data_width = 32, - depth = 64, - base = 0, - write_port = self.write_port, - read_port = self.read_port, - read_threshold = 8, - write_threshold = 64-8 - ) - - # memory - self.memory = DRAMMemory(32, 256) - - class TestFIFO(unittest.TestCase): + @passive + def fifo_ctrl_flag_checker(self, fifo_ctrl, write_threshold, read_threshold): + while True: + level = (yield fifo_ctrl.level) + self.assertEqual((yield fifo_ctrl.writable), level < write_threshold) + self.assertEqual((yield fifo_ctrl.readable), level > read_threshold) + yield + + # _LiteDRAMFIFOCtrl ---------------------------------------------------------------------------- + + def test_fifo_ctrl(self): + dut = _LiteDRAMFIFOCtrl(base=0, depth=16, read_threshold=0, write_threshold=16) + + def main_generator(): + self.assertEqual((yield dut.level), 0) + self.assertEqual((yield dut.write_address), 0) + self.assertEqual((yield dut.read_address), 0) + + # ignore readable/writable + # write address + yield dut.write.eq(1) + yield + # write_address gets updated 1 cycle later + for i in range(24 - 1): + self.assertEqual((yield dut.write_address), i % 16) + yield + yield dut.write.eq(0) + yield + self.assertEqual((yield dut.write_address), 24 % 16) + + # read address + yield dut.read.eq(1) + yield + for i in range(24 - 1): + self.assertEqual((yield dut.read_address), i % 16) + yield + yield dut.read.eq(0) + yield + self.assertEqual((yield dut.read_address), 24 % 16) + + # level + def check_level_diff(write, read, diff): + level = (yield dut.level) + yield dut.write.eq(write) + yield dut.read.eq(read) + yield + yield dut.write.eq(0) + yield dut.read.eq(0) + yield + self.assertEqual((yield dut.level), level + diff) + + check_level_diff(write=1, read=0, diff=+1) + check_level_diff(write=1, read=1, diff=+0) + check_level_diff(write=0, read=1, diff=-1) + + generators = [ + main_generator(), + self.fifo_ctrl_flag_checker(dut, write_threshold=16, read_threshold=0), + ] + run_simulation(dut, generators) + + # _LiteDRAMFIFOWriter -------------------------------------------------------------------------- + + def fifo_writer_test(self, depth, sequence_len, write_threshold): + class DUT(Module): + def __init__(self): + self.port = LiteDRAMNativeWritePort(address_width=32, data_width=32) + ctrl = _LiteDRAMFIFOCtrl(base=8, depth=depth, read_threshold=0, + write_threshold=write_threshold) + writer = _LiteDRAMFIFOWriter(data_width=32, port=self.port, ctrl=ctrl) + self.submodules.ctrl = ctrl + self.submodules.writer = writer + + self.memory = DRAMMemory(32, 128) + assert 8 + sequence_len <= len(self.memory.mem) + + write_data = [seed_to_data(i) for i in range(sequence_len)] + + def generator(dut): + for data in write_data: + yield dut.writer.sink.valid.eq(1) + yield dut.writer.sink.data.eq(data) + yield + while (yield dut.writer.sink.ready) == 0: + yield + yield dut.writer.sink.valid.eq(0) + + for _ in range(16): + yield + + dut = DUT() + generators = [ + generator(dut), + dut.memory.write_handler(dut.port), + self.fifo_ctrl_flag_checker(dut.ctrl, + write_threshold=write_threshold, read_threshold=0), + timeout_generator(1500), + ] + run_simulation(dut, generators) + + mem_expected = [0] * len(dut.memory.mem) + for i, data in enumerate(write_data): + mem_expected[8 + i%depth] = data + self.assertEqual(dut.memory.mem, mem_expected) + + def test_fifo_writer_sequence(self): + self.fifo_writer_test(sequence_len=48, depth=64, write_threshold=64) + + def test_fifo_writer_address_wraps(self): + self.fifo_writer_test(sequence_len=48, depth=32, write_threshold=64) + + def test_fifo_writer_stops_after_threshold(self): + with self.assertRaises(TimeoutError): + self.fifo_writer_test(sequence_len=48, depth=32, write_threshold=32) + + # _LiteDRAMFIFOReader -------------------------------------------------------------------------- + + def fifo_reader_test(self, depth, sequence_len, read_threshold, inital_writes=0): + memory_data = [seed_to_data(i) for i in range(128)] + read_data = [] + + class DUT(Module): + def __init__(self): + self.port = LiteDRAMNativeReadPort(address_width=32, data_width=32) + ctrl = _LiteDRAMFIFOCtrl(base=8, depth=depth, read_threshold=read_threshold, + write_threshold=depth) + reader = _LiteDRAMFIFOReader(data_width=32, port=self.port, ctrl=ctrl) + self.submodules.ctrl = ctrl + self.submodules.reader = reader + + self.memory = DRAMMemory(32, len(memory_data), init=memory_data) + assert 8 + sequence_len <= len(self.memory.mem) + + def reader(dut): + # fake writing to fifo + yield dut.ctrl.write.eq(1) + for _ in range(inital_writes): + yield + yield dut.ctrl.write.eq(0) + yield + + for _ in range(sequence_len): + # fake single write + yield dut.ctrl.write.eq(1) + yield + yield dut.ctrl.write.eq(0) + + while (yield dut.reader.source.valid) == 0: + yield + read_data.append((yield dut.reader.source.data)) + yield dut.reader.source.ready.eq(1) + yield + yield dut.reader.source.ready.eq(0) + yield + + dut = DUT() + generators = [ + reader(dut), + dut.memory.read_handler(dut.port), + self.fifo_ctrl_flag_checker(dut.ctrl, write_threshold=depth, + read_threshold=read_threshold), + timeout_generator(1500), + ] + run_simulation(dut, generators) + + read_data_expected = [memory_data[8 + i%depth] for i in range(sequence_len)] + self.assertEqual(read_data, read_data_expected) + + def test_fifo_reader_sequence(self): + self.fifo_reader_test(sequence_len=48, depth=64, read_threshold=0) + + def test_fifo_reader_address_wraps(self): + self.fifo_reader_test(sequence_len=48, depth=32, read_threshold=0) + + def test_fifo_reader_requires_threshold(self): + with self.assertRaises(TimeoutError): + self.fifo_reader_test(sequence_len=48, depth=32, read_threshold=8) + # will work after we perform the initial writes + self.fifo_reader_test(sequence_len=48, depth=32, read_threshold=8, inital_writes=8) + + # LiteDRAMFIFO --------------------------------------------------------------------------------- + + def test_fifo_default_thresholds(self): + # defaults: read_threshold=0, write_threshold=depth + read_threshold, write_threshold = (0, 128) + write_port = LiteDRAMNativeWritePort(address_width=32, data_width=32) + read_port = LiteDRAMNativeReadPort(address_width=32, data_width=32) + fifo = LiteDRAMFIFO(data_width=32, base=0, depth=write_threshold, + write_port=write_port, read_port=read_port) + + def generator(): + yield write_port.cmd.ready.eq(1) + yield write_port.wdata.ready.eq(1) + for i in range(write_threshold): + yield fifo.sink.valid.eq(1) + yield fifo.sink.data.eq(0) + yield + while (yield fifo.sink.ready) == 0: + yield + yield + + checker = self.fifo_ctrl_flag_checker(fifo.ctrl, write_threshold, read_threshold) + run_simulation(fifo, [generator(), checker]) + def test_fifo(self): + class DUT(Module): + def __init__(self): + self.write_port = LiteDRAMNativeWritePort(address_width=32, data_width=32) + self.read_port = LiteDRAMNativeReadPort(address_width=32, data_width=32) + self.submodules.fifo = LiteDRAMFIFO( + data_width = 32, + depth = 32, + base = 16, + write_port = self.write_port, + read_port = self.read_port, + read_threshold = 8, + write_threshold = 32 - 8 + ) + + self.memory = DRAMMemory(32, 128) + def generator(dut, valid_random=90): prng = random.Random(42) - for i in range(128 + 8): + # we need 8 more writes to account for read_threshold=8 + for i in range(64 + 8): while prng.randrange(100) < valid_random: yield yield dut.fifo.sink.valid.eq(1) @@ -52,7 +253,7 @@ class TestFIFO(unittest.TestCase): def checker(dut, ready_random=90): prng = random.Random(42) - for i in range(128): + for i in range(64): yield dut.fifo.source.ready.eq(0) yield while (yield dut.fifo.source.valid) != 1: @@ -63,7 +264,7 @@ class TestFIFO(unittest.TestCase): self.assertEqual((yield dut.fifo.source.data), i) yield - dut = FIFODUT() + dut = DUT() generators = [ generator(dut), checker(dut),