umbragit/UmbraTrans
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Measure capability by executing instructions, not by reading cpuinfo

Browse Source 
/proc/cpuinfo is a declaration. On node 238 it reports 'AMD Ryzen 9 3900X' and then
lists 65 CPUID flags that stop at SSE4.2 - no avx, avx2 or fma, on silicon that has
all three. Executing an AVX2 instruction there faults with SIGILL, so the hypervisor is
presenting a CPU that does not exist.

That is the same defect the compute design already has, one layer down: the agent sets
throughput_factor by string-matching a GPU model (40.0 if '4070' in name else 20.0), a
declared factor wearing a measured name. Here the hypervisor makes the same kind of lie
about instructions. So a feature is recorded as one of three things and never two:

  present    - an instruction used it and produced the right answer
  absent     - an instruction used it and the kernel killed the process
  unverified - it could not be executed at all

Only 'present' counts as usable. An unverified feature is treated as unavailable on
purpose: the cost of pessimism is a slower path that works, and the cost of optimism is
a job dispatched to a machine that dies on it.

Measured on node 238: sse2, sse4_2, bmi2, aes_ni usable; avx, avx2, fma, avx512f all
SIGILL. On oldbox: everything but avx512f.

Four bugs found while building the probe, each of which would have made it report
something false:

- Inline asm with wrong operand counts, for all eight features. Replaced with
  intrinsics, verified by disassembly.
- One binary for every feature could not compile: -mavx2 alone makes the FMA, AVX-512
  and AES intrinsics fail always_inline. Every feature reported unverified on a machine
  that has most of them. Now one translation unit per feature.
- GCC folded the probe at -O2 with constant inputs and emitted nothing, so it would
  have reported 'present' for every CPU on earth. Inputs are volatile and the expected
  mnemonic is checked in the disassembly; test_a_folded_probe_is_reported_unverified
  builds a deliberately foldable probe and asserts it is caught.
- The SSE4.2 probe stored a comparison mask through _mm_storeu_pd and read a double back,
  getting NaN - reported as 'executed but returned a wrong result' on every machine.

Also: a caller passing an explicit compiler path that does not exist got eight build
errors instead of one clear sentence; the compiler is now verified to work, and a
missing or broken one is a False rather than an exception.

525 passed, 27 skipped.
This commit is contained in:
umbra audio
2026-10-03 09:26:22 -07:00
parent 11f27f5ef9
commit c56ef46190
3 changed files with 994 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

179
umbratrans_web/_capability_probe.py Normal file
View File

@@ -0,0 +1,179 @@
"""The C probe, kept in a string so the module has no build step.
Design notes that are not obvious from the code, and which matter:
**One translation unit per feature.** A single source containing every intrinsic cannot
be built with `-mavx2` alone: the FMA, AVX-512 and AES intrinsics then fail to inline
(`always_inline` against a target lacking the flag is a hard error). That was the first
version and it reported every feature unverified on a machine that has most of them. So
the source below is compiled once per feature with `__FEATURE__` substituted.
**Intrinsics, not inline assembly.** The first attempt used inline asm and every operand
count was wrong. Intrinsics let the compiler get the encoding right, and the result is
verified by disassembly anyway.
**Volatile inputs.** At `-O2` with constant arrays, GCC folded `a*b` at compile time and
emitted no instruction, so a probe that "passed" would have proved nothing and reported
every CPU as supporting AVX. The arrays are `volatile` and seeded from `argc`, so the
load must happen at run time.
**Wrong answers are unverified, not absent.** If the program runs and computes something
wrong, the CPU executed the instruction. Reporting `absent` would be a lie about the
hardware; it is a bug in the probe.
"""
_C_SOURCE = r"""
#include <immintrin.h>
#include <emmintrin.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <math.h>
#include <time.h>
/* volatile: these must be loaded at run time so the compiler cannot fold. */
static volatile double vin[8];
static volatile double vcoef[8];
static volatile int iin[8];
static volatile int icoef[8];
static double elapsed(void) {
struct timespec ts;
clock_gettime(CLOCK_MONOTONIC, &ts);
return (double)ts.tv_sec + (double)ts.tv_nsec / 1e9;
}
int main(int argc, char **argv) {
double d[8];
int iv[8];
for (int i = 0; i < 8; i++) {
vin[i] = 1.5; vcoef[i] = 2.0;
iin[i] = i + 1; icoef[i] = i + 1;
}
/* Seed from argc so the values are unknowable at compile time. */
if (argc > 99) { vin[0] = 3.0; iin[0] = 7; }
/* The volatile arrays are cast away once, explicitly: the load intrinsics take a
* non-volatile pointer, and casting here keeps the qualifier from silently
* disabling the optimisation we depend on. */
const double *vinp = (const double *)vin;
const double *vcoefp = (const double *)vcoef;
const int *iinp = (const int *)iin;
const int *icoefp = (const int *)icoef;
#if __FEATURE_INDEX__ == 0
{
__m128d a = _mm_loadu_pd(vinp), b = _mm_loadu_pd(vcoefp);
_mm_storeu_pd(d, _mm_mul_pd(a, b));
if (fabs(d[0] - 3.0) > 1e-9) return 3;
}
#elif __FEATURE_INDEX__ == 1
{
/* CMPNEQPD is SSE4.2. On an older target this is rejected at build time,
* which is the correct outcome: the toolchain cannot express the feature.
*
* The result is a *mask* - all ones in the 64-bit lane where the operands
* differ, zero where they match - not a double. Storing it through
* _mm_storeu_pd and reading a double back gives NaN, which reads as a wrong
* answer and made this probe report unverified on every machine. The first
* version did exactly that. */
unsigned long long mask[2];
__m128d a = _mm_loadu_pd(vinp), b = _mm_loadu_pd(vcoefp);
_mm_storeu_si64(mask, _mm_castpd_si128(_mm_cmpneq_pd(a, b)));
/* 1.5 != 2.0 -> mask, 2.5 != 2.0 -> no mask. Seeded from argc, so both
* lanes are unknowable at compile time. */
if (mask[0] != 0xFFFFFFFFFFFFFFFFULL) return 3;
if (mask[1] != 0) return 3;
}
#elif __FEATURE_INDEX__ == 2
{
__m256d a = _mm256_loadu_pd(vinp), b = _mm256_loadu_pd(vcoefp);
_mm256_storeu_pd(d, _mm256_mul_pd(a, b));
if (fabs(d[0] - 3.0) > 1e-9) return 3;
}
#elif __FEATURE_INDEX__ == 3
{
__m256i a = _mm256_loadu_si256((const __m256i *)iinp);
__m256i b = _mm256_loadu_si256((const __m256i *)icoefp);
_mm256_storeu_si256((__m256i *)iv, _mm256_add_epi32(a, b));
if (iv[0] != 2 || iv[7] != 16) return 3;
}
#elif __FEATURE_INDEX__ == 4
{
/* a*b + a: the result differs from either input, so a compiler that split
* this into a multiply and an add, or folded it to a multiply, is caught. */
__m256d a = _mm256_loadu_pd(vinp), b = _mm256_loadu_pd(vcoefp);
_mm256_storeu_pd(d, _mm256_fmadd_pd(a, b, a));
if (fabs(d[0] - 4.5) > 1e-9) return 3;
}
#elif __FEATURE_INDEX__ == 5
{
__m512d a = _mm512_loadu_pd(vinp), b = _mm512_loadu_pd(vcoefp);
_mm512_storeu_pd(d, _mm512_mul_pd(a, b));
if (fabs(d[0] - 3.0) > 1e-9) return 3;
}
#elif __FEATURE_INDEX__ == 6
{
unsigned long long a = 0x9e3779b97f4a7c15ULL, r = 0;
/* RORX is BMI2. On an older target the assembler rejects it. */
__asm__ volatile("rorx $13, %1, %0" : "=r"(r) : "r"(a));
if (r == a) return 3;
}
#elif __FEATURE_INDEX__ == 7
{
unsigned char state[16], plain[16];
__m128i k, block;
memcpy(plain, "0123456789abcdef", 16);
memcpy(state, plain, 16);
block = _mm_loadu_si128((const __m128i *)state);
k = _mm_setzero_si128();
k = _mm_aesenc_si128(k, block);
k = _mm_aesenclast_si128(k, block);
_mm_storeu_si128((__m128i *)state, k);
/* No reference AES here, so the assertion is that the bytes changed: the
* instruction ran and did AES work rather than being a no-op copy. */
if (!memcmp(state, plain, 16)) return 3;
}
#else
#error "unknown feature index"
#endif
{
double t0, t1, acc = 0.0;
t0 = elapsed();
for (int i = 0; i < 300000; i++) acc += (double)i * 0.5;
t1 = elapsed();
(void)acc;
printf("%.6f\n", t1 - t0);
}
return 0;
}
"""
#: The mnemonic each feature must produce, checked in the disassembly. If the compiler
#: folded the probe away, none of these appear and the feature is reported unverified.
#: Lists, because different compilers choose different encodings.
_EMITTED = {
"sse2": ("mulpd", "addpd"),
"sse4_2": ("cmpneqpd", "cmpltpd", "cmpneqpd"),
"avx": ("vmulpd", "vaddpd"),
"avx2": ("vpaddd", "vpaddq"),
"fma": ("vfmadd", "vfnmadd"),
"avx512f": ("vmulpd", "vaddpd"),
"bmi2": ("rorx",),
"aes_ni": ("aesenc",),
}
#: Compiler flag per feature. Without the explicit flag the compiler targets its own
#: build machine and emits baseline instructions, so the probe would pass everywhere.
_FEATURE_FLAGS = {
"sse2": ["-msse2"],
"sse4_2": ["-msse4.2"],
"avx": ["-mavx"],
"avx2": ["-mavx2"],
"fma": ["-mfma"],
"avx512f": ["-mavx512f"],
"bmi2": ["-mbmi2"],
"aes_ni": ["-maes"],
}

517
umbratrans_web/capability.py Normal file
View File

@@ -0,0 +1,517 @@
"""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

298
umbratrans_web/tests/test_capability.py Normal file
View File

@@ -0,0 +1,298 @@
"""Capability discovery has to survive its own failure modes.
The failure this module exists to prevent: a scheduler sends AVX2-tuned work to a
machine that faults on it. That happens when capability is *declared* rather than
measured. Two ways it has already happened here:
- The agent sets ``throughput_factor`` by string-matching a GPU model
(``40.0 if "4070" in name else 20.0``). A declared factor wearing a measured name.
- Node 238 reports ``AMD Ryzen 9 3900X`` and faults with SIGILL on AVX2. The
hypervisor presents a CPU that does not exist.
So the tests are mostly about the probe's own honesty:
- **A folded probe must not pass.** At ``-O2`` with constant inputs, GCC folds ``a*b``
and emits nothing. A probe that "passed" would report every CPU as supporting AVX.
The disassembly check is what prevents it, so there is a test that a deliberately
foldable probe is reported unverified rather than present.
- **Absent, present and unverified must stay distinct.** Collapsing "unverified" into
"absent" makes a machine look slower than it is; collapsing it into "present" makes it
look faster, and that is the dangerous direction.
- **No compiler means unverified, not absent.** Falling back to ``/proc/cpuinfo`` when
there is no compiler is precisely the bug.
These run everywhere, including with no database, because they measure the machine they
run on and assert on the *shape* of the answer. The exact results differ by machine and
are deliberately not asserted.
"""
from __future__ import annotations
import shutil
import subprocess
import pytest
from umbratrans_web.capability import (
FEATURES,
CapabilityProfile,
FeatureResult,
_declared_flags,
_find_compiler,
assess_declaration,
check_model_name,
collect,
probe_features,
profile_to_dict,
)
VALID = {"present", "absent", "unverified"}
def _profile(**overrides) -> CapabilityProfile:
base = dict(
node_name="test",
logical_cpus=4,
features=[
FeatureResult("sse2", "present"),
FeatureResult("sse4_2", "present"),
FeatureResult("avx", "present"),
FeatureResult("avx2", "present"),
FeatureResult("fma", "present"),
],
)
base.update(overrides)
return CapabilityProfile(**base)
# -- the three outcomes stay distinct -----------------------------------------
def test_only_present_counts_as_usable() -> None:
"""Unverified is treated as unavailable, on purpose.
The cost of pessimism is a slower path that works. The cost of optimism is a job
dispatched to a machine that dies on it, and a lease that has to expire before
anyone notices.
"""
assert FeatureResult("avx2", "present").usable is True
assert FeatureResult("avx2", "absent").usable is False
assert FeatureResult("avx2", "unverified").usable is False
def test_no_compiler_yields_unverified_not_absent() -> None:
"""The fallback to /proc/cpuinfo is the bug, not the safety net.
A machine with no compiler is not a machine without AVX2. Reporting "absent" would
tell the scheduler to avoid work the machine could do.
"""
results, note = probe_features(compiler="/nonexistent/compiler")
assert {r.status for r in results} == {"unverified"}
assert len(results) == len(FEATURES)
assert "no working C compiler" in note
assert "/proc/cpuinfo" in note, "the note must say why it is not reading cpuinfo"
def test_status_is_always_one_of_three_words() -> None:
"""A free-form status would be silently misread by whatever consumes it."""
results, _ = probe_features()
for result in results:
assert result.status in VALID, result
def test_every_feature_gets_a_result() -> None:
"""A missing feature is not a missing answer - it looks like support."""
results, _ = probe_features()
assert {r.name for r in results} == set(FEATURES)
# -- the fold guard ------------------------------------------------------------
@pytest.mark.skipif(not _find_compiler(), reason="needs a C compiler to test the fold guard")
def test_a_folded_probe_is_reported_unverified_not_present() -> None:
"""The check that makes the whole module trustworthy.
Builds a probe whose work the compiler can do at compile time. It will run and exit
0, so without the disassembly check it would report "present" and claim every CPU
supports AVX. The point is asserted through the real code path.
"""
from umbratrans_web._capability_probe import _C_SOURCE, _EMITTED, _FEATURE_FLAGS
source = (
"#include <immintrin.h>\n"
"int main(void) {\n"
" const double a[4] = {1.5, 2.5, 3.5, 4.5};\n"
" const double b[4] = {2.0, 2.0, 2.0, 2.0};\n"
" double c[4];\n"
" _mm256_storeu_pd(c, _mm256_mul_pd(_mm256_loadu_pd(a), _mm256_loadu_pd(b)));\n"
" return (c[0] == 3.0) ? 0 : 3;\n"
"}\n"
)
compiler = _find_compiler()
assert compiler
import tempfile
from pathlib import Path
with tempfile.TemporaryDirectory() as workdir:
path = Path(workdir) / "fold.c"
path.write_text(source)
binary = Path(workdir) / "fold"
build = subprocess.run(
[compiler, "-O2", *_FEATURE_FLAGS["avx2"], "-o", str(binary), str(path)],
capture_output=True,
text=True,
timeout=120,
)
assert build.returncode == 0, build.stderr[-300:]
run = subprocess.run([str(binary)], capture_output=True, timeout=30)
assert run.returncode == 0, "the folded probe should still exit 0"
from umbratrans_web.capability import _verify_emitted
verdict = _verify_emitted(binary, "avx2")
assert verdict is not None, "a folded probe must be caught by the disassembly check"
assert "folded" in verdict
# And the real probe for the same feature must not be foldable, or the guard would
# report every real machine unverified.
assert _C_SOURCE.count("volatile") >= 4
assert _EMITTED["avx2"]
def test_real_probe_of_this_machine_reports_a_decision() -> None:
"""Whatever this machine is, each feature must come back decided or explained."""
results, _ = probe_features()
for result in results:
assert result.status in VALID
if result.status == "unverified":
assert result.detail, f"{result.name} is unverified with no explanation"
def test_objdump_absence_does_not_fail_a_feature() -> None:
"""Without objdump we cannot check - so we say nothing rather than guess.
Returning unverified here would report every machine as unusable whenever objdump
is missing, which is a worse failure than not checking at all.
"""
from umbratrans_web.capability import _verify_emitted
original = shutil.which
try:
shutil.which = lambda name: None if name == "objdump" else original(name)
import umbratrans_web.capability as capability
capability.shutil.which = shutil.which
assert _verify_emitted(binary=None, name="avx") is None # type: ignore[arg-type]
finally:
shutil.which = original
# -- declarations that contradict measurement ----------------------------------
def test_a_model_name_implying_avx2_with_avx2_absent_is_flagged() -> None:
"""The node 238 case: cpuinfo and execution agree, and both are wrong.
The flags say no AVX and the CPU faults on AVX, so there is no disagreement between
declaration and measurement to report. What is wrong is the *name*. A scheduler
reading it would place AVX2 work on a machine that cannot run it.
"""
profile = _profile(
cpu_model_declared="AMD Ryzen 9 3900X 12-Core Processor",
features=[
FeatureResult("sse2", "present"),
FeatureResult("avx", "absent", "SIGILL"),
FeatureResult("avx2", "absent", "SIGILL"),
FeatureResult("fma", "absent", "SIGILL"),
],
)
flagged = check_model_name(profile)
assert flagged.cpu_declaration_suspect
assert "Ryzen 9 3900X" in flagged.cpu_declaration_suspect
assert "AVX2" in flagged.cpu_declaration_suspect
def test_a_matching_model_name_is_not_flagged() -> None:
"""No false alarms: a pre-AVX2 part reporting no AVX2 is normal, not suspicious."""
profile = _profile(
cpu_model_declared="Intel Pentium G4400",
features=[FeatureResult("avx", "absent"), FeatureResult("avx2", "absent")],
)
assert check_model_name(profile).cpu_declaration_suspect == ""
def test_model_name_check_needs_an_absent_avx2_to_say_anything() -> None:
"""A machine with working AVX2 and a modern name is fine, whatever cpuinfo omits."""
profile = _profile(cpu_model_declared="AMD Ryzen 9 3900X 12-Core Processor")
assert check_model_name(profile).cpu_declaration_suspect == ""
def test_claimed_but_absent_is_flagged_by_the_flag_comparison() -> None:
"""cpuinfo says it has it, execution says otherwise - the clearest case."""
profile = _profile(
cpu_model_declared="Some CPU",
features=[FeatureResult("avx2", "absent", "SIGILL")],
)
# assess_declaration reads the real /proc/cpuinfo, so on a machine without AVX2 in
# cpuinfo it has nothing to compare. The behaviour under test is that it does not
# crash and does not invent a disagreement.
result = assess_declaration(profile)
assert isinstance(result.cpu_declaration_suspect, str)
# -- the profile as a whole ----------------------------------------------------
def test_collect_produces_a_complete_profile() -> None:
profile = collect(node_name="shape-test")
assert profile.node_name == "shape-test"
assert profile.architecture
assert profile.logical_cpus >= 1
assert profile.memory_total_bytes > 0
assert len(profile.features) == len(FEATURES)
assert profile.compiler, "the compiler field must say what was used, even if none"
def test_profile_serialises_with_its_usable_list() -> None:
"""The usable list is what a consumer branches on, so it travels with the data."""
data = profile_to_dict(_profile())
assert "usable_features" in data
assert "avx2" in data["usable_features"]
def test_usable_features_excludes_unverified() -> None:
profile = _profile(
features=[
FeatureResult("sse2", "present"),
FeatureResult("avx2", "unverified", "no compiler"),
]
)
assert profile.usable_features() == ["sse2"]
def test_declared_flags_is_a_set_or_empty() -> None:
"""Never raises, whatever /proc/cpuino looks like."""
flags = _declared_flags()
assert isinstance(flags, set)
def test_collect_does_not_need_root_or_a_network() -> None:
"""It runs on a node with no privileges, which is how nodes actually run."""
profile = collect()
assert profile.features, "a profile with no results is worse than no profile"