Coverage for src/sparkle/CLI/run_parallel_portfolio.py: 73%
247 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-08 12:00 +0000
1#!/usr/bin/env python3
2# -*- coding: UTF-8 -*-
3"""Sparkle command to execute a parallel algorithm portfolio."""
5import sys
6import argparse
7import random
8import time
9import shutil
10import itertools
11from operator import mod
12from pathlib import Path
14from tqdm import tqdm
16import runrunner as rrr
17from runrunner.base import Runner, Run
18from runrunner.slurm import Status, SlurmRun
20from sparkle.CLI.help import logging as sl
21from sparkle.CLI.help import global_variables as gv
22from sparkle.CLI.initialise import check_for_initialise
23from sparkle.CLI.help import argparse_custom as ac
24from sparkle.CLI.help.nicknames import resolve_object_name
25from sparkle.platform.settings_objects import Settings
26from sparkle.solver import Solver
27from sparkle.instance import Instance_Set, InstanceSet, resolve_instance_pair
28from sparkle.types import SolverStatus, resolve_objective, UseTime
29from sparkle.structures import PerformanceDataFrame
32def parser_function() -> argparse.ArgumentParser:
33 """Define the command line arguments.
35 Returns:
36 parser: The parser with the parsed command line arguments
37 """
38 parser = argparse.ArgumentParser(
39 description="Run a portfolio of solvers on an instance set in parallel."
40 )
41 parser.add_argument(
42 *ac.InstanceSetPathsArgument.names, **ac.InstanceSetPathsArgument.kwargs
43 )
44 parser.add_argument(
45 *ac.NicknamePortfolioArgument.names, **ac.NicknamePortfolioArgument.kwargs
46 )
47 parser.add_argument(*ac.SolversArgument.names, **ac.SolversArgument.kwargs)
48 # Settings arguments
49 parser.add_argument(*ac.SettingsFileArgument.names, **ac.SettingsFileArgument.kwargs)
50 parser.add_argument(
51 *Settings.OPTION_objectives.args, **Settings.OPTION_objectives.kwargs
52 )
53 parser.add_argument(
54 *Settings.OPTION_solver_cutoff_time.args,
55 **Settings.OPTION_solver_cutoff_time.kwargs,
56 )
57 parser.add_argument(
58 *Settings.OPTION_parallel_portfolio_number_of_seeds_per_solver.args,
59 **Settings.OPTION_parallel_portfolio_number_of_seeds_per_solver.kwargs,
60 )
61 parser.add_argument(*Settings.OPTION_run_on.args, **Settings.OPTION_run_on.kwargs)
62 return parser
65def create_performance_dataframe(
66 solvers: list[Solver], instances_set: InstanceSet, portfolio_path: Path
67) -> PerformanceDataFrame:
68 """Create a PerformanceDataFrame for the given solvers and instances.
70 Args:
71 solvers: List of solvers to include in the PerformanceDataFrame.
72 instances_set: Set of instances to include in the PerformanceDataFrame.
73 portfolio_path: Path to save the CSV file.
75 Returns:
76 pdf: PerformanceDataFrame object initialized with solvers and instances.
77 """
78 instance_pairs = instances_set.instance_pairs
79 solvers = [str(solver.directory) for solver in solvers]
80 objectives = gv.settings().objectives
81 csv_path = portfolio_path / "results.csv"
82 return PerformanceDataFrame(
83 csv_filepath=csv_path,
84 solvers=solvers,
85 objectives=objectives,
86 instance_pairs=instance_pairs,
87 )
90def init_default_objectives() -> list:
91 """Initialize default objective values and key names.
93 Returns:
94 default_objective_values: Dictionary with default values for each objective.
95 cpu_time_key: Key for CPU time in the default values.
96 status_key: Key for status in the default values.
97 wall_time_key: Key for wall clock time in the default values.
98 """
99 # We record the 'best' of all seed results per solver-instance,
100 # setting start values for objectives that are always present
101 objectives = gv.settings().objectives
102 cutoff = gv.settings().solver_cutoff_time
103 cpu_time_key = [
104 objective.name
105 for objective in objectives
106 if objective.name.startswith("cpu_time")
107 ][0]
108 status_key = [
109 objective.name for objective in objectives if objective.name.startswith("status")
110 ][0]
111 wall_time_key = [
112 objective.name
113 for objective in objectives
114 if objective.name.startswith("wall_time")
115 ][0]
116 default_objective_values = {}
118 for objective in objectives:
119 default_value = float(sys.maxsize) if objective.minimise else 0
120 # Default values for time objectives can be linked to cutoff time
121 if objective.time and objective.post_process:
122 default_value = objective.post_process(
123 default_value, cutoff, SolverStatus.KILLED
124 )
125 default_objective_values[objective.name] = default_value
126 default_objective_values[status_key] = SolverStatus.UNKNOWN # Overwrite status
127 return default_objective_values, cpu_time_key, status_key, wall_time_key
130def monitor_jobs(
131 run: Run,
132 instances_set: InstanceSet,
133 solvers: list[Solver],
134 default_objective_values: dict,
135 run_on: Runner = Runner.SLURM,
136) -> dict:
137 """Monitor job progress and update job output dictionary.
139 Args:
140 run: The run object containing the submitted jobs.
141 instances_set: Set of instances to run on.
142 solvers: List of solvers to run on the instances.
143 default_objective_values: Default objective values for each solver-instance.
144 run_on: Unused
146 Returns:
147 job_output_dict: Dictionary containing the job output for each instance-solver
148 combination.
149 """
150 num_solvers, num_instances = len(solvers), len(instances_set._instance_paths)
151 seeds_per_solver = gv.settings().parallel_portfolio_num_seeds_per_solver
152 n_instance_jobs = num_solvers * seeds_per_solver
154 job_output_dict = {
155 instance_name: {
156 solver.name: default_objective_values.copy() for solver in solvers
157 }
158 for instance_name in instances_set._instance_names
159 }
161 check_interval = gv.settings().parallel_portfolio_check_interval
162 instances_done = [False] * num_instances
164 with tqdm(total=len(instances_done)) as pbar:
165 pbar.set_description("Instances done")
166 while not all(instances_done):
167 prev_done = sum(instances_done)
168 time.sleep(check_interval)
169 job_status_list = [r.status for r in run.jobs]
170 job_status_completed = [
171 status == Status.COMPLETED for status in job_status_list
172 ]
173 # The jobs are sorted by instance
174 for i, instance in enumerate(instances_set._instance_paths):
175 if instances_done[i]:
176 continue
177 instance_job_slice = slice(
178 i * n_instance_jobs, (i + 1) * n_instance_jobs
179 )
180 if any(job_status_completed[instance_job_slice]):
181 instances_done[i] = True
182 # Kill remaining jobs for this instance.
183 solver_kills = [0] * num_solvers
184 for job_index in range(
185 i * n_instance_jobs, (i + 1) * n_instance_jobs
186 ):
187 if not job_status_completed[job_index]:
188 run.jobs[job_index].kill()
189 solver_index = int(
190 (mod(job_index, n_instance_jobs)) // seeds_per_solver
191 )
192 solver_kills[solver_index] += 1
193 for solver_index in range(num_solvers):
194 # All seeds of a solver were killed on instance, set status kill
195 if solver_kills[solver_index] == seeds_per_solver:
196 solver_name = solvers[solver_index].name
197 # Use the set's canonical instance name (aligned by index with
198 # the paths), matching the key job_output_dict was built with;
199 # instance.stem would miss it for suffix-kept collisions.
200 instance_name = instances_set.instance_names[i]
201 job_output_dict[instance_name][solver_name]["status"] = (
202 SolverStatus.KILLED
203 )
204 pbar.update(sum(instances_done) - prev_done)
205 return job_output_dict
208def wait_for_logs(cmd_list: list[str]) -> None:
209 """Wait for all log files to be written.
211 Args:
212 cmd_list: List of command strings for all instance-solver-seed combinations.
213 """
214 # Attempt to verify that all logs have been written (Slurm I/O latency)
215 check_interval = gv.settings().parallel_portfolio_check_interval
216 for cmd in cmd_list:
217 runsolver_configuration = cmd.split(" ")[:11]
218 logs = [
219 Path(config)
220 for config in runsolver_configuration
221 if Path(config).suffix in [".log", ".val", ".rawres"]
222 ]
223 if not all(p.exists() for p in logs):
224 time.sleep(check_interval)
227def update_results_from_logs(
228 cmd_list: list[str],
229 run: Run,
230 solvers: list[Solver],
231 job_output_dict: dict,
232 cpu_time_key: str,
233) -> dict:
234 """Parse logs to update job output dictionary with best objective values.
236 Args:
237 cmd_list: List of command strings for all instance-solver-seed combinations.
238 run: The run object containing the submitted jobs.
239 solvers: List of solvers to run on the instances.
240 job_output_dict: Dictionary containing the job output for each intsance-solver
241 combination.
242 cpu_time_key: Key for CPU time in the job output dictionary.
244 Returns:
245 job_output_dict: Updated job output dictionary with best objective values.
246 """
247 seeds_per_solver = gv.settings().parallel_portfolio_num_seeds_per_solver
248 num_solvers = len(solvers)
249 n_instance_jobs = num_solvers * seeds_per_solver
250 objectives = gv.settings().objectives
252 for index, cmd in enumerate(cmd_list):
253 solver_index = (mod(index, n_instance_jobs)) // seeds_per_solver
254 solver_obj = solvers[solver_index]
255 solver_output = Solver.parse_solver_output(
256 run.jobs[index].stdout,
257 cmd.split(" "),
258 objectives=objectives,
259 verifier=solver_obj.verifier,
260 )
261 instance_name = list(job_output_dict.keys())[index // n_instance_jobs]
262 cpu_time = solver_output[cpu_time_key]
263 cmd_output = job_output_dict[instance_name][solver_obj.name]
264 if cpu_time > 0.0 and cpu_time < cmd_output[cpu_time_key]:
265 for key, value in solver_output.items():
266 if key in [objective.name for objective in objectives]:
267 job_output_dict[instance_name][solver_obj.name][key] = value
268 if cmd_output.get("status") != SolverStatus.KILLED:
269 cmd_output["status"] = solver_output.get("status")
270 return job_output_dict
273def fix_missing_times(
274 job_output_dict: dict, status_key: str, cpu_time_key: str, wall_time_key: str
275) -> dict:
276 """Fix CPU and wall clock times for solvers that did not produce logs.
278 Args:
279 job_output_dict: Dictionary containing the job output for each instance-solver
280 combination.
281 status_key: Key for status in the job output dictionary.
282 cpu_time_key: Key for CPU time in the job output dictionary.
283 wall_time_key: Key for wall clock time in the job output dictionary.
285 Returns:
286 job_output_dict: Updated job output dictionary with fixed CPU and wall clock
287 times.
288 """
289 cutoff = gv.settings().solver_cutoff_time
290 check_interval = gv.settings().parallel_portfolio_check_interval
292 # Fix the CPU/WC time for non existent logs to instance min time + check_interval
293 for instance in job_output_dict.keys():
294 no_log_solvers = []
295 min_time = cutoff
296 for solver in job_output_dict[instance].keys():
297 cpu_time = job_output_dict[instance][solver][cpu_time_key]
298 if cpu_time == -1.0 or cpu_time == float(sys.maxsize):
299 no_log_solvers.append(solver)
300 elif cpu_time < min_time:
301 min_time = cpu_time
302 for solver in no_log_solvers:
303 job_output_dict[instance][solver][cpu_time_key] = min_time + check_interval
304 job_output_dict[instance][solver][wall_time_key] = min_time + check_interval
305 # Fix runtime objectives with resolved CPU/Wall times
306 for key, value in job_output_dict[instance][solver].items():
307 objective = resolve_objective(key)
308 if objective is not None and objective.time:
309 value = (
310 job_output_dict[instance][solver][cpu_time_key]
311 if objective.use_time == UseTime.CPU_TIME
312 else job_output_dict[instance][solver][wall_time_key]
313 )
314 if objective.post_process is not None:
315 status = job_output_dict[instance][solver][status_key]
316 value = objective.post_process(value, cutoff, status)
317 job_output_dict[instance][solver][key] = value
318 return job_output_dict
321def print_and_write_results(
322 job_output_dict: dict,
323 solvers: list[Solver],
324 instances_set: InstanceSet,
325 portfolio_path: Path,
326 status_key: str,
327 cpu_time_key: str,
328 wall_time_key: str,
329 pdf: PerformanceDataFrame,
330) -> None:
331 """Print results to console and write the CSV file."""
332 num_instances = len(job_output_dict)
333 num_solvers = len(solvers)
334 objectives = gv.settings().objectives
335 for index, instance_name in enumerate(job_output_dict.keys()):
336 index_str = f"[{index + 1}/{num_instances}] "
337 instance_output = job_output_dict[instance_name]
338 if all(
339 instance_output[output][status_key] == SolverStatus.TIMEOUT
340 for output in instance_output
341 ):
342 print(f"\n{index_str}{instance_name} was not solved within the cutoff-time.")
343 continue
344 print(f"\n{index_str}{instance_name} yielded the following Solver results:")
345 for sindex in range(index * num_solvers, (index + 1) * num_solvers):
346 solver_name = solvers[mod(sindex, num_solvers)].name
347 job_info = job_output_dict[instance_name][solver_name]
348 print(
349 f"\t- {solver_name} ended with status {job_info[status_key]} in "
350 f"{job_info[cpu_time_key]}s CPU-Time ({job_info[wall_time_key]}s "
351 "Wall clock time)"
352 )
354 # Every job in this run belongs to instances_set, so pair each result with that set
355 # name directly. A name-only map would collapse identically named instances from
356 # different sets onto one key and let them overwrite each other; scoping to the known
357 # set keeps (SetA, inst1) and (SetB, inst1) distinct.
358 valid_pairs = set(pdf.instance_pairs)
359 solver_map = {Path(solver).name: solver for solver in pdf.solvers}
360 for instance, instance_dict in job_output_dict.items():
361 instance_pair = (instances_set.name, instance)
362 if instance_pair not in valid_pairs:
363 continue
364 for solver, objective_dict in instance_dict.items():
365 solver_name = Path(solver).name
366 solver_full_path = solver_map.get(solver_name, solver)
367 for objective in objectives:
368 obj_name = objective.name
369 obj_val = objective_dict.get(
370 obj_name, PerformanceDataFrame.missing_value
371 )
372 pdf.set_value(
373 value=obj_val,
374 solver=solver_full_path,
375 instance_pair=instance_pair,
376 objective=obj_name,
377 )
378 pdf.save_csv()
381def build_command_list(
382 instances_set: InstanceSet,
383 solvers: list[Solver],
384 portfolio_path: Path,
385 performance_data: PerformanceDataFrame,
386) -> list[str]:
387 """Build the list of command strings for all instance-solver-seed combinations.
389 Args:
390 instances_set: Set of instances to run on.
391 solvers: List of solvers to run on the instances.
392 portfolio_path: Path to the parallel portfolio.
393 performance_data: PerformanceDataFrame object.
395 Returns:
396 cmd_list: List of command strings for all instance-solver-seed combinations.
397 """
398 cutoff = gv.settings().solver_cutoff_time
399 objectives = gv.settings().objectives
400 seeds_per_solver = gv.settings().parallel_portfolio_num_seeds_per_solver
401 cmd_list = []
403 # Create a command for each instance-solver-seed combination. The pdf is keyed by the
404 # canonical (set_name, instance_name) pair, which cannot be derived from the path with
405 # stem/name (each InstanceSet subclass names its instances differently), so re-derive
406 # it from the path via resolve_instance_pair.
407 for instance, solver in itertools.product(instances_set.instance_paths, solvers):
408 # instance is a single path, so this resolves directly to its pair.
409 instance_pair = resolve_instance_pair(instance)
410 for _ in range(seeds_per_solver):
411 seed = int(random.getrandbits(32))
412 solver_call_list = solver.build_cmd(
413 instance.absolute(),
414 objectives=objectives,
415 seed=seed,
416 cutoff_time=cutoff,
417 log_dir=portfolio_path,
418 )
420 cmd_list.append(" ".join(solver_call_list))
421 for objective in objectives:
422 performance_data.set_value(
423 value=seed,
424 solver=str(solver.directory),
425 instance_pair=instance_pair,
426 objective=objective.name,
427 solver_fields=["Seed"],
428 )
429 return cmd_list
432def submit_jobs(
433 cmd_list: list[str],
434 solvers: list[Solver],
435 instances_set: InstanceSet,
436 run_on: Runner = Runner.SLURM,
437) -> SlurmRun:
438 """Submit jobs to the runner and return the run object.
440 Args:
441 cmd_list: List of command strings for all instance-solver-seed combinations.
442 solvers: List of solvers to run on the instances.
443 instances_set: Set of instances to run on.
444 run_on: Runner to use for submitting the jobs.
446 Returns:
447 run: The run object containing the submitted jobs.
448 """
449 seeds_per_solver = gv.settings().parallel_portfolio_num_seeds_per_solver
450 num_solvers, num_instances = len(solvers), len(instances_set._instance_paths)
451 num_jobs = num_solvers * num_instances * seeds_per_solver
452 parallel_jobs = min(gv.settings().slurm_jobs_in_parallel, num_jobs)
453 if parallel_jobs > num_jobs:
454 print(
455 "WARNING: Not all jobs will be started at the same time due to the "
456 "limitation of number of Slurm jobs that can be run in parallel. Check"
457 " your Sparkle Slurm Settings."
458 )
459 print(
460 f"Sparkle parallel portfolio is running {seeds_per_solver} seed(s) per solver "
461 f"on {num_solvers} solvers for {num_instances} instances ..."
462 )
464 sbatch_options = gv.settings().sbatch_settings
465 solver_names = ", ".join([solver.name for solver in solvers])
466 # Jobs are added in to the runrunner object in the same order they are provided
467 return rrr.add_to_queue(
468 runner=run_on,
469 cmd=cmd_list,
470 name=f"Parallel Portfolio {solver_names}",
471 parallel_jobs=parallel_jobs,
472 base_dir=sl.caller_log_dir,
473 srun_options=["-N1", "-n1"] + sbatch_options,
474 sbatch_options=sbatch_options,
475 prepend=gv.settings().slurm_job_prepend,
476 )
479def main(argv: list[str]) -> None:
480 """Main method of run parallel portfolio command."""
481 # Define command line arguments
482 parser = parser_function()
484 # Process command line arguments
485 args = parser.parse_args(argv)
486 settings = gv.settings(args)
488 # Log command call
489 sl.log_command(sys.argv, settings.random_state)
490 check_for_initialise()
492 # Compare current settings to latest.ini
493 prev_settings = Settings(Settings.DEFAULT_previous_settings_path)
494 Settings.check_settings_changes(settings, prev_settings)
496 if args.solvers is not None:
497 solver_paths = [
498 resolve_object_name("".join(solver), target_dir=settings.DEFAULT_solver_dir)
499 for solver in args.solvers
500 ]
501 if None in solver_paths:
502 print("Some solvers not recognised! Check solver names:")
503 for i, name in enumerate(solver_paths):
504 if solver_paths[i] is None:
505 print(f'\t- "{solver_paths[i]}" ')
506 sys.exit(-1)
507 solvers = [Solver(solver_path) for solver_path in solver_paths]
508 else:
509 solvers = [
510 Solver(solver)
511 for solver in settings.DEFAULT_solver_dir.iterdir()
512 if solver.is_dir()
513 ]
515 portfolio_path = args.portfolio_name
517 run_on = settings.run_on
518 if run_on == Runner.LOCAL:
519 print("Parallel Portfolio is not fully supported yet for Local runs. Exiting.")
520 sys.exit(-1)
522 # Retrieve instance sets
523 instances = [
524 resolve_object_name(
525 instance_path,
526 gv.file_storage_data_mapping[gv.instances_nickname_path],
527 gv.settings().DEFAULT_instance_dir,
528 Instance_Set,
529 )
530 for instance_path in args.instance_path
531 ]
532 # Join them into one
533 if len(instances) > 1:
534 print(
535 "WARNING: More than one instance set specified. "
536 "Currently only supporting one."
537 )
538 instances = instances[0]
540 print(f"Running on {instances.size} instance(s)...")
542 if not settings.objectives[0].time:
543 print(
544 "ERROR: Parallel Portfolio is currently only relevant for "
545 "RunTime objectives. In all other cases, use validation"
546 )
547 sys.exit(-1)
549 if args.portfolio_name is not None: # Use a nickname
550 portfolio_path = settings.DEFAULT_parallel_portfolio_output / args.portfolio_name
551 else: # Generate a timestamped nickname
552 timestamp = time.strftime("%Y-%m-%d-%H.%M.%S", time.gmtime(time.time()))
553 randintstamp = int(random.getrandbits(32))
554 portfolio_path = (
555 settings.DEFAULT_parallel_portfolio_output / f"{timestamp}_{randintstamp}"
556 )
557 if portfolio_path.exists():
558 print(
559 f"[WARNING] Portfolio path {portfolio_path} already exists! "
560 "Overwrite? [y/n] ",
561 end="",
562 )
563 user_input = input()
564 if user_input != "y":
565 sys.exit()
566 shutil.rmtree(portfolio_path)
568 portfolio_path.mkdir(parents=True)
569 pdf = create_performance_dataframe(solvers, instances, portfolio_path)
570 returned_cmd = build_command_list(instances, solvers, portfolio_path, pdf)
571 default_objective_values, cpu_time_key, status_key, wall_time_key = (
572 init_default_objectives()
573 )
574 returned_run = submit_jobs(returned_cmd, solvers, instances, Runner.SLURM)
575 job_output_dict = monitor_jobs(
576 returned_run, instances, solvers, default_objective_values
577 )
578 wait_for_logs(returned_cmd)
579 job_output_dict = update_results_from_logs(
580 returned_cmd, returned_run, solvers, job_output_dict, cpu_time_key
581 )
582 job_output_dict = fix_missing_times(
583 job_output_dict, status_key, cpu_time_key, wall_time_key
584 )
585 print_and_write_results(
586 job_output_dict,
587 solvers,
588 instances,
589 portfolio_path,
590 status_key,
591 cpu_time_key,
592 wall_time_key,
593 pdf,
594 )
596 # Write used settings to file
597 settings.write_used_settings()
598 print("Running Sparkle parallel portfolio is done!")
599 sys.exit(0)
602if __name__ == "__main__":
603 main(sys.argv[1:])