Enabled code coverage in AutoTuner smoke tests

lint fixes

Signed-off-by: Jeff Ng <[email protected]>

Author: jeffng-or

tools/AutoTuner/src/autotuner/distributed.py

@@ -216,11 +216,13 @@ def evaluate(self, metrics):
         error = "ERR" in metrics.values() or "ERR" in reference.values()
         not_found = "N/A" in metrics.values() or "N/A" in reference.values()
         if error or not_found:
-            return ERROR_METRIC
+            return (ERROR_METRIC, "-", "-")
         ppa = self.get_ppa(metrics)
         gamma = ppa / 10
         score = ppa * (self.step_ / 100) ** (-1) + (gamma * metrics["num_drc"])
-        return score
+        effective_clk_period = metrics["clk_period"] - metrics["worst_slack"]
+        num_drc = metrics["num_drc"]
+        return (score, effective_clk_period, num_drc)
 
 
 def parse_arguments():
@@ -464,32 +466,34 @@ def parse_arguments():
     return args
 
 
-def set_algorithm(experiment_name, config):
+def set_algorithm(
+    algorithm_name, experiment_name, best_params, seed, perturbation, jobs, config
+):
     """
     Configure search algorithm.
     """
     # Pre-set seed if user sets seed to 0
-    if args.seed == 0:
+    if seed == 0:
         print(
             "Warning: you have chosen not to set a seed. Do you wish to continue? (y/n)"
         )
         if input().lower() != "y":
             sys.exit(0)
-        args.seed = None
+        seed = None
     else:
-        torch.manual_seed(args.seed)
-        np.random.seed(args.seed)
-        random.seed(args.seed)
+        torch.manual_seed(seed)
+        np.random.seed(seed)
+        random.seed(seed)
 
-    if args.algorithm == "hyperopt":
+    if algorithm_name == "hyperopt":
         algorithm = HyperOptSearch(
             points_to_evaluate=best_params,
-            random_state_seed=args.seed,
+            random_state_seed=seed,
         )
-    elif args.algorithm == "ax":
+    elif algorithm_name == "ax":
         ax_client = AxClient(
             enforce_sequential_optimization=False,
-            random_seed=args.seed,
+            random_seed=seed,
         )
         AxClientMetric = namedtuple("AxClientMetric", "minimize")
         ax_client.create_experiment(
@@ -498,25 +502,25 @@ def set_algorithm(experiment_name, config):
             objectives={METRIC: AxClientMetric(minimize=True)},
         )
         algorithm = AxSearch(ax_client=ax_client, points_to_evaluate=best_params)
-    elif args.algorithm == "optuna":
-        algorithm = OptunaSearch(points_to_evaluate=best_params, seed=args.seed)
-    elif args.algorithm == "pbt":
-        print("Warning: PBT does not support seed values. args.seed will be ignored.")
+    elif algorithm_name == "optuna":
+        algorithm = OptunaSearch(points_to_evaluate=best_params, seed=seed)
+    elif algorithm_name == "pbt":
+        print("Warning: PBT does not support seed values. seed will be ignored.")
         algorithm = PopulationBasedTraining(
             time_attr="training_iteration",
-            perturbation_interval=args.perturbation,
+            perturbation_interval=perturbation,
             hyperparam_mutations=config,
             synch=True,
         )
-    elif args.algorithm == "random":
+    elif algorithm_name == "random":
         algorithm = BasicVariantGenerator(
-            max_concurrent=args.jobs,
-            random_state=args.seed,
+            max_concurrent=jobs,
+            random_state=seed,
         )
 
     # A wrapper algorithm for limiting the number of concurrent trials.
-    if args.algorithm not in ["random", "pbt"]:
-        algorithm = ConcurrencyLimiter(algorithm, max_concurrent=args.jobs)
+    if algorithm_name not in ["random", "pbt"]:
+        algorithm = ConcurrencyLimiter(algorithm, max_concurrent=jobs)
 
     return algorithm
 
@@ -607,7 +611,15 @@ def main():
 
     if args.mode == "tune":
         best_params = set_best_params(args.platform, args.design)
-        search_algo = set_algorithm(args.experiment, config_dict)
+        search_algo = set_algorithm(
+            args.algorithm,
+            args.experiment,
+            best_params,
+            args.seed,
+            args.perturbation,
+            args.jobs,
+            config_dict,
+        )
         TrainClass = set_training_class(args.eval)
         # PPAImprov requires a reference file to compute training scores.
         if args.eval == "ppa-improv":

tools/AutoTuner/test/autotuner_test_utils.py

@@ -0,0 +1,25 @@
+#!/usr/bin/env python3
+
+import os
+
+
+class AutoTunerTestUtils:
+    @staticmethod
+    def get_exec_cmd():
+        """
+        Returns the execution command based on whether this is a coverage run or
+        not.
+
+        Note that you need to run coverage combine after the runs complete to
+        get the coverage of the parent plus the child invocations
+        """
+
+        if "COVERAGE_RUN" in os.environ:
+            exec = "coverage run --parallel-mode --omit=*/site-packages/*,*/dist-packages/*"
+        else:  # pragma: no cover
+            exec = "python3"
+        return exec + " -m autotuner.distributed"
+
+
+if __name__ == "__main__":  # pragma: no cover
+    print(AutoTunerTestUtils.get_exec_cmd())

tools/AutoTuner/test/ref_file_check.py

@@ -1,38 +1,70 @@
 import unittest
 import subprocess
 import os
-
-cur_dir = os.path.dirname(os.path.abspath(__file__))
-src_dir = os.path.join(cur_dir, "../src/autotuner")
-orfs_dir = os.path.join(cur_dir, "../../../flow")
-os.chdir(src_dir)
+from autotuner_test_utils import AutoTunerTestUtils
 
 
 class RefFileCheck(unittest.TestCase):
-    # only test 1 platform/design.
-    platform = "asap7"
-    design = "gcd"
+    """
+    Tests situations where a referenced file (SDC or FastRoute) is not
+    defined in the AutoTuner config
+    """
 
     def setUp(self):
-        configs = [
-            "../../test/files/no_sdc_ref.json",
-            "../../test/files/no_fr_ref.json",
-        ]
-        self.commands = [
-            f"python3 distributed.py"
-            f" --design {self.design}"
-            f" --platform {self.platform}"
-            f" --config {c}"
-            f" tune --samples 1"
-            for c in configs
-        ]
-
-    # Make this a test case
-    def test_files(self):
-        for c in self.commands:
-            out = subprocess.run(c, shell=True)
-            failed = out.returncode != 0
-            self.assertTrue(failed)
+        self._cur_dir = os.path.dirname(os.path.abspath(__file__))
+        src_dir = os.path.join(self._cur_dir, "../src")
+        os.chdir(src_dir)
+
+        self._exec = AutoTunerTestUtils.get_exec_cmd()
+
+    def _execute_autotuner(self, platform, design, config_file, error_code=None):
+        full_path = os.path.abspath(os.path.join(self._cur_dir, config_file))
+
+        cmd = f"{self._exec} --design {design} --platform {platform} --config {full_path} tune --samples 1"
+
+        out = subprocess.run(cmd, shell=True, text=True, capture_output=True)
+        failed = out.returncode != 0
+        self.assertTrue(failed, f"AT run with {config_file} passed")
+        if error_code:
+            self.assertTrue(
+                error_code in out.stdout,
+                f"Didn't find error code {error_code} in output '{out.stdout}'",
+            )
+
+    def test_asap_gcd_no_sdc(self):
+        """
+        Tests when SDC file is not defined, which is an error for all
+        platforms and designs
+        """
+
+        platform = "asap7"
+        design = "gcd"
+        config_file = "files/no_sdc_ref.json"
+        error_code = "[ERROR TUN-0020] No SDC reference"
+        self._execute_autotuner(platform, design, config_file, error_code)
+
+    def test_asap_gcd_no_fr(self):
+        """
+        Tests when FastRoute file is not defined, which is not an error for
+        asap platform. This test fails anyway
+        """
+
+        platform = "asap7"
+        design = "gcd"
+        config_file = "files/no_fr_ref.json"
+        self._execute_autotuner(platform, design, config_file)
+
+    def test_ihp_gcd_no_fr(self):
+        """
+        Tests when FastRoute file is not defined, which is not an error for
+        any non-asap7 platform.
+        """
+
+        platform = "ihp-sg13g2"
+        design = "gcd"
+        config_file = "files/no_fr_ref.json"
+        error_code = "[ERROR TUN-0021] No FastRoute Tcl"
+        self._execute_autotuner(platform, design, config_file, error_code)
 
 
 if __name__ == "__main__":

tools/AutoTuner/test/resume_check.py

@@ -2,11 +2,12 @@
 import subprocess
 import os
 import time
+from autotuner_test_utils import AutoTunerTestUtils
 
 from contextlib import contextmanager
 
 cur_dir = os.path.dirname(os.path.abspath(__file__))
-src_dir = os.path.join(cur_dir, "../src/autotuner")
+src_dir = os.path.join(cur_dir, "../src")
 orfs_dir = os.path.join(cur_dir, "../../../flow")
 os.chdir(src_dir)
 
@@ -45,8 +46,9 @@ def setUp(self):
         # Cast to 1 decimal place
         res_per_trial = float("{:.1f}".format(self.num_cpus / self.samples))
         options = ["", "--resume"]
+        self.exec = AutoTunerTestUtils.get_exec_cmd()
         self.commands = [
-            f"python3 distributed.py"
+            f"{self.exec}"
             f" --design {self.design}"
             f" --platform {self.platform}"
             f" --config {self.config}"

tools/AutoTuner/test/smoke_test_algo_eval.py

@@ -1,6 +1,7 @@
 import unittest
 import subprocess
 import os
+from autotuner_test_utils import AutoTunerTestUtils
 
 cur_dir = os.path.dirname(os.path.abspath(__file__))
 orfs_dir = os.path.join(cur_dir, "../../../flow")
@@ -19,8 +20,9 @@ def setUp(self):
         _algo = ["hyperopt", "ax", "optuna", "pbt", "random"]
         _eval = ["default", "ppa-improv"]
         self.matrix = [(a, e) for a in _algo for e in _eval]
+        self.exec = AutoTunerTestUtils.get_exec_cmd()
         self.commands = [
-            f"python3 -m autotuner.distributed"
+            f"{self.exec}"
             f" --design {self.design}"
             f" --platform {self.platform}"
             f" --experiment {self.experiment}"

tools/AutoTuner/test/smoke_test_sample_iteration.py

@@ -1,6 +1,7 @@
 import unittest
 import subprocess
 import os
+from autotuner_test_utils import AutoTunerTestUtils
 
 cur_dir = os.path.dirname(os.path.abspath(__file__))
 
@@ -16,8 +17,9 @@ def setUp(self):
         )
         self.experiment = f"smoke-test-sample-iteration-{self.platform}"
         self.matrix = [(5, 1), (1, 5), (2, 2), (1, 1)]
+        self.exec = AutoTunerTestUtils.get_exec_cmd()
         self.commands = [
-            f"python3 -m autotuner.distributed"
+            f"{self.exec}"
             f" --design {self.design}"
             f" --platform {self.platform}"
             f" --experiment {self.experiment}"

tools/AutoTuner/test/smoke_test_sweep.py

@@ -2,6 +2,7 @@
 import subprocess
 import os
 import json
+from autotuner_test_utils import AutoTunerTestUtils
 
 cur_dir = os.path.dirname(os.path.abspath(__file__))
 
@@ -30,8 +31,9 @@ def setUp(self):
         core = os.cpu_count()
         self.jobs = 4 if core >= 4 else core
         self.experiment = f"smoke-test-sweep-{self.platform}"
+        self.exec = AutoTunerTestUtils.get_exec_cmd()
         self.command = (
-            "python3 -m autotuner.distributed"
+            f"{self.exec}"
             f" --design {self.design}"
             f" --platform {self.platform}"
             f" --experiment {self.experiment}"

tools/AutoTuner/test/smoke_test_tune.py

@@ -1,6 +1,7 @@
 import unittest
 import subprocess
 import os
+from autotuner_test_utils import AutoTunerTestUtils
 
 cur_dir = os.path.dirname(os.path.abspath(__file__))
 
@@ -15,8 +16,9 @@ def setUp(self):
             f"../../../flow/designs/{self.platform}/{self.design}/autotuner.json",
         )
         self.experiment = f"smoke-test-tune-{self.platform}"
+        self.exec = AutoTunerTestUtils.get_exec_cmd()
         self.command = (
-            "python3 -m autotuner.distributed"
+            f"{self.exec}"
             f" --design {self.design}"
             f" --platform {self.platform}"
             f" --experiment {self.experiment}"