"""Base class for exceptions in this module."""

def __init__(self, message: str):

lines = [line.strip()

for line in configFile.open().readlines() if uncomment(line)]

try:

return json.loads('\n'.join(lines))

except JSONDecodeError as e:

error(

if key in dict:

return dict[key]

else:

if key in dict:

return dict[key]

else:

'iterations': 'i', 'forks': 'f', 'time': 'r', 'timeout': 'to',

'timeunit': 'tu', 'verbosity': 'v',

'warmupbatchsize': 'wbs', 'warmupforks': 'wf',

'warmupiterations': 'wi', 'warmuptime': 'w',

path = pathlib.Path('.')

path_str = optional('result.path', config)

if path_str:

path = pathlib.Path(path_str)

path = output_dir_path(config)

try:

path.mkdir(parents=True)

except FileExistsError:

try:

arch = platform.machine()

system = platform.system()

kernel = platform.release()

cpu_model = ""

ram_info = ""

os_info = ""

java_info = ""

try:

java_version_out = subprocess.check_output(['java', '-version'], stderr=subprocess.STDOUT).decode().strip()

# The first line usually contains the version information

java_info = java_version_out.splitlines()[0]

except Exception:

java_info = "Unknown Java"

if system == "Darwin":

try:

cpu_model = subprocess.check_output(['sysctl', '-n', 'machdep.cpu.brand_string']).decode().strip()

except Exception:

cpu_model = platform.processor()

try:

mem_bytes = int(subprocess.check_output(['sysctl', '-n', 'hw.memsize']).decode().strip())

ram_info = f"{mem_bytes // (1024**3)}GB RAM"

except Exception:

ram_info = "Unknown RAM"

os_info = f"macOS {platform.mac_ver()[0]}"

elif system == "Linux":

try:

with open("/proc/cpuinfo", "r") as f:

for line in f:

if "model name" in line:

cpu_model = line.split(":")[1].strip()

break

except Exception:

cpu_model = platform.processor()

try:

with open("/proc/meminfo", "r") as f:

for line in f:

if "MemTotal" in line:

mem_kb = int(line.split(":")[1].strip().split()[0])

ram_info = f"{mem_kb // (1024**2)}GB RAM"

break

except Exception:

ram_info = "Unknown RAM"

try:

import lsb_release

os_info = lsb_release.get_distro_information()['DESCRIPTION']

except Exception:

try:

with open("/etc/os-release", "r") as f:

for line in f:

if line.startswith("PRETTY_NAME="):

os_info = line.split("=")[1].strip().strip('"')

break

except Exception:

os_info = f"Linux {platform.release()}"

else:

cpu_model = platform.processor()

os_info = f"{system} {platform.release()}"

return f"{arch} - {cpu_model} - {ram_info} - {os_info} - Kernel: {kernel} - {java_info}"

except Exception as e:

cmd = ["java"]

jvm_args = optional('jvmargs', config)

if jvm_args:

if not type(jvm_args) is list:

error('jvmargs field must be a list of arguments')

for arg_value in jvm_args:

cmd.append(f'-{arg_value}')

java_library_path = optional('java.library.path', config)

if java_library_path:

cmd.append(f'-Djava.library.path={java_library_path}')

jar = optional('jar', config)

if jar:

cmd.append('-jar')

cmd.append(jar)

help = optional('help', config)

if help:

cmd.append('-h')

benchmark = required('benchmark', config)

cmd.append(str(benchmark))

params = optional('params', config)

if params:

if not type(params) is dict:

error('params field must be a dictionary of parameters')

for key, value in params.items():

if type(value) is int or type(value) is float:

value = str(value)

if type(value) is list:

value_str = ','.join([str(v) for v in value])

cmd.append('-p')

cmd.append(f'{key}={value_str}')

elif type(value) is str:

cmd.append('-p')

cmd.append(f'{key}={str(value)}')

else:

error(f'field {key} does not have a string or list value')

flags = optional('flags', config)

if flags:

if not type(flags) is list:

error('Flags field must be a list of flags')

for flag_value in flags:

cmd.append(f'-{flag_value}')

options = optional('options', config)

if options:

if not type(options) is dict:

error('Options field must be a dictionary of parameters')

for key, value in options.items():

if key not in option_map:

error(f'Option {key} is not a valid/known option')

if type(value) is int or type(value) is float:

value = str(value)

if type(value) is not str:

error(

f'Options field {key} must have a string value, not: {value}')

cmd.append(f'-{option_map[key]}')

cmd.append(value)

cmd.extend(output_options(config))

output_file = pathlib.Path(output_log_file(config))

if output_file.exists():

error(f'Output file {output_file} already exists')

with output_file.open(mode='w', encoding='UTF-8') as log:

log.writelines(line + '\n' for line in

['## JMH Run', f'This JMH run was generated on {const_datetime_str}'])

log.writelines(line + '\n' for line in

[f'#### Config', f'The configuration was read from `{config_file}`', '```json'])

json.dump(config, fp=log, indent=4)

log.write('\n')

log.writelines(line + '\n' for line in

['```', '#### Command', 'The java command executed to run the tests', '```', ' '.join(cmd), '```'])

# Save system info

system_info_file = output_dir_path(config).joinpath('system_info.json')

with system_info_file.open(mode='w', encoding='UTF-8') as f:

cmd_str = ' '.join(cmd)

if help_requested:

print(f'JMH Help requested, command: {cmd_str}')

else:

print(f'Execute: {cmd_str}')

proc = subprocess.run(cmd, start_new_session=True)

parser = argparse.ArgumentParser(description='Run configured jmh tests.')

parser.add_argument(

'-c', '--config', help='A JSON configuration file for the JMH run', default='jmh_run.json')

args = parser.parse_args()

try:

config_file = pathlib.Path(args.config)

if not config_file.exists():

raise RunnerError(

f'The config file {config_file} does not exist')

if not config_file.is_file():

raise RunnerError(

f'The config file {config_file} is not a text file')

config = read_config_file(config_file)

cmd_list = build_jmh_command(config)

create_output_dir(config)

log_jmh_session(cmd_list, config, f'{config_file.resolve()}')

exec_jmh_cmd(cmd_list, optional('help', config))

except RunnerError as error:

print(