Fix test result message exception

This refactoring eliminates race conditions to
raise an exception when high concurrency.

The TaskManager doesn't have a queue. Replace
TaskManager-based test result message processing
with a threading approach.

Changes:
- Replace TaskManager with a dedicated background
  thread for processing test result messages
- Add message processing initialization to
  RootRunner.__init__()
- Move wait_for_test_result_messages() call from
  lisa_runner.py to runner.py for one time setup.

Author: squirrelsc

lisa/runner.py

@@ -15,6 +15,10 @@
 from lisa.messages import TestResultMessage, TestResultMessageBase, TestStatus
 from lisa.notifier import register_notifier
 from lisa.parameter_parser.runbook import RunbookBuilder
+from lisa.testsuite import (
+    start_test_result_message_processing,
+    wait_for_test_result_messages,
+)
 from lisa.util import BaseClassMixin, InitializableMixin, LisaException, constants
 from lisa.util.logger import create_file_handler, get_logger, remove_handler
 from lisa.util.parallel import Task, TaskManager, cancel, set_global_task_manager
@@ -239,6 +243,8 @@ def __init__(self, runbook_builder: RunbookBuilder) -> None:
         self._runner_count: int = 0
         self._idle_logged: bool = False
 
+        start_test_result_message_processing()
+
     async def start(self) -> None:
         await super().start()
 
@@ -472,6 +478,10 @@ def _cleanup(self) -> None:
         except Exception as e:
             self._log.warning(f"error on close runner: {e}")
 
+        # wait for all test result messages are processed and notified
+        self._log.debug("waiting for all test result messages to be processed")
+        wait_for_test_result_messages()
+
         try:
             transformer.run(self._runbook_builder, constants.TRANSFORMER_PHASE_CLEANUP)
         except Exception as e:

lisa/runners/lisa_runner.py

@@ -27,12 +27,7 @@
 from lisa.platform_ import PlatformMessage, load_platform
 from lisa.runner import BaseRunner
 from lisa.testselector import select_testcases
-from lisa.testsuite import (
-    TestCaseRequirement,
-    TestResult,
-    TestSuite,
-    wait_for_test_result_messages,
-)
+from lisa.testsuite import TestCaseRequirement, TestResult, TestSuite
 from lisa.util import (
     KernelPanicException,
     LisaException,
@@ -181,10 +176,6 @@ def fetch_task(self) -> Optional[Task[None]]:
         return None
 
     def close(self) -> None:
-        # wait for all test result messages are processed and notified
-        self._log.debug("waiting for all test result messages to be processed")
-        wait_for_test_result_messages()
-
         if hasattr(self, "environments") and self.environments:
             for environment in self.environments:
                 self._delete_environment_task(environment, [])

lisa/testsuite.py

@@ -5,10 +5,12 @@
 
 import copy
 import logging
+import threading
 import traceback
 from dataclasses import dataclass, field
-from functools import partial, wraps
+from functools import wraps
 from pathlib import Path
+from queue import Queue
 from time import sleep
 from typing import Any, Callable, Dict, List, Optional, Type, Union
 
@@ -42,7 +44,6 @@
     get_logger,
     remove_handler,
 )
-from lisa.util.parallel import Task, TaskManager
 from lisa.util.perf_timer import Timer, create_timer
 
 _all_suites: Dict[str, TestSuiteMetadata] = {}
@@ -371,7 +372,7 @@ def _send_result_message(self, stacktrace: Optional[str] = None) -> None:
         result_message.suite_full_name = self.runtime_data.metadata.suite.full_name
         result_message.stacktrace = stacktrace
 
-        _queue_test_message(result=self, result_message=result_message)
+        _queue_test_message(result_message=result_message)
 
     @retry(exceptions=FileExistsError, tries=30, delay=0.1)  # type: ignore
     def __create_case_log_path(self) -> Path:
@@ -893,8 +894,25 @@ def get_cases_metadata() -> Dict[str, TestCaseMetadata]:
     return _all_cases
 
 
+def start_test_result_message_processing() -> None:
+    global __message_thread
+    if not __message_thread:
+        # clear any left messages for UT
+        global __message_queue
+        __message_queue = Queue()
+
+        __message_thread = threading.Thread(target=_message_worker)
+        __message_thread.start()
+    else:
+        raise LisaException("message thread is already created and started")
+
+
 def wait_for_test_result_messages() -> None:
-    __test_message_tasks.wait_for_all_workers()
+    __message_queue.put(None)
+    global __message_thread
+    assert __message_thread is not None
+    __message_thread.join()
+    __message_thread = None
 
 
 def _add_suite_metadata(metadata: TestSuiteMetadata) -> None:
@@ -954,23 +972,27 @@ def _add_case_to_suite(
     test_suite.cases.append(test_case)
 
 
-# process test results message in an order, so the max_workers is 1
-__test_message_tasks: TaskManager[None] = TaskManager(max_workers=1)
+def _message_worker() -> None:
+    while True:
+        message = __message_queue.get()
+        if message is None:
+            break
+        _send_result_message(message)
+
+
+# process test results message in an order and in background
+__message_thread: Optional[threading.Thread] = None
+__message_queue: Queue[Optional[TestResultMessage]] = Queue()
 
 
 def _send_result_message(result_message: TestResultMessage) -> None:
     plugin_manager.hook.update_test_result_message(message=result_message)
     notifier.notify(message=result_message)
 
 
-def _queue_test_message(result: TestResult, result_message: TestResultMessage) -> None:
-    __test_message_tasks.submit_task(
-        Task(
-            task_id=0,
-            task=partial(_send_result_message, result_message),
-            parent_logger=result.log,
-        )
-    )
+def _queue_test_message(result_message: TestResultMessage) -> None:
+    assert result_message is not None
+    __message_queue.put(result_message)
 
 
 plugin_manager.add_hookspecs(TestResult)

selftests/runners/test_lisa_runner.py

@@ -13,7 +13,12 @@
 from lisa.parameter_parser.runbook import RunbookBuilder
 from lisa.runner import RunnerResult
 from lisa.runners.lisa_runner import LisaRunner
-from lisa.testsuite import TestResult, simple_requirement, wait_for_test_result_messages
+from lisa.testsuite import (
+    TestResult,
+    simple_requirement,
+    start_test_result_message_processing,
+    wait_for_test_result_messages,
+)
 from lisa.util.parallel import Task
 from selftests import test_platform, test_testsuite
 from selftests.test_environment import generate_runbook as generate_env_runbook
@@ -59,9 +64,6 @@ def setUp(self) -> None:
         lisa.environment._global_environment_id = 0
 
     def tearDown(self) -> None:
-        # wait for all test result messages to be processed in the current test case.
-        wait_for_test_result_messages()
-
         test_testsuite.cleanup_cases_metadata()  # Necessary side effects!
 
     def test_merge_req_create_on_new(self) -> None:
@@ -742,13 +744,20 @@ def _run_all_tests(self, runner: LisaRunner) -> List[TestResultMessage]:
         results_collector = RunnerResult(schema.Notifier())
         register_notifier(results_collector)
 
-        runner.initialize()
+        start_test_result_message_processing()
+
+        try:
+            runner.initialize()
 
-        while not runner.is_done:
-            task = runner.fetch_task()
-            if task:
-                if isinstance(task, Task):
-                    task()
+            while not runner.is_done:
+                task = runner.fetch_task()
+                if task:
+                    if isinstance(task, Task):
+                        task()
+        finally:
+            # wait for all test result messages to be processed in the current
+            # test case.
+            wait_for_test_result_messages()
 
         runner.close()
         _notifiers.clear()