WIP: Dynamic scheduling

Author: LarsAsplund

vunit/test/list.py

@@ -64,9 +64,6 @@ def __iter__(self):
     def __len__(self):
         return len(self._test_suites)
 
-    def __getitem__(self, idx):
-        return self._test_suites[idx]
-
 
 class TestSuiteWrapper(object):
     """

vunit/test/runner.py

@@ -45,6 +45,8 @@ def __init__(  # pylint: disable=too-many-arguments
         fail_fast=False,
         dont_catch_exceptions=False,
         no_color=False,
+        latest_dependency_updates={},
+        test_suite_history={},
     ):
         self._lock = threading.Lock()
         self._fail_fast = fail_fast
@@ -64,6 +66,8 @@ def __init__(  # pylint: disable=too-many-arguments
         self._stderr = sys.stderr
         self._dont_catch_exceptions = dont_catch_exceptions
         self._no_color = no_color
+        self._latest_dependency_updates = latest_dependency_updates
+        self._test_suite_history = test_suite_history
 
         ostools.PROGRAM_STATUS.reset()
 
@@ -75,11 +79,69 @@ def _is_verbose(self):
     def _is_quiet(self):
         return self._verbosity == self.VERBOSITY_QUIET
 
+    def _create_test_suite_sets(self, test_suites):
+        # Priority is set to fail fast:
+        # 1. Tests that failed before and for which no updates have been made. Expected to fail again. Shortest   #    test first (STF).
+        # 2. Tests that failed before but updates have been made that can change that. STF.
+        # 3. Tests without a history. New tests are more likely to fail and should be run early. Without
+        #    any execution time order, there is no further sorting within this group
+        # 4. Tests that passed before but depends on updates. There is a risk that we've introduced new bugs. STF.
+        # 5. Tests that passed before and for which there are no updates. They are expected to pass again.
+        #    Longest test first to optimize completion time when running with multiple threads
+        #
+        # Each test is given a priority number where the integer part is according to the priority above and
+        # the decimal part seperates the tests within the group based on execution time.
+
+        test_suite_sets = []
+        for idx in range(5):
+            test_suite_sets.append(dict(test_suites=[], total_exec_time=None if idx == 2 else 0))
+
+        for test_suite in test_suites:
+            test_suite_data = self._test_suite_history.get(test_suite.name, False)
+            if not test_suite_data:
+                test_suite_sets[2]["test_suites"].append(dict(test_suite=test_suite, exec_time=None))
+            else:
+                highest_priority = None
+                exec_time = 0
+                for test_name in test_suite.test_names:
+                    test_data = test_suite_data.get(test_name, False)
+                    exec_time += test_data["total_time"]
+                    if not test_data:
+                        priority = 3
+                    else:
+                        updated_dependency = (
+                            self._latest_dependency_updates[test_suite.file_name] > test_data["start_time"]
+                        )
+                        if test_data["failed"]:
+                            if not updated_dependency:
+                                priority = 1
+                            else:
+                                priority = 2
+                        elif test_data["skipped"]:
+                            priority = 3
+                        elif updated_dependency:
+                            priority = 4
+                        else:
+                            priority = 5
+
+                    highest_priority = priority if not highest_priority else min(highest_priority, priority)
+
+                test_suite_sets[highest_priority - 1]["test_suites"].append(
+                    dict(test_suite=test_suite, exec_time=exec_time)
+                )
+                test_suite_sets[highest_priority - 1]["total_exec_time"] += exec_time
+
+        for idx, test_suite_set in enumerate(test_suite_sets):
+            if idx == 2:
+                continue
+            test_suite_set["test_suites"].sort(key=lambda item: item["exec_time"])
+
+        return test_suite_sets
+
     def run(self, test_suites):
         """
         Run a list of test suites
         """
-
         if not Path(self._output_path).exists():
             os.makedirs(self._output_path)
 
@@ -98,7 +160,9 @@ def run(self, test_suites):
 
         self._report.set_expected_num_tests(num_tests)
 
-        scheduler = TestScheduler(test_suites)
+        test_suite_sets = self._create_test_suite_sets(test_suites)
+
+        scheduler = TestScheduler(test_suite_sets, self._num_threads)
 
         threads = []
 
@@ -110,16 +174,16 @@ def run(self, test_suites):
             sys.stderr = ThreadLocalOutput(self._local, self._stdout)
 
             # Start P-1 worker threads
-            for _ in range(self._num_threads - 1):
+            for thread_id in range(1, self._num_threads):
                 new_thread = threading.Thread(
                     target=self._run_thread,
-                    args=(write_stdout, scheduler, num_tests, False),
+                    args=(write_stdout, scheduler, num_tests, False, thread_id),
                 )
                 threads.append(new_thread)
                 new_thread.start()
 
             # Run one worker in main thread such that P=1 is not multithreaded
-            self._run_thread(write_stdout, scheduler, num_tests, True)
+            self._run_thread(write_stdout, scheduler, num_tests, True, 0)
 
             scheduler.wait_for_finish()
 
@@ -136,7 +200,7 @@ def run(self, test_suites):
             sys.stderr = self._stderr
             LOGGER.debug("TestRunner: Leaving")
 
-    def _run_thread(self, write_stdout, scheduler, num_tests, is_main):
+    def _run_thread(self, write_stdout, scheduler, num_tests, is_main, thread_id):
         """
         Run worker thread
         """
@@ -145,7 +209,7 @@ def _run_thread(self, write_stdout, scheduler, num_tests, is_main):
         while True:
             test_suite = None
             try:
-                test_suite = scheduler.next()
+                test_suite = scheduler.next(thread_id)
 
                 output_path = self._get_output_path(test_suite.name)
                 output_file_name = str(Path(output_path) / "output.txt")
@@ -171,7 +235,7 @@ def _run_thread(self, write_stdout, scheduler, num_tests, is_main):
 
             finally:
                 if test_suite is not None:
-                    scheduler.test_done()
+                    scheduler.test_done(thread_id)
 
     def _get_output_path(self, test_suite_name):
         """
@@ -398,40 +462,86 @@ def flush(self):
             self._stdout.flush()
 
 
+class TestIterator:
+    def __init__(self, test_suite_sets, n_threads):
+        self._test_suite_sets = test_suite_sets
+        self._set_idx = 0
+        self._n_threads = n_threads
+        self._thread_status = [dict(start_time=None, exec_time=None) for _ in range(n_threads)]
+
+        # Estimate remaing test time
+        self._exec_time_for_remaining_tests = 0
+        for test_suite_set in test_suite_sets:
+            if total_exec_time := test_suite_set["total_exec_time"]:
+                self._exec_time_for_remaining_tests += total_exec_time
+
+    def thread_done(self, thread_id):
+        self._thread_status[thread_id]["start_time"] = None
+        self._thread_status[thread_id]["exec_time"] = None
+
+    def next(self, thread_id):
+        # Get the first non-empty test suite set or raise StopIteration
+        test_suite_set = next(
+            (test_suite_set for test_suite_set in self._test_suite_sets if test_suite_set["test_suites"])
+        )
+
+        # Estimate time to completion if we can achieve perfect load-balancing over threads
+        remaining_exec_time_for_ongoing_tests = 0
+        for status in self._thread_status:
+            if start_time := status["start_time"]:
+                remaining_exec_time_for_ongoing_tests += min(0, start_time + status["exec_time"] - time.time())
+
+        time_to_completion = (
+            remaining_exec_time_for_ongoing_tests + self._exec_time_for_remaining_tests
+        ) / self._n_threads
+
+        # Select the longest test if delaying it would risk exceeding the ideal time to completion
+        test_suites = test_suite_set["test_suites"]
+        longest_test_exec_time = test_suites[-1]["exec_time"]
+        if longest_test_exec_time is not None and longest_test_exec_time >= 0.9 * time_to_completion:
+            test_suite_data = test_suites.pop(-1)
+        else:
+            test_suite_data = test_suites.pop(0)
+
+        if exec_time := test_suite_data["exec_time"]:
+            self._exec_time_for_remaining_tests -= exec_time
+
+        self._thread_status[thread_id]["exec_time"] = exec_time
+        self._thread_status[thread_id]["start_time"] = time.time()
+
+        return test_suite_data["test_suite"]
+
+
 class TestScheduler(object):
     """
     Schedule tests to different treads
     """
 
-    def __init__(self, tests):
+    def __init__(self, test_suite_sets, n_threads):
         self._lock = threading.Lock()
-        self._tests = tests
-        self._idx = 0
+        self._num_tests = sum(len(test_suite_set["test_suites"]) for test_suite_set in test_suite_sets)
+        self._test_iterator = TestIterator(test_suite_sets, n_threads)
         self._num_done = 0
 
-    def next(self):
+    def next(self, thread_id):
         """
         Return the next test
         """
         ostools.PROGRAM_STATUS.check_for_shutdown()
         with self._lock:  # pylint: disable=not-context-manager
-            if self._idx < len(self._tests):
-                idx = self._idx
-                self._idx += 1
-                return self._tests[idx]
-
-            raise StopIteration
+            return self._test_iterator.next(thread_id)
 
-    def test_done(self):
+    def test_done(self, thread_id):
         """
         Signal that a test has been done
         """
         with self._lock:  # pylint: disable=not-context-manager
+            self._test_iterator.thread_done(thread_id)
             self._num_done += 1
 
     def is_finished(self):
         with self._lock:  # pylint: disable=not-context-manager
-            return self._num_done >= len(self._tests)
+            return self._num_done >= self._num_tests
 
     def wait_for_finish(self):
         """