parallel_programming_course/scripts/run_tests.py at f5ce70d1d8bb47b8a388ff2bb88c070c231140db · aobolensk/parallel_programming_course · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
import os
import subprocess
import platform
from pathlib import Path


def init_cmd_args():
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--running-type",
        required=True,
        choices=["threads", "processes", "performance", "performance-list"],
        help="Specify the execution mode. Choose 'threads' for multithreading or 'processes' for multiprocessing."
    )
    parser.add_argument(
        "--additional-mpi-args",
        required=False,
        default="",
        help="Additional MPI arguments to pass to the mpirun command (optional)."
    )
    args = parser.parse_args()
    _args_dict = vars(args)
    return _args_dict


class PPCRunner:
    def __init__(self):
        self.work_dir = None
        self.valgrind_cmd = "valgrind --error-exitcode=1 --leak-check=full --show-leak-kinds=all"

        if platform.system() == "Windows":
            self.mpi_exec = "mpiexec"
        else:
            self.mpi_exec = "mpirun"

    @staticmethod
    def __get_project_path():
        script_path = Path(__file__).resolve()  # Absolute path of the script
        script_dir = script_path.parent  # Directory containing the script
        return script_dir.parent

    def setup_env(self):
        if (Path(self.__get_project_path()) / "install").exists():
            self.work_dir = Path(self.__get_project_path()) / "install" / "bin"
        else:
            self.work_dir = Path(self.__get_project_path()) / "build" / "bin"

    @staticmethod
    def __run_exec(command):
        result = subprocess.run(command, shell=True, env=os.environ)
        if result.returncode != 0:
            raise Exception(f"Subprocess return {result.returncode}.")

    @staticmethod
    def __get_gtest_settings(repeats_count):
        command = f"--gtest_repeat={repeats_count} "
        command += "--gtest_recreate_environments_when_repeating "
        command += "--gtest_color=0 "
        return command

    def run_threads(self):
        if platform.system() == "Linux" and not os.environ.get("PPC_ASAN_RUN"):
            self.__run_exec(f"{self.valgrind_cmd} {self.work_dir / 'seq_func_tests'} {self.__get_gtest_settings(1)}")
            self.__run_exec(f"{self.valgrind_cmd} {self.work_dir / 'stl_func_tests'} {self.__get_gtest_settings(1)}")

        self.__run_exec(f"{self.work_dir / 'seq_func_tests'} {self.__get_gtest_settings(3)}")
        self.__run_exec(f"{self.work_dir / 'stl_func_tests'} {self.__get_gtest_settings(3)}")
        self.__run_exec(f"{self.work_dir / 'tbb_func_tests'} {self.__get_gtest_settings(3)}")
        self.__run_exec(f"{self.work_dir / 'omp_func_tests'} {self.__get_gtest_settings(3)}")

    def run_core(self):
        if platform.system() == "Linux" and not os.environ.get("PPC_ASAN_RUN"):
            self.__run_exec(f"{self.valgrind_cmd} {self.work_dir / 'core_func_tests'} {self.__get_gtest_settings(1)}")

        self.__run_exec(f"{self.work_dir / 'core_func_tests'} {self.__get_gtest_settings(1)}")

    def run_processes(self, additional_mpi_args):
        PPC_NUM_PROC = os.environ.get("PPC_NUM_PROC")
        if PPC_NUM_PROC is None:
            raise EnvironmentError("Required environment variable 'PPC_NUM_PROC' is not set.")

        mpi_running = f"{self.mpi_exec} {additional_mpi_args} -np {PPC_NUM_PROC}"
        if not os.environ.get("PPC_ASAN_RUN"):
            self.__run_exec(f"{mpi_running} {self.work_dir / 'all_func_tests'} {self.__get_gtest_settings(10)}")
            self.__run_exec(f"{mpi_running} {self.work_dir / 'mpi_func_tests'} {self.__get_gtest_settings(10)}")

    def run_performance(self):
        if not os.environ.get("PPC_ASAN_RUN"):
            PPC_NUM_PROC = os.environ.get("PPC_NUM_PROC")
            if PPC_NUM_PROC is None:
                raise EnvironmentError("Required environment variable 'PPC_NUM_PROC' is not set.")
            mpi_running = f"{self.mpi_exec} -np {PPC_NUM_PROC}"
            self.__run_exec(f"{mpi_running} {self.work_dir / 'all_perf_tests'} {self.__get_gtest_settings(1)}")
            self.__run_exec(f"{mpi_running} {self.work_dir / 'mpi_perf_tests'} {self.__get_gtest_settings(1)}")

        self.__run_exec(f"{self.work_dir / 'omp_perf_tests'} {self.__get_gtest_settings(1)}")
        self.__run_exec(f"{self.work_dir / 'seq_perf_tests'} {self.__get_gtest_settings(1)}")
        self.__run_exec(f"{self.work_dir / 'stl_perf_tests'} {self.__get_gtest_settings(1)}")
        self.__run_exec(f"{self.work_dir / 'tbb_perf_tests'} {self.__get_gtest_settings(1)}")

    def run_performance_list(self):
        for task_type in ["all", "mpi", "omp", "seq", "stl", "tbb"]:
            self.__run_exec(f"{self.work_dir / f'{task_type}_perf_tests'} --gtest_list_tests")


if __name__ == "__main__":
    args_dict = init_cmd_args()

    ppc_runner = PPCRunner()
    ppc_runner.setup_env()

    if args_dict["running_type"] in ["threads", "processes"]:
        ppc_runner.run_core()

    if args_dict["running_type"] == "threads":
        ppc_runner.run_threads()
    elif args_dict["running_type"] == "processes":
        ppc_runner.run_processes(args_dict["additional_mpi_args"])
    elif args_dict["running_type"] == "performance":
        ppc_runner.run_performance()
    elif args_dict["running_type"] == "performance-list":
        ppc_runner.run_performance_list()
    else:
        raise Exception("running-type is wrong!")