standardise and fix smoke_test_timeout

Signed-off-by: Jack Luar <[email protected]>

Author: luarss

tools/AutoTuner/src/autotuner/distributed.py

@@ -515,18 +515,19 @@ def parse_arguments():
         args.timeout_per_trial = round(args.timeout_per_trial * 3600)
     if args.timeout is not None:
         args.timeout = round(args.timeout * 3600)
+    args.timeout = set_timeout(args.timeout, args.timeout_per_trial)
 
     # Calculate timeout based on cpu_budget
     if args.cpu_budget != -1:
         args.timeout = round(args.cpu_budget / os.cpu_count() * 3600)
         args.timeout_per_trial = round(
             args.cpu_budget / (args.jobs * args.resources_per_trial) * 3600
         )
-        overall_timeout = min(args.timeout, args.timeout_per_trial)
+        args.timeotu = set_timeout(args.timeout, args.timeout_per_trial)
         if args.mode == "tune":
-            template = calculate_expected_numbers(overall_timeout, args.samples)
+            template = calculate_expected_numbers(args.timeout, args.samples)
         else:
-            template = calculate_expected_numbers(overall_timeout, 1)
+            template = calculate_expected_numbers(args.timeout, 1)
         print(template)
         if not args.yes:
             print(
@@ -620,6 +621,17 @@ def set_training_class(function):
     return None
 
 
+def set_timeout(timeout, timeout_per_trial):
+    """
+    Set timeout for experiment.
+    """
+    return (
+        min(timeout, timeout_per_trial)
+        if (timeout and timeout_per_trial)
+        else (timeout or timeout_per_trial)
+    )
+
+
 @ray.remote
 def save_best(results):
     """

tools/AutoTuner/test/smoke_test_timeout.py

@@ -3,8 +3,6 @@
 import os
 
 cur_dir = os.path.dirname(os.path.abspath(__file__))
-src_dir = os.path.join(cur_dir, "../src/autotuner")
-os.chdir(src_dir)
 
 
 class BaseTimeoutSmokeTest(unittest.TestCase):
@@ -20,9 +18,8 @@ def setUp(self):
 
         # 0.001 hour translates to 3.6 seconds, which will definitely cause failure.
         timeout_flags = ["--timeout 0.001", "--timeout_per_trial 0.001"]
-        self.timeout_limit = 60  # 60 second upper limit (Ray needs time to shutdown)
         self.commands = [
-            "python3 distributed.py"
+            "python3 -m autotuner.distributed"
             f" --design {self.design}"
             f" --platform {self.platform}"
             f" --experiment {self.experiment}-{idx}"
@@ -34,49 +31,28 @@ def setUp(self):
         ]
 
     def test_timeout(self):
-        raise NotImplementedError(
-            "This method needs to be implemented in the derivative classes."
-        )
+        if not (self.platform and self.design):
+            raise unittest.SkipTest("Platform and design have to be defined")
+        for command in self.commands:
+            out = subprocess.run(command, shell=True, check=False)
+            failed = out.returncode != 0
+            self.assertTrue(failed)
 
 
 class asap7TimeoutSmokeTest(BaseTimeoutSmokeTest):
     platform = "asap7"
     design = "gcd"
 
-    def test_timeout(self):
-        for command in self.commands:
-            out = subprocess.run(
-                command, shell=True, check=False, timeout=self.timeout_limit
-            )
-            failed = out.returncode == 1
-            self.assertTrue(failed)
-
 
 class sky130hdTimeoutSmokeTest(BaseTimeoutSmokeTest):
     platform = "sky130hd"
     design = "gcd"
 
-    def test_timeout(self):
-        for command in self.commands:
-            out = subprocess.run(
-                command, shell=True, check=False, timeout=self.timeout_limit
-            )
-            failed = out.returncode == 1
-            self.assertTrue(failed)
-
 
 class ihpsg13g2TimeoutSmokeTest(BaseTimeoutSmokeTest):
     platform = "ihp-sg13g2"
     design = "gcd"
 
-    def test_timeout(self):
-        for command in self.commands:
-            out = subprocess.run(
-                command, shell=True, check=False, timeout=self.timeout_limit
-            )
-            failed = out.returncode == 1
-            self.assertTrue(failed)
-
 
 if __name__ == "__main__":
     unittest.main()