lib.cdc: extract AsyncFFSynchronizer.

In some cases, it is necessary to synchronize a reset-like signal but
a new clock domain is not desirable. To address these cases, extract
the implementation of ResetSynchronizer into AsyncFFSynchronizer,
and replace ResetSynchronizer with a thin wrapper around it.

nmigen/hdl/cd.py

@@ -21,6 +21,8 @@ class ClockDomain:
         If ``True``, the domain does not use a reset signal. Registers within this domain are
         still all initialized to their reset state once, e.g. through Verilog `"initial"`
         statements.
+    clock_edge : str
+        The edge of the clock signal on which signals are sampled. Must be one of "pos" or "neg".
     async_reset : bool
         If ``True``, the domain uses an asynchronous reset, and registers within this domain
         are initialized to their reset state when reset level changes. Otherwise, registers

nmigen/lib/cdc.py

@@ -2,7 +2,7 @@
 from .. import *
 
 
-__all__ = ["FFSynchronizer", "ResetSynchronizer", "PulseSynchronizer"]
+__all__ = ["FFSynchronizer", "AsyncFFSynchronizer", "ResetSynchronizer", "PulseSynchronizer"]
 
 
 def _check_stages(stages):
@@ -95,6 +95,73 @@ def elaborate(self, platform):
         return m
 
 
+class AsyncFFSynchronizer(Elaboratable):
+    """Synchronize deassertion of an asynchronous signal.
+
+    The signal driven by the :class:`AsyncFFSynchronizer` is asserted asynchronously and deasserted
+    synchronously, eliminating metastability during deassertion.
+
+    This synchronizer is primarily useful for resets and reset-like signals.
+
+    Parameters
+    ----------
+    i : Signal(1), in
+        Asynchronous input signal, to be synchronized.
+    o : Signal(1), out
+        Synchronously released output signal.
+    domain : str
+        Name of clock domain to reset.
+    stages : int, >=2
+        Number of synchronization stages between input and output. The lowest safe number is 2,
+        with higher numbers reducing MTBF further, at the cost of increased deassertion latency.
+    async_edge : str
+        The edge of the input signal which causes the output to be set. Must be one of "pos" or "neg".
+    """
+    def __init__(self, i, o, *, domain="sync", stages=2, async_edge="pos", max_input_delay=None):
+        _check_stages(stages)
+
+        self.i = i
+        self.o = o
+
+        self._domain = domain
+        self._stages = stages
+
+        if async_edge not in ("pos", "neg"):
+            raise ValueError("AsyncFFSynchronizer async edge must be one of 'pos' or 'neg', not {!r}"
+                             .format(async_edge))
+        self._edge = async_edge
+
+        self._max_input_delay = max_input_delay
+
+    def elaborate(self, platform):
+        if hasattr(platform, "get_async_ff_sync"):
+            return platform.get_async_ff_sync(self)
+
+        if self._max_input_delay is not None:
+            raise NotImplementedError("Platform '{}' does not support constraining input delay "
+                                      "for AsyncFFSynchronizer"
+                                      .format(type(platform).__name__))
+
+        m = Module()
+        m.domains += ClockDomain("async_ff", async_reset=True, local=True)
+        flops = [Signal(1, name="stage{}".format(index), reset=1)
+                 for index in range(self._stages)]
+        for i, o in zip((0, *flops), flops):
+            m.d.async_ff += o.eq(i)
+
+        if self._edge == "pos":
+            m.d.comb += ResetSignal("async_ff").eq(self.i)
+        else:
+            m.d.comb += ResetSignal("async_ff").eq(~self.i)
+
+        m.d.comb += [
+            ClockSignal("async_ff").eq(ClockSignal(self._domain)),
+            self.o.eq(flops[-1])
+        ]
+
+        return m
+
+
 class ResetSynchronizer(Elaboratable):
     """Synchronize deassertion of a clock domain reset.
 
@@ -109,7 +176,7 @@ class ResetSynchronizer(Elaboratable):
 
     Parameters
     ----------
-    arst : Signal(1), out
+    arst : Signal(1), in
         Asynchronous reset signal, to be synchronized.
     domain : str
         Name of clock domain to reset.
@@ -133,29 +200,11 @@ def __init__(self, arst, *, domain="sync", stages=2, max_input_delay=None):
         self._domain = domain
         self._stages = stages
 
-        self._max_input_delay = None
+        self._max_input_delay = max_input_delay
 
     def elaborate(self, platform):
-        if hasattr(platform, "get_reset_sync"):
-            return platform.get_reset_sync(self)
-
-        if self._max_input_delay is not None:
-            raise NotImplementedError("Platform '{}' does not support constraining input delay "
-                                      "for ResetSynchronizer"
-                                      .format(type(platform).__name__))
-
-        m = Module()
-        m.domains += ClockDomain("reset_sync", async_reset=True, local=True)
-        flops = [Signal(1, name="stage{}".format(index), reset=1)
-                 for index in range(self._stages)]
-        for i, o in zip((0, *flops), flops):
-            m.d.reset_sync += o.eq(i)
-        m.d.comb += [
-            ClockSignal("reset_sync").eq(ClockSignal(self._domain)),
-            ResetSignal("reset_sync").eq(self.arst),
-            ResetSignal(self._domain).eq(flops[-1])
-        ]
-        return m
+        return AsyncFFSynchronizer(self.arst, ResetSignal(self._domain), domain=self._domain,
+                stages=self._stages, max_input_delay=self._max_input_delay)
 
 
 class PulseSynchronizer(Elaboratable):

nmigen/test/test_lib_cdc.py

@@ -54,6 +54,97 @@ def process():
         sim.run()
 
 
+class AsyncFFSynchronizerTestCase(FHDLTestCase):
+    def test_stages_wrong(self):
+        with self.assertRaises(TypeError,
+                msg="Synchronization stage count must be a positive integer, not 0"):
+            ResetSynchronizer(Signal(), stages=0)
+        with self.assertRaises(ValueError,
+                msg="Synchronization stage count may not safely be less than 2"):
+            ResetSynchronizer(Signal(), stages=1)
+
+    def test_edge_wrong(self):
+        with self.assertRaises(ValueError,
+                msg="AsyncFFSynchronizer async edge must be one of 'pos' or 'neg', not 'xxx'"):
+            AsyncFFSynchronizer(Signal(), Signal(), domain="sync", async_edge="xxx")
+
+    def test_pos_edge(self):
+        i = Signal()
+        o = Signal()
+        m = Module()
+        m.domains += ClockDomain("sync")
+        m.submodules += AsyncFFSynchronizer(i, o)
+
+        sim = Simulator(m)
+        sim.add_clock(1e-6)
+        def process():
+            # initial reset
+            self.assertEqual((yield i), 0)
+            self.assertEqual((yield o), 1)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 1)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 0)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 0)
+            yield Tick(); yield Delay(1e-8)
+
+            yield i.eq(1)
+            yield Delay(1e-8)
+            self.assertEqual((yield o), 1)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 1)
+            yield i.eq(0)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 1)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 0)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 0)
+            yield Tick(); yield Delay(1e-8)
+        sim.add_process(process)
+        with sim.write_vcd("test.vcd"):
+            sim.run()
+
+    def test_neg_edge(self):
+        i = Signal(reset=1)
+        o = Signal()
+        m = Module()
+        m.domains += ClockDomain("sync")
+        m.submodules += AsyncFFSynchronizer(i, o, async_edge="neg")
+
+        sim = Simulator(m)
+        sim.add_clock(1e-6)
+        def process():
+            # initial reset
+            self.assertEqual((yield i), 1)
+            self.assertEqual((yield o), 1)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 1)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 0)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 0)
+            yield Tick(); yield Delay(1e-8)
+
+            yield i.eq(0)
+            yield Delay(1e-8)
+            self.assertEqual((yield o), 1)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 1)
+            yield i.eq(1)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 1)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 0)
+            yield Tick(); yield Delay(1e-8)
+            self.assertEqual((yield o), 0)
+            yield Tick(); yield Delay(1e-8)
+        sim.add_process(process)
+        with sim.write_vcd("test.vcd"):
+            sim.run()
+
+
 class ResetSynchronizerTestCase(FHDLTestCase):
     def test_stages_wrong(self):
         with self.assertRaises(TypeError,

nmigen/vendor/intel.py

@@ -400,15 +400,24 @@ def get_ff_sync(self, ff_sync):
             o_dout=ff_sync.o,
         )
 
-    def get_reset_sync(self, reset_sync):
+    def get_async_ff_sync(self, async_ff_sync):
         m = Module()
-        rst_n = Signal()
-        m.submodules += Instance("altera_std_synchronizer",
-            p_depth=reset_sync._stages,
-            i_clk=ClockSignal(reset_sync._domain),
-            i_reset_n=~reset_sync.arst,
-            i_din=Const(1),
-            o_dout=rst_n,
-        )
-        m.d.comb += ResetSignal(reset_sync._domain).eq(~rst_n)
+        sync_output = Signal()
+        if async_ff_sync.edge == "pos":
+            m.submodules += Instance("altera_std_synchronizer",
+                p_depth=async_ff_sync._stages,
+                i_clk=ClockSignal(async_ff_sync._domain),
+                i_reset_n=~async_ff_sync.i,
+                i_din=Const(1),
+                o_dout=sync_output,
+            )
+        else:
+            m.submodules += Instance("altera_std_synchronizer",
+                p_depth=async_ff_sync._stages,
+                i_clk=ClockSignal(async_ff_sync._domain),
+                i_reset_n=async_ff_sync.i,
+                i_din=Const(1),
+                o_dout=sync_output,
+            )
+        m.d.comb += async_ff_sync.o.eq(~sync_output)
         return m

nmigen/vendor/xilinx_7series.py

@@ -407,21 +407,27 @@ def get_ff_sync(self, ff_sync):
         m.d.comb += ff_sync.o.eq(flops[-1])
         return m
 
-    def get_reset_sync(self, reset_sync):
+    def get_async_ff_sync(self, async_ff_sync):
         m = Module()
-        m.domains += ClockDomain("reset_sync", async_reset=True, local=True)
+        m.domains += ClockDomain("async_ff", async_reset=True, local=True)
         flops = [Signal(1, name="stage{}".format(index), reset=1,
                         attrs={"ASYNC_REG": "TRUE"})
-                 for index in range(reset_sync._stages)]
-        if reset_sync._max_input_delay is None:
+                 for index in range(async_ff_sync._stages)]
+        if async_ff_sync._max_input_delay is None:
             flops[0].attrs["nmigen.vivado.false_path"] = "TRUE"
         else:
-            flops[0].attrs["nmigen.vivado.max_delay"] = str(reset_sync._max_input_delay * 1e9)
+            flops[0].attrs["nmigen.vivado.max_delay"] = str(async_ff_sync._max_input_delay * 1e9)
         for i, o in zip((0, *flops), flops):
-            m.d.reset_sync += o.eq(i)
+            m.d.async_ff += o.eq(i)
+
+        if self._edge == "pos":
+            m.d.comb += ResetSignal("async_ff").eq(asnyc_ff_sync.i)
+        else:
+            m.d.comb += ResetSignal("async_ff").eq(~asnyc_ff_sync.i)
+
         m.d.comb += [
-            ClockSignal("reset_sync").eq(ClockSignal(reset_sync._domain)),
-            ResetSignal("reset_sync").eq(reset_sync.arst),
-            ResetSignal(reset_sync._domain).eq(flops[-1])
+            ClockSignal("async_ff").eq(ClockSignal(asnyc_ff_sync._domain)),
+            async_ff_sync.o.eq(flops[-1])
         ]
+
         return m

nmigen/vendor/xilinx_spartan_3_6.py

@@ -437,24 +437,30 @@ def get_ff_sync(self, ff_sync):
         m.d.comb += ff_sync.o.eq(flops[-1])
         return m
 
-    def get_reset_sync(self, reset_sync):
-        if reset_sync._max_input_delay is not None:
+    def get_async_ff_sync(self, async_ff_sync):
+        if self._max_input_delay is not None:
             raise NotImplementedError("Platform '{}' does not support constraining input delay "
-                                      "for ResetSynchronizer"
+                                      "for AsyncFFSynchronizer"
                                       .format(type(self).__name__))
 
         m = Module()
-        m.domains += ClockDomain("reset_sync", async_reset=True, local=True)
+        m.domains += ClockDomain("async_ff", async_reset=True, local=True)
         flops = [Signal(1, name="stage{}".format(index), reset=1,
                         attrs={"ASYNC_REG": "TRUE"})
-                 for index in range(reset_sync._stages)]
+                 for index in range(async_ff_sync._stages)]
         for i, o in zip((0, *flops), flops):
-            m.d.reset_sync += o.eq(i)
+            m.d.async_ff += o.eq(i)
+
+        if self._edge == "pos":
+            m.d.comb += ResetSignal("async_ff").eq(asnyc_ff_sync.i)
+        else:
+            m.d.comb += ResetSignal("async_ff").eq(~asnyc_ff_sync.i)
+
         m.d.comb += [
-            ClockSignal("reset_sync").eq(ClockSignal(reset_sync._domain)),
-            ResetSignal("reset_sync").eq(reset_sync.arst),
-            ResetSignal(reset_sync._domain).eq(flops[-1])
+            ClockSignal("async_ff").eq(ClockSignal(asnyc_ff_sync._domain)),
+            async_ff_sync.o.eq(flops[-1])
         ]
+
         return m
 
 XilinxSpartan3APlatform = XilinxSpartan3Or6Platform

nmigen/vendor/xilinx_ultrascale.py

@@ -403,21 +403,27 @@ def get_ff_sync(self, ff_sync):
         m.d.comb += ff_sync.o.eq(flops[-1])
         return m
 
-    def get_reset_sync(self, reset_sync):
+    def get_async_ff_sync(self, async_ff_sync):
         m = Module()
-        m.domains += ClockDomain("reset_sync", async_reset=True, local=True)
+        m.domains += ClockDomain("async_ff", async_reset=True, local=True)
         flops = [Signal(1, name="stage{}".format(index), reset=1,
                         attrs={"ASYNC_REG": "TRUE"})
-                 for index in range(reset_sync._stages)]
-        if reset_sync._max_input_delay is None:
+                 for index in range(async_ff_sync._stages)]
+        if async_ff_sync._max_input_delay is None:
             flops[0].attrs["nmigen.vivado.false_path"] = "TRUE"
         else:
-            flops[0].attrs["nmigen.vivado.max_delay"] = str(reset_sync._max_input_delay * 1e9)
+            flops[0].attrs["nmigen.vivado.max_delay"] = str(async_ff_sync._max_input_delay * 1e9)
         for i, o in zip((0, *flops), flops):
-            m.d.reset_sync += o.eq(i)
+            m.d.async_ff += o.eq(i)
+
+        if self._edge == "pos":
+            m.d.comb += ResetSignal("async_ff").eq(asnyc_ff_sync.i)
+        else:
+            m.d.comb += ResetSignal("async_ff").eq(~asnyc_ff_sync.i)
+
         m.d.comb += [
-            ClockSignal("reset_sync").eq(ClockSignal(reset_sync._domain)),
-            ResetSignal("reset_sync").eq(reset_sync.arst),
-            ResetSignal(reset_sync._domain).eq(flops[-1])
+            ClockSignal("async_ff").eq(ClockSignal(asnyc_ff_sync._domain)),
+            async_ff_sync.o.eq(flops[-1])
         ]
+
         return m