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()
 
@@ -79,7 +83,6 @@ 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 +101,9 @@ def run(self, test_suites):
 
         self._report.set_expected_num_tests(num_tests)
 
-        scheduler = TestScheduler(test_suites)
+        scheduler = TestScheduler(
+            test_suites, self._num_threads, self._latest_dependency_updates, self._test_suite_history
+        )
 
         threads = []
 
@@ -110,16 +115,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 +141,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 +150,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 +176,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):
         """
@@ -403,35 +408,163 @@ class TestScheduler(object):
     Schedule tests to different treads
     """
 
-    def __init__(self, tests):
+    def _create_test_suite_sets(self, test_suites):
+        """
+        Create static priority based on test result and file change history.
+        """
+
+        # Test suites are divided into sets which are executed in order. The internal order within a set
+        # is decided dynamically at run-time.
+        #
+        # The 5 sets contains:
+        #
+        # 0. Test suites that failed before and for which no updates have been made. Expected to fail again.
+        # 1. Test suites that failed before but updates have been made that can change that.
+        # 2. Test suites without a history. New tests are more likely to fail and should be executed early.
+        # 3. Test suites that passed before but depends on updates. There is a risk that we've introduced new bugs.
+        # 4. Test suites that passed before and for which there are no updates. They are expected to pass again.
+        #
+        # Within sets, test suites are sorted in execution time order starting with the fastest test.
+        # This is in preparation for the dynamic scheduling that decides the final order within a set.
+        # The exception is set 2 which has no history of execution time.
+
+        # A test suite set keeps the sorted test suite list as well as the total estimated execution time (if available)
+        # for the test suites within the list.
+        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:
+                # Test suites with multiple tests are placed in the set where the highest priority test belongs
+                highest_priority_set = 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:
+                        set_idx = 2
+                    else:
+                        updated_dependency = (
+                            self._latest_dependency_updates[test_suite.file_name] > test_data["start_time"]
+                        )
+                        if test_data["failed"]:
+                            if not updated_dependency:
+                                set_idx = 0
+                            else:
+                                set_idx = 1
+                        elif test_data["skipped"]:
+                            set_idx = 2
+                        elif updated_dependency:
+                            set_idx = 3
+                        else:
+                            set_idx = 4
+
+                    highest_priority_set = set_idx if not highest_priority_set else min(highest_priority_set, set_idx)
+
+                test_suite_sets[highest_priority_set]["test_suites"].append(
+                    dict(test_suite=test_suite, exec_time=exec_time)
+                )
+                test_suite_sets[highest_priority_set]["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 __init__(self, test_suites, num_threads, latest_dependency_updates, test_suite_history):
+        self._num_threads = num_threads
+        self._latest_dependency_updates = latest_dependency_updates
+        self._test_suite_history = test_suite_history
+        self._test_suite_sets = self._create_test_suite_sets(test_suites)
         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 self._test_suite_sets)
         self._num_done = 0
+        self._thread_status = [dict(start_time=None, exec_time=None) for _ in range(num_threads)]
 
-    def next(self):
+        # Estimate remaing test time
+        self._exec_time_for_remaining_tests = 0
+        for test_suite_set in self._test_suite_sets:
+            if total_exec_time := test_suite_set["total_exec_time"]:
+                self._exec_time_for_remaining_tests += total_exec_time
+
+    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]
+            # 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 of threads
+            remaining_exec_time_for_ongoing_tests = 0
+            for idx, status in enumerate(self._thread_status):
+                if status["start_time"] and status["exec_time"]:
+                    remaining_exec_time_for_ongoing_tests += max(
+                        0, status["start_time"] + status["exec_time"] - time.time()
+                    )
+
+            time_to_completion = (
+                remaining_exec_time_for_ongoing_tests + self._exec_time_for_remaining_tests
+            ) / self._num_threads
+
+            # Estimate when next thread will complete
+            time_to_next_thread_completion = None
+            for idx, status in enumerate(self._thread_status):
+                if idx != thread_id:
+                    if not status["start_time"]:
+                        time_to_next_thread_completion = 0
+
+                    elif status["start_time"] and status["exec_time"]:
+                        time_to_thread_completion = max(0, status["start_time"] + status["exec_time"] - time.time())
+                        time_to_next_thread_completion = (
+                            time_to_thread_completion
+                            if time_to_next_thread_completion is None
+                            else min(time_to_next_thread_completion, time_to_thread_completion)
+                        )
+
+            # 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 (time_to_completion <= longest_test_exec_time):
+                test_suite_data = test_suites.pop(-1)
+            elif (
+                (longest_test_exec_time is not None)
+                and (time_to_next_thread_completion is not None)
+                and (time_to_completion - time_to_next_thread_completion <= longest_test_exec_time)
+            ):
+                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()
 
-            raise StopIteration
+            return test_suite_data["test_suite"]
 
-    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._thread_status[thread_id]["start_time"] = None
+            self._thread_status[thread_id]["exec_time"] = None
             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):
         """