Refactoring and added tests

Author: LarsAsplund

tests/unit/test_modelsim_interface.py

@@ -18,6 +18,7 @@
 from vunit.sim_if.modelsim import ModelSimInterface
 from vunit.project import Project
 from vunit.ostools import renew_path, write_file
+from vunit.test.bench import Configuration
 from vunit.vhdl_standard import VHDL
 
 
@@ -316,14 +317,60 @@ def test_overwrites_modelsim_ini_file_from_user(self):
         with open(modelsim_ini, "r") as fptr:
             self.assertEqual(fptr.read(), "user")
 
+    @mock.patch("vunit.sim_if.modelsim.LOGGER", autospec=True)
+    @mock.patch("vunit.sim_if.check_output", autospec=True, return_value="")
+    @mock.patch("vunit.sim_if.modelsim.Process", autospec=True)
+    @mock.patch("vunit.sim_if.vsim_simulator_mixin.Process", autospec=True)
+    def test_optimize(self, vsim_simulator_mixin_process, modelsim_process, check_output, LOGGER):
+        simif = ModelSimInterface(prefix=self.prefix_path, output_path=self.output_path, persistent=False)
+        project = Project()
+        project.add_library("lib", str(Path(self.libraries_path) / "lib"))
+        write_file("file.vhd", "")
+        project.add_source_file("file.vhd", "lib", file_type="vhdl", vhdl_standard=VHDL.standard("2008"))
+        simif.compile_project(project)
+        config = make_config(sim_options={"modelsim.three_step_flow":True})
+
+        # First call should optimize design
+        simif.simulate(self.simulation_output_path, "test_suite_name", config, False)
+        design_to_optimize = "lib.tb(test)"
+        expected_calls = [
+            mock.call("%s scheduled for optimization.", design_to_optimize),
+            mock.call("Acquired library lock for %s to optimize %s.", "lib", design_to_optimize),
+            mock.call("Optimizing %s.", design_to_optimize),
+            mock.call("%s optimization completed.", design_to_optimize),
+        ]
+        self.assertEqual(LOGGER.debug.call_count, len(expected_calls))
+        LOGGER.debug.assert_has_calls(expected_calls)
+
+        # Second call should reuse the already optimized design
+        LOGGER.reset_mock()
+        simif.simulate(self.simulation_output_path, "test_suite_name", config, False)
+        LOGGER.debug.assert_called_once_with("Reusing optimized %s.", "lib.tb(test)")
+
+        # Fake that design is being optimized and that it is being waited for
+        LOGGER.reset_mock()
+        simif._optimized_designs[design_to_optimize]["optimized_design"] = None
+        simif.simulate(self.simulation_output_path, "test_suite_name", config, False)
+        expected_calls = [
+            mock.call("Waiting for %s to be optimized.", design_to_optimize),
+            mock.call("Done waiting for %s to be optimized.", design_to_optimize),
+        ]
+        self.assertEqual(LOGGER.debug.call_count, len(expected_calls))
+        LOGGER.debug.assert_has_calls(expected_calls)
+
+
     def setUp(self):
         self.test_path = str(Path(__file__).parent / "test_modelsim_out")
 
         self.output_path = str(Path(self.test_path) / "modelsim")
         self.prefix_path = str(Path(self.test_path) / "prefix" / "bin")
+        self.libraries_path = str(Path(self.output_path) / "libraries")
+        self.simulation_output_path = str(Path(self.test_path) / "test_output" / "lib.tb")
         renew_path(self.test_path)
         renew_path(self.output_path)
         renew_path(self.prefix_path)
+        renew_path(self.libraries_path)
+        renew_path(self.simulation_output_path)
         installed_modelsim_ini = str(Path(self.prefix_path) / ".." / "modelsim.ini")
         write_file(installed_modelsim_ini, "[Library]")
         self.project = Project()
@@ -334,3 +381,23 @@ def tearDown(self):
         os.chdir(self.cwd)
         if Path(self.test_path).exists():
             rmtree(self.test_path)
+
+def make_config(sim_options=None, generics=None, verilog=False):
+    """
+    Utility to reduce boiler plate in tests
+    """
+    cfg = mock.Mock(spec=Configuration)
+    cfg.library_name = "lib"
+
+    if verilog:
+        cfg.entity_name = "tb"
+        cfg.architecture_name = None
+    else:
+        cfg.entity_name = "tb"
+        cfg.architecture_name = "test"
+
+    cfg.sim_options = {} if sim_options is None else sim_options
+    cfg.generics = {} if generics is None else generics
+    cfg.vhdl_configuration_name = None
+    cfg.vhdl_assert_stop_level = "error"
+    return cfg

vunit/sim_if/modelsim.py

@@ -102,8 +102,13 @@ def __init__(self, prefix, output_path, persistent=False, gui=False):
         self._coverage_files = set()
         assert not (persistent and gui)
         self._create_modelsim_ini()
+        # Contains design already optimized, i.e. the optimized design can be reused
         self._optimized_designs = {}
+        # Contains locks for each library. If locked, a design belonging to the library
+        # is being optimized and no other design in that library can be optimized at the
+        # same time (from another thread)
         self._library_locks = {}
+        # Lock to access the two shared variables above
         self._shared_state_lock = Lock()
 
     def _create_modelsim_ini(self):
@@ -216,7 +221,7 @@ def create_library(self, library_name, path, mapped_libraries=None):
             os.makedirs(apath)
 
         if not file_exists(path):
-            proc = Process([str(Path(self._prefix) / "vlib"), "-unix", "-type", "directory", path], env=self.get_env())
+            proc = Process([str(Path(self._prefix) / "vlib"), "-unix", path], env=self.get_env())
             proc.consume_output(callback=None)
 
         if library_name in mapped_libraries and mapped_libraries[library_name] == path:
@@ -238,7 +243,7 @@ def _get_mapped_libraries(self):
 
     def _optimize_design(self, config):
         """
-        Return if design shall be optimized.
+        Return True if design shall be optimized.
         """
 
         return config.sim_options.get("modelsim.three_step_flow", False)
@@ -309,7 +314,7 @@ def _create_optimize_function(self, config):
         return true
     }}
 
-    return False
+    return false
 }}
 """.format(
             vopt_flags=" ".join(vopt_flags)
@@ -364,7 +369,7 @@ def _wait_for_file_lock(library):
         log_waiting = True
         while (Path(library.directory) / "_lock").exists():
             if log_waiting:
-                LOGGER.debug("Waiting for %s to be removed", Path(library.directory) / "_lock")
+                LOGGER.debug("Waiting for %s to be removed.", Path(library.directory) / "_lock")
                 log_waiting = False
             sleep(0.05)
 
@@ -376,13 +381,13 @@ def _acquire_library_lock(self, library, config, design_to_optimize):
             library_lock = self._library_locks[config.library_name]
 
         if library_lock.locked():
-            LOGGER.debug("Waiting for library lock for %s to optimize %s", config.library_name, design_to_optimize)
+            LOGGER.debug("Waiting for library lock for %s to optimize %s.", config.library_name, design_to_optimize)
         # Do not completely block to allow for Ctrl+C
         while not library_lock.acquire(timeout=0.05):
             pass
 
         self._wait_for_file_lock(library)
-        LOGGER.debug("Acquired library lock for %s to optimize %s", config.library_name, design_to_optimize)
+        LOGGER.debug("Acquired library lock for %s to optimize %s.", config.library_name, design_to_optimize)
 
     def _release_library_lock(self, library, config):
         """
@@ -393,6 +398,9 @@ def _release_library_lock(self, library, config):
             self._library_locks[config.library_name].release()
 
     def _optimize(self, config, script_path):
+        """
+        Optimize design and return simulation target or False if optimization failed.
+        """
         design_to_optimize = self._design_to_optimize(config)
 
         libraries = {lib.name: lib for lib in self._libraries}
@@ -401,7 +409,7 @@ def _optimize(self, config, script_path):
         optimize = False
         with self._shared_state_lock:
             if design_to_optimize not in self._optimized_designs:
-                LOGGER.debug("%s scheduled for optimization", design_to_optimize)
+                LOGGER.debug("%s scheduled for optimization.", design_to_optimize)
                 self._optimized_designs[design_to_optimize] = {
                     "optimized_design": None,
                     "optimization_completed": Event(),
@@ -416,7 +424,7 @@ def _optimize(self, config, script_path):
         if optimize:
             self._acquire_library_lock(library, config, design_to_optimize)
 
-            LOGGER.debug("Optimizing %s", design_to_optimize)
+            LOGGER.debug("Optimizing %s.", design_to_optimize)
 
             optimized_design = self._to_optimized_design(design_to_optimize)
 
@@ -428,11 +436,11 @@ def _optimize(self, config, script_path):
 
             else:
                 tcl = f"""\
-        onerror {{quit -code 1}}
-        source "{fix_path(str(optimize_file_name))!s}"
-        set failed [vunit_optimize]
-        if {{$failed}} {{quit -code 1}}
-        quit -code 0
+onerror {{quit -code 1}}
+source "{fix_path(str(optimize_file_name))!s}"
+set failed [vunit_optimize]
+if {{$failed}} {{quit -code 1}}
+quit -code 0
         """
                 batch_file_name = script_path / "batch_optimize.do"
                 write_file(str(batch_file_name), tcl)
@@ -449,7 +457,7 @@ def _optimize(self, config, script_path):
 
             with self._shared_state_lock:
                 self._optimized_designs[design_to_optimize]["optimized_design"] = optimized_design
-            self._optimized_designs[design_to_optimize]["optimization_completed"].set()
+                self._optimized_designs[design_to_optimize]["optimization_completed"].set()
 
         elif not optimized_design:
             LOGGER.debug("Waiting for %s to be optimized.", design_to_optimize)

vunit/sim_if/vsim_simulator_mixin.py

@@ -342,17 +342,16 @@ def _run_persistent(self, common_file_name, load_only=False):
         except Process.NonZeroExitCode:
             return False
 
-    def _optimize(self, config, script_path):  # pylint: disable=unused-argument
+    def _optimize_design(self, config):  # pylint: disable=unused-argument
         """
-        Return simulation target or False if optimization failed.
+        Return True if design shall be optimized.
         """
-        return "Optimize not supported"
+        return False
 
-    def _optimize_design(self, config):  # pylint: disable=unused-argument
+    def _optimize(self, config, script_path):  # pylint: disable=unused-argument
         """
-        Return if design shall be optimized.
+        Optimize design and return simulation target or False if optimization failed.
         """
-
         return False
 
     def simulate(self, output_path, test_suite_name, config, elaborate_only):