Add sanity checks for wirevector_by_name

This revealed a few bugs in passes where the wirevector_by_name dictionary was not being updated.

Also updated code’s developer notes for clarity.

Author: mdko

pyrtl/core.py

@@ -279,7 +279,7 @@ def add_wirevector(self, wirevector):
         self.wirevector_by_name[wirevector.name] = wirevector
 
     def remove_wirevector(self, wirevector):
-        """ Remove a wirevector object to the block."""
+        """ Remove a wirevector object from the block."""
         self.wirevector_set.remove(wirevector)
         del self.wirevector_by_name[wirevector.name]
 
@@ -505,7 +505,6 @@ def sanity_check(self):
         built according to the assumptions stated in the Block comments.
         """
 
-        # TODO: check that the wirevector_by_name is sane
         from .wire import Input, Const, Output
         from .helperfuncs import get_stack, get_stacks
 
@@ -529,9 +528,6 @@ def sanity_check(self):
                              'or "const_" as a signal name because those are reserved for '
                              'internal use)' % repr(wirevector_names_list))
 
-        # check for dead input wires (not connected to anything)
-        all_input_and_consts = self.wirevector_subset((Input, Const))
-
         # The following line also checks for duplicate wire drivers
         wire_src_dict, wire_dst_dict = self.net_connections()
         dest_set = set(wire_src_dict.keys())
@@ -542,9 +538,12 @@ def sanity_check(self):
             bad_wire_names = '\n    '.join(str(x) for x in connected_minus_allwires)
             raise PyrtlError('Unknown wires found in net:\n %s \n\n %s' % (bad_wire_names,
                              get_stacks(*connected_minus_allwires)))
+
+        all_input_and_consts = self.wirevector_subset((Input, Const))
+
+        # Check for wires that aren't connected to anything (inputs and consts can be unconnected)
         allwires_minus_connected = self.wirevector_set.difference(full_set)
         allwires_minus_connected = allwires_minus_connected.difference(all_input_and_consts)
-        #   ^ allow inputs and consts to be unconnected
         if len(allwires_minus_connected) > 0:
             bad_wire_names = '\n    '.join(str(x) for x in allwires_minus_connected)
             raise PyrtlError('Wires declared but not connected:\n %s \n\n %s' % (bad_wire_names,
@@ -561,6 +560,24 @@ def sanity_check(self):
         # Check for async memories not specified as such
         self.sanity_check_memory_sync(wire_src_dict)
 
+        # Check that all mappings in wirevector_by_name are consistent
+        bad_wv_by_name = [w for n, w in self.wirevector_by_name.items() if n != w.name]
+        if bad_wv_by_name:
+            raise PyrtlInternalError('Wires with inconsistent entry in wirevector_by_name '
+                                     'dict: %s' % [w.name for w in bad_wv_by_name])
+
+        # Check that all wires are in wirevector_by_name
+        wv_by_name_set = set(self.wirevector_by_name.keys())
+        missing_wires = wirevector_names_set.difference(wv_by_name_set)
+        if missing_wires:
+            raise PyrtlInternalError('Missing entries in wirevector_by_name for the '
+                                     'following wires: %s' % missing_wires)
+
+        unknown_wires = wv_by_name_set.difference(wirevector_names_set)
+        if unknown_wires:
+            raise PyrtlInternalError('Unknown wires found in wirevector_by_name: %s'
+                                     % unknown_wires)
+
         if debug_mode:
             # Check for wires that are destinations of a logicNet, but are not outputs and are never
             # used as args.

pyrtl/passes.py

@@ -98,8 +98,7 @@ def _remove_wire_nets(block):
     # now update the block with the new logic and remove wirevectors
     block.logic = new_logic
     for dead_wirevector in wire_removal_set:
-        del block.wirevector_by_name[dead_wirevector.name]
-        block.wirevector_set.remove(dead_wirevector)
+        block.remove_wirevector(dead_wirevector)
 
     block.sanity_check()
 
@@ -446,6 +445,7 @@ def _remove_unused_wires(block, keep_inputs=True):
             PyrtlInternalError("Output wire, " + removed_wire.name + " not driven")
 
     block.wirevector_set = valid_wires
+    block.wirevector_by_name = {wire.name: wire for wire in valid_wires}
 
 # --------------------------------------------------------------------
 #    __           ___       ___  __     __
@@ -825,7 +825,8 @@ def direct_connect_outputs(block=None):
 
     block.logic.difference_update(nets_to_remove)
     block.logic.update(nets_to_add)
-    block.wirevector_set.difference_update(wirevectors_to_remove)
+    for w in wirevectors_to_remove:
+        block.remove_wirevector(w)
 
 
 def _make_tree(wire, block, curr_fanout):

tests/test_core.py

@@ -476,8 +476,8 @@ class TestSanityCheck(unittest.TestCase):
     def setUp(self):
         pyrtl.reset_working_block()
 
-    def sanity_error(self, msg):
-        with self.assertRaisesRegexp(pyrtl.PyrtlError, msg):
+    def sanity_error(self, msg, error_type=pyrtl.PyrtlError):
+        with self.assertRaisesRegexp(error_type, msg):
             pyrtl.working_block().sanity_check()
 
     def test_missing_bitwidth(self):
@@ -510,6 +510,26 @@ def test_not_driven(self):
         out <<= w
         self.sanity_error("used but never driven")
 
+    def test_inconsistent_wirevector_by_name(self):
+        c = pyrtl.Const(42)
+        inp = pyrtl.Input(8, 'inp')
+        out = pyrtl.Output(8, 'out')
+        out <<= inp & c
+        pyrtl.working_block().wirevector_by_name['inp'] = c
+        self.sanity_error("inconsistent entry in wirevector_by_name", pyrtl.PyrtlInternalError)
+
+    def test_missing_wire_in_wirevector_by_name(self):
+        inp = pyrtl.Input(8, 'inp')
+        out = pyrtl.Output(8, 'out')
+        out <<= inp
+        del pyrtl.working_block().wirevector_by_name['inp']
+        self.sanity_error("Missing entries in wirevector_by_name", pyrtl.PyrtlInternalError)
+
+    def test_extra_wire_in_wirevector_by_name(self):
+        inp = pyrtl.Input(8, 'inp')
+        pyrtl.working_block().wirevector_set.discard(inp)
+        self.sanity_error("Unknown wires found in wirevector_by_name", pyrtl.PyrtlInternalError)
+
 
 class TestLogicNets(unittest.TestCase):
     def setUp(self):