• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 os
19import logging
20import re
21import shutil
22import json
23from pathlib import Path
24from typing import Optional, Iterable, Tuple, List, Dict
25from abc import ABC, abstractmethod
26from enum import Flag, auto
27from subprocess import TimeoutExpired
28from vmb.unit import BenchUnit
29from vmb.helpers import StringEnum
30from vmb.shell import ShellDevice, ShellUnix, ShellResult
31from vmb.target import Target
32from vmb.x_shell import CrossShell
33
34log = logging.getLogger('vmb')
35
36
37class VmbToolExecError(Exception):
38    """VMB Error. Tool execution failed."""
39
40    def __init__(self, message: str,
41                 res: Optional[ShellResult] = None) -> None:
42        super().__init__(message)
43        self.out = f'{message}\n\nout:\n{res.out}\n\nerr:\n{res.err}\n' \
44                   if res is not None else message
45
46
47class ToolMode(StringEnum):
48    AOT = 'aot'
49    INT = 'int'
50    JIT = 'jit'
51    INT_CPP = 'int-cpp'
52    INT_IRTOC = 'int-irtoc'
53    INT_LLVM = 'int-llvm'
54    LLVMAOT = 'llvmaot'
55    AOTPGO = 'aot-pgo'
56    DEFAULT = 'default'
57
58
59class OptFlags(Flag):
60    NONE = auto()
61    GC_STATS = auto()
62    JIT_STATS = auto()
63    AOT_STATS = auto()
64    AOT_SKIP_LIBS = auto()
65    DRY_RUN = auto()
66    DISABLE_INLINING = auto()
67    AOT = auto()
68    INT = auto()
69    JIT = auto()
70    INT_CPP = auto()
71    INT_IRTOC = auto()
72    INT_LLVM = auto()
73    LLVMAOT = auto()
74    AOTPGO = auto()
75
76
77class ToolBase(CrossShell, ABC):
78
79    sh_: ShellUnix
80    andb_: ShellDevice
81    hdc_: ShellDevice
82    dev_dir: Path
83    libs: Path
84
85    def __init__(self,
86                 target: Target = Target.HOST,
87                 flags: OptFlags = OptFlags.NONE,
88                 custom_opts: Optional[List[str]] = None):
89        self._target = target
90        self.flags = flags
91        self.custom_opts = custom_opts if custom_opts else []
92
93    def __call__(self, bu: BenchUnit) -> None:
94        self.exec(bu)
95
96    @property
97    @abstractmethod
98    def name(self) -> str:
99        return ''
100
101    @property
102    def version(self) -> str:
103        return 'version n/a'
104
105    @property
106    def target(self) -> Target:
107        return self._target
108
109    @property
110    def sh(self) -> ShellUnix:
111        return ToolBase.sh_
112
113    @property
114    def andb(self) -> ShellDevice:
115        return ToolBase.andb_
116
117    @property
118    def hdc(self) -> ShellDevice:
119        return ToolBase.hdc_
120
121    @property
122    def custom(self) -> str:
123        return ' '.join(self.custom_opts)
124
125    @staticmethod
126    def rename_suffix(old: Path, new_suffix: str) -> Path:
127        if new_suffix == old.suffix:
128            return old
129        new = old.with_suffix(new_suffix)
130        old.rename(new)
131        return new
132
133    @staticmethod
134    def get_cmd_path(cmd: str, env_var: str = '') -> Optional[str]:
135        # use specifically requested
136        p: Optional[str] = os.environ.get(env_var, '')
137        # or use default
138        if not p:
139            p = shutil.which(cmd)
140        if not p or (not os.path.isfile(p)):
141            extra_msg = f' or set via {env_var} env var' if env_var else ''
142            raise RuntimeError(
143                f'{cmd} not found. Add it to PATH{extra_msg}')
144        log.info('Using %s as %s', p, cmd)
145        return p
146
147    @staticmethod
148    def ensure_file(*args, err: str = '') -> str:
149        f = os.path.join(*args)
150        if not os.path.isfile(f):
151            raise RuntimeError(f'File "{f}" not found! {err}')
152        return str(f)
153
154    @staticmethod
155    def ensure_dir(*args, err: str = '') -> str:
156        d = os.path.join(*args)
157        if not os.path.isdir(d):
158            raise RuntimeError(f'Dir "{d}" not found! {err}')
159        return str(d)
160
161    @staticmethod
162    def ensure_dir_env(var_name: str) -> str:
163        return ToolBase.ensure_dir(os.environ.get(var_name, ''),
164                                   err=f'Please set {var_name} env var.')
165
166    @abstractmethod
167    def exec(self, bu: BenchUnit) -> None:
168        pass
169
170    def kill(self) -> None:
171        """Kill tool process(es).
172
173        For host target there is os.killpg() in Shell,
174        but on device tool process needs remote pkill <tool>
175        """
176
177    def x_run(self, cmd: str, measure_time: bool = True,
178              timeout: Optional[float] = None, cwd: str = '') -> ShellResult:
179        try:
180            res = self.x_sh.run(
181                cmd, measure_time=measure_time, timeout=timeout, cwd=cwd)
182        except TimeoutExpired as e:
183            self.kill()
184            raise e
185        if not res or res.ret != 0:
186            raise VmbToolExecError(f'{self.name} failed', res)
187        return res
188
189    def x_src(self, bu: BenchUnit, *ext) -> Path:
190        if self.target == Target.HOST:
191            return bu.src(*ext)
192        return bu.src_device(*ext)
193
194    def x_libs(self, bu: BenchUnit, *ext) -> Iterable[Path]:
195        if self.target == Target.HOST:
196            return bu.libs(*ext)
197        return bu.libs_device(*ext)
198
199    def get_bu_opts(self, bu: BenchUnit) -> Tuple[OptFlags, str]:
200        conf = bu.path.joinpath('config.json')
201        flags: OptFlags = self.flags
202        aot_opts: str = ''
203        if conf.exists():
204            with open(conf, 'r', encoding='utf-8') as f:
205                conf_data = json.load(f)
206                if conf_data.get('disable_inlining', False):
207                    flags |= OptFlags.DISABLE_INLINING
208                aot_opts = conf_data.get('aot_opts', '')
209        return flags, aot_opts
210
211    def custom_opts_obj(self) -> Dict[str, str]:
212        re_opts = re.compile(r'^(-+)?(?P<opt>[\w\-]+)(=|\s+)(?P<val>.+)$')
213        opts = {}
214        for opt in self.custom_opts:
215            m = re.search(re_opts, opt.strip("'\""))
216            if m:
217                opts[m.group("opt")] = m.group("val")
218            else:
219                log.warning('Custom option malformed: %s', opt)
220        return opts
221