Skip to content

Commit

Permalink
Merge pull request cocotb#727 from eric-wieser/edge-weakref
Browse files Browse the repository at this point in the history
 Break reference cycles in the edge triggers
  • Loading branch information
imphil committed Dec 17, 2018
2 parents b0919cf + 1fa2841 commit 9f0cc56
Show file tree
Hide file tree
Showing 2 changed files with 40 additions and 58 deletions.
4 changes: 0 additions & 4 deletions cocotb/handle.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,6 @@
from cocotb.binary import BinaryValue
from cocotb.log import SimLog
from cocotb.result import TestError
from cocotb.triggers import _RisingEdge, _FallingEdge, _Edge
from cocotb.utils import get_python_integer_types

# Only issue a warning for each deprecated attribute access
Expand Down Expand Up @@ -546,9 +545,6 @@ def __init__(self, handle, path):
_handle [integer] : vpi/vhpi handle to the simulator object
"""
NonHierarchyIndexableObject.__init__(self, handle, path)
self._r_edge = _RisingEdge(self)
self._f_edge = _FallingEdge(self)
self._e_edge = _Edge(self)

def drivers(self):
"""
Expand Down
94 changes: 40 additions & 54 deletions cocotb/triggers.py
Original file line number Diff line number Diff line change
Expand Up @@ -206,86 +206,72 @@ def NextTimeStep():
return _nxts


class _Edge(GPITrigger):
class _EdgeBase(GPITrigger):
"""
Execution will resume when an edge occurs on the provided signal
"""
def __init__(self, signal):
GPITrigger.__init__(self)
self.signal = signal

def prime(self, callback):
"""Register notification of a value change via a callback"""
if self.cbhdl == 0:
self.cbhdl = simulator.register_value_change_callback(self.signal.
_handle,
callback,
3,
self)
if self.cbhdl == 0:
raise_error(self, "Unable set up %s Trigger" % (str(self)))
Trigger.prime(self)

def __str__(self):
return self.__class__.__name__ + "(%s)" % self.signal._name
@classmethod
@property
def _edge_type(self):
"""
The edge type, as understood by the C code. Must be set in subclasses
"""
raise NotImplementedError

def Edge(signal):
return signal._e_edge
# 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

class _RisingOrFallingEdge(_Edge):
def __init__(self, signal, rising):
_Edge.__init__(self, signal)
if rising is True:
self._rising = 1
else:
self._rising = 2
def __init__(self, signal):
super(_EdgeBase, self).__init__()
self.signal = signal

def prime(self, callback):
"""Register notification of a value change via a callback"""
if self.cbhdl == 0:
self.cbhdl = simulator.register_value_change_callback(self.signal.
_handle,
callback,
self._rising,
self)
self.cbhdl = simulator.register_value_change_callback(
self.signal._handle, callback, type(self)._edge_type, self
)
if self.cbhdl == 0:
raise_error(self, "Unable set up %s Trigger" % (str(self)))
Trigger.prime(self)
super(_EdgeBase, self).prime()

def __str__(self):
return self.__class__.__name__ + "(%s)" % self.signal._name


class _RisingEdge(_RisingOrFallingEdge):
"""
Execution will resume when a rising edge occurs on the provided signal
"""
def __init__(self, signal):
_RisingOrFallingEdge.__init__(self, signal, rising=True)
class RisingEdge(_EdgeBase):
""" Triggers on the rising edge of the provided signal """
_edge_type = 1


def RisingEdge(signal):
return signal._r_edge
class FallingEdge(_EdgeBase):
""" Triggers on the falling edge of the provided signal """
_edge_type = 2


class _FallingEdge(_RisingOrFallingEdge):
"""
Execution will resume when a falling edge occurs on the provided signal
"""
def __init__(self, signal):
_RisingOrFallingEdge.__init__(self, signal, rising=False)

class Edge(_EdgeBase):
""" Triggers on either edge in a signal """
_edge_type = 3

def FallingEdge(signal):
return signal._f_edge


class ClockCycles(_Edge):
class ClockCycles(GPITrigger):
"""
Execution will resume after N rising edges or N falling edges
"""
def __init__(self, signal, num_cycles, rising=True):
_Edge.__init__(self, signal)
super(ClockCycles, self).__init__()
self.signal = signal
self.num_cycles = num_cycles
if rising is True:
self._rising = 1
Expand Down

0 comments on commit 9f0cc56

Please sign in to comment.