Merge pull request cocotb#733 from eric-wieser/trigger-metaclass

Fix a regression introduced in cocotbgh-727 and cocotbgh-723, caused by a misunderstanding of how __new__ works

cocotb/triggers.py

@@ -38,7 +38,10 @@
     import simulator
 from cocotb.log import SimLog
 from cocotb.result import raise_error
-from cocotb.utils import get_sim_steps, get_time_from_sim_steps
+from cocotb.utils import (
+    get_sim_steps, get_time_from_sim_steps, with_metaclass,
+    ParametrizedSingleton
+)
 
 
 class TriggerException(Exception):
@@ -206,7 +209,7 @@ def NextTimeStep():
     return _nxts
 
 
-class _EdgeBase(GPITrigger):
+class _EdgeBase(with_metaclass(ParametrizedSingleton, GPITrigger)):
     """
     Execution will resume when an edge occurs on the provided signal
     """
@@ -218,19 +221,9 @@ def _edge_type(self):
         """
         raise NotImplementedError
 
-    # Ensure that each signal has at most one edge trigger per edge type.
-    # Using a weak dictionary ensures we don't create a reference cycle
-    _instances = weakref.WeakValueDictionary()
-
-    def __new__(cls, signal):
-        # find the existing instance, if possible - else create a new one
-        key = (signal, cls._edge_type)
-        try:
-            return cls._instances[key]
-        except KeyError:
-            instance = super(_EdgeBase, cls).__new__(cls)
-            cls._instances[key] = instance
-            return instance
+    @classmethod
+    def __singleton_key__(cls, signal):
+        return signal
 
     def __init__(self, signal):
         super(_EdgeBase, self).__init__()
@@ -507,22 +500,13 @@ def prime(self, callback):
         callback(self)
 
 
-class Join(PythonTrigger):
+class Join(with_metaclass(ParametrizedSingleton, PythonTrigger)):
     """
     Join a coroutine, firing when it exits
     """
-    # Ensure that each coroutine has at most one join trigger.
-    # Using a weak dictionary ensures we don't create a reference cycle
-    _instances = weakref.WeakValueDictionary()
-
-    def __new__(cls, coroutine):
-        # find the existing instance, if possible - else create a new one
-        try:
-            return cls._instances[coroutine]
-        except KeyError:
-            instance = super(Join, cls).__new__(cls)
-            cls._instances[coroutine] = instance
-            return instance
+    @classmethod
+    def __singleton_key__(cls, coroutine):
+        return coroutine
 
     def __init__(self, coroutine):
         super(Join, self).__init__()

cocotb/utils.py

@@ -33,6 +33,7 @@
 import math
 import os
 import sys
+import weakref
 
 # For autodocumentation don't need the extension modules
 if "SPHINX_BUILD" in os.environ:
@@ -417,6 +418,72 @@ def exec_(_code_, _globs_=None, _locs_=None):
         exec("""exec _code_ in _globs_, _locs_""")
 
 
+# this is six.with_metaclass, with a clearer docstring
+def with_metaclass(meta, *bases):
+    """
+    This provides:
+
+        class Foo(with_metaclass(Meta, Base1, Base2)): pass
+
+    which is a unifying syntax for:
+
+        # python 3
+        class Foo(Base1, Base2, metaclass=Meta): pass
+
+        # python 2
+        class Foo(Base1, Base2)
+            __metaclass__ = Meta
+
+    """
+    # This requires a bit of explanation: the basic idea is to make a dummy
+    # metaclass for one level of class instantiation that replaces itself with
+    # the actual metaclass.
+    class metaclass(type):
+
+        def __new__(cls, name, this_bases, d):
+            return meta(name, bases, d)
+
+        @classmethod
+        def __prepare__(cls, name, this_bases):
+            return meta.__prepare__(name, bases)
+    return type.__new__(metaclass, 'temporary_class', (), {})
+
+
+class ParametrizedSingleton(type):
+    """
+    A metaclass that allows class construction to reuse an existing instance
+
+    We use this so that `RisingEdge(sig)` and `Join(coroutine)` always return
+    the same instance, rather than creating new copies.
+    """
+
+    def __init__(cls, *args, **kwargs):
+        # Attach a lookup table to this class.
+        # Weak such that if the instance is no longer referenced, it can be
+        # collected.
+        cls.__instances = weakref.WeakValueDictionary()
+
+    def __singleton_key__(cls, *args, **kwargs):
+        """
+        Convert the construction arguments into a normalized representation that
+        uniquely identifies this singleton.
+        """
+        # Once we drop python 2, we can implement a default like the following,
+        # which will work in 99% of cases:
+        # return tuple(inspect.Signature(cls).bind(*args, **kwargs).arguments.items())
+        raise NotImplementedError
+
+    def __call__(cls, *args, **kwargs):
+        key = cls.__singleton_key__(*args, **kwargs)
+        try:
+            return cls.__instances[key]
+        except KeyError:
+            # construct the object as normal
+            self = super(ParametrizedSingleton, cls).__call__(*args, **kwargs)
+            cls.__instances[key] = self
+            return self
+
+
 if __name__ == "__main__":
     import random
     a = ""

tests/test_cases/test_cocotb/test_cocotb.py

@@ -763,6 +763,24 @@ def test_join_identity(dut):
     clk_gen.kill()
 
 
+@cocotb.test()
+def test_edge_identity(dut):
+    """
+    Test that Edge triggers returns the same object each time
+    """
+
+    re = RisingEdge(dut.clk)
+    fe = FallingEdge(dut.clk)
+    e = Edge(dut.clk)
+
+    assert re is RisingEdge(dut.clk)
+    assert fe is FallingEdge(dut.clk)
+    assert e is Edge(dut.clk)
+
+    # check they are all unique
+    assert len({re, fe, e}) == 3
+    yield Timer(1)
+
 
 if sys.version_info[:2] >= (3, 3):
     # this would be a syntax error in older python, so we do the whole