Benchmark_Driver

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# ===--- Benchmark_Driver ------------------------------------------------===//
#
# This source file is part of the Swift.org open source project
#
# Copyright (c) 2014 - 2017 Apple Inc. and the Swift project authors
# Licensed under Apache License v2.0 with Runtime Library Exception
#
# See https://swift.org/LICENSE.txt for license information
# See https://swift.org/CONTRIBUTORS.txt for the list of Swift project authors
#
# ===---------------------------------------------------------------------===//
"""
Benchmark_Driver is a tool for running and analysing Swift Benchmarking Suite.

Example:
 $ Benchmark_Driver run

Use `Benchmark_Driver -h` for help on available commands and options.

class `BenchmarkDriver` runs performance tests and implements the `run` COMMAND.
class `BenchmarkDoctor` analyzes performance tests, implements `check` COMMAND.

"""

importargparse
importfunctools
importglob
importlogging
importmath
importos
importplatform
importre
importsubprocess
importsys
importtime

fromcompare_perf_testsimportLogParser

DRIVER_DIR=os.path.dirname(os.path.realpath(__file__))


classBenchmarkDriver(object):
"""Executes tests from Swift Benchmark Suite.

 It's a higher level wrapper for the Benchmark_X family of binaries
 (X = [O, Onone, Osize]).
 """

def__init__(self, args, tests=None, _subprocess=None, parser=None):
"""Initialize with command line arguments.

 Optional parameters are for injecting dependencies -- used for testing.
 """
self.args=args
self.run_env=os.environ.copy()

# Set a constant hash seed. Some tests are highly sensitive to
# fluctuations in the number of hash collisions.
self.run_env["SWIFT_DETERMINISTIC_HASHING"] ="1"

ifhasattr(args, 'libdir') andargs.libdir:
# The benchmark binaries should pick up the built swift libraries
# automatically, because their RPATH should point to ../lib/swift
# But it does not harm to additionally set the dynamic library path,
# e.g. as a workaround to rdar://78584073
ifplatform.system() =="Darwin":
self.run_env["DYLD_LIBRARY_PATH"] =args.libdir
elifplatform.system() =="Linux":
self.run_env["LD_LIBRARY_PATH"] =args.libdir

self._subprocess=_subprocessorsubprocess
self.all_tests= []
self.test_number= {}
self.tests=testsorself._get_tests()
self.parser=parserorLogParser()
self.results= {}

def_invoke(self, cmd):
returnself._subprocess.check_output(
cmd, stderr=self._subprocess.STDOUT, universal_newlines=True,
env=self.run_env
 )

@property
deftest_harness(self):
"""Full path to test harness binary."""
suffix=self.args.optimizationifhasattr(self.args, "optimization") else"O"
suffix+="-"
ifhasattr(self.args, "architecture") andself.args.architecture:
suffix+=self.args.architecture
pattern=os.path.join(self.args.tests, "Benchmark_"+suffix+"*")
executables= []
ifhasattr(self._subprocess, "test_mode") andself._subprocess.test_mode:
executables= [pattern]
else:
executables=glob.glob(pattern)
iflen(executables) ==0:
raiseValueError(
"No benchmark executable for file name pattern "+
pattern+" found")
iflen(executables) >1:
raiseValueError(
"Multiple benchmark executables for file name pattern "+
pattern+" found\n"+
str(executables) +
"\nSpecify the architecture to select the right benchmark executable")
returnexecutables[0]

def_git(self, cmd):
"""Execute the Git command in the `swift-repo`."""
returnself._invoke(
 ("git -C {0} ".format(self.args.swift_repo) +cmd).split()
 ).strip()

@property
deflog_file(self):
"""Full path to log file.

 If `swift-repo` is set, log file is tied to Git branch and revision.
 """
ifnotself.args.output_dir:
returnNone
log_dir=self.args.output_dir
harness_name=os.path.basename(self.test_harness)
suffix="-"+time.strftime("%Y%m%d%H%M%S", time.localtime())
ifself.args.swift_repo:
log_dir=os.path.join(
log_dir, self._git("rev-parse --abbrev-ref HEAD")
 ) # branch
suffix+="-"+self._git("rev-parse --short HEAD") # revision
returnos.path.join(log_dir, harness_name+suffix+".log")

@property
def_cmd_list_benchmarks(self):
# TODO: Switch to JSON format: add "--json" here
return [self.test_harness, "--list"] + (
 ["--skip-tags="] if (self.args.benchmarksorself.args.filters) else []
 )

def_get_tests(self):
"""Return a list of performance tests to run."""
lines=self._invoke(self._cmd_list_benchmarks).split("\n")
json_tests= []
forlineinlines:
columns=re.split(r'[ ,]+', line.strip())
try:
number=int(columns[0])
name=columns[1]
json_descr= {"number": number, "name": name}
json_tests.append(json_descr)
exceptException:
continue
# TODO: Replace the above with the following to
# use the JSON output from the benchmark driver
# directly
# if line.strip() != "":
# json_tests.append(json.loads(line))
self.all_tests= [json["name"] forjsoninjson_tests]
test_numbers= [json["number"] forjsoninjson_tests]
self.test_number=dict([(json["name"], json["number"]) forjsoninjson_tests])
ifself.args.filters:
returnself._tests_matching_patterns()
ifself.args.benchmarks:
returnself._tests_by_name_or_number(test_numbers)
returnself.all_tests

def_tests_matching_patterns(self):
matches=set()
forfilinself.args.filters:
pattern=re.compile(fil)
new_matches=filter(pattern.match, self.all_tests)
matches=matches.union(new_matches)
returnsorted(list(matches))

def_tests_by_name_or_number(self, test_numbers):
benchmarks=set(self.args.benchmarks)
numbers=list(map(str, test_numbers))
number_to_name=dict(zip(numbers, self.all_tests))
tests_by_number= [
number_to_name[i] foriinbenchmarks.intersection(numbers)
 ]
returnsorted(
list(benchmarks.intersection(set(self.all_tests)).union(tests_by_number))
 )

defrun(
self,
test=None,
num_samples=None,
num_iters=None,
sample_time=None,
verbose=None,
measure_memory=False
 ):
"""Execute benchmark and gather results."""
num_samples=num_samplesor0
num_iters=num_itersor0# automatically determine N to run for 1s
sample_time=sample_timeor0# default is 1s

cmd=self._cmd_run(
test, num_samples, num_iters, sample_time, verbose, measure_memory
 )
output=self._invoke(cmd)
results=self.parser.results_from_string(output)
iftest:
returnlist(results.items())[0][1]
else:
returnresults

def_cmd_run(
self,
test,
num_samples,
num_iters,
sample_time,
verbose,
measure_memory
 ):
cmd= [self.test_harness]
iftest:
cmd.append(test)
else:
cmd.extend([str(self.test_number.get(name, name)) fornameinself.tests])
ifnum_samples>0:
cmd.append("--num-samples={0}".format(num_samples))
else:
cmd.append("--min-samples=2")
ifnum_iters>0:
cmd.append("--num-iters={0}".format(num_iters))
ifsample_time>0:
cmd.append("--sample-time={0}".format(sample_time))
ifverbose:
cmd.append("--verbose")
ifmeasure_memory:
cmd.append("--memory")
# TODO: Uncomment this as soon as the new Benchmark Swift logic is available everywhere
# cmd.append("--json")
returncmd

defrun_independent_samples(self, test):
"""Run benchmark multiple times, gathering independent samples.

 Returns the aggregated result of independent benchmark invocations.
 """

defmerge_results(a, b):
a.merge(b)
returna

returnfunctools.reduce(
merge_results,
 [
self.run(test, measure_memory=True, num_iters=1)
for_inrange(self.args.independent_samples)
 ],
 )

deflog_results(self, results, log_file=None):
"""Log output to `log_file`.

 Creates `args.output_dir` if it doesn't exist yet.
 """
log_file=log_fileorself.log_file
dir=os.path.dirname(log_file)
ifnotos.path.exists(dir):
os.makedirs(dir)
print("Logging results to: %s"%log_file)
withopen(log_file, "w") asf:
forrinresults:
print(r, file=f)

RESULT="{:>3} {:<40} {:>7} {:>7} {:>6} {:>10} {:>6} {:>7} {:>10}"

defrun_and_log(self, csv_console=True):
"""Run benchmarks and continuously log results to the console.

 There are two console log formats: CSV and justified columns. Both are
 compatible with `LogParser`. Depending on the `csv_console` parameter,
 the CSV log format is either printed to console or returned as a string
 from this method. When `csv_console` is False, the console output
 format is justified columns.
 """
format= (
 (lambdavalues: ",".join(values))
ifcsv_console
else (lambdavalues: self.RESULT.format(*values))
 ) # justified columns

defconsole_log(values):
print(format(values))

defsummary(r):
returnlist(
map(
str,
 [
r.test_num,
r.name,
r.num_samples,
r.min_value,
r.q1,
r.median,
r.q3,
r.max_value,
r.max_rss,
 ],
 )
 )

summary_header= [
"#",
"TEST",
"SAMPLES",
"MIN(μs)",
"Q1(μs)",
"MEDIAN(μs)",
"Q3(μs)",
"MAX(μs)",
"MAX_RSS(B)",
 ]
console_log(summary_header)
results= []
fortestinself.tests:
result=self.run_independent_samples(test)
console_log(summary(result))
results.append(result)

print("\nTotal performance tests executed: {0}".format(len(self.tests)))
returnresults

@staticmethod
defrun_benchmarks(args):
"""Run benchmarks and log results."""
driver=BenchmarkDriver(args)
results=driver.run_and_log(csv_console=(args.output_dirisNone))
ifargs.output_dir:
driver.log_results([r.jsonforrinresults])
return0


classLoggingReportFormatter(logging.Formatter):
"""Format logs as plain text or with colors on the terminal.

 Plain text outputs level, category and massage: 'DEBUG category: Hi!'
 Colored output uses color coding based on the level.
 """

importloggingaslog

colors= {
log.DEBUG: "9",
log.INFO: "2",
log.WARNING: "3",
log.ERROR: "1",
log.CRITICAL: "5",
 }

def__init__(self, use_color=False):
"""Specify if report should use colors; defaults to False."""
super(LoggingReportFormatter, self).__init__("%(message)s")
self.use_color=use_color

defformat(self, record):
"""Format the log record with level and category."""
msg=super(LoggingReportFormatter, self).format(record)
category= (record.name.split(".")[-1] +": ") if"."inrecord.nameelse""
return (
"\033[1;3{0}m{1}{2}\033[1;0m".format(
self.colors[record.levelno], category, msg
 )
ifself.use_color
else"{0} {1}{2}".format(record.levelname, category, msg)
 )


classMarkdownReportHandler(logging.StreamHandler):
r"""Write custom formatted messages from BenchmarkDoctor to the stream.

 It works around StreamHandler's hardcoded '\n' and handles the custom
 level and category formatting for BenchmarkDoctor's check report.
 """

def__init__(self, stream):
"""Initialize the handler and write a Markdown table header."""
super(MarkdownReportHandler, self).__init__(stream)
self.setLevel(logging.INFO)
self.stream.write("\n✅ | Benchmark Check Report\n---|---")
self.stream.flush()

levels= {
logging.WARNING: "\n⚠️",
logging.ERROR: "\n⛔️",
logging.INFO: " <br><sub> ",
 }
categories= {"naming": "🔤", "runtime": "⏱", "memory": "Ⓜ️"}
quotes_re=re.compile("'")

defformat(self, record):
msg=super(MarkdownReportHandler, self).format(record)
return (
self.levels.get(record.levelno, "")
+ (
""
ifrecord.levelno==logging.INFO
elseself.categories.get(record.name.split(".")[-1], "") +" | "
 )
+self.quotes_re.sub("`", msg)
 )

defemit(self, record):
msg=self.format(record)
stream=self.stream
try:
ifisinstance(msg, str) and (getattr(stream, "encoding", None) and
getattr(stream, "buffer", None)):
stream.buffer.write(msg.encode(stream.encoding))
else:
stream.write(msg)
exceptUnicodeError:
stream.write(msg.encode("UTF-8"))
self.flush()

defclose(self):
self.stream.write("\n\n")
self.stream.flush()
super(MarkdownReportHandler, self).close()


classBenchmarkDoctor(object):
"""Checks that the benchmark conforms to the standard set of requirements.

 Benchmarks that are part of Swift Benchmark Suite are required to follow
 a set of rules that ensure quality measurements. These include naming
 convention, robustness when varying execution parameters like
 `num-iters` and `num-samples` (no setup overhead, constant memory
 consumption).
 """

log=logging.getLogger("BenchmarkDoctor")
log_naming=log.getChild("naming")
log_runtime=log.getChild("runtime")
log_memory=log.getChild("memory")
log.setLevel(logging.DEBUG)

def__init__(self, args, driver=None):
"""Initialize with command line parameters.

 Optional `driver` parameter for injecting dependency; used for testing.
 """
super(BenchmarkDoctor, self).__init__()
self.results= {}

ifhasattr(args, "markdown") andargs.markdown:
self.console_handler=MarkdownReportHandler(sys.stdout)
else:
self.console_handler=logging.StreamHandler(sys.stdout)
self.console_handler.setFormatter(
LoggingReportFormatter(use_color=sys.stdout.isatty())
 )
self.console_handler.setLevel(
logging.DEBUGifargs.verboseelselogging.INFO
 )
self.driver=driverorBenchmarkDriver(args)
self.log.addHandler(self.console_handler)
self.log.debug("Checking tests: %s", ", ".join(self.driver.tests))
self.requirements= [
self._name_matches_benchmark_naming_convention,
self._name_is_at_most_40_chars_long,
self._no_setup_overhead,
self._reasonable_setup_time,
self._optimized_runtime_in_range,
self._constant_memory_use,
 ]

def__del__(self):
"""Close log handlers on exit."""
forhandlerinlist(self.log.handlers):
handler.close()
self.log.removeHandler(self.console_handler)

benchmark_naming_convention_re=re.compile(r"[A-Z][a-zA-Z0-9\-.!?]+")
camel_humps_re=re.compile(r"[a-z][A-Z]")

@staticmethod
def_name_matches_benchmark_naming_convention(measurements):
name=measurements["name"]
match=BenchmarkDoctor.benchmark_naming_convention_re.match(name)
matched=match.group(0) ifmatchelse""
composite_words=len(BenchmarkDoctor.camel_humps_re.findall(name)) +1

ifname!=matched:
BenchmarkDoctor.log_naming.error(
"'%s' name doesn't conform to benchmark naming convention.", name
 )
BenchmarkDoctor.log_naming.info("See http://bit.ly/BenchmarkNaming")

ifcomposite_words>4:
BenchmarkDoctor.log_naming.warning(
"'%s' name is composed of %d words.", name, composite_words
 )
BenchmarkDoctor.log_naming.info(
"Split '%s' name into dot-separated groups and variants. "
"See http://bit.ly/BenchmarkNaming",
name,
 )

@staticmethod
def_name_is_at_most_40_chars_long(measurements):
name=measurements["name"]

iflen(name) >40:
BenchmarkDoctor.log_naming.error(
"'%s' name is %d characters long.", name, len(name)
 )
BenchmarkDoctor.log_naming.info(
"Benchmark name should not be longer than 40 characters."
 )

@staticmethod
def_select(measurements, num_iters=None, opt_level="O"):
prefix=measurements["name"] +" "+opt_level
prefix+=""ifnum_itersisNoneelse (" i"+str(num_iters))
return [
seriesforname, seriesinmeasurements.items() ifname.startswith(prefix)
 ]

@staticmethod
def_optimized_runtime_in_range(measurements):
name=measurements["name"]
setup, ratio=BenchmarkDoctor._setup_overhead(measurements)
setup=0ifratio<0.05elsesetup

runtimes= []
foriinrange(1, 3):
correction=setup/i
i_series=BenchmarkDoctor._select(measurements, num_iters=i)
forresultini_series:
runtimes.append(result.min_value-correction)
runtime=min(runtimes)

threshold=1000
ifthreshold<runtime:
log= (
BenchmarkDoctor.log_runtime.warning
ifruntime<10000
elseBenchmarkDoctor.log_runtime.error
 )
caveat=""ifsetup==0else" (excluding the setup overhead)"
log("'%s' execution took at least %d μs%s.", name, runtime, caveat)

deffactor(base): # suitable divisor that's integer power of base
returnint(
pow(base, math.ceil(math.log(runtime/float(threshold), base)))
 )

BenchmarkDoctor.log_runtime.info(
"Decrease the workload of '%s' by a factor of %d (%d), to be "
"less than %d μs.",
name,
factor(2),
factor(10),
threshold,
 )

threshold=20
ifruntime<threshold:
log= (
BenchmarkDoctor.log_runtime.error
ifruntime==0
elseBenchmarkDoctor.log_runtime.warning
 )
log("'%s' execution took %d μs.", name, runtime)

BenchmarkDoctor.log_runtime.info(
"Ensure the workload of '%s' has a properly measurable size"
" (runtime > %d μs) and is not eliminated by the compiler (use"
" `blackHole` function if necessary)."
ifruntime==0
else"Increase the workload of '%s' to be more than %d μs.",
name,
threshold,
 )

@staticmethod
def_setup_overhead(measurements):
select=BenchmarkDoctor._select
ti1, ti2= [
float(min(mins))
forminsin [
 [result.min_valueforresultini_series]
fori_seriesin [select(measurements, num_iters=i) foriin [1, 2]]
 ]
 ]
setup=int(round(2.0* (ti1-ti2))) ifti2>20else0# limit of accuracy
ratio= (setup/ti1) ifti1>0else0
return (setup, ratio)

@staticmethod
def_no_setup_overhead(measurements):
setup, ratio=BenchmarkDoctor._setup_overhead(measurements)
ifratio>0.05:
BenchmarkDoctor.log_runtime.error(
"'%s' has setup overhead of %d μs (%.1f%%).",
measurements["name"],
setup,
round((100*ratio), 1),
 )
BenchmarkDoctor.log_runtime.info(
"Move initialization of benchmark data to the `setUpFunction` "
"registered in `BenchmarkInfo`."
 )

@staticmethod
def_reasonable_setup_time(measurements):
setup=min(
 [result.setupor0forresultinBenchmarkDoctor._select(measurements)]
 )
if200000<setup: # 200 ms
BenchmarkDoctor.log_runtime.error(
"'%s' setup took at least %d μs.", measurements["name"], setup
 )
BenchmarkDoctor.log_runtime.info(
"The `setUpFunction` should take no more than 200 ms."
 )

@staticmethod
def_constant_memory_use(measurements):
select=BenchmarkDoctor._select
name=measurements["name"]

memory_uses= [
 [r.mem_pagesforrini_seriesifr.mem_pagesisnotNone]
fori_seriesin [select(measurements, num_iters=i) foriin [1, 2]]
 ]
memory_uses= [mforminmemory_usesifm]
ifnotmemory_uses:
BenchmarkDoctor.log_memory.info(
"unable to compute memory footprint of '%s'",
name,
 )
return

 (min_i1, max_i1), (min_i2, max_i2) = [
 (min(memory_use), max(memory_use))
formemory_useinmemory_uses
 ]
range_i1, range_i2=max_i1-min_i1, max_i2-min_i2
normal_range=15# pages
more_info=False

ifabs(min_i1-min_i2) >max(range_i1, range_i2, normal_range):
more_info=True
BenchmarkDoctor.log_memory.error(
"'%s' varies the memory footprint of the base "
"workload depending on the `num-iters`.",
name,
 )

ifmax(range_i1, range_i2) >normal_range:
more_info=True
BenchmarkDoctor.log_memory.warning(
"'%s' has very wide range of memory used between "
"independent, repeated measurements.",
name,
 )

ifmore_info:
BenchmarkDoctor.log_memory.info(
"'%s' mem_pages [i1, i2]: min=[%d, %d] 𝚫=%d R=[%d, %d]",
name,
*[min_i1, min_i2, abs(min_i1-min_i2), range_i1, range_i2]
 )

@staticmethod
def_adjusted_1s_samples(runtime):
u"""Return sample count that can be taken in approximately 1 second.

 Based on the runtime (μs) of one sample taken with num-iters=1.
 """
ifruntime==0:
return2
s=1000000/float(runtime) # samples for 1s run
s=int(pow(2, round(math.log(s, 2)))) # rounding to power of 2
returnsifs>2else2# always take at least 2 samples

defmeasure(self, benchmark):
"""Measure benchmark with varying iterations and optimization levels.

 Returns a dictionary with benchmark name and `PerformanceTestResult`s.
 """
self.log.debug("Calibrating num-samples for {0}:".format(benchmark))
r=self.driver.run(
benchmark, num_samples=3, num_iters=1, verbose=True
 ) # calibrate
num_samples=self._adjusted_1s_samples(r.min_value)

defcapped(s):
returnmin(s, 200)

run_args= [(capped(num_samples), 1), (capped(num_samples/2), 2)]
opts=self.driver.args.optimization
opts=optsifisinstance(opts, list) else [opts]
self.log.debug(
"Runtime {0} μs yields {1} adjusted samples per second.".format(
r.min_value, num_samples
 )
 )
self.log.debug(
"Measuring {0}, 5 x i1 ({1} samples), 5 x i2 ({2} samples)".format(
benchmark, run_args[0][0], run_args[1][0]
 )
 )

measurements=dict(
 [
 (
"{0} {1} i{2}{3}".format(benchmark, o, i, suffix),
self.driver.run(
benchmark,
num_samples=s,
num_iters=i,
verbose=True,
measure_memory=True,
 ),
 )
foroinopts
fors, iinrun_args
forsuffixinlist("abcde")
 ]
 )
measurements["name"] =benchmark
returnmeasurements

defanalyze(self, benchmark_measurements):
"""Analyze whether benchmark fulfills all requirements."""
self.log.debug("Analyzing %s", benchmark_measurements["name"])
forruleinself.requirements:
rule(benchmark_measurements)

defcheck(self):
"""Measure and analyse all enabled tests."""
fortestinself.driver.tests:
self.analyze(self.measure(test))

@staticmethod
defrun_check(args):
"""Validate benchmarks conform to health rules, report violations."""
doctor=BenchmarkDoctor(args)
doctor.check()
# TODO non-zero error code when errors are logged
# See https://stackoverflow.com/a/31142078/41307
return0


defformat_name(log_path):
"""Return the filename and directory for a log file."""
return"/".join(log_path.split("/")[-2:])


defcompare_logs(compare_script, new_log, old_log, log_dir, opt):
"""Return diff of log files at paths `new_log` and `old_log`."""
print("Comparing %s %s ..."% (format_name(old_log), format_name(new_log)))
subprocess.call(
 [
compare_script,
"--old-file",
old_log,
"--new-file",
new_log,
"--format",
"markdown",
"--output",
os.path.join(log_dir, "latest_compare_{0}.md".format(opt)),
 ]
 )


defcompare(args):
log_dir=args.log_dir
compare_script=args.compare_script
baseline_branch=args.baseline_branch
current_branch=BenchmarkDriver(args, tests=[""])._git(
"rev-parse --abbrev-ref HEAD"
 )
current_branch_dir=os.path.join(log_dir, current_branch)
baseline_branch_dir=os.path.join(log_dir, baseline_branch)

ifcurrent_branch!=baseline_branchandnotos.path.isdir(baseline_branch_dir):
print(
 (
"Unable to find benchmark logs for {baseline_branch} branch. "
+"Set a baseline benchmark log by passing --benchmark to "
+"build-script while on {baseline_branch} branch."
 ).format(baseline_branch=baseline_branch)
 )
return1

recent_logs= {}
forbranch_dirin [current_branch_dir, baseline_branch_dir]:
foroptin ["O", "Onone"]:
recent_logs[os.path.basename(branch_dir) +"_"+opt] =sorted(
glob.glob(os.path.join(branch_dir, "Benchmark_"+opt+"-*.log")),
key=os.path.getctime,
reverse=True,
 )

ifcurrent_branch==baseline_branch:
if (
len(recent_logs[baseline_branch+"_O"]) >1
andlen(recent_logs[baseline_branch+"_Onone"]) >1
 ):
compare_logs(
compare_script,
recent_logs[baseline_branch+"_O"][0],
recent_logs[baseline_branch+"_O"][1],
log_dir,
"O",
 )
compare_logs(
compare_script,
recent_logs[baseline_branch+"_Onone"][0],
recent_logs[baseline_branch+"_Onone"][1],
log_dir,
"Onone",
 )
else:
print(
 (
"{baseline_branch}/{baseline_branch} comparison "
+"skipped: no previous {baseline_branch} logs"
 ).format(baseline_branch=baseline_branch)
 )
else:
# TODO: Check for outdated baseline branch log
if (
len(recent_logs[current_branch+"_O"]) ==0
orlen(recent_logs[current_branch+"_Onone"]) ==0
 ):
print("branch sanity failure: missing branch logs")
return1

if (
len(recent_logs[current_branch+"_O"]) ==1
orlen(recent_logs[current_branch+"_Onone"]) ==1
 ):
print("branch/branch comparison skipped: no previous branch logs")
else:
compare_logs(
compare_script,
recent_logs[current_branch+"_O"][0],
recent_logs[current_branch+"_O"][1],
log_dir,
"O",
 )
compare_logs(
compare_script,
recent_logs[current_branch+"_Onone"][0],
recent_logs[current_branch+"_Onone"][1],
log_dir,
"Onone",
 )

if (
len(recent_logs[baseline_branch+"_O"]) ==0
orlen(recent_logs[baseline_branch+"_Onone"]) ==0
 ):
print(
 (
"branch/{baseline_branch} failure: no {baseline_branch} "+"logs"
 ).format(baseline_branch=baseline_branch)
 )
return1
else:
compare_logs(
compare_script,
recent_logs[current_branch+"_O"][0],
recent_logs[baseline_branch+"_O"][0],
log_dir,
"O",
 )
compare_logs(
compare_script,
recent_logs[current_branch+"_Onone"][0],
recent_logs[baseline_branch+"_Onone"][0],
log_dir,
"Onone",
 )

# TODO: Fail on large regressions

return0


defpositive_int(value):
"""Verify the value is a positive integer."""
ivalue=int(value)
ifnot (ivalue>0):
raiseValueError
returnivalue


defparse_args(args):
"""Parse command line arguments and set default values."""
parser=argparse.ArgumentParser(
epilog="Example: ./Benchmark_Driver run -i 5 -f Prefix -f .*Suffix.*"
 )
subparsers=parser.add_subparsers(
title="Swift benchmark driver commands",
help="See COMMAND -h for additional arguments",
metavar="COMMAND",
 )
subparsers.required=True

shared_benchmarks_parser=argparse.ArgumentParser(add_help=False)
benchmarks_group=shared_benchmarks_parser.add_mutually_exclusive_group()
benchmarks_group.add_argument(
"benchmarks",
default=[],
help="benchmark to run (default: all)",
nargs="*",
metavar="BENCHMARK",
 )
benchmarks_group.add_argument(
"-f",
"--filter",
dest="filters",
action="append",
help="run all tests whose name match regular expression PATTERN, "
+"multiple filters are supported",
metavar="PATTERN",
 )
shared_benchmarks_parser.add_argument(
"-t",
"--tests",
help="directory containing Benchmark_O{,none,size} "+"(default: DRIVER_DIR)",
default=DRIVER_DIR,
 )
shared_benchmarks_parser.add_argument(
"-o",
"--optimization",
metavar="OPT",
choices=["O", "Onone", "Osize"],
help="optimization level to use: {O,Onone,Osize}, (default: O)",
default="O",
 )
shared_benchmarks_parser.add_argument(
"--architecture",
metavar="architecture",
help="current architecture (e.g., x86_64, arm64, etc)",
default=None,
 )

run_parser=subparsers.add_parser(
"run",
help="Run benchmarks and output results to stdout",
parents=[shared_benchmarks_parser],
 )
run_parser.add_argument(
"-i",
"--independent-samples",
help="number of times to run each test (default: 1)",
type=positive_int,
default=1,
 )
run_parser.add_argument(
"--output-dir", help="log results to directory (default: no logging)"
 )
run_parser.add_argument(
"--swift-repo", help="absolute path to the Swift source repository"
 )
run_parser.set_defaults(func=BenchmarkDriver.run_benchmarks)

check_parser=subparsers.add_parser(
"check", help="", parents=[shared_benchmarks_parser]
 )
check_group=check_parser.add_mutually_exclusive_group()
check_group.add_argument(
"-v",
"--verbose",
action="store_true",
help="show more details during benchmark analysis",
 )
check_group.add_argument(
"-md", "--markdown", action="store_true", help="format report as Markdown table"
 )
check_parser.set_defaults(func=BenchmarkDoctor.run_check)

compare_parser=subparsers.add_parser("compare", help="Compare benchmark results")
compare_parser.add_argument(
"--log-dir", required=True, help="directory containing benchmark logs"
 )
compare_parser.add_argument(
"--swift-repo",
required=True,
help="absolute path to the Swift source repository",
 )
compare_parser.add_argument(