# Copyright 2022 The IREE Authors # # Licensed under the Apache License v2.0 with LLVM Exceptions. # See https://llvm.org/LICENSE.txt for license information. # SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception import json import pathlib import time from typing import List, Optional, Sequence, Set, Tuple from common.benchmark_suite import BenchmarkCase, BenchmarkSuite from common.benchmark_config import BenchmarkConfig from common.benchmark_definition import ( BenchmarkInfo, BenchmarkResults, BenchmarkMetrics, BenchmarkRun, DeviceInfo, ) class BenchmarkDriver(object): """Abstract driver runs the whole benchmark flow.""" def __init__( self, device_info: DeviceInfo, benchmark_config: BenchmarkConfig, benchmark_suite: BenchmarkSuite, benchmark_grace_time: float = 0.0, verbose: bool = False, ): self.device_info = device_info self.config = benchmark_config self.benchmark_suite = benchmark_suite self.benchmark_grace_time = benchmark_grace_time self.verbose = verbose self.finished_benchmarks: List[Tuple[BenchmarkInfo, pathlib.Path]] = [] self.finished_captures: List[pathlib.Path] = [] self.benchmark_errors = [] self._seen_benchmark_names: Set[str] = set() def run_benchmark_case( self, benchmark_case: BenchmarkCase, benchmark_results_filename: Optional[pathlib.Path], capture_filename: Optional[pathlib.Path], ) -> None: """Runs the benchmark case and serializes the results. Args: benchmark_case: the benchmark_case. benchmark_results_filename: the path to store the serialized BenchmarkMetrics. Benchmarking is required if set. capture_filename: the path to store captured trace. Trace capturing is required if set. Raises: Exception during benchmarking. """ raise NotImplementedError("Should be overwritten by a subclass.") def run(self) -> None: """Execute the benchmark flow. It performs the following steps: 1. Enumerate and filter benchmark cases. 2. Call 'run_benchmark_case' for each benchmark case. 3. Collect the benchmark results and captures. """ self.config.benchmark_results_dir.mkdir(parents=True, exist_ok=True) if self.config.trace_capture_config is not None: self.config.trace_capture_config.capture_tmp_dir.mkdir( parents=True, exist_ok=True ) cpu_target_arch = self.device_info.get_cpu_arch() gpu_target_arch = self.device_info.get_gpu_arch() detected_architectures = [ arch for arch in [cpu_target_arch, gpu_target_arch] if arch is not None ] if self.config.use_compatible_filter: if cpu_target_arch is None: print( "INFO: Detected unsupported CPU architecture in" f' "{self.device_info}", CPU benchmarking is disabled.' ) if gpu_target_arch is None: print( "INFO: Detected unsupported GPU architecture in" f' "{self.device_info}", GPU benchmarking is disabled.' ) compatible_arch_filter = detected_architectures else: # No compatible filter on the target architectures. compatible_arch_filter = None drivers, loaders = self.__get_available_drivers_and_loaders() benchmark_cases = self.benchmark_suite.filter_benchmarks( available_drivers=drivers, available_loaders=loaders, target_architectures=compatible_arch_filter, driver_filter=self.config.driver_filter, mode_filter=self.config.mode_filter, model_name_filter=self.config.model_name_filter, ) for benchmark_case in benchmark_cases: benchmark_info = self.__get_benchmark_info_from_case( benchmark_case=benchmark_case ) benchmark_name = str(benchmark_info) if benchmark_case.target_arch not in detected_architectures: print( f"WARNING: Benchmark '{benchmark_name}' may be incompatible" f" with the detected architectures '{detected_architectures}'" f" on the device. Pass --compatible-only to skip incompatible" f" benchmarks." ) # Sanity check for the uniqueness of benchmark names. if benchmark_name in self._seen_benchmark_names: raise ValueError( f"Found duplicate benchmark {benchmark_name} in the suites." ) self._seen_benchmark_names.add(benchmark_name) results_path, capture_path = self.__get_output_paths(benchmark_name) # If we continue from the previous results, check and skip if the result # files exist. if self.config.continue_from_previous: if results_path is not None and results_path.exists(): self.finished_benchmarks.append((benchmark_info, results_path)) results_path = None if capture_path is not None and capture_path.exists(): self.finished_captures.append(capture_path) capture_path = None # Skip if no need to benchmark and capture. if results_path is None and capture_path is None: continue print(f"--> Benchmark started: {benchmark_name} <--") try: self.run_benchmark_case(benchmark_case, results_path, capture_path) except Exception as e: # Delete unfinished results if they exist. if results_path is not None: results_path.unlink(missing_ok=True) if capture_path is not None: capture_path.unlink(missing_ok=True) if not self.config.keep_going: raise e print(f"Processing of benchmark failed with: {e}") self.benchmark_errors.append(e) continue finally: # Some grace time. time.sleep(self.benchmark_grace_time) print("Benchmark completed") if results_path: self.finished_benchmarks.append((benchmark_info, results_path)) if capture_path: self.finished_captures.append(capture_path) def get_benchmark_results(self) -> BenchmarkResults: """Returns the finished benchmark results.""" results = BenchmarkResults() results.set_commit(self.config.git_commit_hash) finished_benchmarks = sorted( self.finished_benchmarks, key=lambda pair: str(pair[0]) ) for info, path in finished_benchmarks: benchmark_metrics_json_object = json.loads(path.read_text()) benchmark_run = BenchmarkRun( info=info, metrics=BenchmarkMetrics.from_json_object( benchmark_metrics_json_object ), ) results.benchmarks.append(benchmark_run) return results def get_benchmark_result_filenames(self) -> Sequence[pathlib.Path]: """Returns the json file paths of finished benchmarks.""" return [path for info, path in self.finished_benchmarks] def get_capture_filenames(self) -> Sequence[pathlib.Path]: """Returns the tracy file paths of finished captures.""" return self.finished_captures def get_benchmark_errors(self): """Returns the exceptions captured during benchmarking.""" return self.benchmark_errors def __get_output_paths(self, benchmark_name: str): """Get output paths for the results and capture. The path of results/capture is None if the benchmark/capture doesn't need to be run. """ benchmark_results_filename = None if self.config.normal_benchmark_tool_dir: benchmark_results_filename = ( self.config.benchmark_results_dir / f"{benchmark_name}.json" ) capture_filename = None if self.config.trace_capture_config: capture_filename = ( self.config.trace_capture_config.capture_tmp_dir / f"{benchmark_name}.tracy" ) return (benchmark_results_filename, capture_filename) def __get_benchmark_info_from_case( self, benchmark_case: BenchmarkCase ) -> BenchmarkInfo: run_config = benchmark_case.run_config run_tags = run_config.module_execution_config.tags gen_config = run_config.module_generation_config model_source = str(gen_config.imported_model.model.source_type) compile_tags = gen_config.compile_config.tags return BenchmarkInfo( name=run_config.name, model_name=benchmark_case.model_name, model_tags=benchmark_case.model_tags, model_source=model_source, bench_mode=run_tags, compile_tags=compile_tags, driver_info=benchmark_case.driver_info, device_info=self.device_info, run_config_id=run_config.composite_id, ) def __get_available_drivers_and_loaders( self, ) -> Tuple[Sequence[str], Sequence[str]]: any_tool_dir = ( self.config.normal_benchmark_tool_dir if self.config.normal_benchmark_tool_dir else self.config.trace_capture_config.traced_benchmark_tool_dir ) config_txt_file_path = any_tool_dir / "build_config.txt" config_txt_file_lines = config_txt_file_path.read_text().splitlines() available_drivers = [] available_loaders = [] for line in config_txt_file_lines: name, value = line.strip().split("=") if value != "ON": continue if name == "IREE_HAL_DRIVER_CUDA": available_drivers.append("cuda") elif name == "IREE_HAL_DRIVER_LOCAL_SYNC": available_drivers.append("local-sync") elif name == "IREE_HAL_DRIVER_LOCAL_TASK": available_drivers.append("local-task") elif name == "IREE_HAL_DRIVER_VULKAN": available_drivers.append("vulkan") elif name == "IREE_HAL_EXECUTABLE_LOADER_EMBEDDED_ELF": available_loaders.append("embedded-elf") elif name == "IREE_HAL_EXECUTABLE_LOADER_SYSTEM_LIBRARY": available_loaders.append("system-library") elif name == "IREE_HAL_EXECUTABLE_LOADER_VMVX_MODULE": available_loaders.append("vmvx-module") else: continue if self.verbose: available_drivers_str = ", ".join(available_drivers) print(f"Available drivers: {available_drivers_str}") available_loaders_str = ", ".join(available_loaders) print(f"Available loaders: {available_loaders_str}") return available_drivers, available_loaders