diff --git a/SofaRegressionProgram/SofaRegressionProgram.py b/SofaRegressionProgram/SofaRegressionProgram.py index 5d2ae85..30339d1 100644 --- a/SofaRegressionProgram/SofaRegressionProgram.py +++ b/SofaRegressionProgram/SofaRegressionProgram.py @@ -13,12 +13,12 @@ import Sofa import SofaRuntime # importing SofaRuntime will add the py3 loader to the scene loaders import tools.RegressionSceneList as RegressionSceneList -from tools import ProgressBarHandler as pbh +import tools.RegressionWorker as RegressionWorker regression_file_extension = ".regression-tests" class RegressionProgram: - def __init__(self, input_folder, filter = None, disable_progress_bar = False, verbose = False): + def __init__(self, input_folder, filter = None, disable_progress_bar = False, verbose = False, nbr_jobs = 1): """Initialize the RegressionProgram Args: @@ -26,18 +26,20 @@ def __init__(self, input_folder, filter = None, disable_progress_bar = False, ve filter (str): Regex pattern to filter scene files (e.g., '^demo.*.scn$'). If None, no filter is applied. Defaults to None. disable_progress_bar (bool, optional): If True, disable progress bars. Defaults to False. verbose (bool, optional): If True, enable verbose output. Defaults to False. + nbr_jobs (int, optional): Number of scenes to write/compare at the same time. 0 means one per logical core. Defaults to 1. """ self.scene_sets = [] # List self.disable_progress_bar = disable_progress_bar self.verbose = verbose self.legacy_mode = False + self.nbr_jobs = RegressionWorker.resolve_nbr_jobs(nbr_jobs) for root, dirs, files in os.walk(input_folder): for file in files: if file.endswith(regression_file_extension): file_path = os.path.join(root, file) - scene_list = RegressionSceneList.RegressionSceneList(file_path, filter, self.disable_progress_bar, verbose) + scene_list = RegressionSceneList.RegressionSceneList(file_path, filter, self.disable_progress_bar, verbose, self.nbr_jobs) scene_list.process_file() self.scene_sets.append(scene_list) @@ -58,26 +60,32 @@ def log_errors_in_sets(self): for scene_list in self.scene_sets: scene_list.log_scenes_errors() + def run_all_sets(self, mode, description): + """Run every scene of every set in `mode` ("write" or "compare"). + + When several jobs are allowed, the scenes of all the sets are scheduled + in a single pool: a set holding fewer scenes than the number of jobs + would otherwise leave most of the workers idle. + """ + tasks = [] + for scene_list in self.scene_sets: + scene_list.legacy_mode = self.legacy_mode + tasks.extend(scene_list.build_tasks(mode)) + + return RegressionWorker.run_scene_tasks( + tasks, + nbr_jobs=self.nbr_jobs, + on_result=lambda task, result: task["scene_list"].apply_result(task, result), + description=description, + disable_progress_bar=self.disable_progress_bar) + def write_sets_references(self, id_set=0): scene_list = self.scene_sets[id_set] nbr_scenes = scene_list.write_all_references() return nbr_scenes def write_all_sets_references(self): - nbr_sets = len(self.scene_sets) - - pbar_sets = pbh.ProgressBarHandler(total=nbr_sets, disable=self.disable_progress_bar) - pbar_sets.set_description("Write All sets") - - nbr_scenes = 0 - for i in range(0, nbr_sets): - nbr_scenes = nbr_scenes + self.write_sets_references(i) - pbar_sets.update(1) - - if not self.disable_progress_bar: - pbar_sets.close() - - return nbr_scenes + return self.run_all_sets("write", "Write All sets") def compare_sets_references(self, id_set=0): scene_list = self.scene_sets[id_set] @@ -86,18 +94,7 @@ def compare_sets_references(self, id_set=0): return nbr_scenes def compare_all_sets_references(self): - nbr_sets = len(self.scene_sets) - pbar_sets = pbh.ProgressBarHandler(total=nbr_sets, disable=self.disable_progress_bar) - pbar_sets.set_description("Compare All sets") - - nbr_scenes = 0 - for i in range(0, nbr_sets): - nbr_scenes = nbr_scenes + self.compare_sets_references(i) - pbar_sets.update(1) - - pbar_sets.close() - - return nbr_scenes + return self.run_all_sets("compare", "Compare All sets") def replay_references(self, id_scene, id_set=0): scene_list = self.scene_sets[id_set] @@ -128,7 +125,15 @@ def make_parser(): help="A regex filter to select scenes to test (e.g., '^demo.*.scn$')", type=str) - parser.add_argument('--replay', + parser.add_argument('-j', '--jobs', + dest='jobs', + help="Number of scenes to process at the same time (each one still runs in its own\n" + "isolated process, so the results are unchanged). 0 means one job per logical\n" + "core. Default: 1 (sequential).", + type=int, + default=1) + + parser.add_argument('--replay', dest='replay', help=f"Will launch runSofa on the scene number X (input number) in the input the list of the {regression_file_extension} file given as input and display the scene references aside from the simulation", type=int) @@ -169,6 +174,8 @@ def make_parser(): python SofaRegressionProgram.py --input ./scenes python SofaRegressionProgram.py --input ./scenes --filter \"$demo.*.scn\" python SofaRegressionProgram.py --input ./scenes --replay 5 + python SofaRegressionProgram.py --input ./scenes --jobs 8 + python SofaRegressionProgram.py --input ./scenes --write-references -j 0 ''' return parser @@ -181,7 +188,7 @@ def make_parser(): # 2- Process file if args.input is not None: - reg_prog = RegressionProgram(args.input, args.filter, args.progress_bar_is_disabled, args.verbose) + reg_prog = RegressionProgram(args.input, args.filter, args.progress_bar_is_disabled, args.verbose, args.jobs) else: parser.print_help() exit("Error: Argument is required ! Quitting.") @@ -191,7 +198,11 @@ def make_parser(): if args.legacy_mode: print("Legacy regression mode activated.") reg_prog.legacy_mode = True - + + if reg_prog.nbr_jobs > 1: + print(f"Processing up to {reg_prog.nbr_jobs} scenes at the same time.") + + if args.replay is not None: replayId = int(args.replay) reg_prog.replay_references(replayId) diff --git a/SofaRegressionProgram/tools/RegressionSceneList.py b/SofaRegressionProgram/tools/RegressionSceneList.py index 93f4eb3..b8c2b43 100644 --- a/SofaRegressionProgram/tools/RegressionSceneList.py +++ b/SofaRegressionProgram/tools/RegressionSceneList.py @@ -2,14 +2,13 @@ import tools.RegressionSceneData as RegressionSceneData import tools.RegressionHelper as helper import tools.RegressionWorker as RegressionWorker -from tools import ProgressBarHandler as pbh import re ## This class is responsible for loading a file.regression-tests to gather the list of scene to test with all arguments ## It will provide the API to launch the tests or write refs on all scenes contained in this file class RegressionSceneList: - def __init__(self, file_path, filter, disable_progress_bar = False, verbose = False): + def __init__(self, file_path, filter, disable_progress_bar = False, verbose = False, nbr_jobs = 1): """ /// Path to the file.regression-tests containing the list of scene to tests with all arguments std::string filePath; @@ -24,6 +23,7 @@ def __init__(self, file_path, filter, disable_progress_bar = False, verbose = Fa self.disable_progress_bar = disable_progress_bar self.verbose = verbose self.legacy_mode = False + self.nbr_jobs = nbr_jobs # number of scenes simulated at the same time def get_nbr_scenes(self): @@ -197,33 +197,73 @@ def process_file(self): self.scenes_data_sets.append(scene_data) + def build_task(self, id_scene, mode): + """Return the task descriptor handed over to RegressionWorker for one scene. + + Each scene is run in its own process to guarantee a clean SOFA state + (SOFA does not fully reset its global state between load/unload), which + also makes it safe to run several of them at the same time. + """ + return { + "scene_list": self, + "id_scene": id_scene, + "scene_data": self.scenes_data_sets[id_scene], + "mode": mode, + "legacy": self.legacy_mode, + "verbose": self.verbose, + } + + + def build_tasks(self, mode): + """Return the task descriptors of every scene of this list.""" + return [self.build_task(i, mode) for i in range(len(self.scenes_data_sets))] + + + def apply_result(self, task, result): + """Collect the outcome reported by a worker process for one scene.""" + scene = self.scenes_data_sets[task["id_scene"]] + + if task["mode"] == "write": + if not result.get("ok", False): + helper.writeError(f"While writing references for {scene.file_scene_path}: {result.get('error')}") + return + + if not result.get("ok", False): + # Hard failure (scene could not be loaded / worker crashed). + self.nbr_errors = self.nbr_errors + 1 + helper.writeError(f"While trying to compare {scene.file_scene_path}: {result.get('error')}") + return + + # Bring the worker's outcome back so log_errors() reports it as usual. + scene.apply_worker_result(result) + if not result.get("result", False): + self.nbr_errors = self.nbr_errors + 1 + + + def _run_tasks(self, mode, description): + tasks = self.build_tasks(mode) + return RegressionWorker.run_scene_tasks( + tasks, + nbr_jobs=self.nbr_jobs, + on_result=self.apply_result, + description=description, + disable_progress_bar=self.disable_progress_bar) + + def write_references(self, id_scene, print_log = False): scene = self.scenes_data_sets[id_scene] if self.verbose: helper.writeLog(f'Writing reference files for {scene.file_scene_path}.') - # Each scene is written in its own process to guarantee a clean SOFA - # state (SOFA does not fully reset global state between load/unload). + task = self.build_task(id_scene, "write") result = RegressionWorker.run_scene_in_subprocess( scene, mode="write", disable_progress_bar=self.disable_progress_bar, verbose=self.verbose) + self.apply_result(task, result) - if not result.get("ok", False): - helper.writeError(f"While writing references for {scene.file_scene_path}: {result.get('error')}") def write_all_references(self): - nbr_scenes = len(self.scenes_data_sets) - - pbar_scenes = pbh.ProgressBarHandler(total=nbr_scenes, disable=self.disable_progress_bar) - pbar_scenes.set_description("Write all scenes from: " + self.file_path) - - for i in range(0, nbr_scenes): - self.write_references(i) - pbar_scenes.update(1) - - pbar_scenes.close() - - return nbr_scenes + return self._run_tasks("write", "Write all scenes from: " + self.file_path) def compare_references(self, id_scene): @@ -231,35 +271,15 @@ def compare_references(self, id_scene): if self.verbose: scene.print_info() - # Each scene is compared in its own process to guarantee a clean SOFA - # state, identical to the one used when the references were written. + task = self.build_task(id_scene, "compare") result = RegressionWorker.run_scene_in_subprocess( scene, mode="compare", legacy=self.legacy_mode, disable_progress_bar=self.disable_progress_bar, verbose=self.verbose) + self.apply_result(task, result) - if not result.get("ok", False): - # Hard failure (scene could not be loaded / worker crashed). - self.nbr_errors = self.nbr_errors + 1 - helper.writeError(f"While trying to compare {scene.file_scene_path}: {result.get('error')}") - return - - # Bring the worker's outcome back so log_errors() reports it as usual. - scene.apply_worker_result(result) - if not result.get("result", False): - self.nbr_errors = self.nbr_errors + 1 - def compare_all_references(self): - nbr_scenes = len(self.scenes_data_sets) - pbar_scenes = pbh.ProgressBarHandler(total=nbr_scenes, disable=self.disable_progress_bar) - pbar_scenes.set_description("Compare all scenes from: " + self.file_path) - - for i in range(0, nbr_scenes): - self.compare_references(i) - pbar_scenes.update(1) - pbar_scenes.close() - - return nbr_scenes + return self._run_tasks("compare", "Compare all scenes from: " + self.file_path) def replay_references(self, id_scene): diff --git a/SofaRegressionProgram/tools/RegressionWorker.py b/SofaRegressionProgram/tools/RegressionWorker.py index f34e032..70cfcb7 100644 --- a/SofaRegressionProgram/tools/RegressionWorker.py +++ b/SofaRegressionProgram/tools/RegressionWorker.py @@ -15,11 +15,17 @@ This module has two roles: * Parent side: `run_scene_in_subprocess()` spawns a child for one scene and - marshals the result back through a temporary JSON file. + marshals the result back through a temporary JSON file. `run_scene_tasks()` + schedules a list of scenes over a pool of such children, so that several + scenes are simulated at the same time. * Child side: executed as `python RegressionWorker.py ...`, it sets up the SOFA environment, runs a single scene (write or compare) and writes its result to the file given by `--result-file`. +Because every scene already runs in its own process, running several of them +concurrently changes nothing to the results: children never share any SOFA +state. The parent only has to schedule them and collect their outcome. + Only the standard library is imported at module top-level so that importing this module in the parent does NOT import SOFA (the parent must never load or simulate a scene, otherwise the isolation would be defeated). @@ -31,6 +37,7 @@ import argparse import subprocess import tempfile +from concurrent.futures import ThreadPoolExecutor, as_completed def _safe_remove(path): @@ -45,7 +52,8 @@ def _safe_remove(path): # -------------------------------------------------- def run_scene_in_subprocess(scene_data, mode, legacy=False, disable_progress_bar=False, verbose=False, - format="JSON", python_exe=None): + format="JSON", python_exe=None, + capture_output=False): """Run a single scene (write or compare) in an isolated child process. Args: @@ -57,6 +65,10 @@ def run_scene_in_subprocess(scene_data, mode, legacy=False, format (str): reference file format ("JSON" or "CSV"). python_exe (str): interpreter to use for the child (defaults to the current one). + capture_output (bool): if True, the child output is captured and + returned in the "stdout"/"stderr" keys of the result instead of + being interleaved with the output of the other children. Used when + several scenes run concurrently. Returns: dict: the result reported by the child. Always contains an "ok" key. @@ -89,10 +101,13 @@ def run_scene_in_subprocess(scene_data, mode, legacy=False, if disable_progress_bar: cmd.append("--disable-progress-bar") - # stdout/stderr are inherited so SOFA logs and progress bars behave exactly - # as before (and the parent's --quiet redirection propagates to the child). + # When a single scene runs at a time, stdout/stderr are inherited so SOFA + # logs and progress bars behave exactly as before (and the parent's --quiet + # redirection propagates to the child). When several children run at the + # same time their output is captured instead, and replayed as one block by + # the caller, otherwise the logs of all the scenes would be interleaved. try: - completed = subprocess.run(cmd) + completed = subprocess.run(cmd, capture_output=capture_output, text=capture_output) except Exception as e: _safe_remove(result_path) return {"ok": False, "error": f"Failed to launch worker subprocess: {e}"} @@ -106,11 +121,122 @@ def run_scene_in_subprocess(scene_data, mode, legacy=False, _safe_remove(result_path) if result is None: - return {"ok": False, - "error": f"Worker produced no result (exit code {completed.returncode})."} + result = {"ok": False, + "error": f"Worker produced no result (exit code {completed.returncode})."} + + if capture_output: + result["stdout"] = completed.stdout + result["stderr"] = completed.stderr return result +# -------------------------------------------------- +# Parent side: schedule several scenes concurrently +# -------------------------------------------------- +def resolve_nbr_jobs(nbr_jobs): + """Turn the user-provided job count into a usable number of workers. + + 0 (or a negative value) means "one job per logical core". + """ + if nbr_jobs is None: + return 1 + nbr_jobs = int(nbr_jobs) + if nbr_jobs <= 0: + return os.cpu_count() or 1 + return nbr_jobs + + +def _echo_captured_output(header, result): + """Print in one block the output captured from a child process.""" + out = result.get("stdout") + err = result.get("stderr") + if not (out or err): + return + + if out: + sys.stdout.write(header + "\n") + sys.stdout.write(out if out.endswith("\n") else out + "\n") + sys.stdout.flush() + if err: + sys.stderr.write(header + "\n") + sys.stderr.write(err if err.endswith("\n") else err + "\n") + sys.stderr.flush() + + +def run_scene_tasks(tasks, nbr_jobs=1, format="JSON", on_result=None, + description=None, disable_progress_bar=False): + """Run a list of scenes, up to `nbr_jobs` of them at the same time. + + Args: + tasks (list): task descriptors. Each one is a dict containing at least + "scene_data" (RegressionSceneData), "mode" ("write" or "compare"), + and optionally "legacy" and "verbose". Any other key is ignored + here and simply handed back to `on_result`, which lets the caller + attach whatever context it needs to identify the task. + nbr_jobs (int): maximum number of scenes simulated concurrently. + format (str): reference file format ("JSON" or "CSV"). + on_result (callable): called as `on_result(task, result)` for every + finished task, always from the calling thread so that the callback + does not need any locking. + description (str): label of the progress bar. + disable_progress_bar (bool): disable the progress bar of this run. + + Returns: + int: the number of tasks that were run. + """ + from tools import ProgressBarHandler as pbh + + nbr_jobs = max(1, resolve_nbr_jobs(nbr_jobs)) + # Never spawn more workers than there is work to do. + nbr_jobs = min(nbr_jobs, len(tasks)) if tasks else 1 + + pbar = pbh.ProgressBarHandler(total=len(tasks), disable=disable_progress_bar) + if description is not None: + pbar.set_description(description) + + def _run(task): + return run_scene_in_subprocess( + task["scene_data"], + mode=task["mode"], + legacy=task.get("legacy", False), + # In parallel the per-step progress bars of the children are + # captured along with their output: they would only produce noise. + disable_progress_bar=disable_progress_bar or nbr_jobs > 1, + verbose=task.get("verbose", False), + format=format, + capture_output=nbr_jobs > 1, + ) + + try: + if nbr_jobs == 1: + for task in tasks: + result = _run(task) + if on_result is not None: + on_result(task, result) + pbar.update(1) + else: + with ThreadPoolExecutor(max_workers=nbr_jobs) as executor: + # The threads only wait on their child process: all the result + # handling happens here, in the calling thread. + futures = {executor.submit(_run, task): task for task in tasks} + try: + for future in as_completed(futures): + task = futures[future] + result = future.result() + _echo_captured_output( + f"--- {task['mode']}: {task['scene_data'].file_scene_path}", result) + if on_result is not None: + on_result(task, result) + pbar.update(1) + except (KeyboardInterrupt, SystemExit): + executor.shutdown(wait=False, cancel_futures=True) + raise + finally: + pbar.close() + + return len(tasks) + + # -------------------------------------------------- # Child side: run one scene in a fresh SOFA process # --------------------------------------------------