1#!/usr/bin/env python3 2# -*- coding: utf-8 -*- 3 4# Copyright (c) 2024-2025 Huawei Device Co., Ltd. 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16# 17 18import re 19import logging 20from typing import Optional, List, Union, Iterable 21from pathlib import Path 22from vmb.shell import ShellResult 23from vmb.helpers import remove_prefix, norm_list, create_file 24from vmb.result import TestResult, RunResult, BUStatus 25 26BENCH_PREFIX = 'bench_' 27UNIT_PREFIX = 'bu_' 28LIB_PREFIX = 'lib' 29log = logging.getLogger('vmb') 30 31 32class BenchUnit: 33 """BU represents single VM Benchmark test. 34 35 Simply put, Unit consists of: 36 1) directory containing program to run + libs + resources 37 2) results of building and running this bench 38 """ 39 40 __result_patt = re.compile( 41 r'.*: ([\w-]+) (-?\d+(\.\d+)?([eE][+\-]\d+)?)') 42 __warmup_patt = re.compile( 43 r'.*Warmup \d+:.* (?P<value>\d+(\.\d+)?) ns/op') 44 __iter_patt = re.compile( 45 r'.*Iter \d+:.* (?P<value>\d+(\.\d+)?) ns/op') 46 __real_tm_patt = re.compile( 47 r"(?:Elapsed.*\(h:mm:ss or m:ss\)|Real time)[^:]*:\s*" 48 r"(?:(\d*):)?(\d*)(?:.(\d*))?") 49 50 def __init__(self, 51 path: Union[str, Path], 52 src: Optional[Union[str, Path]] = None, 53 libs: Optional[Iterable[Union[str, Path]]] = None, 54 tags: Optional[Iterable[str]] = None, 55 bugs: Optional[Iterable[str]] = None) -> None: 56 self.path: Path = Path(path) 57 self.__src: Optional[Path] = self.path.joinpath(src) if src else None 58 self.__libs: List[Path] = [ 59 self.path.joinpath(lib) for lib in libs] if libs else [] 60 self.name: str = remove_prefix(self.path.name, UNIT_PREFIX) 61 # default main binary is abc 62 self.__main_bin: Path = self.path.joinpath(f'{BENCH_PREFIX}{self.name}.abc') 63 self.__main_lib: Path = self.path.joinpath(f'lib{self.name}.abc') 64 self.binaries: List[Path] = [] # binaries to clean up after test 65 self.device_path: Optional[Path] = None 66 self.result: TestResult = TestResult( 67 self.name, 68 tags=norm_list(tags), 69 bugs=norm_list(bugs)) 70 self.run_out: str = '' 71 self.crash_info: str = '' 72 73 @property 74 def doclet_src(self) -> Optional[Path]: 75 return self.__src 76 77 @property 78 def status(self) -> BUStatus: 79 return self.result._status # pylint: disable=protected-access 80 81 @status.setter 82 def status(self, stat: BUStatus) -> None: 83 self.result._status = stat # pylint: disable=protected-access 84 85 @property 86 def main_bin(self) -> Path: 87 return self.__main_bin 88 89 @main_bin.setter 90 def main_bin(self, path: Path) -> None: 91 self.__main_bin = path 92 93 @property 94 def main_lib(self) -> Path: 95 return self.__main_lib 96 97 @main_lib.setter 98 def main_lib(self, path: Path) -> None: 99 self.__main_lib = path 100 101 def parse_ext_time(self, rss_out: str) -> float: 102 m = re.search(self.__real_tm_patt, rss_out) 103 if m is None: 104 return 0.0 105 tmp = m.groups() 106 if tmp[0] is None: 107 return round(float(str(tmp[1]) + "." + tmp[2]), 5) 108 return round(int(tmp[0]) * 60 + float(str(tmp[1]) + "." + tmp[2]), 5) 109 110 def parse_run_output(self, res: ShellResult) -> None: 111 if not res.out: 112 return 113 self.run_out = res.out + "\n" + res.err 114 mtch = re.search(self.__result_patt, res.out) 115 if mtch: 116 if mtch.groups()[0] != self.name: 117 log.warning('Name mismatch: %s vs %s', 118 mtch.groups()[0], self.name) 119 avg_time = float(mtch.groups()[1]) 120 warms = [ 121 float(m.group("value")) 122 for m in re.finditer(self.__warmup_patt, res.out)] 123 iters = [ 124 float(m.group("value")) 125 for m in re.finditer(self.__iter_patt, res.out)] 126 self.result.execution_forks.append( 127 RunResult(avg_time, iters, warms)) 128 self.result.mem_bytes = res.rss 129 self.result.execution_status = \ 130 int(res.ret) if res.ret is not None else -1 131 self.status = BUStatus.PASS \ 132 if (0 == self.result.execution_status 133 and len(self.result.execution_forks) > 0) \ 134 else BUStatus.EXECUTION_FAILED 135 if not res.err: 136 return 137 # pylint: disable-next=protected-access 138 self.result._ext_time = self.parse_ext_time(res.err) 139 140 def src(self, *ext, die_on_zero_matches=False) -> Path: 141 files = [f for f in self.path.glob(f'{BENCH_PREFIX}{self.name}*') 142 if (not ext or f.suffix in ext)] 143 if len(files) > 0: 144 return files[0] 145 if die_on_zero_matches: 146 raise RuntimeError(f'Files {self.path}/{BENCH_PREFIX}{self.name}*[{",".join(ext)}] not found!') 147 # fallback: return unexistent 148 return self.path.joinpath(f'{BENCH_PREFIX}{self.name}') 149 150 def libs(self, *ext) -> Iterable[Path]: 151 if self.__libs: 152 return [lib for lib in self.__libs 153 if (not ext or lib.suffix in ext)] 154 # search all libs on filesystem: 155 return {lib for lib in self.path.glob(f'{LIB_PREFIX}*') 156 if (not ext or lib.suffix in ext)} 157 158 def src_device(self, *ext) -> Path: 159 if self.device_path is None: 160 raise RuntimeError(f'Device path not set for {self.name}') 161 return self.device_path.joinpath(self.src(*ext).name) 162 163 def libs_device(self, *ext) -> Iterable[Path]: 164 if self.device_path is None: 165 raise RuntimeError(f'Device path not set for {self.name}') 166 return [self.device_path.joinpath(lib.name) for lib in self.libs(*ext)] 167 168 def save_fail_log(self, root: str, msg: str = '') -> None: 169 fail_log = Path(root).joinpath(self.name).with_suffix('.txt') 170 msg = msg if msg else self.run_out 171 with create_file(fail_log) as f: 172 f.write(msg) 173 if self.crash_info: 174 f.write(self.crash_info) 175