Initial CPUOpt-Kernel scaffold

Author: manishklach

.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+*.pyc
+.test-state/

README.md

@@ -0,0 +1,111 @@
+# CPUOpt-Kernel
+
+CPUOpt-Kernel translates vendor CPU performance controls into safe Linux policy profiles.
+It prefers existing kernel interfaces over raw register writes and treats thermal and fan
+control as platform concerns, not as a shortcut for unsafe CPU tuning.
+
+Version `v0.1` is intentionally conservative:
+
+- Intel-first, user-space-first implementation
+- No raw MSR writes
+- No thermal protection bypass
+- No voltage or overclock manipulation
+- Reversible sysfs writes with snapshot/restore
+- Dry-run support for every profile application
+
+## Why this exists
+
+Linux CPU performance policy is spread across multiple interfaces such as `cpufreq`,
+`intel_pstate`, `amd-pstate`, EPP/EPB, cpuidle, thermal zones, cooling devices, and
+platform-specific hwmon hooks. CPUOpt-Kernel unifies those surfaces into profile-driven,
+workload-aware policy choices without pretending every platform exposes the same knobs.
+
+Fan control is platform-specific and is not treated as a CPU register. Performance mode
+optimizes for sustained boost and reduced throttling, not unsafe thermal bypass. Quiet mode
+reduces CPU aggressiveness before considering fan changes.
+
+## Intel MVP scope
+
+The first release focuses on safe Intel profile application through sysfs:
+
+- `discover` and `status` report CPUFreq, `intel_pstate`, EPP, EPB, turbo, cpuidle,
+  thermal, cooling, hwmon, SMT, topology, and NUMA metadata when present
+- `profile` maps high-level policies to existing sysfs knobs
+- `monitor` provides lightweight live telemetry
+- `restore` replays the last reversible snapshot
+
+AMD and ARM support are scaffolded for future milestones but remain discovery-first.
+
+## Repository layout
+
+```text
+cpuopt-kernel/
+  cpuoptctl/
+  docs/
+  examples/
+  kernel/
+  scripts/
+  tests/
+```
+
+## Commands
+
+```bash
+python3 cpuoptctl/cpuoptctl.py status
+python3 cpuoptctl/cpuoptctl.py discover --json
+sudo python3 cpuoptctl/cpuoptctl.py profile performance --dry-run
+sudo python3 cpuoptctl/cpuoptctl.py profile balanced
+sudo python3 cpuoptctl/cpuoptctl.py profile latency --allow-idle-tuning
+sudo python3 cpuoptctl/cpuoptctl.py profile quiet
+sudo python3 cpuoptctl/cpuoptctl.py profile ai-inference
+sudo python3 cpuoptctl/cpuoptctl.py monitor --interval 2
+sudo python3 cpuoptctl/cpuoptctl.py restore
+python3 cpuoptctl/cpuoptctl.py export-json
+```
+
+For fixture-backed tests:
+
+```bash
+python3 cpuoptctl/cpuoptctl.py status --sysfs-root tests/fixtures/intel_hwp
+python3 cpuoptctl/cpuoptctl.py profile performance --dry-run --sysfs-root tests/fixtures/intel_hwp
+python3 -m unittest
+```
+
+## Safety model
+
+- CPUOpt prefers existing kernel interfaces over raw register writes.
+- Every write validates path existence, writability intent, and candidate value selection.
+- Unknown sysfs paths are skipped, not guessed.
+- Failed writes are logged and are non-fatal.
+- Modified values are snapshotted to `last_state.json` before application.
+- v0.1 never writes `/dev/cpu/*/msr`, firmware registers, BIOS settings, or BMC controls.
+- Fan writes are intentionally unimplemented in v0.1 even if fan inspection is enabled.
+
+See [docs/SAFETY.md](C:\Users\ManishKL\Documents\Playground\cpuopt-kernel\docs\SAFETY.md) for the full model.
+
+## Example status output
+
+```text
+## CPUOpt Status
+
+Vendor: GenuineIntel
+Model: Intel(R) Core(TM) i7-1280P
+Kernel: Linux 6.8.0-test
+Scaling driver: intel_pstate
+HWP/EPP exposed: yes
+Turbo control: intel_pstate/no_turbo
+Policies:
+  policy0 CPUs=0-7 governor=powersave min=400000 max=4700000 epp=balance_performance
+Thermals:
+  x86_pkg_temp temp=51000
+Warnings:
+  Fan control is platform-specific; not directly controlled unless safely exposed.
+```
+
+## Roadmap highlights
+
+- AMD `amd-pstate` profile support
+- ARM SCMI and heterogeneous topology handling
+- Optional kernel module under `/sys/kernel/cpuopt/`
+- Read-only MSR decoding with model-aware guardrails
+- Model-specific write allowlists only after documentation and fallback validation

cpuoptctl/__init__.py

@@ -0,0 +1 @@
+"""CPUOpt-Kernel user-space controller package."""

cpuoptctl/cpuopt_amd.py

@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def annotate(discovery: dict[str, Any]) -> dict[str, Any]:
+    discovery.setdefault("vendor_notes", []).append("AMD profile writes are not implemented in v0.1.")
+    return discovery

cpuoptctl/cpuopt_arm.py

@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+def annotate(discovery: dict[str, Any]) -> dict[str, Any]:
+    discovery.setdefault("vendor_notes", []).append("ARM policy writes are not implemented in v0.1.")
+    return discovery

cpuoptctl/cpuopt_discovery.py

@@ -0,0 +1,190 @@
+from __future__ import annotations
+
+import platform
+from pathlib import Path
+from typing import Any
+
+try:
+    from .cpuopt_amd import annotate as annotate_amd
+    from .cpuopt_arm import annotate as annotate_arm
+    from .cpuopt_thermal import discover_hwmon, discover_thermal
+    from .cpuopt_utils import first_existing, list_dirs, normalize_whitespace, read_int, read_text
+except ImportError:
+    from cpuopt_amd import annotate as annotate_amd
+    from cpuopt_arm import annotate as annotate_arm
+    from cpuopt_thermal import discover_hwmon, discover_thermal
+    from cpuopt_utils import first_existing, list_dirs, normalize_whitespace, read_int, read_text
+
+
+def _parse_cpuinfo(cpuinfo_path: Path) -> dict[str, str]:
+    parsed: dict[str, str] = {}
+    text = read_text(cpuinfo_path)
+    if not text:
+        return parsed
+    for line in text.splitlines():
+        if ":" not in line:
+            continue
+        key, value = line.split(":", 1)
+        key = key.strip()
+        if key not in parsed:
+            parsed[key] = value.strip()
+    return parsed
+
+
+def _detect_cpuinfo(sysfs_root: Path) -> dict[str, str]:
+    return _parse_cpuinfo(
+        first_existing(
+            [
+                sysfs_root / "proc" / "cpuinfo",
+                sysfs_root.parent / "proc" / "cpuinfo",
+                Path("/proc/cpuinfo"),
+            ]
+        )
+        or Path("/proc/cpuinfo")
+    )
+
+
+def _detect_kernel_release(sysfs_root: Path) -> str:
+    value = read_text(
+        first_existing(
+            [
+                sysfs_root / "proc" / "sys" / "kernel" / "osrelease",
+                sysfs_root.parent / "proc" / "sys" / "kernel" / "osrelease",
+                Path("/proc/sys/kernel/osrelease"),
+            ]
+        )
+        or Path("/proc/sys/kernel/osrelease")
+    )
+    return value or platform.release()
+
+
+def _policy_data(policy: Path) -> dict[str, Any]:
+    available_governors = (read_text(policy / "scaling_available_governors") or "").split()
+    available_freqs = (read_text(policy / "scaling_available_frequencies") or "").split()
+    available_epp = (read_text(policy / "energy_performance_available_preferences") or "").split()
+    return {
+        "name": policy.name,
+        "path": str(policy.resolve()),
+        "related_cpus": read_text(policy / "related_cpus"),
+        "scaling_driver": read_text(policy / "scaling_driver"),
+        "scaling_governor": read_text(policy / "scaling_governor"),
+        "available_governors": available_governors,
+        "available_frequencies": available_freqs,
+        "cpuinfo_min_freq": read_int(policy / "cpuinfo_min_freq"),
+        "cpuinfo_max_freq": read_int(policy / "cpuinfo_max_freq"),
+        "scaling_min_freq": read_int(policy / "scaling_min_freq"),
+        "scaling_max_freq": read_int(policy / "scaling_max_freq"),
+        "scaling_cur_freq": read_int(policy / "scaling_cur_freq"),
+        "energy_performance_preference": read_text(policy / "energy_performance_preference"),
+        "energy_performance_available_preferences": available_epp,
+        "energy_perf_bias": read_text(policy / "energy_perf_bias"),
+    }
+
+
+def _cpuidle_data(sysfs_root: Path) -> list[dict[str, Any]]:
+    cpu_root = sysfs_root / "devices" / "system" / "cpu"
+    states: list[dict[str, Any]] = []
+    for cpu in list_dirs(cpu_root, "cpu[0-9]*"):
+        for state in list_dirs(cpu / "cpuidle", "state*"):
+            states.append(
+                {
+                    "cpu": cpu.name,
+                    "state": state.name,
+                    "name": read_text(state / "name"),
+                    "desc": read_text(state / "desc"),
+                    "latency": read_int(state / "latency"),
+                    "disable": read_text(state / "disable"),
+                }
+            )
+    return states
+
+
+def _topology_data(sysfs_root: Path) -> list[dict[str, Any]]:
+    cpu_root = sysfs_root / "devices" / "system" / "cpu"
+    items: list[dict[str, Any]] = []
+    for cpu in list_dirs(cpu_root, "cpu[0-9]*"):
+        topo = cpu / "topology"
+        items.append(
+            {
+                "cpu": cpu.name,
+                "core_id": read_text(topo / "core_id"),
+                "physical_package_id": read_text(topo / "physical_package_id"),
+                "thread_siblings_list": read_text(topo / "thread_siblings_list"),
+            }
+        )
+    return items
+
+
+def _numa_data(sysfs_root: Path) -> list[dict[str, Any]]:
+    node_root = sysfs_root / "devices" / "system" / "node"
+    items: list[dict[str, Any]] = []
+    for node in list_dirs(node_root, "node*"):
+        items.append(
+            {
+                "name": node.name,
+                "cpulist": read_text(node / "cpulist"),
+                "distance": read_text(node / "distance"),
+            }
+        )
+    return items
+
+
+def discover(sysfs_root: str = "/sys") -> dict[str, Any]:
+    root = Path(sysfs_root)
+    cpuinfo = _detect_cpuinfo(root)
+    cpu_root = root / "devices" / "system" / "cpu"
+    cpufreq_root = cpu_root / "cpufreq"
+    intel_pstate_root = cpu_root / "intel_pstate"
+    amd_pstate_root = cpu_root / "amd_pstate"
+
+    policies = [_policy_data(policy) for policy in list_dirs(cpufreq_root, "policy*")]
+    thermal = discover_thermal(root)
+    hwmon = discover_hwmon(root)
+
+    vulnerabilities: dict[str, str] = {}
+    vuln_root = cpu_root / "vulnerabilities"
+    if vuln_root.exists():
+        for candidate in sorted(vuln_root.iterdir()):
+            if candidate.is_file():
+                value = read_text(candidate)
+                if value is not None:
+                    vulnerabilities[candidate.name] = value
+
+    data: dict[str, Any] = {
+        "sysfs_root": str(root),
+        "vendor": cpuinfo.get("vendor_id") or cpuinfo.get("CPU implementer") or platform.machine(),
+        "arch": platform.machine(),
+        "model_name": normalize_whitespace(cpuinfo.get("model name") or cpuinfo.get("Processor")),
+        "family": cpuinfo.get("cpu family"),
+        "model": cpuinfo.get("model"),
+        "stepping": cpuinfo.get("stepping"),
+        "kernel": _detect_kernel_release(root),
+        "smt_control": read_text(cpu_root / "smt" / "control"),
+        "intel_pstate": {
+            "exists": intel_pstate_root.exists(),
+            "status": read_text(intel_pstate_root / "status"),
+            "no_turbo": read_text(intel_pstate_root / "no_turbo"),
+        },
+        "amd_pstate": {
+            "exists": amd_pstate_root.exists(),
+            "status": read_text(amd_pstate_root / "status"),
+        },
+        "cpufreq_boost": read_text(cpufreq_root / "boost"),
+        "policies": policies,
+        "cpuidle_states": _cpuidle_data(root),
+        "thermal": thermal,
+        "hwmon": hwmon,
+        "topology": _topology_data(root),
+        "numa": _numa_data(root),
+        "vulnerabilities": vulnerabilities,
+        "warnings": ["Fan control is platform-specific; not directly controlled unless safely exposed."],
+        "vendor_notes": [],
+    }
+
+    vendor = str(data["vendor"])
+    if "AuthenticAMD" in vendor:
+        data = annotate_amd(data)
+    if "ARM" in vendor or "aarch64" in vendor or "arm" in str(data["arch"]).lower():
+        data = annotate_arm(data)
+
+    return data

cpuoptctl/cpuopt_intel.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from typing import Iterable
+
+
+PROFILE_EPP_ORDER: dict[str, list[str]] = {
+    "performance": ["performance"],
+    "balanced": ["balance_performance", "balance_power", "default", "performance"],
+    "quiet": ["power", "balance_power", "balance_performance"],
+    "latency": ["performance"],
+    "ai-inference": ["performance", "balance_performance"],
+}
+
+PROFILE_GOVERNOR_ORDER: dict[str, list[str]] = {
+    "performance": ["performance", "schedutil", "powersave"],
+    "balanced": ["schedutil", "powersave", "performance"],
+    "quiet": ["powersave", "schedutil"],
+    "latency": ["performance", "schedutil"],
+    "ai-inference": ["performance", "schedutil"],
+}
+
+
+def choose_preference(available: Iterable[str], preferred: Iterable[str]) -> str | None:
+    available_list = [item for item in available if item]
+    for candidate in preferred:
+        if candidate in available_list:
+            return candidate
+    return None
+
+
+def epp_for_profile(profile_name: str, available: Iterable[str]) -> str | None:
+    return choose_preference(available, PROFILE_EPP_ORDER.get(profile_name, []))
+
+
+def governor_for_profile(profile_name: str, available: Iterable[str]) -> str | None:
+    return choose_preference(available, PROFILE_GOVERNOR_ORDER.get(profile_name, []))
+
+
+def epb_for_profile(profile_name: str) -> str | None:
+    mapping = {
+        "performance": "0",
+        "balanced": "6",
+        "quiet": "15",
+        "latency": "0",
+        "ai-inference": "0",
+    }
+    return mapping.get(profile_name)