Skip to content

[Python] Parallelize regression {compare|write} processes #122

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
fredroy wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[Python] Parallelize regression {compare|write} processes #122

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 43 additions & 32 deletions SofaRegressionProgram/SofaRegressionProgram.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,31 +13,33 @@
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:
input_folder (str): Path to the folder containing regression test files.
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 <RegressionSceneList>
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)
Expand All @@ -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]
Expand All @@ -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]
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand All @@ -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.")
Expand All @@ -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)
Expand Down
102 changes: 61 additions & 41 deletions SofaRegressionProgram/tools/RegressionSceneList.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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):
Expand Down Expand Up @@ -197,69 +197,89 @@ 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):
scene = self.scenes_data_sets[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):
Expand Down
Loading