lib.cdc: Add ElasticBuffer and smoke test

Author: Wren6991

nmigen/lib/cdc.py

@@ -7,6 +7,7 @@
     "ResetSynchronizer",
     "PulseSynchronizer",
     "BusSynchronizer",
+    "ElasticBuffer",
     "Gearbox"
 ]
 
@@ -261,6 +262,66 @@ def elaborate(self, platform):
 
         return m
 
+class ElasticBuffer(Elaboratable):
+    """Pass data between two clock domains with the same frequency, and bounded phase difference.
+
+    Increasing the storage depth increases tolerance for clock wander and jitter, but still within
+    some bound. For less-well-behaved clocks, consider AsyncFIFO.
+
+    Parameters
+    ----------
+    width : int > 0
+        Width of databus to be resynchronized
+    depth : int > 1
+        Number of storage elements in buffer
+    idomain : str
+        Name of input clock domain
+    odomain : str
+        Name of output clock domain
+
+    Attributes
+    ----------
+    i : Signal(width)
+        Input data bus
+    o : Signal(width)
+        Output data bus
+    """
+    def __init__(self, width, depth, idomain, odomain):
+        if not isinstance(width, int) or width < 1:
+            raise TypeError("width must be a positive integer, not '{!r}'".format(width))
+        if not isinstance(depth, int) or depth <= 1:
+            raise TypeError("depth must be an integer > 1, not '{!r}'".format(depth))
+
+        self.i = Signal(width)
+        self.o = Signal(width)
+        self.width = width
+        self.depth = depth
+        self.idomain = idomain
+        self.odomain = odomain
+
+    def elaborate(self, platform):
+        m = Module()
+
+        wptr = Signal(max=self.depth, reset=self.depth // 2)
+        rptr = Signal(max=self.depth)
+        m.d[self.idomain] += wptr.eq(_incr(wptr, self.depth))
+        m.d[self.odomain] += rptr.eq(_incr(rptr, self.depth))
+
+        storage = Memory(self.width, self.depth)
+        wport = m.submodules.wport = storage.write_port(domain=self.idomain)
+        rport = m.submodules.rport = storage.read_port(domain=self.odomain)
+
+        m.d.comb += [
+            wport.en.eq(1),
+            wport.addr.eq(wptr),
+            wport.data.eq(self.i),
+            rport.addr.eq(rptr),
+            self.o.eq(rport.data)
+        ]
+
+        return m
+
+
 class Gearbox(Elaboratable):
     """Adapt the width of a continous datastream.
 

nmigen/test/test_lib_cdc.py

@@ -200,6 +200,47 @@ def process():
             sim.add_process(process)
             sim.run()
 
+class ElasticBufferTestCase(FHDLTestCase):
+    def test_paramcheck(self):
+        with self.assertRaises(TypeError):
+            e = ElasticBuffer(0, 2, "i", "o")
+        with self.assertRaises(TypeError):
+            e = ElasticBuffer("i", 2, "i", "o")
+        with self.assertRaises(TypeError):
+            e = ElasticBuffer(1, 1, "i", "o")
+        with self.assertRaises(TypeError):
+            e = ElasticBuffer(1, "i", "i", "o")
+        e = ElasticBuffer(1, 2, "i", "o")
+
+    def test_smoke_po2(self):
+        self.check_smoke(8, 8)
+
+    def test_smoke_nonpo2(self):
+        self.check_smoke(8, 7)
+
+    def check_smoke(self, width, depth):
+        m = Module()
+        m.domains += ClockDomain("sync")
+        e = m.submodules.dut = ElasticBuffer(width, depth, "sync", "sync")
+
+        with Simulator(m, vcd_file = open("test.vcd", "w")) as sim:
+            sim.add_clock(1e-6)
+            def process():
+                pipeline_filled = False
+                expected_out = 1
+                yield Tick()
+                for i in range(depth * 4):
+                    yield e.i.eq(i)
+                    if (yield e.o):
+                        pipeline_filled = True
+                    if pipeline_filled:
+                        self.assertEqual((yield e.o), expected_out)
+                        expected_out = (expected_out + 1) % (2 ** width)
+                    yield Tick()
+                self.assertEqual(pipeline_filled, True)
+            sim.add_process(process)
+            sim.run()
+
 # TODO: test with distinct clocks
 # (since we can currently only test symmetric aspect ratio)
 class GearboxTestCase(FHDLTestCase):