[CONFORMANCE] Solve the problem to generate filters (#16713)

* [CONFORMANCE] Solve the problem to generate filters

* trigger linux build

* Add handling of not run tests

* Remove extra

src/tests/ie_test_utils/functional_test_utils/layer_tests_summary/run_conformance.py

@@ -50,6 +50,7 @@ def parse_arguments():
     ov_config_path_helper = "Specify path to file contains plugin config"
     dump_conformance_help = "Set '1' if you want to create Conformance IRs from custom/downloaded models. In other cases, set 0. The default value is '1'"
     shape_mode_help = "Specify shape mode for conformance. Default value is ``. Possible values: `static`, `dynamic`, ``"
+    parallel_help = "Parallel over HW devices. For example run tests over GPU.0, GPU.1 and etc"
 
     parser.add_argument("-m", "--models_path", help=models_path_help, type=str, required=False, default=NO_MODEL_CONSTANT)
     parser.add_argument("-d", "--device", help= device_help, type=str, required=False, default="CPU")
@@ -61,6 +62,7 @@ def parse_arguments():
     parser.add_argument("-c", "--ov_config_path", help=ov_config_path_helper, type=str, required=False, default="")
     parser.add_argument("-s", "--dump_conformance", help=dump_conformance_help, type=int, required=False, default=0)
     parser.add_argument("-sm", "--shape_mode", help=shape_mode_help, type=str, required=False, default="")
+    parser.add_argument("-p", "--parallel_devices", help=parallel_help, type=int, required=False, default=0)
 
     return parser.parse_args()
 
@@ -167,7 +169,7 @@ class Conformance:
                              f"--report_unique_name", f'--output_folder="{parallel_report_dir}"',
                              f'--gtest_filter={self._gtest_filter}', f'--config_path="{self._ov_config_path}"',
                              f'--shape_mode={self._shape_mode}']
-        conformance = TestParallelRunner(f"{conformance_path}", command_line_args, self._workers, logs_dir, "")
+        conformance = TestParallelRunner(f"{conformance_path}", command_line_args, self._workers, logs_dir, "", True)
         conformance.run()
         conformance.postprocess_logs()
 

src/tests/ie_test_utils/functional_test_utils/layer_tests_summary/run_parallel.py

@@ -5,10 +5,11 @@ from utils.conformance_utils import get_logger, progressbar
 from utils import constants
 from utils import file_utils
 from argparse import ArgumentParser
-from subprocess import Popen, STDOUT, TimeoutExpired, run
+from subprocess import Popen, STDOUT, TimeoutExpired, run, call
 from hashlib import sha256
 from pathlib import Path
 from shutil import rmtree
+from signal import SIGKILL
 
 import os
 import sys
@@ -48,7 +49,7 @@ def parse_arguments():
     parser.add_argument("-e", "--exec_file", help=exec_file_path_help, type=str, required=True)
     parser.add_argument("-c", "--cache_path", help=cache_path_help, type=str, required=False, default="")
     parser.add_argument("-j", "--workers", help=worker_num_help, type=int, required=False, default=(os.cpu_count() - 1) if os.cpu_count() > 2 else 1)
-    parser.add_argument("-p", "--parallel_devices", help=parallel_help, type=int, required=False, default=1)
+    parser.add_argument("-p", "--parallel_devices", help=parallel_help, type=int, required=False, default=0)
     parser.add_argument("-w", "--working_dir", help=working_dir_num_help, type=str, required=False, default=".")
     parser.add_argument("-t", "--process_timeout", help=process_timeout_help, type=int, required=False, default=DEFAULT_PROCESS_TIMEOUT)
     return parser.parse_args()
@@ -133,6 +134,18 @@ class TaskManager:
                 log_file.close()
             # logger.info(f"{self._idx}/{len(self._command_list)} is started")
             self._idx += 1
+
+    @staticmethod
+    def kill_process_tree(pid):
+        try:
+            if not constants.IS_WIN:
+                os.killpg(pid, SIGKILL)
+            else:
+                call(['taskkill', '/F', '/T', '/PID', str(pid)])
+        except OSError as err:
+            # logger.warning(f"Impossible to kill process {pid} with error: {err}")
+            pass
+
     
     def __find_free_process(self):
         while True:
@@ -140,7 +153,9 @@ class TaskManager:
                 try:
                     if float((datetime.datetime.now() - self._timers[pid]).total_seconds()) > self.process_timeout:
                         logger.warning(f"Process {pid} exceed time limetattion per process")
+                        self.kill_process_tree(self._process_list[pid].pid)
                         self._process_list[pid].kill()
+                        self._process_list[pid].wait(timeout=1)
                     self._process_list[pid].wait(timeout=0)
                     device = get_device_by_args(self._process_list[pid].args)
                     # logger.info(f"{self._idx}/{len(self._command_list)} is started")
@@ -172,6 +187,7 @@ class TaskManager:
                 try:
                     if float((datetime.datetime.now() - self._timers[pid]).total_seconds()) > self.process_timeout:
                         logger.warning(f"Process {pid} exceed time limetation per process. The process will be killed")
+                        self.kill_process_tree(self._process_list[pid].pid)
                         self._process_list[pid].kill()
                         self._process_list[pid].wait(timeout=1)
                     self._process_list[pid].wait(timeout=0)
@@ -184,7 +200,7 @@ class TaskManager:
         return self._idx
 
 class TestParallelRunner:
-    def __init__(self, exec_file_path: os.path, test_command_line: list, worker_num: int, working_dir: os.path, cache_path: os.path):
+    def __init__(self, exec_file_path: os.path, test_command_line: list, worker_num: int, working_dir: os.path, cache_path: os.path, is_parallel_devices: False):
         self._exec_file_path = exec_file_path
         self._working_dir = working_dir
         self._command = self.__init_basic_command_line_for_exec_file(test_command_line)
@@ -202,7 +218,7 @@ class TestParallelRunner:
         self._total_test_cnt = 0
         self._available_devices = None
         self._device = get_device_by_args(self._command.split())
-        if has_python_api:
+        if has_python_api and is_parallel_devices:
             self._available_devices = get_available_devices(self._device)
         else:
             self._available_devices = [self._device] if not self._device is None else []
@@ -294,7 +310,7 @@ class TestParallelRunner:
         cached_test_list = list()
         runtime_test_test = list()
         cached_test_list_names = list()
-        it = 0
+
         for test in test_list_cache:
             if test._name in test_list_runtime:
                 cached_test_list.append(test)
@@ -320,33 +336,42 @@ class TestParallelRunner:
         
         # Run crashed tests in a separed thread
         if idx < len(proved_test_list):
-            while proved_test_list[idx]._time == -1 :
-                proved_test_list.pop(idx)
+            while proved_test_list[idx]._time == -1:
+                test = proved_test_list.pop(idx)
+                res_test_filters.append(test._name)
                 if idx >= len(proved_test_list):
                     break
 
+        longest_device = ""
+        for device in self._available_devices:
+            if len(device) > len(longest_device):
+                longest_device = device
+
         # prepare gtest filters per worker according command line length limitation
         while len(proved_test_list) > 0:
             test_times = []
             is_not_full = True
             worker_test_filters = list()
 
-            for _ in range(self._worker_num):
+            real_worker_num = self._worker_num * len(self._available_devices)
+
+            for _ in range(real_worker_num):
                 if len(proved_test_list) == 0:
                     break
                 worker_test_filters.append(f'"{self.__replace_restricted_symbols(proved_test_list[0]._name)}":')
-                test_times.append(proved_test_list[0]._time)
-                proved_test_list.pop(0)
+                test = proved_test_list.pop(0)
+                test_times.append(test._time)
             while is_not_full and len(proved_test_list) > 0:
-                for i in range(self._worker_num):
+                for i in range(real_worker_num):
                     if i >= len(proved_test_list):
                         break
                     if i == 0:
                         continue
                     while test_times[0] > test_times[i] + proved_test_list[len(proved_test_list) - 1]._time:
                         final_pos = len(proved_test_list) - 1
-                        if len(worker_test_filters[i]) + def_length + len(proved_test_list[final_pos]._name) < MAX_LENGHT:
-                            worker_test_filters[i] += f'"{self.__replace_restricted_symbols(proved_test_list[final_pos]._name)}":'
+                        filter = proved_test_list[final_pos]._name
+                        if len(worker_test_filters[i]) + def_length + len(filter.replace(self._device, longest_device)) < MAX_LENGHT:
+                            worker_test_filters[i] += f'"{self.__replace_restricted_symbols(filter)}":'
                             test_times[i] += proved_test_list[final_pos]._time
                             proved_test_list.pop(final_pos)
                         else:
@@ -355,17 +380,22 @@ class TestParallelRunner:
                         if len(proved_test_list) == 0:
                             break
                 if is_not_full and len(proved_test_list) > 0:
-                    worker_test_filters[0] += f'"{self.__replace_restricted_symbols(proved_test_list[0]._name)}":'
-                    test_times[0] += proved_test_list[0]._time
-                    proved_test_list.pop(0)
+                    filter = proved_test_list[0]._name
+                    if len(worker_test_filters[0]) + def_length + len(filter.replace(self._device, longest_device)) < MAX_LENGHT:
+                        worker_test_filters[0] += f'"{self.__replace_restricted_symbols(filter)}":'
+                        test_times[0] += proved_test_list[0]._time
+                        proved_test_list.pop(0)
+                    else:
+                        is_not_full = False
             for filter in worker_test_filters:
                 res_test_filters.append(filter)
             is_not_full = True
         # logging for debug
-        # for i in range(len(res_test_filters)):
-        #     filter = res_test_filters[i]
-        #     cnt = filter.count('\":')
-        #     logger.info(f"Number of tests in job_{i}: {cnt}")
+        for i in range(len(res_test_filters)):
+            filter = res_test_filters[i]
+            cnt = filter.count('\":')
+            self._total_test_cnt += cnt
+            # logger.info(f"Number of tests in job_{i}: {cnt}")
         return res_test_filters
             
     def __get_filters(self):
@@ -380,7 +410,6 @@ class TestParallelRunner:
 
         if len(cached_test_list) > 0:
             self._is_save_cache = False
-            self._total_test_cnt += len(cached_test_list)
             cached_test_list = self.__prepare_smart_filters(cached_test_list)
         if len(runtime_test_list) > 0:
             self._is_save_cache = True
@@ -391,7 +420,10 @@ class TestParallelRunner:
         
     def __execute_tests(self, filters: list(), prev_worker_cnt = 0):
         commands = [f'{self._command} --gtest_filter={filter}' for filter in filters]
-        task_manager = TaskManager(commands, self._working_dir, prev_worker_cnt, self._device, self._available_devices)
+        tmp_log_dir = os.path.join(self._working_dir, "temp")
+        if not os.path.isdir(tmp_log_dir):
+            os.mkdir(tmp_log_dir)
+        task_manager = TaskManager(commands, tmp_log_dir, prev_worker_cnt, self._device, self._available_devices)
         for _ in progressbar(range(self._worker_num), "Worker initialization: ", 40):
             task_manager.init_worker()
         for _ in progressbar(range(len(commands) - self._worker_num), "Worker execution: ", 40):
@@ -399,6 +431,26 @@ class TestParallelRunner:
                 break
         return task_manager.compelete_all_processes()
 
+    def __find_not_runned_tests(self):
+        test_names = set()
+        for log in Path(os.path.join(self._working_dir, "temp")).rglob("log_*.log"):
+            log_filename = os.path.join(self._working_dir, log)
+            with open(log_filename, "r") as log_file:
+                test_name = None
+                try:
+                    lines = log_file.readlines()
+                except:
+                    lines = log.read_text(encoding='ascii', errors='ignore').split('\n')
+
+                for line in lines:
+                    if constants.RUN in line:
+                        test_name = line[line.find(constants.RUN) + len(constants.RUN) + 1:-1:]
+                        if test_name is not None:
+                            test_names.add(f'"{test_name}":')
+                log_file.close()
+        test_list_runtime = set(self.__get_test_list_by_runtime())
+        return list(test_list_runtime.difference(test_names))
+
     def run(self):
         if TaskManager.process_timeout == -1:
             TaskManager.process_timeout = DEFAULT_PROCESS_TIMEOUT
@@ -410,15 +462,19 @@ class TestParallelRunner:
         filters_cache, filters_runtime = self.__get_filters()
 
         worker_cnt = 0
-        if len(filters_runtime):
-            logger.info(f"Execute jobs taken from runtime")
-            worker_cnt = self.__execute_tests(filters_runtime, worker_cnt)
+        if len(filters_cache):
+            logger.info(f"Execute jobs taken from cache")
+            worker_cnt = self.__execute_tests(filters_cache, worker_cnt)
         # 15m for one test in one process
         if TaskManager.process_timeout == -1 or TaskManager.process_timeout == DEFAULT_PROCESS_TIMEOUT:
             TaskManager.process_timeout = DEFAULT_TEST_TIMEOUT
-        if len(filters_cache):
-            logger.info(f"Execute jobs taken from cache")
-            self.__execute_tests(filters_cache, worker_cnt)
+        if len(filters_runtime):
+            logger.info(f"Execute jobs taken from runtime")
+            worker_cnt = self.__execute_tests(filters_runtime, worker_cnt)
+        not_runned_test_filter = self.__find_not_runned_tests()
+        if len(not_runned_test_filter) > 0:
+            logger.info(f"Execute not runned {len(not_runned_test_filter)} tests")
+            worker_cnt = self.__execute_tests(not_runned_test_filter, worker_cnt)
 
         t_end = datetime.datetime.now()
         total_seconds = (t_end - t_start).total_seconds()
@@ -599,9 +655,11 @@ if __name__ == "__main__":
     logger.info(f"[ARGUMENTS] --working_dir={args.working_dir}")
     logger.info(f"[ARGUMENTS] --process_timeout={args.process_timeout}")
     logger.info(f"[ARGUMENTS] --cache_path={args.cache_path}")
+    logger.info(f"[ARGUMENTS] --workers={args.workers}")
+    logger.info(f"[ARGUMENTS] --parallel_devices={args.parallel_devices}")
     logger.info(f"[ARGUMENTS] Executable file arguments = {exec_file_args}")
     TaskManager.process_timeout = args.process_timeout
-    conformance = TestParallelRunner(args.exec_file, exec_file_args, args.workers, args.working_dir, args.cache_path)
+    conformance = TestParallelRunner(args.exec_file, exec_file_args, args.workers, args.working_dir, args.cache_path, args.parallel_devices)
     conformance.run()
     if not conformance.postprocess_logs():
         logger.error("Run is not successful")