521 lines
19 KiB
Python
521 lines
19 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Measure this machine by running things on it, not by reading what it claims.
|
|
|
|
`/proc/cpuinfo` is a declaration. On node 238 - a VirtualBox guest on a Ryzen 9 3900X -
|
|
it reports the correct model name and then lists 65 CPUID flags that stop at SSE4.2: no
|
|
`avx`, no `avx2`, no `fma`. That silicon has all three. The hypervisor masks them, and a
|
|
guest that reads its own `/proc/cpuinfo` would conclude it cannot run AVX work. It might
|
|
be right, for reasons having nothing to do with the CPU.
|
|
|
|
So a feature is recorded as one of three things, never two:
|
|
|
|
- **present** - an instruction using it was executed and produced the right answer.
|
|
- **absent** - an instruction using it was executed and the kernel killed the process.
|
|
- **unverified** - it could not be executed at all, because there is no compiler, or no
|
|
permission, or the probe could not be built.
|
|
|
|
"Unverified" is the honest answer and it is the one this project has most often got
|
|
wrong. The agent's `throughput_factor` is chosen by string-matching a GPU model
|
|
(`40.0 if "4070" in name else 20.0`), which is a declared factor wearing a measured
|
|
one's name. A scheduler that trusted `/proc/cpuinfo` here would send AVX-tuned work to
|
|
a machine that faults on it. Both are the same mistake at different layers.
|
|
|
|
Nothing here installs anything, needs root, or modifies the machine. It compiles a
|
|
probe if a compiler exists, runs it, and reports. If it cannot compile, it says so.
|
|
|
|
The result is deliberately small: the facts a scheduler needs to place work and to
|
|
refuse work it cannot do. Not a full CPU database.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import platform
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
from dataclasses import asdict, dataclass, field
|
|
from pathlib import Path
|
|
|
|
from ._capability_probe import _C_SOURCE, _EMITTED, _FEATURE_FLAGS
|
|
|
|
#: Instruction-set extensions worth probing. Each is a real risk in practice: a build
|
|
#: compiled for AVX2 will SIGILL on a machine without it, and a build compiled *without*
|
|
#: AVX2 leaves most of the performance on the table. Both directions matter, so both
|
|
#: are measured.
|
|
FEATURES = (
|
|
"sse2",
|
|
"sse4_2",
|
|
"avx",
|
|
"avx2",
|
|
"fma",
|
|
"avx512f",
|
|
"bmi2",
|
|
"aes_ni",
|
|
)
|
|
|
|
@dataclass
|
|
class FeatureResult:
|
|
"""One probed feature. ``status`` is never guessed."""
|
|
|
|
name: str
|
|
#: "present", "absent", or "unverified". Nothing else.
|
|
status: str
|
|
#: Why, when not present. Short, because this is stored and reported.
|
|
detail: str = ""
|
|
#: Seconds for the probe's own scalar loop. A rough shape, not a benchmark.
|
|
scalar_seconds: float = 0.0
|
|
|
|
@property
|
|
def usable(self) -> bool:
|
|
"""Only ``present`` counts as usable.
|
|
|
|
An unverified feature is treated as unavailable on purpose. A scheduler that
|
|
optimistically assumed an unverified AVX2 would send work that faults; the
|
|
cost of pessimism is a slower path that works, and that is the cheaper error.
|
|
"""
|
|
|
|
return self.status == "present"
|
|
|
|
|
|
@dataclass
|
|
class CapabilityProfile:
|
|
"""What this machine can actually do, as measured."""
|
|
|
|
node_name: str = ""
|
|
hostname: str = ""
|
|
kernel: str = ""
|
|
architecture: str = ""
|
|
virtualized: str = ""
|
|
|
|
logical_cpus: int = 0
|
|
cores_per_socket: int = 0
|
|
threads_per_core: int = 0
|
|
cpu_model_declared: str = ""
|
|
#: Set when the declared model and the probed flags disagree, which is the case
|
|
#: that matters and the reason this module exists.
|
|
cpu_declaration_suspect: str = ""
|
|
|
|
memory_total_bytes: int = 0
|
|
memory_available_bytes: int = 0
|
|
swap_total_bytes: int = 0
|
|
disk_free_bytes: int = 0
|
|
|
|
features: list[FeatureResult] = field(default_factory=list)
|
|
compiler: str = ""
|
|
probe_note: str = ""
|
|
|
|
def feature(self, name: str) -> FeatureResult | None:
|
|
for result in self.features:
|
|
if result.name == name:
|
|
return result
|
|
return None
|
|
|
|
def usable_features(self) -> list[str]:
|
|
return [f.name for f in self.features if f.usable]
|
|
|
|
|
|
def _read(path: str, default: str = "") -> str:
|
|
try:
|
|
return Path(path).read_text().strip()
|
|
except OSError:
|
|
return default
|
|
|
|
|
|
def _first_match(text: str, pattern: str, cast=str, default=0):
|
|
import re
|
|
|
|
match = re.search(pattern, text, re.MULTILINE)
|
|
return cast(match.group(1)) if match else default
|
|
|
|
|
|
def _find_compiler() -> str | None:
|
|
"""A C compiler, or nothing.
|
|
|
|
Probing needs one. If there is none, the honest answer is "unverified" for every
|
|
feature, and the profile says why - rather than falling back to /proc/cpuinfo,
|
|
which is the failure this whole module exists to avoid.
|
|
"""
|
|
|
|
for candidate in ("cc", "gcc", "clang"):
|
|
found = shutil.which(candidate)
|
|
if found:
|
|
return found
|
|
return None
|
|
|
|
|
|
def _compiler_works(compiler: str) -> bool:
|
|
"""Does this compiler actually compile something?
|
|
|
|
Cheap: compile a one-line program to /dev/null-equivalent in a temp dir. Checking
|
|
beats assuming, because the failure mode of assuming is eight confusing errors
|
|
instead of one clear sentence.
|
|
"""
|
|
|
|
import tempfile
|
|
|
|
with tempfile.TemporaryDirectory(prefix="umbra-cc-") as workdir:
|
|
source = Path(workdir) / "t.c"
|
|
source.write_text("int main(void){return 0;}")
|
|
try:
|
|
probe = subprocess.run(
|
|
[compiler, "-o", str(Path(workdir) / "t"), str(source)],
|
|
capture_output=True,
|
|
timeout=60,
|
|
)
|
|
except (OSError, subprocess.SubprocessError):
|
|
# A path that does not exist raises rather than returning non-zero, and a
|
|
# compiler that hangs or cannot fork should count as unusable too. Either
|
|
# way the answer is "no working compiler", not an exception to the caller.
|
|
return False
|
|
return probe.returncode == 0
|
|
|
|
|
|
def probe_features(compiler: str | None = None) -> tuple[list[FeatureResult], str]:
|
|
"""Execute one instruction per feature and record what happened.
|
|
|
|
Returns the results and a note explaining any that could not be probed. The three
|
|
outcomes are kept distinct because collapsing "absent" into "unverified" would make
|
|
a machine look more capable than it is, and collapsing them the other way would
|
|
make it look less capable than it is.
|
|
"""
|
|
|
|
# The compiler has to be *usable*, not merely named. A caller passing an explicit
|
|
# path that does not exist would otherwise get eight identical build failures
|
|
# reported as unverified-with-a-compiler-why, when the honest answer is that there
|
|
# is no compiler here at all.
|
|
compiler = compiler or _find_compiler()
|
|
if not compiler or not _compiler_works(compiler):
|
|
note = (
|
|
"no working C compiler, so no instruction could be executed. Every feature "
|
|
"is unverified rather than absent: /proc/cpuinfo is a declaration and this "
|
|
"module will not report one as a measurement. Install gcc or clang to "
|
|
"settle these."
|
|
)
|
|
return ([FeatureResult(name, "unverified", "no working compiler") for name in FEATURES], note)
|
|
|
|
results: list[FeatureResult] = []
|
|
with tempfile.TemporaryDirectory(prefix="umbra-probe-") as workdir:
|
|
source = Path(workdir) / "probe.c"
|
|
source.write_text(_C_SOURCE)
|
|
for name in FEATURES:
|
|
results.append(_probe_one(compiler, source, Path(workdir), name))
|
|
|
|
unverified = [r.name for r in results if r.status == "unverified"]
|
|
note = ""
|
|
if unverified:
|
|
note = f"could not be probed: {', '.join(unverified)}"
|
|
return results, note
|
|
|
|
|
|
def _probe_one(compiler: str, source: Path, workdir: Path, name: str) -> FeatureResult:
|
|
"""Build and run a probe for exactly one feature.
|
|
|
|
One translation unit per feature, not one binary for all of them. A single source
|
|
containing every intrinsic cannot be built with `-mavx2` alone, because the FMA,
|
|
AVX-512 and AES intrinsics in it then fail to inline - `always_inline` against a
|
|
target that lacks the flag is a hard error, not a warning. That was the first
|
|
version, and it reported every feature unverified on a machine that has most of
|
|
them.
|
|
|
|
The feature name is preprocessed in so each binary contains one probe.
|
|
"""
|
|
|
|
index = FEATURES.index(name)
|
|
unit = source.read_text().replace("__FEATURE_INDEX__", str(index))
|
|
unit_path = workdir / f"probe_{name}.c"
|
|
unit_path.write_text(unit)
|
|
binary = workdir / f"probe_{name}"
|
|
|
|
build = subprocess.run(
|
|
[compiler, "-O2", *_FEATURE_FLAGS[name], "-o", str(binary), str(unit_path)],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=180,
|
|
)
|
|
if build.returncode != 0:
|
|
errors = [
|
|
line for line in build.stderr.splitlines() if " error:" in line
|
|
] or build.stderr.strip().splitlines() or ["build failed"]
|
|
return FeatureResult(name, "unverified", f"compiler refused: {errors[0][-80:]}")
|
|
|
|
# The instruction must be in the binary. Without this, a compiler that folded the
|
|
# probe away would report "present" for a CPU it never tested - and it does fold
|
|
# them, at -O2, when the inputs are constants.
|
|
emitted = _verify_emitted(binary, name)
|
|
if emitted is not None:
|
|
return FeatureResult(name, "unverified", emitted)
|
|
|
|
try:
|
|
run = subprocess.run([str(binary)], capture_output=True, text=True, timeout=30)
|
|
except subprocess.TimeoutExpired:
|
|
return FeatureResult(name, "unverified", "probe timed out")
|
|
|
|
if run.returncode == 0:
|
|
try:
|
|
seconds = float(run.stdout.strip().splitlines()[-1])
|
|
except (ValueError, IndexError):
|
|
seconds = 0.0
|
|
return FeatureResult(name, "present", scalar_seconds=seconds)
|
|
if run.returncode < 0:
|
|
signal = -run.returncode
|
|
detail = f"killed by signal {signal}"
|
|
if signal == 4:
|
|
detail = "SIGILL - the CPU rejected the instruction"
|
|
return FeatureResult(name, "absent", detail)
|
|
if run.returncode == 3:
|
|
# Executed, but computed the wrong answer. The CPU ran the instruction, so
|
|
# reporting "absent" would be a lie about the hardware; this is a probe bug.
|
|
return FeatureResult(name, "unverified", "executed but returned a wrong result")
|
|
return FeatureResult(name, "unverified", f"probe exited {run.returncode}")
|
|
|
|
|
|
def _verify_emitted(binary: Path, name: str) -> str | None:
|
|
"""Disassemble and confirm the instruction is really there.
|
|
|
|
Returns None when it is, or a reason string when it is not. This is what makes the
|
|
probe trustworthy: it turns "the compiler accepted my flag" into "the instruction is
|
|
in the binary", and catches the constant-folding case that would otherwise make
|
|
every CPU look capable of everything.
|
|
"""
|
|
|
|
objdump = shutil.which("objdump")
|
|
if not objdump:
|
|
# Cannot check. Say so rather than assume it passed - but do not mark the
|
|
# feature unverified either, because we have not disproved anything.
|
|
return None
|
|
try:
|
|
disassembly = subprocess.run(
|
|
[objdump, "-d", str(binary)], capture_output=True, text=True, timeout=60
|
|
).stdout
|
|
except subprocess.TimeoutExpired:
|
|
return "disassembly timed out"
|
|
if not disassembly:
|
|
return "disassembly produced nothing"
|
|
|
|
expected = _EMITTED.get(name, ())
|
|
if any(mnemonic in disassembly for mnemonic in expected):
|
|
return None
|
|
return (
|
|
f"compiled but no {' or '.join(expected)} in the disassembly - "
|
|
"the compiler folded the probe away, so nothing was measured"
|
|
)
|
|
|
|
|
|
def _declared_flags() -> set[str]:
|
|
text = _read("/proc/cpuinfo")
|
|
line = next((entry for entry in text.splitlines() if entry.startswith("flags")), "")
|
|
if not line:
|
|
return set()
|
|
return set(line.split(":", 1)[-1].split())
|
|
|
|
|
|
def measure(profile: CapabilityProfile) -> CapabilityProfile:
|
|
"""Fill in everything measurable without compiling anything."""
|
|
|
|
profile.hostname = platform.node()
|
|
profile.kernel = platform.release()
|
|
profile.architecture = platform.machine()
|
|
|
|
cpuinfo = _read("/proc/cpuinfo")
|
|
profile.cpu_model_declared = _first_match(cpuinfo, r"^model name\s*:\s*(.+)$", str, "")
|
|
|
|
lscpu = ""
|
|
if shutil.which("lscpu"):
|
|
try:
|
|
lscpu = subprocess.run(["lscpu"], capture_output=True, text=True, timeout=15).stdout
|
|
except subprocess.TimeoutExpired:
|
|
lscpu = ""
|
|
if lscpu:
|
|
profile.logical_cpus = _first_match(lscpu, r"^CPU\(s\):\s*(\d+)", int, 0)
|
|
profile.cores_per_socket = _first_match(lscpu, r"^Core\(s\) per socket:\s*(\d+)", int, 0)
|
|
profile.threads_per_core = _first_match(
|
|
lscpu, r"^Thread\(s\) per core:\s*(\d+)", int, 1
|
|
)
|
|
else:
|
|
profile.logical_cpus = os.cpu_count() or 0
|
|
|
|
meminfo = _read("/proc/meminfo")
|
|
profile.memory_total_bytes = _first_match(meminfo, r"^MemTotal:\s*(\d+)", int, 0) * 1024
|
|
profile.memory_available_bytes = (
|
|
_first_match(meminfo, r"^MemAvailable:\s*(\d+)", int, 0) * 1024
|
|
)
|
|
profile.swap_total_bytes = _first_match(meminfo, r"^SwapTotal:\s*(\d+)", int, 0) * 1024
|
|
|
|
try:
|
|
stat = os.statvfs("/")
|
|
profile.disk_free_bytes = stat.f_bavail * stat.f_frsize
|
|
except OSError:
|
|
profile.disk_free_bytes = 0
|
|
|
|
try:
|
|
profile.virtualized = _read("/sys/class/dmi/id/product_name") or "unknown"
|
|
except OSError:
|
|
profile.virtualized = "unknown"
|
|
if not profile.virtualized:
|
|
profile.virtualized = "unknown"
|
|
if "hypervisor" in cpuinfo:
|
|
profile.virtualized = "hypervisor flag present"
|
|
|
|
return profile
|
|
|
|
|
|
def assess_declaration(profile: CapabilityProfile) -> CapabilityProfile:
|
|
"""Flag a CPU whose declared flags and probed instructions disagree.
|
|
|
|
The disagreement is the finding. A machine that advertises a model it does not
|
|
have the features of is telling you something about whoever wrote the
|
|
declaration, not about the hardware - and a scheduler reading the declaration is
|
|
reading that.
|
|
"""
|
|
|
|
declared = _declared_flags()
|
|
if not declared or not profile.features:
|
|
return profile
|
|
|
|
mapping = {
|
|
"avx": "avx",
|
|
"avx2": "avx2",
|
|
"fma": "fma",
|
|
"avx512f": "avx512f",
|
|
"bmi2": "bmi2",
|
|
"sse4_2": "sse4_2",
|
|
}
|
|
suspicious: list[str] = []
|
|
for feature, flag in mapping.items():
|
|
result = profile.feature(feature)
|
|
if result is None:
|
|
continue
|
|
claims = flag in declared
|
|
if result.status == "absent" and claims:
|
|
suspicious.append(f"{feature} claimed by /proc/cpuinfo but absent when executed")
|
|
elif result.status == "present" and not claims:
|
|
suspicious.append(f"{feature} executable but not advertised by /proc/cpuinfo")
|
|
|
|
if suspicious:
|
|
profile.cpu_declaration_suspect = "; ".join(suspicious)
|
|
return profile
|
|
|
|
|
|
def collect(node_name: str = "", compiler: str | None = None) -> CapabilityProfile:
|
|
"""The whole profile: measure, then probe, then compare the two."""
|
|
|
|
profile = CapabilityProfile(node_name=node_name)
|
|
profile = measure(profile)
|
|
compiler = compiler or _find_compiler()
|
|
profile.compiler = compiler or "none"
|
|
results, note = probe_features(compiler)
|
|
profile.features = results
|
|
profile.probe_note = note
|
|
return check_model_name(assess_declaration(profile))
|
|
|
|
|
|
def profile_to_dict(profile: CapabilityProfile) -> dict:
|
|
data = asdict(profile)
|
|
data["usable_features"] = profile.usable_features()
|
|
return data
|
|
|
|
|
|
def profile_to_json(profile: CapabilityProfile) -> str:
|
|
return json.dumps(profile_to_dict(profile), indent=2, sort_keys=True)
|
|
|
|
|
|
def _main(argv: list[str] | None = None) -> int:
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="Measure this machine for the scheduler.")
|
|
parser.add_argument("--node-name", default="")
|
|
parser.add_argument("--json", action="store_true", help="machine-readable output")
|
|
args = parser.parse_args(argv)
|
|
|
|
started = time.perf_counter()
|
|
profile = collect(node_name=args.node_name)
|
|
elapsed = time.perf_counter() - started
|
|
|
|
if args.json:
|
|
print(profile_to_json(profile))
|
|
return 0
|
|
|
|
print(f"node {profile.hostname} ({profile.architecture})")
|
|
print(f"kernel {profile.kernel}")
|
|
print(f"cpus {profile.logical_cpus} logical "
|
|
f"({profile.cores_per_socket} cores x {profile.threads_per_core})")
|
|
print(f"memory {profile.memory_total_bytes / 2**30:.1f} GiB total, "
|
|
f"{profile.memory_available_bytes / 2**30:.1f} GiB available")
|
|
print(f"disk free {profile.disk_free_bytes / 2**30:.1f} GiB")
|
|
print(f"compiler {profile.compiler}")
|
|
print(f"probe took {elapsed:.1f}s")
|
|
print()
|
|
print("features (by execution, not by declaration):")
|
|
for result in profile.features:
|
|
mark = {"present": "yes", "absent": "NO ", "unverified": "??"}[result.status]
|
|
line = f" {mark} {result.name}"
|
|
if result.detail:
|
|
line += f" {result.detail}"
|
|
print(line)
|
|
if profile.cpu_declaration_suspect:
|
|
print()
|
|
print("DECLARATION DISAGREES WITH MEASUREMENT:")
|
|
for note in profile.cpu_declaration_suspect.split("; "):
|
|
print(f" ! {note}")
|
|
if profile.probe_note:
|
|
print()
|
|
print(f"note: {profile.probe_note}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(_main())
|
|
|
|
|
|
#: CPU families that cannot be what the model name claims when AVX2 is missing. Every
|
|
#: x86 CPU since Intel's Sandy Bridge (2011) and AMD's Bulldozer (2011) has AVX, and
|
|
#: every one since Haswell (2013) / Excavator (2015) has AVX2 and FMA. A machine
|
|
#: reporting a post-2013 model but faulting on AVX2 is not that model.
|
|
#:
|
|
#: Used only to raise a flag. It never decides a feature - execution already did that.
|
|
_POST_AVX2_MARKERS = (
|
|
"haswell", "broadwell", "skylake", "kaby lake", "coffee lake", "comet lake",
|
|
"ice lake", "rocket lake", "alder lake", "raptor lake", "tiger lake",
|
|
"zen", "zen 2", "zen 3", "zen 4", "ryzen", "epyc", "excavator",
|
|
)
|
|
|
|
|
|
def check_model_name(profile: CapabilityProfile) -> CapabilityProfile:
|
|
"""Flag a model name that the measured features cannot support.
|
|
|
|
Separate from `assess_declaration`, and the reason it exists: on node 238 the
|
|
CPUID flags and the executed instructions *agree* - both say no AVX - so the
|
|
flag-vs-measurement check has nothing to report. What is wrong there is the model
|
|
name. It claims a Ryzen 9 3900X, a 2019 part with AVX2 and FMA in silicon, and the
|
|
guest faults on both. A hypervisor is presenting a CPU that does not exist.
|
|
|
|
That is worth saying plainly, because a scheduler reading the model name would
|
|
place AVX2-tuned work on a machine that cannot run it.
|
|
"""
|
|
|
|
name = (profile.cpu_model_declared or "").lower()
|
|
if not name:
|
|
return profile
|
|
avx2 = profile.feature("avx2")
|
|
if avx2 is None or avx2.status != "absent":
|
|
return profile
|
|
|
|
if not any(marker in name for marker in _POST_AVX2_MARKERS):
|
|
return profile
|
|
|
|
note = (
|
|
f"the model name declares '{profile.cpu_model_declared}', which has AVX2 and "
|
|
"FMA in silicon, but an AVX2 instruction faulted with SIGILL. The name is a "
|
|
"declaration from the hypervisor and does not describe this guest"
|
|
)
|
|
if profile.cpu_declaration_suspect:
|
|
profile.cpu_declaration_suspect += f"; {note}"
|
|
else:
|
|
profile.cpu_declaration_suspect = note
|
|
return profile
|