def inspect_server_benchmarks(server: "Server") -> List[dict]:
"""Generate a list of BenchmarkScore-like dicts for the Server."""
benchmarks = []
framework = "bogomips"
try:
benchmarks.append(
{
**_benchmark_metafields(
server, framework="lscpu", benchmark_id=framework
),
"score": round(float(_server_lscpu_field(server, "BogoMIPS:"))),
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e)
framework = "bw_mem"
try:
with open(_server_framework_stdout_path(server, framework), "r") as lines:
for line in lines:
# filter out error messages
if match(r"^(rd|wr|rdwr) \d+(\.\d+) \d+(\.\d+)$", line):
row = line.strip().split()
benchmarks.append(
{
**_benchmark_metafields(server, framework=framework),
"config": {"operation": row[0], "size": float(row[1])},
"score": float(row[2]),
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e)
framework = "compression_text"
try:
algos = _server_framework_stdout_from_json(server, framework)
for algo, levels in algos.items():
for level, datas in levels.items():
for data in datas:
threads = data["threads"]
config = {
"algo": algo,
"compression_level": None if level == "null" else int(level),
"cores": (
Parallelism.MULTI if threads > 1 else Parallelism.SINGLE
),
}
if data.get("extra_args", {}).get("block_size") is not None:
config["block_size"] = data["extra_args"]["block_size"]
for measurement in ["ratio", "compress", "decompress"]:
if data[measurement]:
benchmarks.append(
{
**_benchmark_metafields(
server,
benchmark_id=":".join([framework, measurement]),
),
"config": config,
"score": float(data[measurement]),
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e, True)
framework = "geekbench"
try:
with open(_server_framework_path(server, framework, "results.json"), "r") as fp:
scores = json.load(fp)
kernel_version = next(
(
m.group(1)
for line in _server_geekbench(server)
if line.startswith("Kernel")
for m in [search(r"(\d+\.\d+\.\d+\S*)", line)]
if m
),
None,
)
for cores, workloads in scores.items():
parallelism = (
Parallelism.SINGLE
if cores == "Single-Core Performance"
else Parallelism.MULTI
)
for workload, values in workloads.items():
workload_fields = {
"config": {"cores": parallelism},
"score": float(values["score"]),
}
if values.get("description"):
workload_fields["note"] = values["description"]
benchmarks.append(
{
**_benchmark_metafields(
server,
benchmark_id=":".join(
[framework, sub(r"\W+", "_", workload.lower())]
),
kernel_version_fallback=kernel_version,
),
**workload_fields,
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e, True)
framework = "passmark"
try:
with open(_server_framework_stdout_path(server, framework), "r") as fp:
scores = yaml_safe_load(fp)
passmark_version = ".".join(
[str(scores["Version"][i]) for i in ["Major", "Minor", "Build"]]
)
kernel_version = next(
(
line.split("Kernel:")[1].strip()
for line in _server_passmark(server)
if line.startswith("Kernel:")
),
None,
)
for key, name in PASSMARK_MAPS.items():
benchmarks.append(
{
**_benchmark_metafields(
server,
benchmark_id=":".join(
[framework, sub(r"\W+", "_", name.lower())]
),
framework_version_fallback=passmark_version,
kernel_version_fallback=kernel_version,
),
"score": float(scores["Results"][key]),
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e, True)
framework = "openssl"
try:
with open(_server_framework_path(server, framework, "parsed.json"), "r") as fp:
workloads = json.load(fp)
for workload in workloads:
benchmarks.append(
{
**_benchmark_metafields(server, framework=framework),
"config": {
"algo": workload["algo"],
"block_size": workload["block_size"],
},
"score": float(workload["speed"]),
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e, True)
framework = "stress_ng"
# TODO deprecate
try:
cores_per_path = {"stressng": server.vcpus, "stressngsinglecore": 1}
for cores_path in cores_per_path.keys():
line = _extract_line_from_file(
_server_framework_stderr_path(server, cores_path),
"bogo-ops-per-second-real-time",
)
benchmarks.append(
{
**_benchmark_metafields(
server,
framework=cores_path,
benchmark_id=":".join([framework, "cpu_all"]),
),
"config": {"cores": cores_per_path[cores_path]},
"score": float(line.split(": ")[1]),
}
)
except Exception:
# backfill with newer method - can be dropped once we deprecate stress_ng:cpu_all
try:
records = []
with open(
_server_framework_stdout_path(server, "stressngfull"), newline=""
) as f:
rows = csv.reader(f, quoting=csv.QUOTE_NONNUMERIC)
for row in rows:
records.append(row)
for i in [0, len(records) - 1]:
benchmarks.append(
{
**_benchmark_metafields(
server,
framework="stressngfull",
benchmark_id=":".join([framework, "cpu_all"]),
),
"config": {"cores": records[i][0]},
"score": records[i][1],
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e, True)
workload = "div16"
try:
records = []
with open(
_server_framework_stdout_path(server, "stressngfull"), newline=""
) as f:
rows = csv.reader(f, quoting=csv.QUOTE_NONNUMERIC)
for row in rows:
records.append(row)
for record in records:
benchmarks.append(
{
**_benchmark_metafields(
server,
framework="stressngfull",
benchmark_id=":".join([framework, workload]),
),
"config": {"cores": record[0]},
"score": record[1],
}
)
# best single and multi core performance
bests = {"best1": records[0][1], "bestn": max([r[1] for r in records])}
for k, v in bests.items():
benchmarks.append(
{
**_benchmark_metafields(
server,
framework="stressngfull",
benchmark_id=":".join([framework, k]),
),
"score": v,
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e, True)
for framework in SERVER_CLIENT_FRAMEWORK_MAPS.keys():
try:
versions = _server_framework_meta(server, framework)["version"]
# drop the build number at the end of the redis server version
if framework == "redis":
versions = sub(r" build=[a-zA-Z0-9]+", "", versions)
records = []
with open(
_server_framework_stdout_path(server, framework), newline=""
) as f:
rows = csv.DictReader(f, quoting=csv.QUOTE_NONNUMERIC)
for row in rows:
if "connections" in row.keys():
row["connections_per_vcpus"] = row["connections"] / server.vcpus
records.append(row)
framework_config = SERVER_CLIENT_FRAMEWORK_MAPS[framework]
keys = framework_config["keys"]
measurements = framework_config["measurements"]
# don't care about threads, keep the records with the highest rps
records = sorted(records, key=lambda x: (*[x[k] for k in keys], -x["rps"]))
records = groupby(records, key=itemgetter(*keys))
records = [next(group) for _, group in records]
for record in records:
for measurement in measurements:
score_field = measurement.split("-")[0]
if score_field == "throughput":
score_field = "rps"
score = record[score_field]
server_usrsys = record["server_usr"] + record["server_sys"]
client_usrsys = record["client_usr"] + record["client_sys"]
note = (
"CPU usage (server/client usr+sys): "
f"{round(server_usrsys, 4)}/{round(client_usrsys, 4)}."
)
if measurement.endswith("-extrapolated"):
note += f" Original RPS: {score}."
score = round(
score / server_usrsys * (server_usrsys + client_usrsys), 2
)
if measurement.startswith("throughput"):
# drop the "k" suffix and multiply by 1024
size = int(record["size"][:-1]) * 1024
score = score * size
benchmarks.append(
{
**_benchmark_metafields(
server,
framework=framework,
benchmark_id=":".join([framework, measurement]),
framework_version_fallback=versions,
override_framework_version=True,
),
"config": {k: record[k] for k in keys},
"score": score,
"note": note,
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e, True)
framework = "membench"
try:
with open(_server_framework_stdout_path(server, framework), newline="") as fp:
reader = csv.DictReader(fp)
ram_scope_dict = {
"read": False,
"write": False,
"copy": False,
"latency": False,
}
cpu_cache_total_kib = (
server.cpu_l3_cache_total
or server.cpu_l2_cache_total
or server.cpu_l1d_cache_total
)
for row in reader:
operation = row["operation"]
size_kb = int(float(row["size_kb"]))
config = {"size_kb": size_kb}
if operation == "latency":
score = float(row["latency_ns"])
if score == 0:
continue
benchmarks.append(
{
**_benchmark_metafields(
server,
framework=framework,
benchmark_id="membench:latency",
),
"config": config,
"score": score,
}
)
if (
cpu_cache_total_kib is not None
and size_kb > cpu_cache_total_kib
and not ram_scope_dict[operation]
):
ram_scope_dict[operation] = True
benchmarks.append(
{
**_benchmark_metafields(
server,
framework=framework,
benchmark_id="membench:latency",
),
"config": {"scope": "RAM"},
"score": score,
}
)
elif operation in ("read", "write", "copy"):
score = float(row["bandwidth_mb_s"])
if score == 0:
continue
benchmarks.append(
{
**_benchmark_metafields(
server,
framework=framework,
benchmark_id=f"membench:bandwidth_{operation}",
),
"config": config,
"score": score,
}
)
if (
cpu_cache_total_kib is not None
and size_kb > cpu_cache_total_kib
and not ram_scope_dict[operation]
):
ram_scope_dict[operation] = True
benchmarks.append(
{
**_benchmark_metafields(
server,
framework=framework,
benchmark_id=f"membench:bandwidth_{operation}",
),
"config": {"scope": "RAM"},
"score": score,
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e, True)
framework = "llm_speed"
try:
assert _server_framework_meta(server, "llm")["exit_code"] == 0
with open(_server_framework_stdout_path(server, "llm"), "r") as fp:
for line in fp:
record = json.loads(line)
model_name = path.basename(record.get("model_filename", "unknown"))
measurement = "text_generation"
if record.get("n_prompt") != 0:
measurement = "prompt_processing"
tokens = record.get("n_prompt") + record.get("n_gen")
config = {
"model": model_name,
"tokens": tokens,
}
benchmarks.append(
{
**_benchmark_metafields(
server,
framework="llm",
benchmark_id=":".join([framework, measurement]),
),
"config": config,
"score": float(record["avg_ts"]),
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e, True)
framework = "nvbandwidth"
try:
with open(_server_framework_stdout_path(server, framework), "r") as fp:
data = json.load(fp)["nvbandwidth"]
# aborted runs (e.g. failed pattern check) still list the testcases
# passed before the error, but the whole run is untrustworthy
if data.get("error"):
raise ValueError(f"nvbandwidth run aborted: {data['error']}")
passed = {
testcase["name"]: testcase
for testcase in data["testcases"]
if testcase.get("status") == "Passed"
}
gpu_count = 0
for testcase in passed.values():
matrix = testcase.get("bandwidth_matrix")
if matrix:
# columns are GPUs in all testcases (rows can be stacked blocks)
gpu_count = max(gpu_count, len(matrix[0]))
extend_environment = {
"driver_version": data["Driver Version"],
"cuda_version": data["CUDA Runtime Version"],
"nvbandwidth_version": data["version"],
"gpu_count": gpu_count,
"p2p_supported": any(
name.startswith("device_to_device_") for name in passed
),
}
for metric, source in (
("host_to_gpu", "host_to_all_memcpy_ce"),
("gpu_to_host", "all_to_host_memcpy_ce"),
):
if source not in passed:
continue
benchmarks.append(
{
**_benchmark_metafields(
server,
framework=framework,
benchmark_id=":".join([framework, "all", metric]),
extend_environment=extend_environment,
),
"score": passed[source]["sum"],
}
)
duplex_sources = [
"host_to_all_bidirectional_memcpy_ce",
"all_to_host_bidirectional_memcpy_ce",
]
if all(source in passed for source in duplex_sources):
benchmarks.append(
{
**_benchmark_metafields(
server,
framework=framework,
benchmark_id=":".join([framework, "all", "duplex"]),
extend_environment=extend_environment,
),
"score": min(passed[source]["sum"] for source in duplex_sources),
}
)
for group, metric, source in (
("slot", "host_to_gpu", "host_to_device_memcpy_ce"),
("slot", "gpu_to_host", "device_to_host_memcpy_ce"),
("slot", "local_copy", "device_local_copy"),
("slot", "latency", "host_device_latency_sm"),
("p2p", "single", "device_to_device_memcpy_write_ce"),
("p2p", "duplex", "device_to_device_bidirectional_memcpy_write_ce"),
("p2p", "latency", "device_to_device_latency_sm"),
("p2p", "gather", "all_to_one_write_ce"),
):
testcase = passed.get(source)
if testcase is None:
continue
matrix = testcase["bandwidth_matrix"]
directions = 1
if len(matrix) == 3 * len(matrix[0]):
# v0.6+ bidirectional tests stack the two directional blocks and
# their Total, with `sum` covering only the Total block
matrix = matrix[-len(matrix[0]) :]
directions = 2
elif len(matrix) > len(matrix[0]):
_log_cannot_load_benchmarks(
server,
":".join([framework, group, metric]),
f"unexpected {len(matrix)}x{len(matrix[0])} matrix",
)
continue
cell_count = sum(cell != "N/A" for row in matrix for cell in row)
if not cell_count:
continue
benchmarks.append(
{
**_benchmark_metafields(
server,
framework=framework,
benchmark_id=":".join([framework, group, metric]),
extend_environment=extend_environment,
),
"score": testcase["sum"] / cell_count / directions,
}
)
sm = passed.get("host_to_device_memcpy_sm")
ce = passed.get("host_to_device_memcpy_ce")
if sm is not None and ce is not None and ce["sum"]:
benchmarks.append(
{
**_benchmark_metafields(
server,
framework=framework,
benchmark_id=":".join([framework, "efficiency", "sm_ce_ratio"]),
extend_environment=extend_environment,
),
"score": min(sm["sum"] / ce["sum"], 1.0),
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e, True)
framework = "stressng_benchmarks"
try:
# lookup imports tables, which imports this module
from .lookup import stress_ng_benchmark_id
stressors = _server_framework_stdout_from_json(server, framework)["stressors"]
benchmark_metafields = _benchmark_metafields(
server, framework=framework, benchmark_id=framework
)
for method, by_cores in stressors.items():
try:
benchmark_id = stress_ng_benchmark_id(method)
if benchmark_id is None:
_log_cannot_load_benchmarks(
server, framework, f"unknown stressor {method!r}"
)
continue
for cores, result in by_cores.items():
if result is None:
continue
score = float(result["bogo_ops_per_sec"])
if score <= 0:
continue
cores_count = int(cores)
row = {
**benchmark_metafields,
"benchmark_id": benchmark_id,
"config": {
"cores": (
Parallelism.SINGLE
if cores_count == 1
else Parallelism.MULTI
)
},
"score": score,
}
benchmarks.append(row)
# 1-vCPU servers only have a "1" key, also emit MULTI so
# cores=multi filters still include them
if cores_count == 1 and server.vcpus == 1:
benchmarks.append(
{
**row,
"config": {"cores": Parallelism.MULTI},
}
)
except Exception as e:
_log_cannot_load_benchmarks(server, f"{framework}:{method}", e, True)
except Exception as e:
_log_cannot_load_benchmarks(server, framework, e, True)
# add server related pgbench benchmark scores
benchmarks.extend(_pgbench_benchmark_scores(server))
return benchmarks