shawn
4e0ab94ee2
- Added start-llama-server.bat with cross-brand GPU auto-detection: * Detection 1: nvidia-smi for NVIDIA * Detection 2: WMIC for AMD, Intel Arc, Intel HD/Iris/UHD * Detection 3: vulkaninfo as final fallback probe * Falls back to CPU-only build when no GPU found - Downloaded llama-server b9947 builds: * Vulkan build (91 MB) - supports all GPU brands * CPU build (45 MB) - fallback in cpu/ subdir - Downloaded Qwen2.5-Coder-1.5B Q4_K_M model (1.1 GB) - Fixed run-hermes.bat polling: replaces fixed 5s timeout with curl-based readiness polling (up to 30 attempts, 1s apart) - All .bat files verified: ASCII text, CRLF line terminators - Added .gitignore for build artifacts and large binaries
991 lines
35 KiB
Python
991 lines
35 KiB
Python
"""
|
|
stress.py — RAM, CPU, and GPU stress testing orchestration.
|
|
|
|
Designed for the Hermes Portable Rescue USB environment.
|
|
Each tester class follows the same interface:
|
|
- run() -> dict: structured results for LLM consumption
|
|
- report() -> str: human-readable summary
|
|
|
|
Usage
|
|
-----
|
|
from src.diagnostics.stress import run_all_stress_tests
|
|
results = run_all_stress_tests(duration=60)
|
|
print(results)
|
|
|
|
Or from CLI:
|
|
python -m src.diagnostics.stress --duration 60 --ram --cpu --gpu
|
|
|
|
Requires bundled tools:
|
|
- stress-ng (CPU + RAM stress)
|
|
- memtester (RAM stress fallback)
|
|
- nvidia-smi, glxinfo, glmark2 (GPU)
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass, field, asdict
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
logger = logging.getLogger("stress")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Data structures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class StressResult:
|
|
"""Structured result of a single stress test run."""
|
|
|
|
component: str # "ram", "cpu", "gpu"
|
|
backend: str # tool used ("stress-ng", "memtester", "memtest86+", etc.)
|
|
status: str # "passed", "failed", "error", "skipped"
|
|
duration_seconds: float
|
|
summary: str # one-line human verdict
|
|
details: Dict[str, Any] = field(default_factory=dict)
|
|
errors: List[str] = field(default_factory=list)
|
|
raw_output: str = ""
|
|
|
|
def to_dict(self) -> dict:
|
|
return asdict(self)
|
|
|
|
@staticmethod
|
|
def skipped(component: str, reason: str) -> "StressResult":
|
|
return StressResult(
|
|
component=component,
|
|
backend="none",
|
|
status="skipped",
|
|
duration_seconds=0.0,
|
|
summary=f"Skipped: {reason}",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _find_tool(name: str) -> Optional[str]:
|
|
"""Return absolute path to *name* if it exists on PATH, else None."""
|
|
return shutil.which(name)
|
|
|
|
|
|
def _run(
|
|
cmd: List[str],
|
|
timeout: int = 300,
|
|
env: Optional[Dict[str, str]] = None,
|
|
) -> Tuple[int, str, str]:
|
|
"""Run *cmd* and return (returncode, stdout, stderr)."""
|
|
base_env = os.environ.copy()
|
|
if env:
|
|
base_env.update(env)
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout,
|
|
env=base_env,
|
|
)
|
|
return proc.returncode, proc.stdout, proc.stderr
|
|
except FileNotFoundError:
|
|
return -1, "", f"Command not found: {cmd[0]}"
|
|
except subprocess.TimeoutExpired:
|
|
return -2, "", f"Command timed out after {timeout}s"
|
|
except Exception as exc:
|
|
return -3, "", str(exc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RAM Stress Tester
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class RAMStressTester:
|
|
"""
|
|
RAM stress testing with multiple backends.
|
|
|
|
Backends (tried in order):
|
|
1. stress-ng --vm <n> --vm-bytes <bytes> (user-space RAM stress)
|
|
2. memtester <bytes> <iterations> (user-space RAM test)
|
|
3. memtest86+ result parser (reads existing report)
|
|
"""
|
|
|
|
BACKEND_PRIORITY = ["stress-ng", "memtester", "memtest86+result"]
|
|
|
|
def __init__(
|
|
self,
|
|
duration: int = 60,
|
|
workers: Optional[int] = None,
|
|
vm_bytes: str = "80%",
|
|
memtest_iterations: int = 2,
|
|
memtest86_report_path: Optional[str] = None,
|
|
):
|
|
self.duration = duration
|
|
self.workers = workers or self._default_workers()
|
|
self.vm_bytes = vm_bytes
|
|
self.memtester_iterations = memtest_iterations
|
|
self.memtest86_report = memtest86_report_path
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public interface
|
|
# ------------------------------------------------------------------
|
|
|
|
def run(self) -> StressResult:
|
|
backend, fn = self._pick_backend()
|
|
if fn is None:
|
|
return StressResult.skipped("ram", f"no backend available ({backend})")
|
|
logger.info("RAM testing with backend=%s workers=%d duration=%ds", backend, self.workers, self.duration)
|
|
t0 = time.monotonic()
|
|
result = fn()
|
|
result.duration_seconds = round(time.monotonic() - t0, 1)
|
|
result.backend = backend
|
|
return result
|
|
|
|
def report(self, result: StressResult) -> str:
|
|
lines = [
|
|
f"--- RAM Stress Test ---",
|
|
f" Backend: {result.backend}",
|
|
f" Status: {result.status}",
|
|
f" Duration: {result.duration_seconds}s",
|
|
f" Summary: {result.summary}",
|
|
]
|
|
if result.details:
|
|
for k, v in result.details.items():
|
|
lines.append(f" {k}: {v}")
|
|
if result.errors:
|
|
lines.append(" Errors:")
|
|
for e in result.errors:
|
|
lines.append(f" - {e}")
|
|
return "\n".join(lines)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internal
|
|
# ------------------------------------------------------------------
|
|
|
|
@staticmethod
|
|
def _default_workers() -> int:
|
|
"""Heuristic: use half of available CPUs for VM stress workers."""
|
|
try:
|
|
return max(1, os.cpu_count() or 2 // 2)
|
|
except Exception:
|
|
return 1
|
|
|
|
def _pick_backend(self) -> Tuple[str, Optional[callable]]:
|
|
"""Return (label, callable) for the best available backend."""
|
|
if _find_tool("stress-ng"):
|
|
return "stress-ng", self._run_stress_ng
|
|
if _find_tool("memtester"):
|
|
return "memtester", self._run_memtester
|
|
if self.memtest86_report or _find_tool("memtest86+report"):
|
|
return "memtest86+result", self._parse_memtest86
|
|
return "none", None
|
|
|
|
def _run_stress_ng(self) -> StressResult:
|
|
"""Run stress-ng memory stress (--vm <workers> --vm-bytes <bytes>)."""
|
|
cmd = [
|
|
"stress-ng",
|
|
"--vm", str(self.workers),
|
|
"--vm-bytes", self.vm_bytes,
|
|
"--timeout", f"{self.duration}s",
|
|
"--metrics-brief",
|
|
"--no-rand-seed",
|
|
]
|
|
rc, stdout, stderr = _run(cmd, timeout=self.duration + 30)
|
|
|
|
errors = []
|
|
if stderr:
|
|
errors.append(stderr.strip())
|
|
|
|
details = self._parse_stress_ng_metrics(stdout)
|
|
details["command"] = " ".join(cmd)
|
|
details["workers"] = self.workers
|
|
|
|
if rc == -1:
|
|
return StressResult(
|
|
component="ram",
|
|
status="error",
|
|
summary="stress-ng not found",
|
|
details=details,
|
|
errors=errors,
|
|
raw_output=stdout + "\n" + stderr,
|
|
)
|
|
if rc == -2:
|
|
return StressResult(
|
|
component="ram",
|
|
status="error",
|
|
summary="stress-ng timed out",
|
|
details=details,
|
|
errors=errors,
|
|
raw_output=stdout + "\n" + stderr,
|
|
)
|
|
|
|
# stress-ng exits 0 on clean completion even if errors occurred
|
|
status = "passed" if rc == 0 else "error"
|
|
bogo_ops = details.get("total_bogo_ops", 0)
|
|
summary = (
|
|
f"Completed {bogo_ops} bogo-ops across {self.workers} VM workers"
|
|
if bogo_ops > 0 else
|
|
f"Finished with status={status}"
|
|
)
|
|
return StressResult(
|
|
component="ram",
|
|
status=status,
|
|
summary=summary,
|
|
details=details,
|
|
errors=errors,
|
|
raw_output=stdout + "\n" + stderr,
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_stress_ng_metrics(text: str) -> Dict[str, Any]:
|
|
"""Extract bogo-ops and other metrics from stress-ng --metrics-brief."""
|
|
metrics: Dict[str, Any] = {}
|
|
total_bogo = 0.0
|
|
bogo_per_sec = 0.0
|
|
# stress-ng metrics lines look like:
|
|
# stress-ng: info: [123] stress-ng-vm: 4 runouts, 0 stalled...
|
|
# stress-ng: info: [123] stress-ng-metrics: vm 1234.56 bogo-ops ...
|
|
for line in text.splitlines():
|
|
if "bogo-ops" in line.lower():
|
|
m = re.search(r"vm\s+([\d.]+)\s+bogo-ops", line, re.IGNORECASE)
|
|
if m:
|
|
total_bogo = float(m.group(1))
|
|
m2 = re.search(r"bogo-ops/s\s+real\s+([\d.]+)", line, re.IGNORECASE)
|
|
if m2:
|
|
bogo_per_sec = float(m2.group(1))
|
|
metrics["total_bogo_ops"] = round(total_bogo, 1)
|
|
metrics["bogo_ops_per_sec"] = round(bogo_per_sec, 3) if bogo_per_sec else 0.0
|
|
return metrics
|
|
|
|
def _run_memtester(self) -> StressResult:
|
|
"""Run memtester <vm_bytes> <iterations>."""
|
|
# Convert vm_bytes like "80%" to a usable size for memtester
|
|
# memtester takes size in megabytes directly
|
|
size_mb = self._vm_bytes_to_mb()
|
|
if size_mb <= 0:
|
|
size_mb = 256 # fallback
|
|
cmd = ["memtester", f"{size_mb}M", str(self.memtester_iterations)]
|
|
rc, stdout, stderr = _run(cmd, timeout=max(self.duration, 120))
|
|
|
|
errors = []
|
|
if stderr:
|
|
errors.append(stderr.strip())
|
|
|
|
details = self._parse_memtester_output(stdout)
|
|
details["command"] = " ".join(cmd)
|
|
details["size_mb"] = size_mb
|
|
details["iterations"] = self.memtester_iterations
|
|
|
|
status = "passed" if rc == 0 else "failed"
|
|
if rc < 0:
|
|
status = "error"
|
|
|
|
failure_count = details.get("failures", 0)
|
|
summary = f"memtester: {details.get('loops_completed', 0)} loops, {failure_count} failures" if status != "error" else "memtester run failed"
|
|
return StressResult(
|
|
component="ram",
|
|
status=status,
|
|
summary=summary,
|
|
details=details,
|
|
errors=errors,
|
|
raw_output=stdout + "\n" + stderr,
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_memtester_output(text: str) -> Dict[str, Any]:
|
|
"""Parse memtester output for pass/fail information."""
|
|
details: Dict[str, Any] = {}
|
|
loops = 0
|
|
failures = 0
|
|
for line in text.splitlines():
|
|
m = re.match(r"Loop\s+(\d+)", line, re.IGNORECASE)
|
|
if m:
|
|
loops = max(loops, int(m.group(1)))
|
|
if "FAILURE" in line.upper():
|
|
failures += 1
|
|
details["loops_completed"] = loops
|
|
details["failures"] = failures
|
|
return details
|
|
|
|
def _parse_memtest86(self) -> StressResult:
|
|
"""Parse existing MemTest86+ report (result XML or text file)."""
|
|
report_path = self.memtest86_report
|
|
if report_path is None:
|
|
# Common locations
|
|
for candidate in [
|
|
"/MemTest86-Report.xml",
|
|
"/boot/MemTest86-Report.xml",
|
|
"/memtest86/report.xml",
|
|
"~/MemTest86-Report.xml",
|
|
]:
|
|
expanded = os.path.expanduser(candidate)
|
|
if os.path.isfile(expanded):
|
|
report_path = expanded
|
|
break
|
|
|
|
if not report_path or not os.path.isfile(report_path):
|
|
return StressResult(
|
|
component="ram",
|
|
status="skipped",
|
|
summary="MemTest86+ report not found; run MemTest86+ on next boot",
|
|
details={"suggested_path": "/MemTest86-Report.xml"},
|
|
)
|
|
|
|
try:
|
|
text = Path(report_path).read_text()
|
|
except Exception as exc:
|
|
return StressResult(
|
|
component="ram",
|
|
status="error",
|
|
summary=f"Cannot read report: {exc}",
|
|
)
|
|
|
|
details = self._parse_memtest86_xml(text)
|
|
details["report_path"] = report_path
|
|
summary = (
|
|
f"MemTest86+ report: {details.get('passes', '?')} passes, "
|
|
f"{details.get('errors', 0)} errors"
|
|
)
|
|
status = "passed" if details.get("errors", 0) == 0 else "failed"
|
|
return StressResult(
|
|
component="ram",
|
|
status=status,
|
|
summary=summary,
|
|
details=details,
|
|
raw_output=text,
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_memtest86_xml(text: str) -> Dict[str, Any]:
|
|
"""Naïve extraction of key fields from MemTest86+ XML report."""
|
|
def _tag(tag: str) -> str:
|
|
m = re.search(rf"<{tag}[^>]*>(.*?)</{tag}>", text, re.DOTALL)
|
|
return m.group(1).strip() if m else ""
|
|
|
|
passes_str = _tag("Passes")
|
|
return {
|
|
"passes": int(passes_str) if passes_str.isdigit() else 0,
|
|
"errors": _tag("Errors"),
|
|
"test_start_time": _tag("TestStartTime"),
|
|
"test_end_time": _tag("TestEndTime"),
|
|
"memory_size": _tag("MemorySize"),
|
|
"cpu_type": _tag("CpuType"),
|
|
"cpu_speed": _tag("CpuSpeed"),
|
|
}
|
|
|
|
def _vm_bytes_to_mb(self) -> int:
|
|
"""Convert a vm_bytes string (e.g. '80%' or '512M') to megabytes."""
|
|
s = str(self.vm_bytes).strip().lower()
|
|
if s.endswith("%"):
|
|
try:
|
|
pct = int(s.rstrip("%"))
|
|
# Estimate available RAM via /proc/meminfo
|
|
mem_total_kb = 0
|
|
try:
|
|
for line in Path("/proc/meminfo").read_text().splitlines():
|
|
if line.startswith("MemTotal:"):
|
|
parts = line.split()
|
|
if len(parts) >= 2:
|
|
mem_total_kb = int(parts[1])
|
|
break
|
|
except Exception:
|
|
mem_total_kb = 8 * 1024 * 1024 # fallback: 8 GB
|
|
mem_total_mb = mem_total_kb // 1024
|
|
return max(64, mem_total_mb * pct // 100)
|
|
except (ValueError, TypeError):
|
|
return 512
|
|
if s.endswith("g"):
|
|
try:
|
|
return int(s.rstrip("g")) * 1024
|
|
except ValueError:
|
|
return 512
|
|
if s.endswith("m"):
|
|
try:
|
|
return int(s.rstrip("m"))
|
|
except ValueError:
|
|
return 512
|
|
try:
|
|
return int(s)
|
|
except (ValueError, TypeError):
|
|
return 512
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CPU Stress Tester
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class CPUStressTester:
|
|
"""
|
|
CPU stress testing via stress-ng.
|
|
|
|
Tests several CPU subsystems in sequence or parallel:
|
|
- cpu (integer arithmetic)
|
|
- matrix (floating-point matrix operations)
|
|
- fpu (FPU stress)
|
|
- cache (L1/L2 cache thrashing)
|
|
- context (context switching)
|
|
|
|
Each stressor is run briefly to exercise different CPU aspects.
|
|
"""
|
|
|
|
DEFAULT_STRESSORS = ["cpu", "matrix", "fpu", "cache"]
|
|
|
|
def __init__(
|
|
self,
|
|
duration: int = 60,
|
|
workers: Optional[int] = None,
|
|
stressors: Optional[List[str]] = None,
|
|
sequential: bool = True,
|
|
):
|
|
self.duration = duration
|
|
self.workers = workers or self._default_workers()
|
|
self.stressors = stressors or list(self.DEFAULT_STRESSORS)
|
|
self.sequential = sequential
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public interface
|
|
# ------------------------------------------------------------------
|
|
|
|
def run(self) -> StressResult:
|
|
backend = self._pick_backend()
|
|
if backend is None:
|
|
return StressResult.skipped("cpu", "stress-ng not found on PATH")
|
|
|
|
t0 = time.monotonic()
|
|
if self.sequential:
|
|
result = self._run_sequential(backend)
|
|
else:
|
|
result = self._run_parallel(backend)
|
|
result.duration_seconds = round(time.monotonic() - t0, 1)
|
|
result.backend = backend
|
|
return result
|
|
|
|
def report(self, result: StressResult) -> str:
|
|
lines = [
|
|
f"--- CPU Stress Test ---",
|
|
f" Backend: {result.backend}",
|
|
f" Status: {result.status}",
|
|
f" Duration: {result.duration_seconds}s",
|
|
f" Summary: {result.summary}",
|
|
]
|
|
details = result.details
|
|
if "stressors_run" in details:
|
|
lines.append(f" Stressors run: {', '.join(details['stressors_run'])}")
|
|
if "stressor_results" in details:
|
|
for sres in details["stressor_results"]:
|
|
lines.append(f" - {sres.get('name', '?')}: {sres.get('bogo_ops', 0)} bogo-ops")
|
|
if result.errors:
|
|
lines.append(" Errors:")
|
|
for e in result.errors:
|
|
lines.append(f" - {e}")
|
|
return "\n".join(lines)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internal
|
|
# ------------------------------------------------------------------
|
|
|
|
@staticmethod
|
|
def _default_workers() -> int:
|
|
try:
|
|
return os.cpu_count() or 2
|
|
except Exception:
|
|
return 2
|
|
|
|
@staticmethod
|
|
def _pick_backend() -> Optional[str]:
|
|
path = _find_tool("stress-ng")
|
|
return path
|
|
|
|
def _run_sequential(self, backend: str) -> StressResult:
|
|
"""Run each stressor one at a time so we get per-stressor metrics."""
|
|
per_stressor_duration = max(10, self.duration // len(self.stressors))
|
|
all_stressor_results: List[Dict[str, Any]] = []
|
|
all_errors: List[str] = []
|
|
total_bogo = 0.0
|
|
|
|
for stressor in self.stressors:
|
|
cmd = [
|
|
backend,
|
|
"--" + stressor, str(self.workers),
|
|
"--timeout", f"{per_stressor_duration}s",
|
|
"--metrics-brief",
|
|
"--no-rand-seed",
|
|
]
|
|
logger.info("CPU stressor: %s (workers=%d, duration=%ds)", stressor, self.workers, per_stressor_duration)
|
|
rc, stdout, stderr = _run(cmd, timeout=per_stressor_duration + 30)
|
|
|
|
if rc < 0:
|
|
all_errors.append(f"{stressor}: {stderr.strip() or 'unknown error'}")
|
|
all_stressor_results.append({"name": stressor, "status": "error", "bogo_ops": 0})
|
|
continue
|
|
|
|
metrics = self._parse_single_stressor_metrics(stdout, stressor)
|
|
metrics["name"] = stressor
|
|
metrics["status"] = "passed" if rc == 0 else "error"
|
|
all_stressor_results.append(metrics)
|
|
total_bogo += metrics.get("bogo_ops", 0.0)
|
|
all_errors.extend(self._extract_errors(stdout + "\n" + stderr))
|
|
|
|
details: Dict[str, Any] = {}
|
|
details["stressors_run"] = self.stressors
|
|
details["stressor_results"] = all_stressor_results
|
|
details["total_bogo_ops"] = round(total_bogo, 1)
|
|
|
|
status = "passed" if not all_errors else "failed"
|
|
summary = (
|
|
f"CPU stress: {len(self.stressors)} stressors, "
|
|
f"{details['total_bogo_ops']} total bogo-ops"
|
|
)
|
|
if all_errors:
|
|
summary += f", {len(all_errors)} errors"
|
|
|
|
return StressResult(
|
|
component="cpu",
|
|
status=status,
|
|
summary=summary,
|
|
details=details,
|
|
errors=all_errors,
|
|
)
|
|
|
|
def _run_parallel(self, backend: str) -> StressResult:
|
|
"""Run all stressors in one stress-ng invocation (--all)."""
|
|
# Build command: stress-ng --cpu N --matrix N --fpu N --cache N --timeout Xs --metrics-brief
|
|
cmd = [backend]
|
|
for stressor in self.stressors:
|
|
cmd.extend(["--" + stressor, str(self.workers)])
|
|
cmd.extend(["--timeout", f"{self.duration}s", "--metrics-brief", "--no-rand-seed"])
|
|
|
|
logger.info("CPU parallel stress: stressors=%s workers=%d", self.stressors, self.workers)
|
|
rc, stdout, stderr = _run(cmd, timeout=self.duration + 30)
|
|
|
|
all_errors: List[str] = []
|
|
if stderr:
|
|
all_errors.extend(self._extract_errors(stderr))
|
|
if rc < 0:
|
|
all_errors.append(f"stress-ng rc={rc}")
|
|
|
|
all_stressor_results: List[Dict[str, Any]] = []
|
|
total_bogo = 0.0
|
|
for stressor in self.stressors:
|
|
metrics = self._parse_single_stressor_metrics(stdout, stressor)
|
|
metrics["name"] = stressor
|
|
metrics["status"] = "passed" if rc == 0 else "error"
|
|
all_stressor_results.append(metrics)
|
|
total_bogo += metrics.get("bogo_ops", 0.0)
|
|
|
|
details: Dict[str, Any] = {}
|
|
details["stressors_run"] = self.stressors
|
|
details["stressor_results"] = all_stressor_results
|
|
details["total_bogo_ops"] = round(total_bogo, 1)
|
|
details["command"] = " ".join(cmd)
|
|
details["workers_per_stressor"] = self.workers
|
|
|
|
status = "passed" if rc == 0 else "error"
|
|
summary = f"CPU parallel stress: {details['total_bogo_ops']} bogo-ops"
|
|
if all_errors:
|
|
summary += f", {len(all_errors)} issues"
|
|
status = "failed"
|
|
|
|
return StressResult(
|
|
component="cpu",
|
|
status=status,
|
|
summary=summary,
|
|
details=details,
|
|
errors=all_errors,
|
|
raw_output=stdout + "\n" + stderr,
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_single_stressor_metrics(text: str, stressor: str) -> Dict[str, Any]:
|
|
"""Extract metrics for a single stressor from stress-ng output."""
|
|
bogo_ops = 0.0
|
|
bogo_per_sec = 0.0
|
|
for line in text.splitlines():
|
|
if "bogo-ops" in line.lower() and stressor in line.lower():
|
|
m = re.search(rf"{stressor}\s+([\d.]+)\s+bogo-ops", line, re.IGNORECASE)
|
|
if m:
|
|
bogo_ops = float(m.group(1))
|
|
m2 = re.search(r"bogo-ops/s\s+real\s+([\d.]+)", line, re.IGNORECASE)
|
|
if m2:
|
|
bogo_per_sec = float(m2.group(1))
|
|
return {
|
|
"bogo_ops": round(bogo_ops, 1),
|
|
"bogo_ops_per_sec": round(bogo_per_sec, 3),
|
|
}
|
|
|
|
@staticmethod
|
|
def _extract_errors(text: str) -> List[str]:
|
|
"""Extract error/warning lines from stress-ng output."""
|
|
errors = []
|
|
for line in text.splitlines():
|
|
lower = line.lower()
|
|
for kw in ("error", "warning", "fail", "cannot"):
|
|
if kw in lower:
|
|
errors.append(line.strip())
|
|
break
|
|
return errors
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GPU Stress Tester
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class GPUStressTester:
|
|
"""
|
|
GPU info gathering and basic stress testing.
|
|
|
|
Capabilities (tried in order of preference):
|
|
1. nvidia-smi — NVIDIA GPU info + monitoring
|
|
2. glxinfo — OpenGL vendor/renderer/version
|
|
3. glmark2 — OpenGL benchmark / stress
|
|
4. lspci — GPU model identification fallback
|
|
"""
|
|
|
|
def __init__(self, duration: int = 60, skip_benchmark: bool = False):
|
|
self.duration = duration
|
|
self.skip_benchmark = skip_benchmark
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public interface
|
|
# ------------------------------------------------------------------
|
|
|
|
def run(self) -> StressResult:
|
|
t0 = time.monotonic()
|
|
info = self._gather_gpu_info()
|
|
stress = self._run_stress_test()
|
|
result = self._merge(info, stress)
|
|
result.duration_seconds = round(time.monotonic() - t0, 1)
|
|
return result
|
|
|
|
def report(self, result: StressResult) -> str:
|
|
lines = [
|
|
f"--- GPU Stress Test ---",
|
|
f" Backend: {result.backend}",
|
|
f" Status: {result.status}",
|
|
f" Duration: {result.duration_seconds}s",
|
|
f" Summary: {result.summary}",
|
|
]
|
|
details = result.details
|
|
gpu_info = details.get("gpu_info", {})
|
|
if gpu_info:
|
|
lines.append(" GPU Info:")
|
|
for k, v in gpu_info.items():
|
|
lines.append(f" {k}: {v}")
|
|
if "benchmark_score" in details:
|
|
lines.append(f" Benchmark score: {details['benchmark_score']}")
|
|
if result.errors:
|
|
lines.append(" Errors:")
|
|
for e in result.errors:
|
|
lines.append(f" - {e}")
|
|
return "\n".join(lines)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internal
|
|
# ------------------------------------------------------------------
|
|
|
|
def _gather_gpu_info(self) -> Dict[str, Any]:
|
|
"""Collect GPU info from all available sources."""
|
|
info: Dict[str, Any] = {}
|
|
errors: List[str] = []
|
|
|
|
# 1) nvidia-smi
|
|
nv_path = _find_tool("nvidia-smi")
|
|
if nv_path:
|
|
rc, stdout, stderr = _run([nv_path, "--query-gpu=name,driver_version,temperature.gpu,memory.total,utilization.gpu", "--format=csv,noheader"], timeout=15)
|
|
if rc == 0:
|
|
for line in stdout.strip().splitlines():
|
|
parts = [p.strip() for p in line.split(",")]
|
|
if len(parts) >= 3:
|
|
info["model"] = parts[0]
|
|
info["driver_version"] = parts[1]
|
|
info["temperature_c"] = parts[2]
|
|
if len(parts) >= 4:
|
|
info["vram_total"] = parts[3]
|
|
if len(parts) >= 5:
|
|
info["utilization_pct"] = parts[4]
|
|
else:
|
|
errors.append(f"nvidia-smi: {stderr.strip()}")
|
|
else:
|
|
errors.append("nvidia-smi not found")
|
|
|
|
# 2) glxinfo (OpenGL vendor/renderer)
|
|
glx_path = _find_tool("glxinfo")
|
|
if glx_path:
|
|
rc, stdout, stderr = _run([glx_path, "-B"], timeout=15)
|
|
if rc == 0:
|
|
for line in stdout.splitlines():
|
|
lower = line.lower()
|
|
if "opengl vendor" in lower:
|
|
info["gl_vendor"] = line.split(":", 1)[-1].strip()
|
|
elif "opengl renderer" in lower:
|
|
info["gl_renderer"] = line.split(":", 1)[-1].strip()
|
|
elif "opengl version" in lower:
|
|
info["gl_version"] = line.split(":", 1)[-1].strip()
|
|
|
|
# 3) lspci as fallback for model
|
|
if "model" not in info:
|
|
lspci_path = _find_tool("lspci")
|
|
if lspci_path:
|
|
rc, stdout, stderr = _run([lspci_path], timeout=10)
|
|
if rc == 0:
|
|
for line in stdout.splitlines():
|
|
l = line.lower()
|
|
if any(kw in l for kw in ("vga", "3d controller", "display")):
|
|
info["model"] = line.strip()
|
|
break
|
|
|
|
info["has_nvidia_gpu"] = nv_path is not None
|
|
return info
|
|
|
|
def _run_stress_test(self) -> Dict[str, Any]:
|
|
"""Run GPU benchmark/stress if available."""
|
|
result: Dict[str, Any] = {}
|
|
errors: List[str] = []
|
|
|
|
if self.skip_benchmark:
|
|
result["benchmark_skipped"] = True
|
|
return result
|
|
|
|
# Try glmark2 first (OpenGL benchmark)
|
|
glmark_path = _find_tool("glmark2")
|
|
if glmark_path:
|
|
rc, stdout, stderr = _run(
|
|
[glmark2_path, "--run-forever"],
|
|
timeout=self.duration + 15,
|
|
env={"DISPLAY": os.environ.get("DISPLAY", ":0")},
|
|
)
|
|
if rc == 0:
|
|
score = self._parse_glmark2_score(stdout)
|
|
result["benchmark_score"] = score
|
|
result["benchmark_tool"] = "glmark2"
|
|
else:
|
|
errors.append("glmark2 not found; no GPU benchmark available")
|
|
|
|
if errors:
|
|
result["errors"] = errors
|
|
return result
|
|
|
|
@staticmethod
|
|
def _parse_glmark2_score(text: str) -> Optional[int]:
|
|
"""Extract the final glmark2 score. Looks for 'glmark2 Score: NNN'."""
|
|
m = re.search(r"glmark2\s+Score:\s+(\d+)", text, re.IGNORECASE)
|
|
if m:
|
|
return int(m.group(1))
|
|
return None
|
|
|
|
def _merge(self, info: Dict[str, Any], stress: Dict[str, Any]) -> StressResult:
|
|
"""Merge gathered info and stress results into a single StressResult."""
|
|
all_errors = info.pop("errors", [])
|
|
all_errors.extend(stress.pop("errors", []))
|
|
|
|
details: Dict[str, Any] = {}
|
|
details["gpu_info"] = info
|
|
details.update(stress)
|
|
|
|
model = info.get("model", info.get("gl_renderer", "unknown"))
|
|
summary_parts = [f"GPU: {model}"]
|
|
if "benchmark_score" in details and details["benchmark_score"] is not None:
|
|
summary_parts.append(f"glmark2 score: {details['benchmark_score']}")
|
|
if "temperature_c" in info:
|
|
summary_parts.append(f"temp: {info['temperature_c']}°C")
|
|
|
|
status = "passed" if not all_errors else "failed"
|
|
# Even with all_errors, we always have info from lspci at minimum
|
|
backend_parts = []
|
|
if _find_tool("nvidia-smi"):
|
|
backend_parts.append("nvidia-smi")
|
|
if _find_tool("glxinfo"):
|
|
backend_parts.append("glxinfo")
|
|
if _find_tool("glmark2"):
|
|
backend_parts.append("glmark2")
|
|
if _find_tool("lspci"):
|
|
backend_parts.append("lspci")
|
|
if not backend_parts:
|
|
backend_parts = ["none"]
|
|
|
|
return StressResult(
|
|
component="gpu",
|
|
backend="+".join(backend_parts),
|
|
status=status,
|
|
duration_seconds=0.0, # caller overwrites this with actual wall time
|
|
summary="; ".join(summary_parts),
|
|
details=details,
|
|
errors=all_errors,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Orchestration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def run_all_stress_tests(
|
|
duration: int = 60,
|
|
ram: bool = True,
|
|
cpu: bool = True,
|
|
gpu: bool = True,
|
|
skip_gpu_benchmark: bool = False,
|
|
) -> Dict[str, StressResult]:
|
|
"""
|
|
Run all requested stress tests and return structured results.
|
|
|
|
Parameters
|
|
----------
|
|
duration : int
|
|
Per-test duration in seconds (CPU/RAM get this; GPU benchmark runs separately).
|
|
ram, cpu, gpu : bool
|
|
Toggle individual tests.
|
|
skip_gpu_benchmark : bool
|
|
If True, only gather GPU info, skip the OpenGL benchmark loop.
|
|
|
|
Returns
|
|
-------
|
|
dict of {component: StressResult}
|
|
"""
|
|
results: Dict[str, StressResult] = {}
|
|
logger.info("Stress test suite starting (duration=%ds)", duration)
|
|
|
|
if ram:
|
|
ram_tester = RAMStressTester(duration=duration)
|
|
results["ram"] = ram_tester.run()
|
|
logger.info("RAM test: %s | %s", results["ram"].status, results["ram"].summary)
|
|
|
|
if cpu:
|
|
cpu_tester = CPUStressTester(duration=duration)
|
|
results["cpu"] = cpu_tester.run()
|
|
logger.info("CPU test: %s | %s", results["cpu"].status, results["cpu"].summary)
|
|
|
|
if gpu:
|
|
gpu_tester = GPUStressTester(duration=duration, skip_benchmark=skip_gpu_benchmark)
|
|
results["gpu"] = gpu_tester.run()
|
|
logger.info("GPU test: %s | %s", results["gpu"].status, results["gpu"].summary)
|
|
|
|
return results
|
|
|
|
|
|
def print_report(results: Dict[str, StressResult]) -> None:
|
|
"""Pretty-print test results to stdout."""
|
|
testers = {
|
|
"ram": RAMStressTester,
|
|
"cpu": CPUStressTester,
|
|
"gpu": GPUStressTester,
|
|
}
|
|
for component, tester_cls in testers.items():
|
|
result = results.get(component)
|
|
if result is not None:
|
|
print(tester_cls().report(result))
|
|
print()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def parse_cli_args() -> dict:
|
|
"""Simple manual arg parser (no external deps)."""
|
|
args = {
|
|
"duration": 60,
|
|
"ram": True,
|
|
"cpu": True,
|
|
"gpu": True,
|
|
"skip_gpu_benchmark": False,
|
|
"verbose": False,
|
|
}
|
|
|
|
argv = sys.argv[1:]
|
|
i = 0
|
|
while i < len(argv):
|
|
a = argv[i]
|
|
if a in ("--duration", "-d"):
|
|
if i + 1 < len(argv):
|
|
args["duration"] = int(argv[i + 1])
|
|
i += 1
|
|
elif a == "--ram-only":
|
|
args["ram"] = True
|
|
args["cpu"] = False
|
|
args["gpu"] = False
|
|
elif a == "--cpu-only":
|
|
args["ram"] = False
|
|
args["cpu"] = True
|
|
args["gpu"] = False
|
|
elif a == "--gpu-only":
|
|
args["ram"] = False
|
|
args["cpu"] = False
|
|
args["gpu"] = True
|
|
elif a in ("--no-ram",):
|
|
args["ram"] = False
|
|
elif a in ("--no-cpu",):
|
|
args["cpu"] = False
|
|
elif a in ("--no-gpu",):
|
|
args["gpu"] = False
|
|
elif a in ("--skip-gpu-benchmark",):
|
|
args["skip_gpu_benchmark"] = True
|
|
elif a in ("--verbose", "-v"):
|
|
args["verbose"] = True
|
|
elif a in ("--help", "-h"):
|
|
print("Usage: python -m src.diagnostics.stress [options]")
|
|
print()
|
|
print("Options:")
|
|
print(" -d, --duration SECS Per-test duration (default: 60)")
|
|
print(" --ram-only RAM test only")
|
|
print(" --cpu-only CPU test only")
|
|
print(" --gpu-only GPU test only")
|
|
print(" --no-ram Skip RAM test")
|
|
print(" --no-cpu Skip CPU test")
|
|
print(" --no-gpu Skip GPU test")
|
|
print(" --skip-gpu-benchmark GPU info only, no benchmark")
|
|
print(" -v, --verbose Verbose logging")
|
|
print(" -h, --help This help")
|
|
sys.exit(0)
|
|
i += 1
|
|
return args
|
|
|
|
|
|
def main() -> None:
|
|
"""CLI entry point."""
|
|
config = parse_cli_args()
|
|
if config["verbose"]:
|
|
logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s")
|
|
else:
|
|
logging.basicConfig(level=logging.WARNING, format="%(levelname)s | %(message)s")
|
|
|
|
print("Hermes Portable Rescue — Stress Test Suite")
|
|
print(f" Duration: {config['duration']}s per test")
|
|
print(f" RAM: {'yes' if config['ram'] else 'no'}")
|
|
print(f" CPU: {'yes' if config['cpu'] else 'no'}")
|
|
print(f" GPU: {'yes' if config['gpu'] else 'no'}")
|
|
print(f" GPU benchmark: {'no' if config['skip_gpu_benchmark'] else 'yes'}")
|
|
print()
|
|
|
|
results = run_all_stress_tests(
|
|
duration=config["duration"],
|
|
ram=config["ram"],
|
|
cpu=config["cpu"],
|
|
gpu=config["gpu"],
|
|
skip_gpu_benchmark=config["skip_gpu_benchmark"],
|
|
)
|
|
|
|
print_report(results)
|
|
|
|
# Overall status
|
|
print("=" * 50)
|
|
print("OVERALL")
|
|
for comp, res in results.items():
|
|
print(f" {comp.upper():5s} {res.status:10s} {res.summary}")
|
|
print("=" * 50)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|