Reference

A UML diagram showing the package relationships within the tool.

main

The main function for the HPC MultiBench tool.

get_parser() -> ArgumentParser

Get the argument parser for the tool.

Source code in src/hpc_multibench/main.py

def get_parser() -> ArgumentParser:  # pragma: no cover
    """Get the argument parser for the tool."""
    parser = ArgumentParser(
        description="A Swiss army knife for comparing programs on HPC resources."
    )
    # As an argument of the base tool not subcommands, due to the ergonomics of
    # running `record` then `report` on same yaml file in sequence
    parser.add_argument(
        "-y",
        "--yaml-path",
        required=True,
        type=Path,
        help="the path to the configuration YAML file",
    )
    parser.add_argument(
        "-o",
        "--outputs-directory",
        type=Path,
        default=DEFAULT_BASE_OUTPUTS_DIRECTORY,
        help="the path to the configuration YAML file",
    )

    sub_parsers = parser.add_subparsers(dest="command", required=True)
    parser_record = sub_parsers.add_parser(
        "record", help="record data from running the test benches"
    )
    parser_interactive = sub_parsers.add_parser(
        "interactive", help="show the interactive TUI"
    )
    sub_parsers.add_parser(
        "report", help="report analysis about completed test bench runs"
    )

    parser_record.add_argument(
        "-d",
        "--dry-run",
        action="store_true",
        help="print but don't submit the generated sbatch files",
    )
    parser_record.add_argument(
        "-w",
        "--wait",
        action="store_true",
        help="wait for the submitted jobs to finish to exit",
    )
    for sub_parser in (parser_record, parser_interactive):
        sub_parser.add_argument(
            "-nc",
            "--no-clobber",
            action="store_true",
            help="don't delete any previous run results of the test benches",
        )
    return parser

main() -> None

Run the tool.

Source code in src/hpc_multibench/main.py

def main() -> None:  # pragma: no cover
    """Run the tool."""
    args = get_parser().parse_args()
    test_plan = TestPlan(args.yaml_path, args.outputs_directory)

    if args.command == "record":
        test_plan.record_all(args)

    elif args.command == "report":
        test_plan.report_all(args)

    else:
        args.dry_run = False
        args.wait = False
        UserInterface(test_plan, args).run()

test_plan

A class representing the test plan defined from YAML for a tool run.

TestPlan

The test plan defined from YAML for a tool run.

Source code in src/hpc_multibench/test_plan.py

class TestPlan:
    """The test plan defined from YAML for a tool run."""

    def __init__(self, yaml_path: Path, base_output_directory: Path) -> None:
        """Instantiate the test plan from a YAML file."""
        self.yaml_path = yaml_path
        self.base_output_directory = base_output_directory

        test_plan_model = TestPlanModel.from_yaml(yaml_path)
        self.benches = [
            TestBench(
                name=bench_name,
                run_configuration_models={
                    name: deepcopy(config)
                    for name, config in test_plan_model.run_configurations.items()
                    if name in bench_model.run_configurations
                },
                bench_model=bench_model,
                base_output_directory=base_output_directory,
            )
            for bench_name, bench_model in test_plan_model.benches.items()
        ]

    def record_all(self, args: Namespace) -> None:
        """Run all the enabled test benches in the plan."""
        for bench in self.benches:
            if bench.bench_model.enabled:
                print(f"Recording data from test bench '{bench.name}'")
                bench.record(args)

        if args.wait:
            for bench in self.benches:
                if bench.bench_model.enabled:
                    status = (
                        "timed out while waiting for queued jobs"
                        if bench.wait_for_queue()
                        else "finished all queued jobs"
                    )
                    print(f"Test bench '{bench.name}' {status}!")

    def report_all(self, _args: Namespace) -> None:
        """Analyse all the enabled test benches in the plan."""
        for bench in self.benches:
            if bench.bench_model.enabled:
                print(f"Reporting data from test bench '{bench.name}'")
                bench.report()

__init__(yaml_path: Path, base_output_directory: Path) -> None

Instantiate the test plan from a YAML file.

Source code in src/hpc_multibench/test_plan.py

def __init__(self, yaml_path: Path, base_output_directory: Path) -> None:
    """Instantiate the test plan from a YAML file."""
    self.yaml_path = yaml_path
    self.base_output_directory = base_output_directory

    test_plan_model = TestPlanModel.from_yaml(yaml_path)
    self.benches = [
        TestBench(
            name=bench_name,
            run_configuration_models={
                name: deepcopy(config)
                for name, config in test_plan_model.run_configurations.items()
                if name in bench_model.run_configurations
            },
            bench_model=bench_model,
            base_output_directory=base_output_directory,
        )
        for bench_name, bench_model in test_plan_model.benches.items()
    ]

record_all(args: Namespace) -> None

Run all the enabled test benches in the plan.

Source code in src/hpc_multibench/test_plan.py

def record_all(self, args: Namespace) -> None:
    """Run all the enabled test benches in the plan."""
    for bench in self.benches:
        if bench.bench_model.enabled:
            print(f"Recording data from test bench '{bench.name}'")
            bench.record(args)

    if args.wait:
        for bench in self.benches:
            if bench.bench_model.enabled:
                status = (
                    "timed out while waiting for queued jobs"
                    if bench.wait_for_queue()
                    else "finished all queued jobs"
                )
                print(f"Test bench '{bench.name}' {status}!")

report_all(_args: Namespace) -> None

Analyse all the enabled test benches in the plan.

Source code in src/hpc_multibench/test_plan.py

def report_all(self, _args: Namespace) -> None:
    """Analyse all the enabled test benches in the plan."""
    for bench in self.benches:
        if bench.bench_model.enabled:
            print(f"Reporting data from test bench '{bench.name}'")
            bench.report()