diff --git a/AGENTS.md b/AGENTS.md index 23b2367..d8ab768 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,7 +4,9 @@ Berth is the VRAM leasing service for every GPU you run. One Go daemon per machi hands a single card to many inference engines in turn: engines lease the card, the holder is drained, Berth waits until the driver reports the memory free, then admits the next engine. Processes holding VRAM without a lease are stowaways, -reported, never touched. Linux + NVIDIA. Alpha. +reported, never touched. Linux + NVIDIA; Linux AMD discrete RDNA3/RDNA4 and Phoenix +APU (780M) support is experimental. APU leases are exclusive, with separate reserved +UMA VRAM/GTT accounting and host-RAM admission checks; APU parking is not supported. Alpha. ## Commands @@ -43,8 +45,10 @@ local proof command green before opening a PR; CI must end green at merge. port discovery by response fingerprint, and the 24 h request log. - `internal/export` — the store's rows as CSV or JSON for the `berth export` command. - `internal/fsx` — the one atomic file write (fsync the file and its directory). -- `internal/gpu` — nvidia-smi polling (memory, compute apps) and each card's real - limits; on Windows, per-process VRAM from the PDH counter Task Manager reads. +- `internal/gpu` — the shared Probe contract, NVIDIA nvidia-smi polling, and Linux + AMD sysfs/DRM observations (discrete RDNA3/RDNA4 and Phoenix APUs); on Windows, NVIDIA per-process + VRAM from the PDH counter Task Manager reads. Guarded AMD reads require fresh + driver evidence for admission and release. See contrib/amd/README.md for validation. - `internal/harbor` — the `berth harbor` web UI server and its collectors; UI assets live in `internal/harbor/web/`. - `internal/history` — the node's last hour of samples in memory at 5 s resolution. diff --git a/README.md b/README.md index 461af29..545f29a 100644 --- a/README.md +++ b/README.md @@ -153,8 +153,31 @@ line goes away. - Every reporting command takes `--format json|yaml`. All of them keep results on stdout and diagnostics on stderr, and exit 0, 1 or 2 by one rule. -**Runs on** Linux with an NVIDIA GPU (supported) and Windows 10/11 with an NVIDIA GPU -(experimental). Apache-2.0. +**Runs on** Linux with an NVIDIA GPU (supported), Linux with AMD RDNA3/RDNA4 discrete +GPUs or a Phoenix APU such as the 780M (experimental), and Windows 10/11 with an NVIDIA +GPU (experimental). Apache-2.0. + +### AMD on Linux + +Select the AMD probe explicitly (or let `auto` select it on an AMD-only host): + +```sh +berth daemon --gpu-backend amd --gpu 0 +# Or install the user service with that selection: +berth service install --harbor --gpu-backend amd --gpu 0 +``` + +For AMD, `--gpu N` means Linux DRM **cardN**, not a HIP/ROCm ordinal. The probe matches +processes by PCI address, not PID alone. On mixed NVIDIA/AMD installations, choose +`--gpu-backend` explicitly; auto refuses ambiguity. This selects what Berth observes, +not where the engine loads: configure the engine's GPU placement separately. + +The AMD adapter reads kernel sysfs and DRM `fdinfo`; no ROCm installation, `amd-smi`, +root privileges, or extra monitoring service is required. Phoenix APUs keep reserved +UMA VRAM and shared-RAM GTT separate; use exclusive leases (omit `--needs-mib`). New +APU admissions require host RAM above `--headroom-mib` (default 2048 MiB); parking is +not yet supported on APUs, so engines use the normal drain. Other APU generations and +AMD Windows are not supported. See [scope and validation](contrib/amd/README.md). ## Quick start @@ -230,7 +253,7 @@ The same pages are a site with search at . |---|---| | Linux + NVIDIA | Supported | | Windows 10/11 + NVIDIA | Experimental: leases, process attribution, job-object drains and game claims work, and the test suite runs on Windows in CI. Per-process VRAM comes from the GPU Process Memory counter (Task Manager's source), which also lists graphics apps and games nvidia-smi omits; host RAM and the page file come from GlobalMemoryStatusEx, and root space from GetDiskFreeSpaceEx. `berth service install` sets the node up as logon tasks; still no packaged installer and no locked-down control pipe. `contrib/windows/check.ps1` compares Berth's per-process VRAM with Task Manager's and writes a report for the drain report form. | -| AMD on Linux | Not supported yet; planned for v0.4 (a vendor-neutral probe, then amdgpu) | +| AMD on Linux | Experimental: discrete RDNA3/RDNA4 and Phoenix APUs through amdgpu sysfs/DRM. W7900 and 780M bounded allocation/handoff validated; RDNA4 fixture-tested, hardware validation pending. APU leases are exclusive, with independent UMA/GTT accounting and host-RAM admission checks. [Scope and checks](contrib/amd/README.md). | | Intel, Apple | Not supported yet | | Multiple GPUs | Inventoried with real limits; leases schedule one card per daemon today | diff --git a/contrib/amd/README.md b/contrib/amd/README.md new file mode 100644 index 0000000..f89fc74 --- /dev/null +++ b/contrib/amd/README.md @@ -0,0 +1,246 @@ +# Linux AMD support and validation + +Experimental support for **discrete RDNA3/RDNA4 and Phoenix APUs (780M)**, using the existing +`gpu.Probe` interface. No ROCm, amd-smi, CGO, or device-context dependency in Berth. +The adapter does not allocate GPU memory or change driver settings. + +## Selection and requirements + +```sh +berth daemon --gpu-backend amd --gpu 0 +berth service install --gpu-backend amd --gpu 0 --harbor +berth status --format json +``` + +AMD indices are DRM `cardN` numbers, retained even when unsupported cards are +excluded; they are not compacted and are not HIP device ordinals. Inventory and +process clients are joined by normalized PCI address, including the domain. +`auto` selects AMD when supported AMD cards exist and nvidia-smi is absent; if +both are present it requires an explicit backend. NVIDIA's indices and optional +streaming implementation are unchanged. The swap bench accepts `-gpu-backend` too. + +Supported GC IPs are 11.0.0, 11.0.2, 11.0.3 (RDNA3 discrete), 12.0.0, 12.0.1 +(RDNA4 discrete), and 11.0.1 (Phoenix APU). Other APU IPs, including 11.0.4 and +11.5, remain excluded until validated. APUs are never treated as discrete-VRAM +devices. The kernel must expose +`ip_discovery/die/0/GC/0/{major,minor,revision}`, `mem_info_vram_{total,used}` and +DRM client memory counters. Use an amdgpu-enabled kernel/firmware supporting the +actual card; this feature does not install or upgrade drivers. + +### Phoenix/780M admission policy + +```sh +berth daemon --gpu-backend amd --gpu 1 --headroom-mib 2048 +# Omit --needs-mib: the initial APU implementation supports exclusive leases. +berth run --as my-apu-engine -- MY_ENGINE_COMMAND +``` + +The GPU index must match the actual `cardN`; 1 is this validation host's 780M, +not a portable HIP ordinal. Each client is still matched by full PCI identity. +APU `total`, `used`, `free`, and `baseline` describe **only the firmware-reserved +UMA VRAM pool**. `gtt_total`, `gtt_used`, and `gtt_baseline` describe the independent +shared-system-RAM driver pool. GTT capacity is not additional VRAM, available +host RAM, or reserved memory. GTT-only clients remain visible as GPU holders. +Leases, claims and stowaways expose their shared use as `gtt_used`; human status +labels both pools. Existing VRAM history/graphs remain reserved-pool views, not +a graph of combined APU memory. + +New APU admissions require current `MemAvailable` to exceed `--headroom-mib` +(default 2048 MiB; 0 retains the safety default), and host pressure below the +configured park-pressure threshold. Unknown RAM availability blocks admission. +A low-RAM holder can still drain: the floor gates the next grant, not the drain. +This floor is not a reservation and cannot guarantee an unbounded engine fits. +Other host/dGPU workloads compete for this RAM too. + +Sized APU acquisitions and reattachments are refused because a single +`--needs-mib` cannot specify both allocation domains. **APU parking is currently +refused explicitly**, falling back to normal unload/stop: treating GTT-resident +weights as a successful offload would be unsafe. Discrete-GPU sized leases and +parking retain their existing behavior. + +Run Berth on the host with its real `/proc` and `/sys` views, normally as the +same user as its engines. Restrictive procfs permissions, a PID namespace, or +unobservable cross-user/container processes limit attribution. Do not solve that +by granting blanket root privileges; first check process visibility and run the +opt-in validation on the target deployment. + +## Memory and safety contract + +- Card-wide total/used VRAM comes from the driver, in bytes. Free is their + non-negative difference; GTT/system RAM is never added to dedicated VRAM. +- Per-process data uses resident VRAM (`drm-resident-vram`, falling back to + legacy `drm-memory-vram`), never requested-buffer totals. The two resident + spellings are aliases and are not added together. +- Duplicated/inherited DRM descriptors are deduplicated by PCI device and client + ID. When a client is shared between PIDs, the lowest observed PID represents + it; attribution to separate leases is inherently ambiguous. +- Shared buffer bytes are discounted from per-client residency because fdinfo + has no object identity with which to deduplicate them across clients. Reported + process use is consequently a private-resident lower bound; shared and other + unattributed allocations stay in the card-wide baseline. On older kernels, + unmapped GEM handles can also exist only in that baseline until GPU-VA mapping. +- Same-user PIDs whose fdinfo is inaccessible appear as `unobserved_pids` on the + scheduled GPU in status JSON and as a human status warning. Previously observed + ownership is retained; unreadable is never treated as a confirmed empty scan. + Unreadable foreign-user allocations remain in the baseline without PID details. +- Failed or stale AMD observations block new admissions. A sampled PID/start-time + mismatch invalidates the read instead of assigning old memory to a reused PID. +- Sized admissions must fit both the lease budget and fresh driver free bytes. + Release/park confirmation requires the unattributed driver memory to return to + the lease's pre-load baseline, as well as the existing process and read-order + checks. Missing process data alone cannot confirm release or a park. +- APU release additionally requires the GTT baseline to return independently. + VRAM becoming free cannot conceal lingering shared buffers, and a decrease + in one pool cannot offset a leak in the other. GTT/CPU backing and the kernel's + available-RAM estimate are never added together as capacity. +- This is conservative accounting, not an allocation reservation or an atomic + driver snapshot. Changing shared/unobservable background allocations can delay + confirmation. Whole-card leases retain Berth's existing report-only contested + grant behavior: they do not promise an otherwise empty card or isolate engines. + Configure engine device placement separately and use sized requests when the + required footprint is known. Stowaways are never stopped. +- Optional temperature, utilization, power, clocks and PCIe readings use sysfs/ + hwmon. Unsupported values are -1. RPM/PWM are not mislabelled as fan percent. + +## Milestones and repeatable checks + +### 1. Unchanged baseline + +Run the repository's proof, exact lint, and Windows cross-build/vet commands from +AGENTS.md before changes. These passed on the starting tree. Go 1.27.1 was used; +no dependency, Go version, linter version, or system package changes were needed. + +### 2. Read-only observation + +```sh +BERTH_TEST_AMD_LIVE=1 go test -v ./internal/gpu -run TestAMDLiveReadOnly -count=1 +``` + +This only inspects host sysfs/procfs and defaults to card0. It never allocates or +unloads. Select the 780M with `BERTH_TEST_AMD_CARD=1` on this host. On 2026-10-22, +Linux 6.12.107+deb13-amd64 exposed the W7900's driver totals +and existing ROCm Python and Vulkan llama-server clients at PCI 0000:03:00.0. The +co-located 780M was initially excluded, then validated separately with both UMA +and GTT observations. Some systemd/SSH PIDs were inaccessible and were reported +explicitly. Observation alone does not validate engine unload behavior. + +### 3. Fixture and broker safety gates + +```sh +go test ./internal/gpu ./internal/broker ./internal/cli ./contrib/bench/swap +go test -race ./internal/gpu ./internal/broker ./internal/cli +``` + +Fixtures cover all six admitted GC IPs, legacy/modern resident counters, PCI +domains, multi-card numbering, unvalidated-APU exclusion, shared/duplicate clients, malformed +data, unit overflow, permissions, missing optional metrics and cancellation. +Broker regressions cover driver-delayed release, disappearing/unreadable PIDs, +stale/failed observations, PID reuse, park residual confirmation and unrelated +memory releases that must not conceal lingering buffers. APU regressions cover +GTT-only ownership, independent pool release, unreadable GTT holders, RAM/pressure +admission and recovery, exclusive-only requests, and explicit park refusal. Service tests check that +AMD/card flags reach the generated daemon unit, not the gateway or harbor. +The guarded-probe regression also preserves upstream's VFIO hold/probe-suppression +and return-to-host behavior. The existing VFIO CLI still uses nvidia-smi; AMD +VFIO command support and physical passthrough tests are not part of this feature. + +### 4. Opt-in real allocation and handoff + +**Ask the operator first.** This allocates actual GPU memory; it is not a read-only +check. Build a scratch binary, identify the matching render node by PCI device, +and ensure at least 128 MiB is free: + +```sh +go build -o /tmp/opencode/berth-amd ./cmd/berth +python3 contrib/amd/check.py --allow-allocation \ + --berth /tmp/opencode/berth-amd --gpu 0 --render-node /dev/dri/renderD128 \ + --output /tmp/opencode/berth-amd-validation +``` + +The Python standard-library helper starts a separate daemon/socket/config with +engine discovery and claims disabled, and disables persistent usage files. It +maps a 64 MiB VRAM GEM buffer in a `berth run` child, confirms measured ownership, +requests a higher-priority second child, and checks that the first lease is freed +before the second is granted. Both children and the daemon are its own processes; +it never connects to installed services or stops existing GPU workloads. Evidence +is `report.json` plus logs in the requested directory. Use a new output directory +for each run. This validates the driver/lease path, not model correctness or a +specific engine's ROCm park/resume implementation. + +For the 780M, test a handoff from reserved UMA into GTT: + +```sh +python3 contrib/amd/check.py --allow-allocation \ + --berth /tmp/opencode/berth-amd --gpu 1 --render-node /dev/dri/renderD129 \ + --pool mixed --output /tmp/opencode/berth-780m-validation +``` + +`--pool gtt` tests GTT-to-GTT instead. APU checks also create a separate daemon +with an intentionally impossible RAM floor and verify that its child never starts; +they do not create real memory pressure. The 780M mixed-pool handoff and RAM guard +passed on 2026-10-22. Reported UMA and GTT baselines are checked independently. + +### Real embedding-engine smoke (separate approval) + +`model-check.py` runs an **existing** local Vulkan llama-server and embedding GGUF +under a private Berth lease, sends two embedding requests, validates finite normalized +768-dimensional vectors, then tests an idle `/slots` boundary and handoff to a GTT +buffer. It downloads nothing and never restarts installed services. The default +4 GiB RSS budget requires 6 GiB available host RAM before starting. A smaller +`--host-rss-limit-mib` (512..4096) reduces the test budget, **not** the additional +2 GiB host floor. The validated 1 GiB budget therefore requires 3 GiB available. +It also requires 2 GiB free reserved UMA and monitors model residency against a +2 GiB GPU ceiling and the selected kernel RSS high-water budget. These watchdog +limits are not reservations or a cgroup memory guarantee. + +```sh +python3 contrib/amd/model-check.py --allow-allocation \ + --berth /tmp/opencode/berth-amd --engine /PATH/TO/VULKAN/llama-server \ + --model /PATH/TO/embeddinggemma-2.gguf --gpu 1 --render-node /dev/dri/renderD129 \ + --host-rss-limit-mib 1024 \ + --output /tmp/opencode/berth-780m-model-validation +``` + +This helper uses `device-only.py`, a process-local Landlock allowlist (Linux +x86_64/aarch64, ABI >= 5), so only the selected DRM render node can be opened or +used for device ioctls. It fails closed if isolation is unavailable. Landlock is +only a validation-helper requirement, **not a Berth runtime dependency**. Device +environment hints alone are insufficient: the initial native model attempt was +stopped when the Vulkan loader probed the W7900 despite selecting the 780M. + +The isolated real-model smoke **passed** on the 780M with an existing 175 MiB +EmbeddingGemma 2 GGUF, 512-token context and two CPU threads. Two requests produced +finite normalized 768-dimensional embeddings. The process's only observed GPU +was PCI 0000:0a:00.0. Berth drained the idle model at its `/slots` boundary, confirmed +release from both pools, and admitted a 64 MiB GTT child. UMA and GTT totals and +baselines returned exactly to their pre-test values. Observed kernel RSS high-water +was below 300 MiB and observed GPU residency below 180 MiB. This is a functional +smoke, not a quality/throughput benchmark, full-context stress test, or proof of +every ROCm/HIP memory mode. No existing service was stopped or reconfigured. + +The default-budget retry had first refused when host availability fell to about +2.7 GiB. After RAM recovered, the test budget was reduced to 1 GiB RSS while +preserving the same 2 GiB host floor. Never stop unrelated workloads or reduce the +safety floor merely to make validation pass. + +**Hardware evidence:** W7900 (RDNA3/gfx1100), kernel 6.12.107+deb13-amd64: real +allocation, attribution, priority drain and measured handoff passed. RDNA4 is +fixture-tested but **has not been physically tested** in this environment. +NVIDIA compatibility is covered by its existing tests and Windows build/vet; +no NVIDIA card was available for a live regression test. The 780M has bounded +dual-pool allocation/handoff, RAM-admission, and isolated Vulkan embedding-engine +validation. Engine-specific live park/resume, sized APU sharing, +other APU IPs, AMD Windows and other AMD generations remain outside this claim. + +### 5. Final regression gate + +Run the full Linux proof and exact lint commands, Windows build/vet, and the +opt-in hardware check after the final source change. Keep all checks green; +record unrelated failures rather than changing unrelated code. + +## References + +- [Kernel AMDGPU sysfs memory counters](https://docs.kernel.org/gpu/amdgpu/driver-misc.html#gpu-memory-usage-information) +- [DRM client identity, residency, sharing and units](https://www.kernel.org/doc/html/latest/gpu/drm-usage-stats.html) +- [Linux AMDGPU GC 11 platform handling](https://github.com/torvalds/linux/blob/master/drivers/gpu/drm/amd/amdgpu/soc21.c) +- [Linux AMDGPU GC 12 platform handling](https://github.com/torvalds/linux/blob/master/drivers/gpu/drm/amd/amdgpu/soc24.c) diff --git a/contrib/amd/check.py b/contrib/amd/check.py new file mode 100644 index 0000000..f897c91 --- /dev/null +++ b/contrib/amd/check.py @@ -0,0 +1,216 @@ +# Copyright 2026 Trevin Tindall (T-Crypt) +# SPDX-License-Identifier: Apache-2.0 + +"""Opt-in, isolated Linux AMD allocation/handoff check; no model or ROCm needed. + +Uses two 64 MiB GEM buffers sequentially, on exactly the selected render node. +Never connects to the installed daemon or stops an existing workload. The +operator must explicitly allow allocation. Not part of CI or normal go tests. +""" + +import argparse +import fcntl +import json +import os +from pathlib import Path +import signal +import shutil +import struct +import subprocess +import sys +import tempfile +import time + + +SIZE = 64 << 20 + + +def buffer(node, ready, pool): + # DRM_IOCTL_AMDGPU_GEM_CREATE: 32-byte in/out union, VRAM domain = 4. + # A GEM allocation is real driver memory, not a HIP/Vulkan compute context. + fd = os.open(node, os.O_RDWR | os.O_CLOEXEC) + try: + domain = 2 if pool == "gtt" else 4 + req = bytearray(struct.pack("=QQQQ", SIZE, 4096, domain, 0)) + fcntl.ioctl(fd, 0xC0206440, req, True) + handle = struct.unpack_from("=I", req)[0] + # Legacy amdgpu fdinfo accounts VM-mapped buffers, not every GEM + # handle. Map this object into this file's GPU VA, as runtimes do. + fcntl.ioctl(fd, 0x40286448, struct.pack("=IIIIQQQ", handle, 0, 1, 6, 0x100000000, 0, SIZE)) + stop = False + + def finish(_signum, _frame): + nonlocal stop + stop = True + + signal.signal(signal.SIGTERM, finish) + signal.signal(signal.SIGINT, finish) + Path(ready).write_text(json.dumps({"pid": os.getpid(), "bytes": SIZE, "pool": pool})) + deadline = time.monotonic() + 30 + while not stop and time.monotonic() < deadline: + time.sleep(0.05) + fcntl.ioctl(fd, 0x40286448, struct.pack("=IIIIQQQ", handle, 0, 2, 0, 0x100000000, 0, SIZE)) + # DRM_IOCTL_GEM_CLOSE. Closing fd in finally also releases all handles. + fcntl.ioctl(fd, 0x40086409, struct.pack("=II", handle, 0)) + finally: + os.close(fd) + + +def wait_for(predicate, description): + deadline = time.monotonic() + 20 + while time.monotonic() < deadline: + result = predicate() + if result: + return result + time.sleep(0.1) + raise RuntimeError("timed out waiting for " + description) + + +def check(args): + card = Path(f"/sys/class/drm/card{args.gpu}/device") + render = Path("/sys/class/drm") / Path(args.render_node).name / "device" + if card.resolve(strict=True) != render.resolve(strict=True): + raise RuntimeError("render node and selected DRM card refer to different GPUs") + free = int((card / "mem_info_vram_total").read_text()) - int( + (card / "mem_info_vram_used").read_text() + ) + if free < 2 * SIZE: + raise RuntimeError("less than 128 MiB free; refusing to allocate") + if args.pool != "vram": + gtt_free = int((card / "mem_info_gtt_total").read_text()) - int((card / "mem_info_gtt_used").read_text()) + available = next(int(line.split()[1]) * 1024 for line in Path("/proc/meminfo").read_text().splitlines() if line.startswith("MemAvailable:")) + if gtt_free < 2 * SIZE or available < (2 << 30) + 2 * SIZE: + raise RuntimeError("insufficient GTT/host RAM headroom; refusing to allocate") + directory = Path(args.output) if args.output else Path(tempfile.mkdtemp(prefix="berth-amd-")) + if args.output: + directory.mkdir(parents=True, exist_ok=False) + sock = str(directory / "berth.sock") + config = directory / "berth.toml" + config.write_text("[engines]\ndiscover = false\n[claims]\nbuiltin = false\n[priority]\njuggle_window = '-1s'\n") + binary = str(Path(shutil.which(args.berth) or args.berth).resolve(strict=True)) + processes = [] + logs = [] + + def spawn(name, argv): + log = (directory / (name + ".log")).open("w") + logs.append(log) + proc = subprocess.Popen(argv, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) + processes.append(proc) + return proc + + def status(): + result = subprocess.run( + [binary, "status", "--socket", sock, "--format", "json"], + capture_output=True, text=True, timeout=5, check=True, + ) + return json.loads(result.stdout) + + def holding(name, pool): + st = status() + report["last_observation"] = st + field = "gtt_used" if pool == "gtt" else "used" + matches = [l for l in st["leases"] if l["engine"] == name and l.get(field, 0) >= SIZE] + return st if matches else None + + def released(): + st = status() + return st if not st["leases"] else None + + report = {"result": "failed"} + try: + daemon = spawn("daemon", [binary, "daemon", "--socket", sock, + "--gpu-backend", "amd", "--gpu", str(args.gpu), "--poll", "250ms", + "--config", str(config), "--counters", "", "--usage", "", "--footprint", ""]) + wait_for(lambda: Path(sock).exists() or daemon.poll() is not None, "isolated daemon") + if daemon.poll() is not None: + raise RuntimeError("daemon failed; see " + str(directory / "daemon.log")) + report["initial"] = status() + scheduled = next(g for g in report["initial"]["gpus"] if g["scheduled"]) + if args.pool != "vram" and not scheduled.get("apu"): + raise RuntimeError("GTT checks require an APU; refusing to allocate on another device") + if scheduled.get("apu"): + # Simulate an insufficient floor, without creating actual RAM + # pressure or allocating anything. The child must never start. + guard_sock = str(directory / "ram-guard.sock") + guard = spawn("ram-guard", [binary, "daemon", "--socket", guard_sock, + "--gpu-backend", "amd", "--gpu", str(args.gpu), "--poll", "250ms", + "--headroom-mib", "1048576", "--config", str(config), + "--counters", "", "--usage", "", "--footprint", ""]) + wait_for(lambda: Path(guard_sock).exists() or guard.poll() is not None, "RAM-guard daemon") + marker = directory / "must-not-start" + blocked = subprocess.run([binary, "run", "--socket", guard_sock, "--config", str(config), + "--as", "ram-guard", "--wait", "500ms", "--fail-closed", "--", sys.executable, + "-c", "from pathlib import Path; Path(" + repr(str(marker)) + ").touch()"], + capture_output=True, text=True, timeout=10) + if blocked.returncode == 0 or marker.exists() or "APU host RAM headroom" not in blocked.stderr: + raise RuntimeError("APU RAM guard did not refuse before starting the child: " + blocked.stderr) + report["ram_guard"] = "passed" + guard.terminate() + guard.wait(timeout=10) + pools = ["vram", "gtt"] if args.pool == "mixed" else [args.pool, args.pool] + for (name, priority), pool in zip([("amd-check-a", "batch"), ("amd-check-b", "foreground")], pools): + spawn(name, [binary, "run", "--socket", sock, "--config", str(config), + "--as", name, "--priority", priority, "--fail-closed", "--grace", "2s", + "--wait", "15s", "--", sys.executable, str(Path(__file__).resolve()), + "--allow-allocation", "--buffer", "--render-node", args.render_node, + "--pool", pool, + "--ready", str(directory / (name + ".json"))]) + report[name] = wait_for(lambda: holding(name, pool), name + " " + pool + " allocation attribution") + a_pid = json.loads((directory / "amd-check-a.json").read_text())["pid"] + b_pid = json.loads((directory / "amd-check-b.json").read_text())["pid"] + if Path(f"/proc/{a_pid}/fdinfo").exists(): + raise RuntimeError("first buffer process still exists when the second was admitted") + # Stop only our own wrapper; runner forwards the signal to its child. + processes[-1].terminate() + processes[-1].wait(timeout=10) + report["final"] = wait_for(released, "final driver-confirmed release") + events = subprocess.run([binary, "events", "--socket", sock, "--format", "json"], + capture_output=True, text=True, timeout=5, check=True) + report["events"] = [json.loads(line) for line in events.stdout.splitlines() if line.strip()] + if not any(e["kind"] == "lease.draining" and e.get("engine") == "amd-check-a" for e in report["events"]): + raise RuntimeError("first engine was not drained by the higher-priority waiter") + freed = next(e["seq"] for e in report["events"] if e["kind"] == "lease.freed" and e.get("engine") == "amd-check-a") + granted = next(e["seq"] for e in report["events"] if e["kind"] == "lease.granted" and e.get("engine") == "amd-check-b") + if freed >= granted: + raise RuntimeError("next lease was granted before the previous release was confirmed") + report["result"] = "passed" + report["buffer_pids"] = [a_pid, b_pid] + report["bytes_per_buffer"] = SIZE + report["pools"] = pools + print("PASS: real AMD allocation, PID attribution, priority drain, driver-confirmed release and handoff") + finally: + for proc in reversed(processes): + if proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + for log in logs: + log.close() + (directory / "report.json").write_text(json.dumps(report, indent=2) + "\n") + print("Evidence:", directory) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--allow-allocation", action="store_true") + parser.add_argument("--berth", default="berth") + parser.add_argument("--gpu", type=int, default=0) + parser.add_argument("--render-node", default="/dev/dri/renderD128") + parser.add_argument("--output") + parser.add_argument("--pool", choices=["vram", "gtt", "mixed"], default="vram", help="APU: test reserved VRAM, shared GTT, or a handoff between them") + parser.add_argument("--buffer", action="store_true", help=argparse.SUPPRESS) + parser.add_argument("--ready", help=argparse.SUPPRESS) + args = parser.parse_args() + if not args.allow_allocation: + parser.error("requires --allow-allocation; allocates up to 128 MiB on the selected GPU") + if args.buffer: + buffer(args.render_node, args.ready, args.pool) + else: + check(args) + + +if __name__ == "__main__": + main() diff --git a/contrib/amd/device-only.py b/contrib/amd/device-only.py new file mode 100644 index 0000000..c43b9e2 --- /dev/null +++ b/contrib/amd/device-only.py @@ -0,0 +1,88 @@ +# Copyright 2026 Trevin Tindall (T-Crypt) +# SPDX-License-Identifier: Apache-2.0 + +"""Run a validation child with access to exactly one DRM render node. + +Linux Landlock ABI >= 5 is required; there is no unisolated fallback. This is a +test helper, not a Berth runtime dependency. No sudo, namespaces, driver changes +or system-wide permissions are used. Normal filesystem permissions still apply. +""" + +import ctypes +import os +from pathlib import Path +import platform +import sys + + +def main(): + if len(sys.argv) < 4 or sys.argv[2] != "--": + raise SystemExit("usage: device-only.py /dev/dri/renderDN -- COMMAND [ARGS...]") + if sys.platform != "linux" or platform.machine() not in ["x86_64", "aarch64"]: + raise SystemExit("validated Landlock syscall numbers require Linux x86_64/aarch64") + node = Path(sys.argv[1]).resolve(strict=True) + if node.parent != Path("/dev/dri") or not node.name.startswith("renderD"): + raise SystemExit("select one DRM render node") + libc = ctypes.CDLL(None, use_errno=True) + + def call(number, *args): + result = libc.syscall(number, *args) + if result < 0: + error = ctypes.get_errno() + raise OSError(error, os.strerror(error)) + return result + + abi = call(444, 0, 0, 1) + if abi < 5: + raise SystemExit("Landlock ABI >= 5 is required for device-ioctl isolation") + rights = (1 << 1) | (1 << 2) | (1 << 15) # WRITE_FILE, READ_FILE, IOCTL_DEV + ruleset_attr = ctypes.c_uint64(rights) + ruleset = call(444, ctypes.byref(ruleset_attr), ctypes.sizeof(ruleset_attr), 0) + + class PathRule(ctypes.Structure): + _pack_ = 1 + _fields_ = [("allowed_access", ctypes.c_uint64), ("parent_fd", ctypes.c_int32)] + + def allow(path): + fd = os.open(path, os.O_PATH | os.O_CLOEXEC) + try: + rule = PathRule(rights, fd) + call(445, ruleset, 1, ctypes.byref(rule), 0) + finally: + os.close(fd) + + try: + for entry in Path("/").iterdir(): + if entry.name != "dev": + try: + allow(entry) + except FileNotFoundError: + pass + for name in ["null", "zero", "full", "random", "urandom", "tty", "shm", "pts"]: + path = Path("/dev") / name + if path.exists(): + allow(path) + allow(node) + if libc.prctl(38, 1, 0, 0, 0) != 0: # PR_SET_NO_NEW_PRIVS, child only + error = ctypes.get_errno() + raise OSError(error, os.strerror(error)) + call(446, ruleset, 0) + finally: + os.close(ruleset) + # Fail closed if an unexpected device is accessible. Landlock checks open + # permissions before the DRM driver's open callback creates any context. + for other in Path("/dev/dri").glob("*"): + if other.is_dir() or other.resolve() == node: + continue + try: + fd = os.open(other, os.O_RDWR | os.O_CLOEXEC) + except PermissionError: + continue + else: + os.close(fd) + raise RuntimeError("unexpected GPU node remains accessible: " + str(other)) + os.execvpe(sys.argv[3], sys.argv[3:], os.environ) + + +if __name__ == "__main__": + main() diff --git a/contrib/amd/model-check.py b/contrib/amd/model-check.py new file mode 100644 index 0000000..41ef661 --- /dev/null +++ b/contrib/amd/model-check.py @@ -0,0 +1,209 @@ +# Copyright 2026 Trevin Tindall (T-Crypt) +# SPDX-License-Identifier: Apache-2.0 + +"""Opt-in Phoenix APU Vulkan embedding/handoff smoke using an existing GGUF. + +No downloads or installed-service changes. Requires a local Vulkan llama-server +and embedding model. Uses a private Berth daemon, loopback port and a bounded +follow-up GTT allocation. GPU pools stay separate in the report. This is a +functional smoke test, not an embedding-quality or performance benchmark. +""" + +import argparse +import json +import math +import os +from pathlib import Path +import shutil +import socket +import subprocess +import sys +import time +import urllib.error +import urllib.request + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--allow-allocation", action="store_true") + parser.add_argument("--berth", required=True) + parser.add_argument("--engine", required=True) + parser.add_argument("--model", required=True) + parser.add_argument("--gpu", type=int, default=1) + parser.add_argument("--render-node", default="/dev/dri/renderD129") + parser.add_argument("--output", required=True) + parser.add_argument("--host-rss-limit-mib", type=int, default=4096, help="watchdog budget, 512..4096 MiB; preflight preserves an additional 2 GiB host floor") + args = parser.parse_args() + if not args.allow_allocation: + parser.error("requires --allow-allocation; runs a model on the APU (2 GiB GPU / 4 GiB RSS limits)") + if not 512 <= args.host_rss_limit_mib <= 4096: + parser.error("host RSS budget must stay within 512..4096 MiB") + rss_limit = args.host_rss_limit_mib << 20 + binary = str(Path(shutil.which(args.berth) or args.berth).resolve(strict=True)) + engine = str(Path(args.engine).resolve(strict=True)) + model = Path(args.model).resolve(strict=True) + if model.stat().st_size > (1 << 30): + parser.error("use an embedding GGUF of at most 1 GiB for this bounded smoke") + card = Path(f"/sys/class/drm/card{args.gpu}/device") + render = Path("/sys/class/drm") / Path(args.render_node).name / "device" + if card.resolve(strict=True) != render.resolve(strict=True): + parser.error("DRM card and render node refer to different GPUs") + if (card / "ip_discovery/die/0/GC/0/revision").read_text().strip() != "1": + parser.error("this check is for the validated Phoenix APU, not a discrete GPU") + available = next(int(l.split()[1]) * 1024 for l in Path("/proc/meminfo").read_text().splitlines() if l.startswith("MemAvailable:")) + free = int((card / "mem_info_vram_total").read_text()) - int((card / "mem_info_vram_used").read_text()) + if available < (2 << 30) + rss_limit or free < (2 << 30): + parser.error(f"need {2048 + args.host_rss_limit_mib} MiB available host RAM and 2 GiB free reserved UMA before testing") + pci = card.resolve().name + vendor = (card / "vendor").read_text().strip().removeprefix("0x") + device = (card / "device").read_text().strip().removeprefix("0x") + directory = Path(args.output) + directory.mkdir(parents=True, exist_ok=False) + config = directory / "berth.toml" + config.write_text("[engines]\ndiscover = false\n[claims]\nbuiltin = false\n[priority]\njuggle_window = '-1s'\n") + sock = str(directory / "berth.sock") + with socket.socket() as reservation: + reservation.bind(("127.0.0.1", 0)) + port = reservation.getsockname()[1] + base = f"http://127.0.0.1:{port}" + env = os.environ.copy() + for key in list(env): + if key.startswith("LLAMA_ARG_") or key in ["GGML_VK_PREFER_HOST_MEMORY", "GGML_VK_ALLOW_SYSMEM_FALLBACK"]: + del env[key] + env["MESA_VK_DEVICE_SELECT"] = vendor + ":" + device + "!" + env["GGML_VK_VISIBLE_DEVICES"] = "0" + processes, logs = [], [] + report = {"result": "failed", "pci": pci, "port": port, "host_rss_limit": rss_limit, "peak_observed_rss": 0, "peak_observed_gpu": 0, "observed_gpu_pci": []} + + def spawn(name, argv, child_env=None): + log = (directory / (name + ".log")).open("w") + logs.append(log) + p = subprocess.Popen(argv, env=child_env, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) + processes.append(p) + return p + + def status(): + r = subprocess.run([binary, "status", "--socket", sock, "--format", "json"], capture_output=True, text=True, timeout=5, check=True) + return json.loads(r.stdout) + + def guard(): + if not Path(sock).exists(): + return + st = status() + report["last_observation"] = st + for lease in st["leases"]: + if lease["engine"] != "apu-model": + continue + report["peak_observed_gpu"] = max(report["peak_observed_gpu"], lease["used"] + lease.get("gtt_used", 0)) + if lease["used"] + lease.get("gtt_used", 0) > (2 << 30): + raise RuntimeError("model exceeded the 2 GiB GPU allocation ceiling") + pids = set(lease.get("gpu_pids", [])) | {lease.get("pid", 0)} + for pid in pids - {0}: + try: + rss = int(Path(f"/proc/{pid}/statm").read_text().split()[1]) * os.sysconf("SC_PAGE_SIZE") + for line in Path(f"/proc/{pid}/status").read_text().splitlines(): + if line.startswith("VmHWM:"): + rss = max(rss, int(line.split()[1]) * 1024) + report["peak_observed_rss"] = max(report["peak_observed_rss"], rss) + if rss > rss_limit: + raise RuntimeError("model exceeded its resident host RAM watchdog budget") + for fd in Path(f"/proc/{pid}/fdinfo").iterdir(): + text = fd.read_text() + if "drm-driver:\tamdgpu" in text: + actual = next(l.split(":", 1)[1].strip() for l in text.splitlines() if l.startswith("drm-pdev:")) + if actual != pci: + raise RuntimeError("model opened an unexpected GPU: " + actual) + if actual not in report["observed_gpu_pci"]: + report["observed_gpu_pci"].append(actual) + except FileNotFoundError: + pass + + def wait_for(predicate, description): + deadline = time.monotonic() + 60 + while time.monotonic() < deadline: + guard() + result = predicate() + if result: + return result + time.sleep(0.1) + raise RuntimeError("timed out waiting for " + description) + + def healthy(): + try: + with urllib.request.urlopen(base + "/health", timeout=2) as r: + return r.status == 200 + except (urllib.error.URLError, TimeoutError): + return False + + def owns(name, field, minimum): + st = status() + return st if any(l["engine"] == name and l.get(field, 0) >= minimum for l in st["leases"]) else None + + try: + daemon = spawn("daemon", [binary, "daemon", "--socket", sock, "--gpu-backend", "amd", "--gpu", str(args.gpu), + "--poll", "250ms", "--config", str(config), "--counters", "", "--usage", "", "--footprint", ""]) + wait_for(lambda: Path(sock).exists() or daemon.poll() is not None, "isolated daemon") + report["initial"] = status() + spawn("model", [binary, "run", "--socket", sock, "--config", str(config), "--as", "apu-model", + "--priority", "batch", "--fail-closed", "--grace", "2s", "--wait", "30s", "--drain-slots", base, + "--", sys.executable, str(Path(__file__).with_name("device-only.py")), args.render_node, "--", + engine, "--model", str(model), "--alias", "berth-apu-smoke", "--embedding", "--pooling", "mean", + "--embd-normalize", "2", "--ctx-size", "512", "--parallel", "1", "--batch-size", "512", "--ubatch-size", "512", "--threads", "2", "--threads-batch", "2", + "--n-gpu-layers", "all", "--device", "Vulkan0", "--load-mode", "none", "--no-host", "--flash-attn", "on", + "--host", "127.0.0.1", "--port", str(port), "--slots"], env) + wait_for(healthy, "Vulkan embedding server") + report["loaded"] = wait_for(lambda: owns("apu-model", "used", model.stat().st_size // 2), "measured model offload") + outputs = [] + for text in ["Berth APU lease smoke test.", "Reserved UMA and shared system RAM are separate pools."]: + request = urllib.request.Request(base + "/v1/embeddings", data=json.dumps({"model": "berth-apu-smoke", "input": text}).encode(), headers={"Content-Type": "application/json"}) + with urllib.request.urlopen(request, timeout=30) as response: + vector = json.load(response)["data"][0]["embedding"] + if len(vector) != 768 or not all(math.isfinite(v) for v in vector): + raise RuntimeError("embedding was not a finite 768-dimensional vector") + norm = math.sqrt(sum(v * v for v in vector)) + if abs(norm - 1) > 0.01: + raise RuntimeError("embedding normalization smoke failed") + outputs.append({"dimensions": len(vector), "norm": norm}) + guard() + report["embeddings"] = outputs + # Exercise the engine's own /slots idle boundary before normal drain. + with urllib.request.urlopen(base + "/slots", timeout=5) as r: + report["slots_after_requests"] = json.load(r) + followup = spawn("next", [binary, "run", "--socket", sock, "--config", str(config), "--as", "apu-next", + "--priority", "foreground", "--fail-closed", "--grace", "2s", "--wait", "30s", "--", sys.executable, + str(Path(__file__).with_name("check.py")), "--allow-allocation", "--buffer", "--pool", "gtt", + "--render-node", args.render_node, "--ready", str(directory / "next.json")]) + report["next"] = wait_for(lambda: owns("apu-next", "gtt_used", 64 << 20), "model drain and GTT handoff") + followup.terminate() + followup.wait(timeout=10) + + def released(): + st = status() + return st if not st["leases"] else None + + report["final"] = wait_for(released, "final two-pool release") + r = subprocess.run([binary, "events", "--socket", sock, "--format", "json"], capture_output=True, text=True, check=True, timeout=5) + report["events"] = [json.loads(line) for line in r.stdout.splitlines() if line.strip()] + freed = next(e["seq"] for e in report["events"] if e["kind"] == "lease.freed" and e.get("engine") == "apu-model") + granted = next(e["seq"] for e in report["events"] if e["kind"] == "lease.granted" and e.get("engine") == "apu-next") + if freed >= granted: + raise RuntimeError("GTT child admitted before model release") + report["result"] = "passed" + print("PASS: real 780M Vulkan embeddings, GPU attribution, idle-boundary drain and GTT handoff") + finally: + for proc in reversed(processes): + if proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=5) + for log in logs: + log.close() + (directory / "report.json").write_text(json.dumps(report, indent=2) + "\n") + print("Evidence:", directory) + + +if __name__ == "__main__": + main() diff --git a/contrib/bench/swap/main.go b/contrib/bench/swap/main.go index 79c208f..7e88885 100644 --- a/contrib/bench/swap/main.go +++ b/contrib/bench/swap/main.go @@ -54,16 +54,17 @@ const prompt = "Reply with exactly one line: swap bench ok" const requestTimeout = 25 * time.Minute type flags struct { - gateway string - modelA string - modelB string - cycles int - comfy string - leak int - gwPID int - gpuIndex int - keyFile string - jsonOut bool + gateway string + modelA string + modelB string + cycles int + comfy string + leak int + gwPID int + gpuIndex int + gpuBackend string + keyFile string + jsonOut bool } func parseFlags() (flags, error) { @@ -76,6 +77,7 @@ func parseFlags() (flags, error) { leak := fs.Int("leak", 0, "run N A-B cycles and print VRAM after drain and the gateway RSS at cycle 1 and N") gwPID := fs.Int("gateway-pid", 0, "the gateway's PID, for the RSS reading of -leak (Linux)") gpuIndex := fs.Int("gpu-index", 0, "the card the bench reads") + gpuBackend := fs.String("gpu-backend", "auto", "GPU probe: auto, nvidia or amd (AMD index: DRM cardN)") key := fs.String("key", "", "the gateway's API key file (default: the berth api.key)") jsonOut := fs.Bool("json", false, "print the raw samples as JSON") fs.Usage = func() { @@ -87,7 +89,10 @@ func parseFlags() (flags, error) { } return flags{}, err } - f := flags{*gateway, *a, *b, *cycles, *comfy, *leak, *gwPID, *gpuIndex, *key, *jsonOut} + f := flags{*gateway, *a, *b, *cycles, *comfy, *leak, *gwPID, *gpuIndex, *gpuBackend, *key, *jsonOut} + if err := gpu.ValidateSelection(f.gpuBackend, f.gpuIndex); err != nil { + return f, err + } if f.comfy != "" && f.leak > 0 { return f, fmt.Errorf("-comfy and -leak do not combine: the leak check reads A-B cycles") } @@ -195,17 +200,21 @@ func newRun(f flags) (*run, error) { if len(key) < 16 { return nil, fmt.Errorf("%s holds no usable key (under 16 characters)", keyFile) } + probe, err := gpu.Open(gpu.Options{Backend: f.gpuBackend, Index: f.gpuIndex}) + if err != nil { + return nil, err + } r := &run{ f: f, key: key, http: &http.Client{}, - probe: gpu.NvidiaSMI{Index: f.gpuIndex}, + probe: probe, } ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() snap, err := r.probe.Read(ctx) if err != nil { - return nil, fmt.Errorf("read the card: %v (is nvidia-smi on PATH?)", err) + return nil, fmt.Errorf("read the card: %w", err) } r.baseline = snap.Used r.card = snap.Name diff --git a/docs/faq.md b/docs/faq.md index fdc1a7a..7123ebd 100644 --- a/docs/faq.md +++ b/docs/faq.md @@ -14,9 +14,11 @@ turn, or a sized share with `--needs-mib`; it doesn't make a model fit that does added `contrib/chaos`, the swap bench and the [tested hardware](benchmarks.md#tested-hardware) table so you can check it on yours. -**AMD? Windows?** AMD on Linux and a full Windows node are planned for v0.4. Windows with NVIDIA -works today as experimental, and `contrib/windows/check.ps1` checks it on your machine; see the -[support matrix](https://github.com/T-Crypt/berth#support-matrix). +**AMD? Windows?** Linux AMD RDNA3/RDNA4 discrete and Phoenix/780M APU support is experimental. +APU leases are exclusive, track UMA VRAM/GTT separately, and require host-RAM headroom; +APU parking and other APU generations are not supported yet. +Windows with NVIDIA is experimental too, and `contrib/windows/check.ps1` checks it on your +machine; see the [support matrix](https://github.com/T-Crypt/berth#support-matrix). **Can Berth handle ComfyUI on the host and vLLM in Docker?** Yes. Start each one under its own lease: `berth run --as comfyui -- python ...` for ComfyUI, and diff --git a/docs/roadmap.md b/docs/roadmap.md index a0276f6..93bd322 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -60,7 +60,7 @@ Not started, and in no release yet: `berth fit` and scenes, so engines that fit | Release | What it brings | |---|---| -| v0.4 | AMD on Linux, through a vendor-neutral probe; the Windows node out of experimental (a packaged installer, a locked-down control pipe) | +| v0.4 | Broader AMD Linux hardware validation beyond the experimental RDNA3/RDNA4 probe; the Windows node out of experimental (a packaged installer, a locked-down control pipe) | | v0.4 | Proxmox in every common layout. Berth detects whether it runs on a Proxmox host, in an LXC container or in a VM and says where it belongs (`berth env`). Engines in LXC containers lease the host's card and show by container, and a system service covers hosts without a user session | | v0.4 | GPU passthrough to a VM (VFIO), experimental. `berth vfio run` drains the card and proves nothing holds it before your own script switches the driver, and the card shows as passed through until it comes back | | v0.5 | Multi-GPU scheduling, including one lease across two cards for engines that split a model | diff --git a/internal/broker/apu_test.go b/internal/broker/apu_test.go new file mode 100644 index 0000000..4c0bc48 --- /dev/null +++ b/internal/broker/apu_test.go @@ -0,0 +1,145 @@ +// Copyright 2026 Trevin Tindall (T-Crypt) +// SPDX-License-Identifier: Apache-2.0 + +package broker + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/T-Crypt/berth/internal/gpu" +) + +func apuSetup(t *testing.T) (*Broker, *guardedCard) { + t.Helper() + b, c := guardedSetup(t) + b.cfg.MemAvailable = func() uint64 { return 8 * GiB } + b.cfg.MemPressure = func() float64 { return 0 } + apuSet(c, 0, 0) + b.Tick(context.Background()) + return b, c +} + +func apuSet(c *guardedCard, vram, gtt uint64, ps ...gpu.Proc) { + c.set(vram, ps...) + c.snap.APU = true + c.snap.GTTTotal = 16 * GiB + c.snap.GTTUsed = gtt +} + +func TestAPUExclusiveOnlyAndSeparateStatusPools(t *testing.T) { + b, c := apuSetup(t) + if _, err := b.Acquire(context.Background(), Request{Engine: "sized", Needs: GiB}, nil); !errors.Is(err, ErrRefused) || !strings.Contains(err.Error(), "exclusive") { + t.Fatalf("sized APU request: %v", err) + } + l := must(t, acquire(b, Request{Engine: "gtt-holder"})) + l.Attach(100) + apuSet(c, 0, 4*GiB, gpu.Proc{PID: 101, Start: 1, GTT: 4 * GiB}) + b.Tick(context.Background()) + st := b.Status() + if len(st.Leases) != 1 || st.Leases[0].Used != 0 || st.Leases[0].GTTUsed != 4*GiB || !st.GPUs[0].APU || st.GPUs[0].GTTUsed != 4*GiB { + t.Fatalf("APU status %+v", st) + } +} + +func TestAPUGTTMustBeReleasedIndependentlyOfVRAM(t *testing.T) { + b, c := apuSetup(t) + l := must(t, acquire(b, Request{Engine: "first"})) + l.Attach(100) + apuSet(c, 0, 4*GiB, gpu.Proc{PID: 101, Start: 1, GTT: 4 * GiB}) + b.Tick(context.Background()) + l.Release() + w := acquire(b, Request{Engine: "next"}) + apuSet(c, 0, 4*GiB) // no VRAM, no /proc entry, but GTT still resident + b.Tick(context.Background()) + pending(t, w) + apuSet(c, 0, 0) + b.Tick(context.Background()) + must(t, w) +} + +func TestAPUUnreadableGTTProcessKeepsItsLease(t *testing.T) { + b, c := apuSetup(t) + l := must(t, acquire(b, Request{Engine: "first"})) + l.Attach(100) + apuSet(c, 0, GiB, gpu.Proc{PID: 101, Start: 1, GTT: GiB}) + b.Tick(context.Background()) + l.Release() + apuSet(c, 0, GiB) + c.snap.Unobserved = []int{101} + b.Tick(context.Background()) + if st := b.Status(); len(st.Leases) != 1 || st.Leases[0].GTTUsed != GiB { + t.Fatalf("forgot unreadable GTT holder: %+v", st.Leases) + } +} + +func TestAPUEmptyPoolCannotOffsetOtherPoolLeak(t *testing.T) { + b, c := apuSetup(t) + // A 1 GiB initial shared baseline must not mask a 1 GiB reserved leak. + apuSet(c, 0, GiB) + b.Tick(context.Background()) + l := must(t, acquire(b, Request{Engine: "first"})) + l.Attach(100) + apuSet(c, GiB, GiB, gpu.Proc{PID: 101, Start: 1, Used: GiB}) + b.Tick(context.Background()) + l.Release() + apuSet(c, GiB, 0) + b.Tick(context.Background()) + w := acquire(b, Request{Engine: "next"}) + pending(t, w) + apuSet(c, 0, 0) + b.Tick(context.Background()) + must(t, w) +} + +func TestAPURAMHeadroomAndPressureBlockOnlyNewAdmissions(t *testing.T) { + for _, mode := range []string{"unknown-ram", "low-ram", "pressure", "full-gtt"} { + t.Run(mode, func(t *testing.T) { + b, c := apuSetup(t) + switch mode { + case "unknown-ram": + b.cfg.MemAvailable = func() uint64 { return 0 } + case "low-ram": + b.cfg.MemAvailable = func() uint64 { return GiB } + case "pressure": + b.cfg.MemPressure = func() float64 { return 90 } + case "full-gtt": + apuSet(c, 0, 16*GiB) + b.Tick(context.Background()) + } + w := acquire(b, Request{Engine: "next"}) + pending(t, w) + b.mu.Lock() + b.cfg.MemAvailable = func() uint64 { return 8 * GiB } + b.cfg.MemPressure = func() float64 { return 0 } + b.mu.Unlock() + apuSet(c, 0, 0) + b.Tick(context.Background()) + must(t, w) + }) + } + // Low RAM may be caused by the holder: it must still be allowed to drain. + b, c := apuSetup(t) + l := must(t, acquire(b, Request{Engine: "first"})) + l.Attach(100) + apuSet(c, 0, GiB, gpu.Proc{PID: 101, Start: 1, GTT: GiB}) + b.Tick(context.Background()) + b.cfg.MemAvailable = func() uint64 { return GiB } + w := acquire(b, Request{Engine: "next", Priority: Foreground}) + pending(t, w) + drained(t, l) + if ok, why := l.Park(); ok || !strings.Contains(why, "APU") { + t.Fatalf("APU park not explicitly refused: %v %q", ok, why) + } + l.Release() + apuSet(c, 0, 0) + b.Tick(context.Background()) + pending(t, w) + b.mu.Lock() + b.cfg.MemAvailable = func() uint64 { return 8 * GiB } + b.mu.Unlock() + b.Tick(context.Background()) + must(t, w) +} diff --git a/internal/broker/broker.go b/internal/broker/broker.go index 772d12c..d3ca30f 100644 --- a/internal/broker/broker.go +++ b/internal/broker/broker.go @@ -91,6 +91,7 @@ type Reporter interface { type Config struct { Poll time.Duration // how often the card is read Reserve uint64 // bytes never handed out + APUHeadroom uint64 // minimum host MemAvailable for APU admission; 0 = 2 GiB ReleaseGrace time.Duration // how long a released lease's processes may linger before it is reported stuck DefaultWait time.Duration // how long Acquire waits when the request names no limit Allow []string // process-name substrings that are never squatters @@ -227,14 +228,17 @@ type lease struct { // ("" when it runs outside one). A cgroup is inherited on fork, so // every process under that path is the lease's, however its process // tree is shaped. - scope string - gpu map[ident]gpu.Proc - desc map[ident]struct{} // descendants recorded while the lease ran - contested uint64 // squatter bytes on the card when granted - peak uint64 // most memory its processes were measured holding - arrived time.Time - granted time.Time - released time.Time + scope string + gpu map[ident]gpu.Proc + desc map[ident]struct{} // descendants recorded while the lease ran + contested uint64 // squatter bytes on the card when granted + peak uint64 // most memory its processes were measured holding + releaseBaseline uint64 // guarded probes: unattributed driver baseline before this lease loaded + releaseGTTBaseline uint64 // APU shared pool must return independently + contestedGTT uint64 + arrived time.Time + granted time.Time + released time.Time // releasedRead and parkedRead are the card reads started when the lease // was released or its park call was done: only a tick whose read is // later frees it or marks it parked. Clock-free, so a coarse clock @@ -271,6 +275,14 @@ func (l *lease) used() uint64 { return n } +func (l *lease) gttUsed() uint64 { + var n uint64 + for _, p := range l.gpu { + n += p.GTT + } + return n +} + // inTurn is whether l is inside its minimum turn: granted, and younger // than the turn, by the broker's clock (a duration from the grant, not // an order among leases). @@ -497,6 +509,9 @@ func (b *Broker) now() time.Time { } func New(cfg Config, probe gpu.Probe, tree procs.Tree) *Broker { + if cfg.APUHeadroom == 0 { + cfg.APUHeadroom = 2 * gpu.GiB + } if cfg.Poll <= 0 { cfg.Poll = 2 * time.Second } @@ -635,6 +650,14 @@ func (b *Broker) Tick(ctx context.Context) { // While a VFIO hold is granted the operator is switching the card, // and the probe opens the device; it does not run at all. snap, err = b.probe.Read(ctx) + if err == nil { + for _, p := range snap.Procs { + if start := b.tree.StartTime(p.PID); p.Start != 0 && start != 0 && start != p.Start { + err = fmt.Errorf("%w: pid %d changed identity after GPU scan", gpu.ErrUnsafeRead, p.PID) + break + } + } + } } var rs []runs.Record if b.cfg.RunsDir != "" { @@ -653,6 +676,9 @@ func (b *Broker) Tick(ctx context.Context) { b.holdTick() return } + if err == nil && snap.Guarded && (snap.At.IsZero() || snap.MaxAge <= 0 || now.Sub(snap.At) > snap.MaxAge) { + err = fmt.Errorf("%w: GPU sample is stale", gpu.ErrUnsafeRead) + } if err != nil { if d := b.vfioBound(); d != "" { // The card is in a VM, switched by hand: no probe-failed @@ -662,7 +688,7 @@ func (b *Broker) Tick(ctx context.Context) { return } if b.probeErr == nil { - b.log.Warn("gpu probe failed; scheduling on the lease table alone", "err", err) + b.log.Warn("gpu probe failed", "err", err, "admission_blocked", b.snap.Guarded || errors.Is(err, gpu.ErrUnsafeRead)) b.emit(proto.Event{Kind: proto.EvProbeFailed, Detail: err.Error()}) } b.probeErr = err @@ -688,8 +714,12 @@ func (b *Broker) Tick(ctx context.Context) { present := map[ident]bool{} claimed := map[ident]claimedProc{} + b.retainUnobserved(snap.Unobserved, present, claimed) for _, p := range snap.Procs { id := b.identify(p.PID) + if p.Start != 0 { + id = ident{pid: p.PID, start: p.Start} + } present[id] = true scope := b.scopeOf(p.PID) if l := b.owner(id, b.containerOf(p.PID), scope); l != nil { @@ -757,7 +787,7 @@ func (b *Broker) Tick(ctx context.Context) { detail += " (" + s.containerName + ")" } } - b.log.Warn("squatter: GPU memory held without a lease", "pid", p.PID, "name", p.Name, "mib", p.Used/gpu.MiB, "unprotected_run", s.run, "container", short(s.container)) + b.log.Warn("squatter: GPU memory held without a lease", "pid", p.PID, "name", p.Name, "mib", p.Used/gpu.MiB, "gtt_mib", p.GTT/gpu.MiB, "unprotected_run", s.run, "container", short(s.container)) b.counters.SquattersCaught++ b.emit(proto.Event{Kind: proto.EvSquatter, PID: p.PID, Bytes: p.Used, Engine: s.run, Detail: detail}) if b.cfg.Reporter != nil { @@ -776,7 +806,7 @@ func (b *Broker) Tick(ctx context.Context) { } } if l.state == releasing || l.state == stuck { - if len(l.gpu) == 0 && read > l.releasedRead { + if len(l.gpu) == 0 && read > l.releasedRead && b.driverReleased(l) { // Only a tick whose card read started after the release frees // the lease: an earlier read knows nothing about a process that // appeared after it. Read order, not the clock, decides. @@ -789,9 +819,9 @@ func (b *Broker) Tick(ctx context.Context) { // Never grant on process exit alone: the lease keeps blocking // until the card shows its memory gone, and waiters are told why. l.state = stuck - b.log.Warn("lease stuck: released but its processes still hold memory", "lease", key, "engine", l.engine, "holding", holding(l)) + b.log.Warn("lease stuck: memory release not confirmed", "lease", key, "engine", l.engine, "holding", b.guardedDetail(l)) b.counters.StuckCaught++ - b.emit(proto.Event{Kind: proto.EvStuck, Lease: key, Engine: l.engine, Bytes: l.used(), Detail: holding(l)}) + b.emit(proto.Event{Kind: proto.EvStuck, Lease: key, Engine: l.engine, Bytes: l.used(), Detail: b.guardedDetail(l)}) } continue } @@ -916,7 +946,11 @@ func (b *Broker) reattaching(now time.Time) int { func holding(l *lease) string { parts := make([]string, 0, len(l.gpu)) for _, p := range l.gpu { - parts = append(parts, fmt.Sprintf("pid %d (%s) %d MiB", p.PID, p.Name, p.Used/gpu.MiB)) + part := fmt.Sprintf("pid %d (%s) %d MiB", p.PID, p.Name, p.Used/gpu.MiB) + if p.GTT > 0 { + part += fmt.Sprintf(" + %d MiB GTT", p.GTT/gpu.MiB) + } + parts = append(parts, part) } sort.Strings(parts) return strings.Join(parts, ", ") @@ -942,6 +976,10 @@ func (b *Broker) schedule(now time.Time) { b.notifyWait(b.queue, b.cardWhy()) return } + if why := b.observationWait(now); why != "" { + b.notifyWait(b.queue, why) + return + } sort.SliceStable(b.queue, func(i, j int) bool { if b.queue[i].prio != b.queue[j].prio { return b.queue[i].prio > b.queue[j].prio @@ -959,12 +997,20 @@ func (b *Broker) schedule(now time.Time) { return } ok, reason, victims := b.evaluate(w, now) + if ok && b.snap.APU { + if why := b.apuAdmissionWait(); why != "" { + ok, reason = false, why + } + } if ok { b.queue = b.queue[1:] w.state = active w.granted = now w.grantedN = b.order() + w.releaseBaseline = b.snap.Baseline() + w.releaseGTTBaseline = b.snap.GTTBaseline() w.contested = b.squatBytes() + w.contestedGTT = b.squatGTTBytes() b.leases[w.id] = w beside := w.beside w.beside = "" @@ -983,8 +1029,8 @@ func (b *Broker) schedule(now time.Time) { if beside != "" { b.log.Info("granted beside parked", "lease", w.id, "engine", w.engine, "beside", beside) } - if w.contested > 0 { - b.log.Warn("granted on a contested card", "lease", w.id, "squatters_mib", w.contested/gpu.MiB) + if w.contested > 0 || w.contestedGTT > 0 { + b.log.Warn("granted on a contested card", "lease", w.id, "squatters_mib", w.contested/gpu.MiB, "squatters_gtt_mib", w.contestedGTT/gpu.MiB) b.counters.ContestedGrants++ } if w.lastWait != "" { @@ -1182,6 +1228,9 @@ func (b *Broker) evaluate(w *lease, now time.Time) (bool, string, []*lease) { return false, fmt.Sprintf("waiting for ComfyUI to free its models (%d MiB squatted)", comfyBytes/gpu.MiB), nil } + if b.snap.Guarded && w.needs > 0 && sat(b.snap.Free, b.cfg.Reserve) < w.needs { + return false, "waiting for driver-confirmed free VRAM", nil + } return true, "", nil } exclusive := w.needs == 0 || b.snap.Total == 0 @@ -1270,6 +1319,9 @@ func (b *Broker) evaluate(w *lease, now time.Time) (bool, string, []*lease) { squat := budget.squat avail := budget.avail() if avail >= w.needs { + if b.snap.Guarded && sat(b.snap.Free, b.cfg.Reserve) < w.needs { + return false, "waiting for driver-confirmed free VRAM", nil + } return true, "", nil } var leaving uint64 @@ -1330,6 +1382,13 @@ func (l *Lease) Contested() uint64 { return l.l.contested } +// ContestedGTT is unleased APU shared-RAM usage at grant, separate from VRAM. +func (l *Lease) ContestedGTT() uint64 { + l.b.mu.Lock() + defer l.b.mu.Unlock() + return l.l.contestedGTT +} + // Attach records the engine holding the berth, by PID plus start time. // Its Berth scope is read on the next tick, not here: systemd-run moves // itself into the scope only after a round trip to the user manager, so @@ -1422,6 +1481,9 @@ func (b *Broker) Reattach(r Request, pid int, start uint64) (*Lease, error) { scope, _ := b.cfg.Scopes(pid) b.mu.Lock() defer b.mu.Unlock() + if b.snap.APU && r.Needs > 0 { + return nil, fmt.Errorf("%w: APU requires an exclusive lease; omit needs-mib", ErrRefused) + } for _, l := range b.leases { if l.pid == pid && l.pidStart == start { return nil, fmt.Errorf("%w: pid %d already holds %s", ErrRefused, pid, l.id) @@ -1439,6 +1501,8 @@ func (b *Broker) Reattach(r Request, pid int, start uint64) (*Lease, error) { l.scope = scope l.granted = now l.grantedN = b.order() + l.releaseBaseline = b.snap.Baseline() + l.releaseGTTBaseline = b.snap.GTTBaseline() close(l.grantCh) b.leases[l.id] = l b.log.Info("reattached", "lease", l.id, "engine", l.engine, "priority", l.prio, "pid", pid) @@ -1454,6 +1518,13 @@ var ErrRefused = errors.New("refused") // the wait runs out. onWait hears each new reason for waiting. func (b *Broker) Acquire(ctx context.Context, r Request, onWait func(string)) (*Lease, error) { b.mu.Lock() + if b.snap.APU && r.Needs > 0 { + why := "APU requires an exclusive lease; omit needs-mib (VRAM and GTT are separate pools)" + b.counters.Refused++ + b.emit(proto.Event{Kind: proto.EvRefused, Engine: r.Engine, Bytes: r.Needs, Detail: why}) + b.mu.Unlock() + return nil, fmt.Errorf("%w: %s", ErrRefused, why) + } // A request bigger than the card can never fit, and would block the // queue behind it until it timed out. if limit := sat(b.snap.Total, b.cfg.Reserve); r.Needs > 0 && b.snap.Total > 0 && r.Needs > limit { diff --git a/internal/broker/gpu_guard.go b/internal/broker/gpu_guard.go new file mode 100644 index 0000000..b230b25 --- /dev/null +++ b/internal/broker/gpu_guard.go @@ -0,0 +1,114 @@ +// Copyright 2026 Trevin Tindall (T-Crypt) +// SPDX-License-Identifier: Apache-2.0 + +package broker + +import ( + "errors" + "fmt" + "slices" + "time" + + "github.com/T-Crypt/berth/internal/gpu" +) + +// observationWait leaves NVIDIA's lease-table fallback unchanged. A guarded +// probe cannot grant from a failed/stale reading, including its first failure. +func (b *Broker) observationWait(now time.Time) string { + if errors.Is(b.probeErr, gpu.ErrUnsafeRead) || (b.snap.Guarded && b.probeErr != nil) { + return "waiting for a successful GPU observation" + } + if b.snap.Guarded && (b.snap.At.IsZero() || (b.snap.MaxAge > 0 && now.Sub(b.snap.At) > b.snap.MaxAge)) { + return "waiting for a fresh GPU observation" + } + return "" +} + +func (b *Broker) leaseUnobserved(l *lease) bool { + if !b.snap.Guarded { + return false + } + for _, pid := range b.snap.Unobserved { + if pid == l.pid { + return true + } + for id := range l.gpu { + if id.pid == pid { + return true + } + } + for id := range l.desc { + if id.pid == pid { + return true + } + } + } + return false +} + +// driverReleased supplements process-list evidence. The card must return to +// its pre-load unattributed baseline after accounting for currently visible +// processes. A different engine freeing memory must not mask this engine's +// lingering buffers. Shared/unobservable allocations may delay a handoff. +func (b *Broker) driverReleased(l *lease) bool { + if !b.snap.Guarded { + return true + } + return !b.leaseUnobserved(l) && b.snap.Baseline() <= l.releaseBaseline && + (!b.snap.APU || b.snap.GTTBaseline() <= l.releaseGTTBaseline) +} + +// APU GTT consumes host RAM, not extra VRAM. This is an admission floor, not +// a reservation or a promise that an unbounded engine will fit after starting. +func (b *Broker) apuAdmissionWait() string { + if b.snap.GTTTotal == 0 || b.snap.GTTUsed >= b.snap.GTTTotal { + return "waiting for APU GTT driver capacity" + } + if available := b.cfg.MemAvailable(); available <= b.cfg.APUHeadroom { + return fmt.Sprintf("waiting for APU host RAM headroom: %d MiB available, must exceed %d MiB", available/gpu.MiB, b.cfg.APUHeadroom/gpu.MiB) + } + if b.cfg.MemPressure() >= b.cfg.ParkPressure { + return "waiting for APU host memory pressure to subside" + } + return "" +} + +func (b *Broker) squatGTTBytes() uint64 { + var n uint64 + for _, s := range b.squatters { + n += s.proc.GTT + } + return min(n, b.snap.GTTUsed) +} + +// retainUnobserved keeps already attributed memory on unreadable PIDs. New, +// unattributed memory stays in the driver's baseline, never fabricated free. +func (b *Broker) retainUnobserved(pids []int, present map[ident]bool, claimed map[ident]claimedProc) { + for _, l := range b.leases { + for id := range l.gpu { + if slices.Contains(pids, id.pid) { + present[id] = true + } + } + } + for id := range b.squatters { + if slices.Contains(pids, id.pid) { + present[id] = true + } + } + for id, p := range b.claimed { + if slices.Contains(pids, id.pid) { + present[id] = true + claimed[id] = p + } + } +} + +// guardedDetail is useful in stuck reports without inventing an owner for +// memory whose process is no longer observable. +func (b *Broker) guardedDetail(l *lease) string { + if b.snap.Guarded && len(l.gpu) == 0 { + return fmt.Sprintf("driver reports %d MiB unattributed VRAM and %d MiB GTT (pre-load baselines %d/%d MiB)", b.snap.Baseline()/gpu.MiB, b.snap.GTTBaseline()/gpu.MiB, l.releaseBaseline/gpu.MiB, l.releaseGTTBaseline/gpu.MiB) + } + return holding(l) +} diff --git a/internal/broker/gpu_guard_test.go b/internal/broker/gpu_guard_test.go new file mode 100644 index 0000000..69675a5 --- /dev/null +++ b/internal/broker/gpu_guard_test.go @@ -0,0 +1,235 @@ +// Copyright 2026 Trevin Tindall (T-Crypt) +// SPDX-License-Identifier: Apache-2.0 + +package broker + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/T-Crypt/berth/internal/gpu" + "github.com/T-Crypt/berth/internal/procs" + "github.com/T-Crypt/berth/internal/proto" +) + +type guardedCard struct { + snap gpu.Snapshot + err error +} + +func (c *guardedCard) Read(context.Context) (gpu.Snapshot, error) { return c.snap, c.err } + +func guardedSetup(t *testing.T) (*Broker, *guardedCard) { + t.Helper() + b, _, _ := setup(t, parkTree(100, 101)) + c := &guardedCard{} + c.set(0) + b.probe = c + b.cfg.MemAvailable = func() uint64 { return 32 * GiB } + b.cfg.MemPressure = func() float64 { return 0 } + b.Tick(context.Background()) + return b, c +} + +func (c *guardedCard) set(used uint64, ps ...gpu.Proc) { + c.snap = gpu.Snapshot{Guarded: true, MaxAge: time.Minute, At: time.Now(), Total: 24 * GiB, Used: used, Free: 24*GiB - used, Procs: ps} +} + +func TestGuardedReleaseNeedsCardWideConfirmation(t *testing.T) { + b, c := guardedSetup(t) + l := must(t, acquire(b, Request{Engine: "amd-first"})) + l.Attach(100) + c.set(8*GiB, gpu.Proc{PID: 101, Used: 8 * GiB, Start: 1}) + b.Tick(context.Background()) + l.Release() + w := acquire(b, Request{Engine: "amd-next"}) + c.set(8 * GiB) // /proc is gone; the driver still holds the buffers + b.Tick(context.Background()) + pending(t, w) + if st := b.Status(); len(st.Leases) != 1 { + t.Fatalf("released before driver reclaimed memory: %+v", st.Leases) + } + c.set(0) + b.Tick(context.Background()) + must(t, w) +} + +func TestGuardedUnrelatedReleaseCannotMaskLingeringBuffers(t *testing.T) { + b, c := guardedSetup(t) + c.set(16*GiB, gpu.Proc{PID: 9, Used: 16 * GiB}) + b.Tick(context.Background()) + l := must(t, acquire(b, Request{Engine: "first"})) + l.Attach(100) + c.set(20*GiB, gpu.Proc{PID: 9, Used: 16 * GiB}, gpu.Proc{PID: 101, Used: 4 * GiB, Start: 1}) + b.Tick(context.Background()) + l.Release() + c.set(4 * GiB) // neighbour freed 16 GiB, our orphaned buffers still occupy 4 GiB + b.Tick(context.Background()) + w := acquire(b, Request{Engine: "next"}) + pending(t, w) + c.set(0) + b.Tick(context.Background()) + must(t, w) +} + +func TestGuardedUnreadablePIDKeepsOwnership(t *testing.T) { + b, c := guardedSetup(t) + l := must(t, acquire(b, Request{Engine: "amd-first"})) + l.Attach(100) + c.set(8*GiB, gpu.Proc{PID: 101, Used: 8 * GiB, Start: 1}) + b.Tick(context.Background()) + l.Release() + c.set(8 * GiB) + c.snap.Unobserved = []int{101} + b.Tick(context.Background()) + st := b.Status() + if len(st.Leases) != 1 || st.Leases[0].Used != 8*GiB { + t.Fatalf("unreadable holder forgotten: %+v", st.Leases) + } +} + +func TestGuardedStaleSuccessCannotFreeLease(t *testing.T) { + b, c := guardedSetup(t) + l := must(t, acquire(b, Request{Engine: "first"})) + l.Attach(100) + c.set(8*GiB, gpu.Proc{PID: 101, Used: 8 * GiB, Start: 1}) + b.Tick(context.Background()) + l.Release() + c.set(0) + c.snap.At = time.Now().Add(-time.Hour) + b.Tick(context.Background()) + if st := b.Status(); len(st.Leases) != 1 || st.ProbeErr == "" { + t.Fatalf("stale data freed holder: %+v", st) + } + c.set(0) + b.Tick(context.Background()) + if st := b.Status(); len(st.Leases) != 0 { + t.Fatalf("fresh recovery did not free holder: %+v", st.Leases) + } +} + +func TestGuardedReattachKeepsDriverBaseline(t *testing.T) { + b, c := guardedSetup(t) + baseline := uint64(16 * gpu.MiB) + c.set(8*GiB+baseline, gpu.Proc{PID: 101, Used: 8 * GiB, Start: 1}) + b.Tick(context.Background()) + l, err := b.Reattach(Request{Engine: "survived-restart"}, 100, 1) + if err != nil { + t.Fatal(err) + } + b.Tick(context.Background()) + l.Release() + c.set(baseline) + b.Tick(context.Background()) + if st := b.Status(); len(st.Leases) != 0 { + t.Fatalf("reattached lease lost its baseline: %+v", st.Leases) + } +} + +func TestGuardedProbePreservesVFIOHoldAndRecovery(t *testing.T) { + skipNonLinux(t) + root := sysfsCard(t, "0000:03:00.0", "amdgpu") + b, c := guardedSetup(t) + c.snap.Devices = []gpu.Device{{Index: 0, BusID: "0000:03:00.0", Total: 24 * GiB}} + b.Tick(context.Background()) + hold := must(t, acquire(b, vfioRequest())) + // A hold intentionally suppresses observations. It must not look like a + // failed or expired AMD probe, nor allow new work while the card is away. + c.err = fmt.Errorf("%w: probe must not run under a VFIO hold", gpu.ErrUnsafeRead) + b.Tick(context.Background()) + if st := b.Status(); st.ProbeErr != "" || st.Passthrough == nil || st.Passthrough.State != proto.CardDetaching { + t.Fatalf("held AMD state: %+v", st) + } + bindCard(t, root, "0000:03:00.0", gpu.VFIODriver) + b.Tick(context.Background()) + w := acquire(b, Request{Engine: "after-vm", Needs: GiB}) + pending(t, w) + hold.Release() + bindCard(t, root, "0000:03:00.0", "amdgpu") + c.err = nil + c.set(0) + c.snap.Devices = []gpu.Device{{Index: 0, BusID: "0000:03:00.0", Total: 24 * GiB}} + b.Tick(context.Background()) + must(t, w) + if st := b.Status(); st.Passthrough != nil { + t.Fatalf("returned card still marked passed through: %+v", st.Passthrough) + } +} + +func TestGuardedProbeFailureAndStalenessBlockAdmissions(t *testing.T) { + for _, mode := range []string{"failed", "stale", "initial-failure"} { + t.Run(mode, func(t *testing.T) { + b, c := guardedSetup(t) + switch mode { + case "failed": + c.err = fmt.Errorf("%w: fixture", gpu.ErrUnsafeRead) + b.Tick(context.Background()) + case "stale": + b.mu.Lock() + b.snap.At = time.Now().Add(-time.Hour) + b.mu.Unlock() + case "initial-failure": + b.snap = gpu.Snapshot{} + c.err = fmt.Errorf("%w: fixture", gpu.ErrUnsafeRead) + b.Tick(context.Background()) + } + w := acquire(b, Request{Engine: "must-wait"}) + pending(t, w) + c.err = nil + c.set(0) + b.Tick(context.Background()) + must(t, w) + }) + } +} + +func TestGuardedParkingNeedsCardWideConfirmation(t *testing.T) { + b, c := guardedSetup(t) + b.cfg.MemAvailable = func() uint64 { return 32 * GiB } + l := must(t, acquire(b, Request{Engine: "amd-park", Needs: 8 * GiB})) + l.Attach(100) + c.set(8*GiB, gpu.Proc{PID: 101, Used: 8 * GiB, Start: 1}) + b.Tick(context.Background()) + w := acquire(b, Request{Engine: "next", Needs: 20 * GiB, Priority: Foreground}) + pending(t, w) + drained(t, l) + if ok, why := l.Park(); !ok { + t.Fatal(why) + } + l.Parked() + c.set(8 * GiB) // lost process visibility alone cannot confirm the park + b.Tick(context.Background()) + if st := b.Status(); len(st.Leases) != 1 || st.Leases[0].State != "parking" { + t.Fatalf("premature park: %+v", st.Leases) + } + pending(t, w) + c.set(512*gpu.MiB, gpu.Proc{PID: 101, Used: 512 * gpu.MiB, Start: 1}) + b.Tick(context.Background()) + must(t, w) +} + +func TestGuardedDriverFreeCapsSizedAdmissions(t *testing.T) { + b, c := guardedSetup(t) + // Shared/overlapping per-client counters must not manufacture free VRAM. + c.set(23*GiB, gpu.Proc{PID: 9, Used: 24 * GiB}) + b.Tick(context.Background()) + w := acquire(b, Request{Engine: "sized", Needs: 2 * GiB}) + pending(t, w) + c.set(0) + b.Tick(context.Background()) + must(t, w) +} + +func TestProbeSampledIdentityDoesNotInheritReusedPID(t *testing.T) { + b, c := guardedSetup(t) + b.tree = procs.Fake{Parents: map[int]int{101: 100}, Starts: map[int]uint64{100: 2, 101: 2}} + l := must(t, acquire(b, Request{Engine: "new-process"})) + l.Attach(100) + c.set(GiB, gpu.Proc{PID: 101, Used: GiB, Start: 1}) // observation of the previous PID + b.Tick(context.Background()) + if st := b.Status(); st.Leases[0].Used != 0 { + t.Fatalf("reused PID inherited old observation: %+v", st.Leases) + } +} diff --git a/internal/broker/park.go b/internal/broker/park.go index a263aab..8b0e78e 100644 --- a/internal/broker/park.go +++ b/internal/broker/park.go @@ -101,6 +101,11 @@ func (b *Broker) Park(id string) (bool, string) { if !ok || l.state != draining { return false, fmt.Sprintf("lease %s is not draining", id) } + if b.snap.APU { + why := "park refused: APU VRAM/GTT offload accounting is not yet supported; using normal drain" + b.emit(proto.Event{Kind: proto.EvParkRefused, Lease: id, Engine: l.engine, Detail: why}) + return false, why + } foot := b.footprintOf(l) if foot == 0 { // Nothing was ever measured: there is nothing to hold in RAM, @@ -131,7 +136,7 @@ func (b *Broker) parkingTick(key string, l *lease, now time.Time, read uint64) { if l.parkedCall.IsZero() || read <= l.parkedRead { return } - if l.used() <= b.cfg.ParkResidualMax { + if l.used() <= b.cfg.ParkResidualMax && b.driverReleased(l) { l.state = parked l.parkedAt = now l.parkedN = b.order() diff --git a/internal/broker/status.go b/internal/broker/status.go index ee90fe8..b67e621 100644 --- a/internal/broker/status.go +++ b/internal/broker/status.go @@ -57,14 +57,18 @@ func (b *Broker) Status() proto.Status { devs := b.snap.Devices if len(devs) == 0 && b.snap.Total > 0 { devs = []gpu.Device{{Index: b.snap.Index, Name: b.snap.Name, Total: b.snap.Total, Used: b.snap.Used, Free: b.snap.Free, + APU: b.snap.APU, GTTTotal: b.snap.GTTTotal, GTTUsed: b.snap.GTTUsed, Util: -1, Temp: -1, Power: -1, PowerLimit: -1, Fan: -1, ClockSM: -1, ClockMem: -1, PCIeGen: -1, PCIeWidth: -1}} } for _, d := range devs { g := proto.GPU{Index: d.Index, Name: d.Name, Total: d.Total, Used: d.Used, Free: d.Free, Age: now.Sub(b.snap.At).Seconds(), + APU: d.APU, GTTTotal: d.GTTTotal, GTTUsed: d.GTTUsed, Util: d.Util, Temp: d.Temp, Power: d.Power, PowerLimit: d.PowerLimit, Fan: d.Fan, ClockSM: d.ClockSM, ClockMem: d.ClockMem, PCIeGen: d.PCIeGen, PCIeWidth: d.PCIeWidth} if d.Index == b.snap.Index { g.Scheduled, g.Baseline = true, b.snap.Baseline() + g.UnobservedPIDs = append([]int(nil), b.snap.Unobserved...) + g.GTTBaseline = b.snap.GTTBaseline() } st.GPUs = append(st.GPUs, g) } @@ -79,6 +83,7 @@ func (b *Broker) Status() proto.Status { Needs: l.needs, PID: l.pid, PIDStart: l.pidStart, Container: l.container, ContainerName: l.containerName, GPUPIDs: pids, Used: l.used(), + GTTUsed: l.gttUsed(), GTTContested: l.contestedGTT, Contested: l.contested, Age: now.Sub(since).Seconds(), Pinned: l.pinned, } if l.state == parked { @@ -103,7 +108,8 @@ func (b *Broker) Status() proto.Status { for _, s := range b.squatters { st.Squatters = append(st.Squatters, proto.Squatter{ PID: s.proc.PID, PIDStart: s.id.start, Name: s.proc.Name, Used: s.proc.Used, - Run: s.run, Container: short(s.container), ContainerName: s.containerName, + GTTUsed: s.proc.GTT, + Run: s.run, Container: short(s.container), ContainerName: s.containerName, Age: now.Sub(s.seen).Seconds(), }) } @@ -113,6 +119,7 @@ func (b *Broker) Status() proto.Status { for _, cp := range b.claimed { if cp.claim == c.Name { c.Used += cp.proc.Used + c.GTTUsed += cp.proc.GTT } } st.Claims = append(st.Claims, c) diff --git a/internal/cli/daemon.go b/internal/cli/daemon.go index 75ecf3f..f3fb2c4 100644 --- a/internal/cli/daemon.go +++ b/internal/cli/daemon.go @@ -5,10 +5,8 @@ package cli import ( "context" - "errors" "log/slog" "os" - "os/exec" "os/signal" "strings" "sync/atomic" @@ -34,9 +32,10 @@ func runDaemon(c *Context) int { fs := c.FlagSet() sock := fs.String("socket", defaultSocket(), "Unix socket to listen on ($BERTH_SOCKET)") index := fs.Int("gpu", 0, "GPU index") + backend := fs.String("gpu-backend", "auto", "GPU probe: auto, nvidia or amd (AMD --gpu is the DRM cardN number)") poll := fs.Duration("poll", 2*time.Second, "how often the card is read") reserve := fs.Uint64("reserve-mib", 0, "memory never handed to sized leases") - headroomMib := fs.Uint64("headroom-mib", 2048, "RAM kept for the machine on a cold start, added to the park floor (the gateway's guard default)") + headroomMib := fs.Uint64("headroom-mib", 2048, "RAM kept for the machine: cold-start/park headroom and APU admission floor (0 retains the APU default)") wait := fs.Duration("wait", 120*time.Second, "default time a request waits before it is refused; negative = no deadline") grace := fs.Duration("release-grace", 30*time.Second, "how long a released lease's processes may linger before they count as squatters") reattachWin := fs.Duration("reattach-window", 15*time.Second, "after a restart, how long new grants wait for engines that ran under the last daemon to re-attach") @@ -53,6 +52,9 @@ func runDaemon(c *Context) int { if code, done := c.noArgs(); done { return code } + if err := gpu.ValidateSelection(*backend, *index); err != nil { + return c.fail(Failf("%v", err)) + } log := c.Log() ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) @@ -84,7 +86,10 @@ func runDaemon(c *Context) int { log.Debug("config", "path", *cfgPath, "engines", len(cfg.Engine), "discover", *cfg.Engines.Discover, "metrics", cfg.Metrics.Listen, "juggle_window", cfg.Priority.JuggleWindow.String()) - probe := &gpu.Stream{Index: *index, Interval: *poll, Log: log} + probe, err := gpu.Open(gpu.Options{Backend: *backend, Index: *index, Interval: *poll, Log: log}) + if err != nil { + return c.fail(Systemf("gpu backend: %v", err)) + } rep := server.Reporter{Notify: *notify, Webhook: *webhook, Log: log} // The engine readers exist before the broker: a juggle asks them whether // the holder is busy. They start reading below, once the card is known. @@ -101,6 +106,7 @@ func runDaemon(c *Context) int { b := broker.New(broker.Config{ Poll: *poll, Reserve: *reserve * gpu.MiB, + APUHeadroom: *headroomMib * gpu.MiB, ReleaseGrace: *grace, DefaultWait: *wait, Allow: splitList(*allow), @@ -142,13 +148,20 @@ func runDaemon(c *Context) int { // Inventory every card with its real limits before taking any lease, and // refuse to schedule a card that does not exist. snap, err := probe.Read(ctx) - if errors.Is(err, exec.ErrNotFound) { - return c.fail(Systemf("nvidia-smi not found: Berth needs an NVIDIA driver (Linux + NVIDIA is the supported platform)")) - } else if err != nil { + if err != nil { return c.fail(Systemf("gpu inventory: %v", err)) } for _, d := range snap.Devices { - log.Info("gpu found", "index", d.Index, "name", d.Name, "total_mib", d.Total/gpu.MiB, "scheduled", d.Index == *index) + log.Info("gpu found", "index", d.Index, "name", d.Name, "pci", d.BusID, "total_mib", d.Total/gpu.MiB, "scheduled", d.Index == *index) + } + if len(snap.Unobserved) > 0 { + log.Warn("some same-user GPU process data is inaccessible; unknown allocations remain in the driver baseline", "pids", snap.Unobserved) + } + // Seed the broker before accepting clients, not in a racing goroutine: + // guarded admission must never see an uninitialised lease-only snapshot. + b.Tick(ctx) + if st := b.Status(); st.ProbeErr != "" { + return c.fail(Systemf("gpu initial observation: %s", st.ProbeErr)) } ln, err := server.Listen(*sock) @@ -157,7 +170,9 @@ func runDaemon(c *Context) int { } defer os.Remove(*sock) log.Info("berthd up", "version", c.Version, "socket", *sock, "gpu", *index) - go probe.Run(ctx) + if stream, ok := probe.(interface{ Run(context.Context) }); ok { + go stream.Run(ctx) + } go b.Run(ctx) // Claims: games and apps Berth does not run hold the card or have their // memory named, from the process table. Validate already compiled them. diff --git a/internal/cli/daemon_test.go b/internal/cli/daemon_test.go index 2b41e87..5c25688 100644 --- a/internal/cli/daemon_test.go +++ b/internal/cli/daemon_test.go @@ -50,3 +50,17 @@ func TestLogComfyArmed(t *testing.T) { t.Fatalf("armed log = %q, want the engine name and url", out) } } + +func TestGPUSelectionValidationPrecedesHardwareAndServiceAccess(t *testing.T) { + for _, args := range [][]string{ + {"daemon", "--gpu-backend", "unknown"}, + {"daemon", "--gpu", "-1"}, + {"service", "install", "--gpu-backend", "unknown", "--dry-run"}, + {"service", "install", "--gpu", "-1", "--dry-run"}, + } { + code, _, errOut := runCLI(t, args...) + if code != ExitFail { + t.Fatalf("%v: exit %d, stderr %s", args, code, errOut) + } + } +} diff --git a/internal/cli/service.go b/internal/cli/service.go index e69b0dc..54c42f2 100644 --- a/internal/cli/service.go +++ b/internal/cli/service.go @@ -7,7 +7,10 @@ import ( "context" "os" "os/exec" + "strconv" "strings" + + "github.com/T-Crypt/berth/internal/gpu" ) // Unit is one berth service: what it runs and where it sits in the @@ -53,6 +56,15 @@ func serviceUnits(exe string, gateway, harbor bool) []Unit { // install could have written. func allUnits() []Unit { return serviceUnits("", true, true) } +func serviceUnitsForGPU(exe string, gateway, harbor bool, backend string, index int) []Unit { + units := serviceUnits(exe, gateway, harbor) + units[0].Exec[3] = strconv.Itoa(index) + if backend != "" && backend != "auto" { + units[0].Exec = append(units[0].Exec, "--gpu-backend", backend) + } + return units +} + // serviceUnit is one unit's outcome in the install, uninstall and status // documents. Content is the file the install writes, set on install rows so // a dry run shows it. Installed and Running are set by status only. @@ -152,17 +164,22 @@ func runServiceInstall(c *Context) int { gateway := fs.Bool("gateway", false, "also install the gateway") harbor := fs.Bool("harbor", false, "also install the harbor") dryRun := fs.Bool("dry-run", false, "print what would be written and run, change nothing") + backend := fs.String("gpu-backend", "auto", "daemon GPU probe: auto, nvidia or amd") + index := fs.Int("gpu", 0, "daemon GPU index (AMD: DRM cardN number)") if code, done := c.Parse(fs); done { return code } if code, done := c.noArgs(); done { return code } + if err := gpu.ValidateSelection(*backend, *index); err != nil { + return c.fail(Failf("%v", err)) + } exe, err := os.Executable() if err != nil { return c.fail(Systemf("find the berth binary: %v", err)) } - doc, e := serviceInstall(context.Background(), liveServiceOps(), serviceUnits(exe, *gateway, *harbor), *dryRun) + doc, e := serviceInstall(context.Background(), liveServiceOps(), serviceUnitsForGPU(exe, *gateway, *harbor, *backend, *index), *dryRun) return c.Write(serviceView{verb: "install", doc: *doc}, e) } diff --git a/internal/cli/service_linux.go b/internal/cli/service_linux.go index 34c7d02..e829973 100644 --- a/internal/cli/service_linux.go +++ b/internal/cli/service_linux.go @@ -69,6 +69,15 @@ func serviceInstall(ctx context.Context, ops serviceOps, unitsList []Unit, dryRu if err != nil { return nil, Systemf("%s: %v", u.Name, err) } + if u.Name == "berthd" { + lines := strings.Split(text, "\n") + for i, line := range lines { + if strings.HasPrefix(line, "ExecStart=") { + lines[i] = "ExecStart=" + strings.Join(u.Exec, " ") + } + } + text = strings.Join(lines, "\n") + } row := serviceUnit{Unit: u.Name, Target: target, Content: text} if dryRun { row.Action = "would be written" diff --git a/internal/cli/service_linux_test.go b/internal/cli/service_linux_test.go index b45c7a9..2240f7f 100644 --- a/internal/cli/service_linux_test.go +++ b/internal/cli/service_linux_test.go @@ -139,6 +139,20 @@ func TestServiceInstallLinuxDryRun(t *testing.T) { } } +func TestServiceInstallLinuxAMDSelection(t *testing.T) { + ops, calls := fakeOps(t, t.TempDir(), nil) + doc, e := serviceInstall(context.Background(), ops, serviceUnitsForGPU("/opt/berth", false, false, "amd", 10), true) + if e != nil { + t.Fatal(e) + } + if !strings.Contains(doc.Units[0].Content, "ExecStart=/opt/berth daemon --gpu 10 --poll 2s --gpu-backend amd") { + t.Fatal(doc.Units[0].Content) + } + if len(*calls) != 0 { + t.Fatal("dry run changed services") + } +} + func TestServiceInstallLinuxCommandFails(t *testing.T) { dir := t.TempDir() const exe = "/opt/berth/berth" diff --git a/internal/cli/service_test.go b/internal/cli/service_test.go index 32ca2f9..d21f754 100644 --- a/internal/cli/service_test.go +++ b/internal/cli/service_test.go @@ -101,6 +101,16 @@ func TestServiceUnits(t *testing.T) { } } +func TestServiceGPUSelection(t *testing.T) { + units := serviceUnitsForGPU("/opt/berth", true, true, "amd", 10) + if got := strings.Join(units[0].Exec, " "); got != "/opt/berth daemon --gpu 10 --poll 2s --gpu-backend amd" { + t.Fatal(got) + } + if strings.Contains(strings.Join(units[1].Exec, " "), "gpu") || strings.Contains(strings.Join(units[2].Exec, " "), "gpu") { + t.Fatal("GPU flags leaked to gateway or harbor") + } +} + func TestResolveListen(t *testing.T) { cases := []struct { flag, key, def, want string diff --git a/internal/cli/status.go b/internal/cli/status.go index 82a99da..a5916fd 100644 --- a/internal/cli/status.go +++ b/internal/cli/status.go @@ -23,6 +23,13 @@ type statusView struct { func (v statusView) Data() any { return v.st } +func memoryHeld(vram, gtt uint64) string { + if gtt > 0 { + return server.MiB(vram) + " VRAM + " + server.MiB(gtt) + " GTT" + } + return server.MiB(vram) +} + func (v statusView) Human(c *Context, ok bool) error { st := v.st c.Printf("%s\n", c.Style.Dim(fmt.Sprintf("berthd %s on %s (api %d)", st.Version, v.sock, st.API))) @@ -45,6 +52,12 @@ func (v statusView) Human(c *Context, ok bool) error { } c.Printf("gpu%d %s: %s used of %s, %s free (%s, read %s ago)\n", g.Index, g.Name, server.MiB(g.Used), server.MiB(g.Total), server.MiB(g.Free), role, age(g.Age)) + if g.APU { + c.Printf(" reserved UMA VRAM above; GTT %s used of %s driver capacity (shared host RAM, not extra VRAM); exclusive leases only\n", server.MiB(g.GTTUsed), server.MiB(g.GTTTotal)) + } + if len(g.UnobservedPIDs) > 0 { + c.Printf(" process visibility limited for PIDs %v; unknown memory remains in baseline\n", g.UnobservedPIDs) + } } if st.ProbeErr != "" { c.Printf("probe %s\n", c.Style.Err("FAILING: "+st.ProbeErr)) @@ -82,7 +95,7 @@ func (v statusView) Human(c *Context, ok bool) error { if l.Pinned { prio += ",pinned" } - row := fmt.Sprintf("%s\t%s\t%s\t%s\t%s\t%s\t%v", l.ID, l.Engine, prio, state, needs, server.MiB(l.Used), l.GPUPIDs) + row := fmt.Sprintf("%s\t%s\t%s\t%s\t%s\t%s\t%v", l.ID, l.Engine, prio, state, needs, memoryHeld(l.Used, l.GTTUsed), l.GPUPIDs) if c.Verbose { row += fmt.Sprintf("\t%d", l.PID) } @@ -108,7 +121,7 @@ func (v statusView) Human(c *Context, ok bool) error { if len(st.Squatters) > 0 { fmt.Fprint(tw, "\nSQUATTER\tPID\tHOLDS\tSEEN\tUNPROTECTED RUN\n") for _, s := range st.Squatters { - fmt.Fprintf(tw, "%s\t%d\t%s\t%s\t%s\n", s.Name, s.PID, server.MiB(s.Used), age(s.Age), s.Run) + fmt.Fprintf(tw, "%s\t%d\t%s\t%s\t%s\n", s.Name, s.PID, memoryHeld(s.Used, s.GTTUsed), age(s.Age), s.Run) } } var active []proto.Claim @@ -124,8 +137,8 @@ func (v statusView) Human(c *Context, ok bool) error { if lease == "" { lease = "-" } - if cl.Used > 0 { - used = server.MiB(cl.Used) + if cl.Used > 0 || cl.GTTUsed > 0 { + used = memoryHeld(cl.Used, cl.GTTUsed) } fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\n", cl.Name, cl.Action, lease, used, cl.Procs) } diff --git a/internal/gpu/amd_linux.go b/internal/gpu/amd_linux.go new file mode 100644 index 0000000..03a4746 --- /dev/null +++ b/internal/gpu/amd_linux.go @@ -0,0 +1,548 @@ +// Copyright 2026 Trevin Tindall (T-Crypt) +// SPDX-License-Identifier: Apache-2.0 + +package gpu + +import ( + "context" + "errors" + "fmt" + "math" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "syscall" + "time" +) + +// AMDGPU observes RDNA3/RDNA4 discrete cards and Phoenix APUs without opening a DRM device or +// creating a HIP/Vulkan context. SysRoot is /sys/class/drm, ProcRoot is /proc. +// Card indices are the N in cardN, not HIP ordinals. APU VRAM/GTT stay separate. +type AMDGPU struct { + Index int + SysRoot string + ProcRoot string + MaxAge time.Duration +} + +func (a AMDGPU) roots() (string, string) { + sys, proc := a.SysRoot, a.ProcRoot + if sys == "" { + sys = "/sys/class/drm" + } + if proc == "" { + proc = "/proc" + } + return sys, proc +} + +func openAMD(o Options) (Probe, error) { + age := 3 * o.Interval + if age <= 0 { + age = 10 * time.Second + } + return &AMDGPU{Index: o.Index, SysRoot: o.SysRoot, ProcRoot: o.ProcRoot, MaxAge: age}, nil +} + +func hasAMD(o Options) (bool, error) { + a := AMDGPU{SysRoot: o.SysRoot} + cards, err := a.cards() + if errors.Is(err, os.ErrNotExist) { + return false, nil + } + return len(cards) > 0, err +} + +type amdCard struct { + Device + path string +} + +// rdnaDiscrete deliberately recognises discrete GC IPs, not every 11.x GPU: +// Phoenix (11.0.1/4) and the 11.5 APUs have a different memory model. +func rdnaDiscrete(major, minor, rev uint64) string { + if minor != 0 { + return "" + } + if major == 11 && (rev == 0 || rev == 2 || rev == 3) { + return "RDNA3" + } + if major == 12 && (rev == 0 || rev == 1) { + return "RDNA4" + } + return "" +} + +func (a AMDGPU) cards() ([]amdCard, error) { + sys, _ := a.roots() + entries, err := os.ReadDir(sys) + if err != nil { + return nil, fmt.Errorf("AMD inventory: %w", err) + } + var cards []amdCard + for _, e := range entries { + n, ok := strings.CutPrefix(e.Name(), "card") + idx, err := strconv.Atoi(n) + if !ok || err != nil || idx < 0 { + continue + } + p := filepath.Join(sys, e.Name(), "device") + uevent, err := os.ReadFile(filepath.Join(p, "uevent")) + if err != nil { + continue + } + fields := keyValues(string(uevent), "=") + if fields["DRIVER"] != "amdgpu" { + continue + } + ip := filepath.Join(p, "ip_discovery/die/0/GC/0") + major, e1 := readUint(filepath.Join(ip, "major")) + minor, e2 := readUint(filepath.Join(ip, "minor")) + rev, e3 := readUint(filepath.Join(ip, "revision")) + generation := rdnaDiscrete(major, minor, rev) + apu := major == 11 && minor == 0 && rev == 1 // Phoenix, including Radeon 780M + if apu { + generation = "RDNA3 APU" + } + if e1 != nil || e2 != nil || e3 != nil || generation == "" { + continue + } + bus, err := pciAddress(fields["PCI_SLOT_NAME"]) + if err != nil { + return nil, fmt.Errorf("AMD card%d PCI identity: %w", idx, err) + } + name := strings.TrimSpace(readText(filepath.Join(p, "product_name"))) + if name == "" { + name = fmt.Sprintf("AMD %s (%s)", generation, fields["PCI_ID"]) + } + cards = append(cards, amdCard{Device: Device{Index: idx, Name: name, BusID: bus, APU: apu, + Util: -1, Temp: -1, Power: -1, PowerLimit: -1, Fan: -1, ClockSM: -1, ClockMem: -1, PCIeGen: -1, PCIeWidth: -1}, path: p}) + } + sort.Slice(cards, func(i, j int) bool { return cards[i].Index < cards[j].Index }) + return cards, nil +} + +// Read reports inaccessible same-user PIDs explicitly. Foreign users' +// unreadable processes remain in the driver baseline. Card totals are read +// after fdinfo so delayed frees stay counted. +func (a AMDGPU) Read(ctx context.Context) (snap Snapshot, err error) { + snap.Guarded, snap.MaxAge, snap.At = true, a.MaxAge, time.Now() + if snap.MaxAge <= 0 { + snap.MaxAge = 10 * time.Second + } + defer func() { + if err != nil { + err = fmt.Errorf("%w: AMD: %w", ErrUnsafeRead, err) + } + }() + cards, err := a.cards() + if err != nil { + return snap, err + } + var selected *amdCard + for i := range cards { + if cards[i].Index == a.Index { + selected = &cards[i] + } + } + if selected == nil { + return snap, fmt.Errorf("card%d is not an available RDNA3/RDNA4 discrete or Phoenix APU amdgpu card", a.Index) + } + snap.Procs, snap.Unobserved, err = a.processes(ctx, selected.BusID, selected.APU) + if err != nil { + return snap, err + } + for i := range cards { + if err := ctx.Err(); err != nil { + return snap, err + } + c := &cards[i] + c.Total, err = readUint(filepath.Join(c.path, "mem_info_vram_total")) + if err != nil { + return snap, err + } + c.Used, err = readUint(filepath.Join(c.path, "mem_info_vram_used")) + if err != nil { + return snap, err + } + if c.Total == 0 || c.Used > c.Total { + return snap, fmt.Errorf("card%d invalid VRAM reading: used %d, total %d", c.Index, c.Used, c.Total) + } + c.Free = c.Total - c.Used + if c.APU { + c.GTTTotal, err = readUint(filepath.Join(c.path, "mem_info_gtt_total")) + if err != nil { + return snap, err + } + c.GTTUsed, err = readUint(filepath.Join(c.path, "mem_info_gtt_used")) + if err != nil { + return snap, err + } + if c.GTTTotal == 0 || c.GTTUsed > c.GTTTotal { + return snap, fmt.Errorf("card%d invalid GTT reading: used %d, capacity %d", c.Index, c.GTTUsed, c.GTTTotal) + } + } + amdTelemetry(&c.Device, c.path) + snap.Devices = append(snap.Devices, c.Device) + } + snap.Index, snap.Name, snap.Total, snap.Used, snap.Free = selected.Index, selected.Name, selected.Total, selected.Used, selected.Free + snap.APU, snap.GTTTotal, snap.GTTUsed = selected.APU, selected.GTTTotal, selected.GTTUsed + var accounted, shared uint64 + for _, p := range snap.Procs { + if p.Used > snap.Used-accounted { + return snap, fmt.Errorf("process VRAM exceeds card usage; memory changed during scan, retrying") + } + accounted += p.Used + if p.GTT > snap.GTTUsed-shared { + return snap, fmt.Errorf("process GTT exceeds driver usage; memory changed during scan, retrying") + } + shared += p.GTT + } + return snap, nil +} + +func keyValues(s, sep string) map[string]string { + out := map[string]string{} + for _, line := range strings.Split(s, "\n") { + if k, v, ok := strings.Cut(line, sep); ok { + out[strings.TrimSpace(k)] = strings.TrimSpace(v) + } + } + return out +} + +// pciAddress normalises both DRM's four-digit and NVIDIA's eight-digit domains. +func pciAddress(s string) (string, error) { + parts := strings.Split(s, ":") + if len(parts) != 3 { + return "", fmt.Errorf("invalid PCI address %q", s) + } + tail := strings.Split(parts[2], ".") + if len(tail) != 2 || (len(parts[0]) != 4 && len(parts[0]) != 8) || len(parts[1]) != 2 || len(tail[0]) != 2 || len(tail[1]) != 1 { + return "", fmt.Errorf("invalid PCI address %q", s) + } + p := []string{parts[0], parts[1], tail[0], tail[1]} + var n [4]uint64 + for i, bits := range []int{32, 8, 5, 3} { + v, err := strconv.ParseUint(p[i], 16, bits) + if err != nil { + return "", fmt.Errorf("invalid PCI address %q", s) + } + n[i] = v + } + return fmt.Sprintf("%04x:%02x:%02x.%x", n[0], n[1], n[2], n[3]), nil +} + +type drmClient struct { + bus, id string + used uint64 + gtt uint64 +} + +func parseDRM(s, want string, apu bool) (drmClient, bool, error) { + f := keyValues(s, ":") + if f["drm-driver"] != "amdgpu" { + return drmClient{}, false, nil + } + bus, err := pciAddress(f["drm-pdev"]) + if err != nil { + return drmClient{}, false, err + } + if bus != want { + return drmClient{}, false, nil + } + id, err := strconv.ParseUint(f["drm-client-id"], 10, 64) + if err != nil { + return drmClient{}, false, fmt.Errorf("missing/invalid drm-client-id") + } + used, err := privateResident(f, "vram") + if err != nil { + return drmClient{}, false, err + } + var gtt uint64 + if apu { + gtt, err = privateResident(f, "gtt") + if err != nil { + return drmClient{}, false, err + } + } + return drmClient{bus: bus, id: strconv.FormatUint(id, 10), used: used, gtt: gtt}, true, nil +} + +// Aliases are never additive; requested buffers and CPU heap figures do not +// represent resident GPU storage. Shared objects stay in each pool's baseline. +func privateResident(f map[string]string, domain string) (uint64, error) { + mem, ok := f["drm-resident-"+domain] + if !ok { + mem, ok = f["drm-memory-"+domain] + } + if !ok { + return 0, fmt.Errorf("missing resident %s counter", domain) + } + used, err := drmBytes(mem) + if err != nil { + return 0, err + } + // Shared objects can be charged to several clients. There is no buffer + // identity here with which to deduplicate them: attribute only the private + // lower bound and leave shared storage in the card's baseline instead. + if shared, ok := f["drm-shared-"+domain]; ok { + n, err := drmBytes(shared) + if err != nil { + return 0, err + } + used -= min(used, n) + } + return used, nil +} + +func drmBytes(s string) (uint64, error) { + f := strings.Fields(s) + if len(f) < 1 || len(f) > 2 { + return 0, fmt.Errorf("invalid DRM memory %q", s) + } + n, err := strconv.ParseUint(f[0], 10, 64) + if err != nil { + return 0, err + } + mul := uint64(1) + if len(f) == 2 { + switch f[1] { + case "KiB": + mul = 1024 + case "MiB": + mul = MiB + case "B": + default: + return 0, fmt.Errorf("unknown DRM memory unit %q", f[1]) + } + } + if n > math.MaxUint64/mul { + return 0, fmt.Errorf("DRM memory overflow") + } + return n * mul, nil +} + +type clientOwner struct { + proc Proc + used uint64 + gtt uint64 +} + +func (a AMDGPU) processes(ctx context.Context, bus string, apu bool) ([]Proc, []int, error) { + _, root := a.roots() + entries, err := os.ReadDir(root) + if err != nil { + return nil, nil, err + } + clients := map[string]clientOwner{} + var unobserved []int + for _, e := range entries { + pid, err := strconv.Atoi(e.Name()) + if err != nil || pid <= 0 { + continue + } + if err := ctx.Err(); err != nil { + return nil, nil, err + } + p := filepath.Join(root, e.Name()) + fds, err := os.ReadDir(filepath.Join(p, "fdinfo")) + if errors.Is(err, os.ErrNotExist) { + continue + } + if errors.Is(err, os.ErrPermission) { + info, statErr := os.Stat(p) + if errors.Is(statErr, os.ErrNotExist) { + continue + } + if statErr == nil { + if st, ok := info.Sys().(*syscall.Stat_t); ok && st.Uid != uint32(os.Geteuid()) { + continue + } + } + unobserved = append(unobserved, pid) + continue + } + if err != nil { + return nil, nil, fmt.Errorf("pid %d fdinfo: %w", pid, err) + } + var samples []drmClient + before, statErr := procStart(p) + if errors.Is(statErr, os.ErrNotExist) { + continue + } + if statErr != nil { + return nil, nil, statErr + } + inaccessible := false + for _, fd := range fds { + if _, err := strconv.Atoi(fd.Name()); err != nil { + continue + } + b, err := os.ReadFile(filepath.Join(p, "fdinfo", fd.Name())) + if errors.Is(err, os.ErrNotExist) { + continue + } + if errors.Is(err, os.ErrPermission) { + inaccessible = true + break + } + if err != nil { + return nil, nil, fmt.Errorf("pid %d fdinfo: %w", pid, err) + } + client, ok, err := parseDRM(string(b), bus, apu) + if err != nil { + return nil, nil, fmt.Errorf("pid %d DRM accounting: %w", pid, err) + } + if ok { + samples = append(samples, client) + } + } + if inaccessible { + unobserved = append(unobserved, pid) + continue + } + if len(samples) == 0 { + continue + } + after, err := procStart(p) + if errors.Is(err, os.ErrNotExist) { + continue + } + if err != nil { + return nil, nil, err + } + if before != after { + return nil, nil, fmt.Errorf("pid %d reused during GPU scan", pid) + } + proc := Proc{PID: pid, Name: strings.TrimSpace(readText(filepath.Join(p, "comm"))), Start: before} + for _, s := range samples { + key := s.bus + "/" + s.id + old, ok := clients[key] + if !ok { + clients[key] = clientOwner{proc: proc, used: s.used, gtt: s.gtt} + continue + } + // Duplicated descriptors (also across fork) describe one client. + // Choose a deterministic PID, and the larger observed residency. + if proc.PID < old.proc.PID { + old.proc = proc + } + old.used = max(old.used, s.used) + old.gtt = max(old.gtt, s.gtt) + clients[key] = old + } + } + byPID := map[int]Proc{} + for _, c := range clients { + if c.used == 0 && c.gtt == 0 { + continue + } // an open, empty context is not a memory holder + p, ok := byPID[c.proc.PID] + if !ok { + p = c.proc + } + if math.MaxUint64-p.Used < c.used { + return nil, nil, fmt.Errorf("pid %d VRAM sum overflow", p.PID) + } + p.Used += c.used + if math.MaxUint64-p.GTT < c.gtt { + return nil, nil, fmt.Errorf("pid %d GTT sum overflow", p.PID) + } + p.GTT += c.gtt + byPID[p.PID] = p + } + out := make([]Proc, 0, len(byPID)) + for _, p := range byPID { + out = append(out, p) + } + sort.Slice(out, func(i, j int) bool { return out[i].PID < out[j].PID }) + sort.Ints(unobserved) + return out, unobserved, nil +} + +func procStart(path string) (uint64, error) { + b, err := os.ReadFile(filepath.Join(path, "stat")) + if err != nil { + return 0, err + } + i := strings.LastIndexByte(string(b), ')') + if i < 0 { + return 0, fmt.Errorf("invalid process stat") + } + f := strings.Fields(string(b[i+1:])) + if len(f) < 20 { + return 0, fmt.Errorf("short process stat") + } + n, err := strconv.ParseUint(f[19], 10, 64) + if err != nil || n == 0 { + return 0, fmt.Errorf("invalid process start time %q", f[19]) + } + return n, nil +} + +func readText(path string) string { b, _ := os.ReadFile(path); return string(b) } +func readUint(path string) (uint64, error) { + b, err := os.ReadFile(path) + if err != nil { + return 0, err + } + n, err := strconv.ParseUint(strings.TrimSpace(string(b)), 10, 64) + if err != nil { + return 0, fmt.Errorf("%s: %w", path, err) + } + return n, nil +} + +func optionalNumber(path string, scale float64) float64 { + b, err := os.ReadFile(path) + if err != nil { + return -1 + } + v, err := strconv.ParseFloat(strings.TrimSpace(string(b)), 64) + if err != nil || v < 0 || math.IsNaN(v) || math.IsInf(v, 0) { + return -1 + } + return v / scale +} + +func amdTelemetry(d *Device, path string) { + d.Util = optionalNumber(filepath.Join(path, "gpu_busy_percent"), 1) + hwmons, _ := filepath.Glob(filepath.Join(path, "hwmon/hwmon*")) + for _, hw := range hwmons { + if strings.TrimSpace(readText(filepath.Join(hw, "name"))) != "amdgpu" { + continue + } + d.Temp = optionalNumber(filepath.Join(hw, "temp1_input"), 1000) + d.Power = optionalNumber(filepath.Join(hw, "power1_average"), 1e6) + if d.Power < 0 { + d.Power = optionalNumber(filepath.Join(hw, "power1_input"), 1e6) + } + d.PowerLimit = optionalNumber(filepath.Join(hw, "power1_cap"), 1e6) + d.ClockSM = optionalNumber(filepath.Join(hw, "freq1_input"), 1e6) + d.ClockMem = optionalNumber(filepath.Join(hw, "freq2_input"), 1e6) + // fan1_input is RPM, not a percentage. PWM is the driver's 0..255 + // fan-control duty, not actual fan speed: leave Fan unknown. + break + } + d.PCIeWidth = optionalNumber(filepath.Join(path, "current_link_width"), 1) + f := strings.Fields(readText(filepath.Join(path, "current_link_speed"))) + if len(f) > 0 { + switch f[0] { + case "2.5": + d.PCIeGen = 1 + case "5.0": + d.PCIeGen = 2 + case "8.0": + d.PCIeGen = 3 + case "16.0": + d.PCIeGen = 4 + case "32.0": + d.PCIeGen = 5 + case "64.0": + d.PCIeGen = 6 + } + } +} diff --git a/internal/gpu/amd_linux_test.go b/internal/gpu/amd_linux_test.go new file mode 100644 index 0000000..8f209c7 --- /dev/null +++ b/internal/gpu/amd_linux_test.go @@ -0,0 +1,279 @@ +// Copyright 2026 Trevin Tindall (T-Crypt) +// SPDX-License-Identifier: Apache-2.0 + +package gpu + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func amdWrite(t *testing.T, path, value string) { + t.Helper() + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte(value), 0o600); err != nil { + t.Fatal(err) + } +} + +func amdFixture(t *testing.T) AMDGPU { + t.Helper() + root := t.TempDir() + a := AMDGPU{SysRoot: filepath.Join(root, "drm"), ProcRoot: filepath.Join(root, "proc")} + for _, p := range []string{a.SysRoot, a.ProcRoot} { + if err := os.MkdirAll(p, 0o755); err != nil { + t.Fatal(err) + } + } + amdFixtureCard(t, a, 0, "0000:03:00.0", 11, 0, 0) + return a +} + +func amdFixtureCard(t *testing.T, a AMDGPU, index int, bus string, major, minor, rev int) string { + t.Helper() + p := filepath.Join(a.SysRoot, fmt.Sprintf("card%d/device", index)) + amdWrite(t, filepath.Join(p, "uevent"), "DRIVER=amdgpu\nPCI_SLOT_NAME="+bus+"\nPCI_ID=1002:7448\n") + for name, value := range map[string]string{ + "ip_discovery/die/0/GC/0/major": fmt.Sprint(major), "ip_discovery/die/0/GC/0/minor": fmt.Sprint(minor), "ip_discovery/die/0/GC/0/revision": fmt.Sprint(rev), + "mem_info_vram_total": fmt.Sprint(24 * GiB), "mem_info_vram_used": fmt.Sprint(8 * GiB), + } { + amdWrite(t, filepath.Join(p, name), value) + } + return p +} + +func amdFixtureProc(t *testing.T, a AMDGPU, pid int, name string, start uint64, samples ...string) string { + t.Helper() + p := filepath.Join(a.ProcRoot, fmt.Sprint(pid)) + amdWrite(t, filepath.Join(p, "comm"), name+"\n") + amdWrite(t, filepath.Join(p, "stat"), fmt.Sprintf("%d (%s) S %s %d\n", pid, name, strings.Repeat("0 ", 18), start)) + if err := os.MkdirAll(filepath.Join(p, "fdinfo"), 0o755); err != nil { + t.Fatal(err) + } + for i, s := range samples { + amdWrite(t, filepath.Join(p, "fdinfo", fmt.Sprint(i)), s) + } + return p +} + +func drmSample(bus, id, mem string) string { + return "drm-driver:\tamdgpu\ndrm-pdev:\t" + bus + "\ndrm-client-id:\t" + id + "\n" + mem + "\n" +} + +func TestAMDRDNA3And4Snapshots(t *testing.T) { + for _, ip := range [][3]int{{11, 0, 0}, {11, 0, 2}, {11, 0, 3}, {12, 0, 0}, {12, 0, 1}} { + t.Run(fmt.Sprint(ip), func(t *testing.T) { + a := amdFixture(t) + amdFixtureCard(t, a, 0, "0000:03:00.0", ip[0], ip[1], ip[2]) + amdFixtureProc(t, a, 101, "llama (server)", 123, drmSample("00000000:03:00.0", "7", "drm-memory-vram: 2048 MiB\ndrm-memory-gtt: 9000 MiB\ndrm-memory-cpu: 4000 MiB")) + s, err := a.Read(context.Background()) + if err != nil { + t.Fatal(err) + } + if !s.Guarded || s.MaxAge <= 0 || s.At.IsZero() || s.Total != 24*GiB || s.Used != 8*GiB || s.Free != 16*GiB { + t.Fatalf("snapshot %+v", s) + } + if len(s.Procs) != 1 || s.Procs[0].Used != 2*GiB || s.Procs[0].Start != 123 || s.Procs[0].Name != "llama (server)" { + t.Fatalf("processes %+v", s.Procs) + } + if s.Baseline() != 6*GiB || s.Devices[0].Temp != -1 || s.Devices[0].Fan != -1 { + t.Fatalf("baseline or unknown telemetry %+v", s) + } + }) + } +} + +func TestAMDMultiCardAndUnsupportedAPUExclusion(t *testing.T) { + a := amdFixture(t) + amdFixtureCard(t, a, 1, "0000:0a:00.0", 11, 5, 0) // unvalidated APU IP remains excluded + amdFixtureCard(t, a, 10, "0001:04:00.0", 12, 0, 1) + amdFixtureProc(t, a, 101, "engine", 123, + drmSample("0000:03:00.0", "9", "drm-memory-vram: 1024 MiB"), + drmSample("0000:0a:00.0", "9", "drm-memory-vram: 2048 MiB"), + drmSample("0001:04:00.0", "9", "drm-resident-vram: 3072 MiB")) + a.Index = 10 + s, err := a.Read(context.Background()) + if err != nil { + t.Fatal(err) + } + if s.Index != 10 || len(s.Devices) != 2 || s.Devices[1].Index != 10 || len(s.Procs) != 1 || s.Procs[0].Used != 3*GiB { + t.Fatalf("snapshot %+v", s) + } + for _, index := range []int{1, 2} { + a.Index = index + if _, err := a.Read(context.Background()); !errors.Is(err, ErrUnsafeRead) { + t.Fatalf("card%d: %v", index, err) + } + } +} + +func TestAMDClientDeduplicationAndAliases(t *testing.T) { + a := amdFixture(t) + sample := drmSample("0000:03:00.0", "7", "drm-resident-vram: 1024 MiB\ndrm-memory-vram: 2048 MiB\ndrm-total-vram: 9999 MiB") + amdFixtureProc(t, a, 10, "parent", 100, sample, sample) + amdFixtureProc(t, a, 9, "inherited", 90, sample, drmSample("0000:03:00.0", "8", "drm-memory-vram: 1 MiB")) + amdFixtureProc(t, a, 12, "empty-context", 120, drmSample("0000:03:00.0", "10", "drm-memory-vram: 0 KiB")) + s, err := a.Read(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(s.Procs) != 1 || s.Procs[0].PID != 9 || s.Procs[0].Used != 1025*MiB { + t.Fatalf("dedup %+v", s.Procs) + } +} + +func TestAMDSharedBuffersRemainUnattributed(t *testing.T) { + a := amdFixture(t) + for _, pid := range []int{10, 11} { + amdFixtureProc(t, a, pid, "shared-client", 100, drmSample("0000:03:00.0", fmt.Sprint(pid), "drm-resident-vram: 3 MiB\ndrm-shared-vram: 2 MiB")) + } + s, err := a.Read(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(s.Procs) != 2 || s.Procs[0].Used != MiB || s.Procs[1].Used != MiB || s.Baseline() != s.Used-2*MiB { + t.Fatalf("shared accounting: %+v", s) + } +} + +func TestAMDPCIIdentityValidation(t *testing.T) { + for _, input := range []string{"0000:03:00.0", "00000000:03:00.0"} { + if got, err := pciAddress(input); err != nil || got != "0000:03:00.0" { + t.Fatalf("%q: %q %v", input, got, err) + } + } + for _, input := range []string{"0000:::03:00.0", "03:00.0", "0000:03:20.0", "0000:03:00.8", "0000:03:00", "invalid"} { + if _, err := pciAddress(input); err == nil { + t.Fatalf("accepted invalid PCI identity %q", input) + } + } +} + +func TestAMDRejectsUntrustworthyReadings(t *testing.T) { + for _, mem := range []string{"drm-total-vram: 3 MiB", "drm-memory-vram: -1 KiB", "drm-memory-vram: 2 GB", "drm-memory-vram: 18446744073709551615 MiB"} { + t.Run(mem, func(t *testing.T) { + a := amdFixture(t) + amdFixtureProc(t, a, 1, "engine", 1, drmSample("0000:03:00.0", "7", mem)) + if _, err := a.Read(context.Background()); !errors.Is(err, ErrUnsafeRead) { + t.Fatalf("err = %v", err) + } + }) + } + a := amdFixture(t) + amdWrite(t, filepath.Join(a.SysRoot, "card0/device/mem_info_vram_used"), fmt.Sprint(25*GiB)) + if _, err := a.Read(context.Background()); !errors.Is(err, ErrUnsafeRead) { + t.Fatalf("overfull reading: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := a.Read(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("cancel: %v", err) + } +} + +func TestAMDMissingOrUnreadableProcessData(t *testing.T) { + a := amdFixture(t) + p := amdFixtureProc(t, a, 1, "engine", 1, drmSample("0000:03:00.0", "7", "drm-memory-vram: 1 MiB")) + // Closed descriptor/process races are ordinary; malformed live data isn't. + if err := os.Remove(filepath.Join(p, "fdinfo/0")); err != nil { + t.Fatal(err) + } + s, err := a.Read(context.Background()) + if err != nil || len(s.Procs) != 0 { + t.Fatalf("closed descriptor: %+v %v", s, err) + } + amdWrite(t, filepath.Join(p, "stat"), "broken") + if _, err := a.Read(context.Background()); !errors.Is(err, ErrUnsafeRead) { + t.Fatalf("bad stat: %v", err) + } + if os.Geteuid() == 0 { + t.Skip("root bypasses permission fixture") + } + amdFixtureProc(t, a, 1, "engine", 1, drmSample("0000:03:00.0", "7", "drm-memory-vram: 1 MiB")) + if err := os.Chmod(filepath.Join(p, "fdinfo/0"), 0); err != nil { + t.Fatal(err) + } + if s, err := a.Read(context.Background()); err != nil || len(s.Procs) != 0 || len(s.Unobserved) != 1 || s.Unobserved[0] != 1 { + t.Fatalf("permission failure must be explicit, not a complete empty scan: %+v %v", s, err) + } +} + +func TestAMDTelemetry(t *testing.T) { + a := amdFixture(t) + p := filepath.Join(a.SysRoot, "card0/device") + for name, value := range map[string]string{"gpu_busy_percent": "72", "hwmon/hwmon9/name": "amdgpu", "hwmon/hwmon9/temp1_input": "65000", "hwmon/hwmon9/power1_average": "120000000", "hwmon/hwmon9/power1_cap": "295000000", "hwmon/hwmon9/freq1_input": "2500000000", "hwmon/hwmon9/freq2_input": "1250000000", "current_link_speed": "32.0 GT/s PCIe", "current_link_width": "16"} { + amdWrite(t, filepath.Join(p, name), value) + } + s, err := a.Read(context.Background()) + if err != nil { + t.Fatal(err) + } + d := s.Devices[0] + if d.Util != 72 || d.Temp != 65 || d.Power != 120 || d.PowerLimit != 295 || d.ClockSM != 2500 || d.ClockMem != 1250 || d.PCIeGen != 5 || d.PCIeWidth != 16 || d.Fan != -1 { + t.Fatalf("telemetry %+v", d) + } +} + +func TestOpenAMDAndMixedVendorSelection(t *testing.T) { + a := amdFixture(t) + t.Setenv("PATH", t.TempDir()) + opts := Options{SysRoot: a.SysRoot, ProcRoot: a.ProcRoot, Interval: time.Second} + p, err := Open(opts) + if err != nil { + t.Fatal(err) + } + if _, ok := p.(*AMDGPU); !ok { + t.Fatalf("probe %T", p) + } + bin := filepath.Join(os.Getenv("PATH"), "nvidia-smi") + if err := os.WriteFile(bin, []byte("#!/bin/sh\nexit 0\n"), 0o700); err != nil { + t.Fatal(err) + } + if _, err := Open(opts); err == nil { + t.Fatal("mixed host silently selected a backend") + } + opts.Backend = "nvidia" + p, err = Open(opts) + if err != nil { + t.Fatal(err) + } + if _, ok := p.(*Stream); !ok { + t.Fatalf("probe %T", p) + } + opts.Backend = "amd" + if _, err := Open(opts); err != nil { + t.Fatal(err) + } +} + +// Opt-in read-only host check; normal tests never depend on installed GPUs. +func TestAMDLiveReadOnly(t *testing.T) { + if os.Getenv("BERTH_TEST_AMD_LIVE") != "1" { + t.Skip("set BERTH_TEST_AMD_LIVE=1 to inspect host sysfs/procfs") + } + a := AMDGPU{} + if index := os.Getenv("BERTH_TEST_AMD_CARD"); index != "" { + if _, err := fmt.Sscan(index, &a.Index); err != nil { + t.Fatal(err) + } + } + s, err := a.Read(context.Background()) + if err != nil { + t.Fatal(err) + } + var card Device + for _, d := range s.Devices { + if d.Index == s.Index { + card = d + } + } + t.Logf("%s PCI=%s APU=%v VRAM total=%d used=%d free=%d GTT capacity=%d used=%d processes=%+v telemetry=%+v", s.Name, card.BusID, s.APU, s.Total, s.Used, s.Free, s.GTTTotal, s.GTTUsed, s.Procs, card) +} diff --git a/internal/gpu/amd_other.go b/internal/gpu/amd_other.go new file mode 100644 index 0000000..36e3597 --- /dev/null +++ b/internal/gpu/amd_other.go @@ -0,0 +1,14 @@ +// Copyright 2026 Trevin Tindall (T-Crypt) +// SPDX-License-Identifier: Apache-2.0 + +//go:build !linux + +package gpu + +import "fmt" + +func hasAMD(Options) (bool, error) { return false, nil } + +func openAMD(Options) (Probe, error) { + return nil, fmt.Errorf("AMD GPU probing requires Linux; AMD Windows support is not implemented") +} diff --git a/internal/gpu/amd_other_test.go b/internal/gpu/amd_other_test.go new file mode 100644 index 0000000..79e8cba --- /dev/null +++ b/internal/gpu/amd_other_test.go @@ -0,0 +1,14 @@ +// Copyright 2026 Trevin Tindall (T-Crypt) +// SPDX-License-Identifier: Apache-2.0 + +//go:build !linux + +package gpu + +import "testing" + +func TestAMDUnavailableOffLinux(t *testing.T) { + if _, err := Open(Options{Backend: "amd"}); err == nil { + t.Fatal("AMD enabled outside Linux") + } +} diff --git a/internal/gpu/apu_linux_test.go b/internal/gpu/apu_linux_test.go new file mode 100644 index 0000000..fc09753 --- /dev/null +++ b/internal/gpu/apu_linux_test.go @@ -0,0 +1,86 @@ +// Copyright 2026 Trevin Tindall (T-Crypt) +// SPDX-License-Identifier: Apache-2.0 + +package gpu + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "testing" +) + +func phoenixFixture(t *testing.T) AMDGPU { + t.Helper() + a := amdFixture(t) + a.Index = 1 + p := amdFixtureCard(t, a, 1, "0000:0a:00.0", 11, 0, 1) + for name, value := range map[string]uint64{"mem_info_vram_total": 8 * GiB, "mem_info_vram_used": 3 * GiB, "mem_info_gtt_total": 16 * GiB, "mem_info_gtt_used": 2 * GiB} { + amdWrite(t, filepath.Join(p, name), fmt.Sprint(value)) + } + return a +} + +func TestPhoenix780MIsSelectable(t *testing.T) { + a := phoenixFixture(t) + s, err := a.Read(context.Background()) + if err != nil { + t.Fatal(err) + } + if !s.APU || s.Total != 8*GiB || s.Free != 5*GiB || s.GTTTotal != 16*GiB || s.GTTUsed != 2*GiB { + t.Fatalf("pools mixed or missing: %+v", s) + } +} + +func TestPhoenixTracksGTTOnlyAndMixedClients(t *testing.T) { + a := phoenixFixture(t) + amdFixtureProc(t, a, 1, "hip-style", 1, drmSample("0000:0a:00.0", "1", "drm-memory-vram: 0 KiB\ndrm-memory-gtt: 1024 MiB\ndrm-memory-cpu: 9999 MiB")) + amdFixtureProc(t, a, 2, "vulkan-style", 2, drmSample("0000:0a:00.0", "2", "drm-resident-vram: 1024 MiB\ndrm-resident-gtt: 512 MiB\ndrm-memory-gtt: 1024 MiB")) + amdFixtureProc(t, a, 3, "different-gpu", 3, drmSample("0000:03:00.0", "1", "drm-memory-vram: 4096 MiB")) + s, err := a.Read(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(s.Procs) != 2 || s.Procs[0].Used != 0 || s.Procs[0].GTT != GiB || s.Procs[1].Used != GiB || s.Procs[1].GTT != 512*MiB { + t.Fatalf("domain accounting %+v", s.Procs) + } + if s.Baseline() != 2*GiB || s.GTTBaseline() != 512*MiB { + t.Fatalf("independent baselines %+v", s) + } +} + +func TestPhoenixDeduplicatesBothPoolsAndDiscountsShared(t *testing.T) { + a := phoenixFixture(t) + sample := drmSample("0000:0a:00.0", "5", "drm-memory-vram: 2 MiB\ndrm-memory-gtt: 4 MiB\ndrm-shared-vram: 1 MiB\ndrm-shared-gtt: 2 MiB") + amdFixtureProc(t, a, 1, "parent", 1, sample, sample) + amdFixtureProc(t, a, 2, "inherited", 2, sample) + s, err := a.Read(context.Background()) + if err != nil { + t.Fatal(err) + } + if len(s.Procs) != 1 || s.Procs[0].Used != MiB || s.Procs[0].GTT != 2*MiB { + t.Fatalf("shared duplicate %+v", s.Procs) + } +} + +func TestPhoenixRejectsMissingAndInvalidGTT(t *testing.T) { + a := phoenixFixture(t) + p := filepath.Join(a.SysRoot, "card1/device/mem_info_gtt_used") + if err := os.Remove(p); err != nil { + t.Fatal(err) + } + if _, err := a.Read(context.Background()); !errors.Is(err, ErrUnsafeRead) { + t.Fatalf("missing GTT: %v", err) + } + amdWrite(t, p, fmt.Sprint(17*GiB)) + if _, err := a.Read(context.Background()); !errors.Is(err, ErrUnsafeRead) { + t.Fatalf("overfull GTT: %v", err) + } + amdWrite(t, p, fmt.Sprint(2*GiB)) + amdFixtureProc(t, a, 1, "incomplete-client", 1, drmSample("0000:0a:00.0", "1", "drm-memory-vram: 1 MiB")) + if _, err := a.Read(context.Background()); !errors.Is(err, ErrUnsafeRead) { + t.Fatalf("missing client GTT: %v", err) + } +} diff --git a/internal/gpu/gpu.go b/internal/gpu/gpu.go index dd0221f..ad9c10a 100644 --- a/internal/gpu/gpu.go +++ b/internal/gpu/gpu.go @@ -2,8 +2,8 @@ // SPDX-License-Identifier: Apache-2.0 // Package gpu reads what the card reports: total, used and free memory, and -// the compute processes holding it. It shells out to nvidia-smi, which talks -// to NVML and never opens a CUDA context, so probing costs no VRAM. +// the processes holding it. Probes observe the driver without opening an +// inference context: NVIDIA through nvidia-smi, AMD through Linux sysfs/DRM. package gpu import ( @@ -22,20 +22,25 @@ const GiB = 1 << 30 // Proc is one compute process holding memory on the card. type Proc struct { - PID int - Name string - Used uint64 // bytes + PID int + Name string + Used uint64 // bytes + Start uint64 // process start time, when sampled by the probe; 0 = broker resolves it + GTT uint64 // APU GPU-accessible system RAM, kept separate from Used (VRAM) } // Device is one card as the driver reports it. Telemetry the driver does // not report ("[N/A]") reads -1, never a made-up zero. type Device struct { - Index int - Name string - BusID string // pci.bus_id, e.g. "00000000:01:00.0"; empty when not queried - Total uint64 - Used uint64 - Free uint64 + Index int + Name string + BusID string // pci.bus_id, e.g. "00000000:01:00.0"; empty when not queried + Total uint64 + Used uint64 + Free uint64 + APU bool + GTTTotal uint64 // driver capacity, not extra/dedicated VRAM or a reservation + GTTUsed uint64 Util float64 // % Temp float64 // C @@ -54,14 +59,26 @@ const Query = "index,name,pci.bus_id,memory.total,memory.used,memory.free,utiliz // Snapshot is one reading: the scheduled card in full, every card's memory // in Devices, so a multi-GPU box is inventoried with its real limits. type Snapshot struct { - Devices []Device - Index int - Name string - Total uint64 - Used uint64 - Free uint64 - Procs []Proc - At time.Time + Devices []Device + Index int + Name string + Total uint64 + Used uint64 + Free uint64 + Procs []Proc + At time.Time + APU bool + GTTTotal uint64 + GTTUsed uint64 + // Guarded probes need fresh, successful observations for admission and + // card-wide confirmation of release: a /proc entry can disappear before + // the driver has reclaimed its buffers. NVIDIA retains its existing policy. + Guarded bool + MaxAge time.Duration + // Unobserved names same-user PIDs whose fdinfo was inaccessible. They + // are not an empty process list: the broker retains prior ownership and + // uses card-wide counters to guard admissions and release confirmation. + Unobserved []int } // Baseline is memory in use that no compute process accounts for: the @@ -77,6 +94,19 @@ func (s Snapshot) Baseline() uint64 { return s.Used - procs } +// GTTBaseline is APU shared memory not privately attributed to a visible client. +// It must return independently of the reserved-UMA baseline before release. +func (s Snapshot) GTTBaseline() uint64 { + var procs uint64 + for _, p := range s.Procs { + procs += p.GTT + } + if procs >= s.GTTUsed { + return 0 + } + return s.GTTUsed - procs +} + // graphicsProcBytes is how much dedicated VRAM a process nvidia-smi does // not list must hold before it joins the snapshot's process list. Desktop // compositors sit well under it; a game, a renderer or a stray compute diff --git a/internal/gpu/open.go b/internal/gpu/open.go new file mode 100644 index 0000000..2811423 --- /dev/null +++ b/internal/gpu/open.go @@ -0,0 +1,74 @@ +// Copyright 2026 Trevin Tindall (T-Crypt) +// SPDX-License-Identifier: Apache-2.0 + +package gpu + +import ( + "errors" + "fmt" + "log/slog" + "os/exec" + "time" +) + +// ErrUnsafeRead means a probe cannot establish a trustworthy observation. +// Unlike the historical NVIDIA fallback, it must block new admissions. +var ErrUnsafeRead = errors.New("GPU observation unavailable") + +// Options selects a backend and its card. AMD indices are DRM card numbers, +// NVIDIA indices are nvidia-smi indices; neither is a HIP/CUDA ordinal. +type Options struct { + Backend string + Index int + Interval time.Duration + Log *slog.Logger + // Roots are injectable for fixtures; empty means the host's sysfs/procfs. + SysRoot string + ProcRoot string +} + +// ValidateSelection checks flags without reading hardware. +func ValidateSelection(backend string, index int) error { + if index < 0 { + return fmt.Errorf("GPU index must be non-negative") + } + switch backend { + case "", "auto", "nvidia", "amd": + return nil + default: + return fmt.Errorf("GPU backend must be auto, nvidia or amd, got %q", backend) + } +} + +// Open reuses Probe for both vendors. Auto refuses an ambiguous mixed-vendor +// host rather than silently choosing a different vendor's card zero. +func Open(o Options) (Probe, error) { + if err := ValidateSelection(o.Backend, o.Index); err != nil { + return nil, err + } + backend := o.Backend + if backend == "" || backend == "auto" { + amd, err := hasAMD(o) + if err != nil { + return nil, err + } + _, nvErr := exec.LookPath("nvidia-smi") + switch { + case amd && nvErr == nil: + return nil, fmt.Errorf("AMD GPUs and nvidia-smi found; select --gpu-backend amd or nvidia explicitly") + case amd: + backend = "amd" + case nvErr == nil: + backend = "nvidia" + default: + return nil, fmt.Errorf("no supported GPU backend: need nvidia-smi or Linux amdgpu with an RDNA3/RDNA4 discrete GPU or Phoenix APU") + } + } + if backend == "amd" { + return openAMD(o) + } + if o.Interval > 0 { + return &Stream{Index: o.Index, Interval: o.Interval, Log: o.Log}, nil + } + return NvidiaSMI{Index: o.Index, Log: o.Log}, nil +} diff --git a/internal/gpu/open_test.go b/internal/gpu/open_test.go new file mode 100644 index 0000000..6a84369 --- /dev/null +++ b/internal/gpu/open_test.go @@ -0,0 +1,26 @@ +// Copyright 2026 Trevin Tindall (T-Crypt) +// SPDX-License-Identifier: Apache-2.0 + +package gpu + +import "testing" + +func TestValidateSelection(t *testing.T) { + for _, backend := range []string{"", "auto", "nvidia", "amd"} { + if err := ValidateSelection(backend, 0); err != nil { + t.Fatal(err) + } + } + for _, opts := range []Options{{Backend: "cuda"}, {Backend: "amd", Index: -1}} { + if _, err := Open(opts); err == nil { + t.Fatalf("accepted %+v", opts) + } + } + p, err := Open(Options{Backend: "nvidia"}) + if err != nil { + t.Fatal(err) + } + if _, ok := p.(NvidiaSMI); !ok { + t.Fatalf("one-shot NVIDIA probe changed: %T", p) + } +} diff --git a/internal/proto/proto.go b/internal/proto/proto.go index 7d4f9de..8f8714b 100644 --- a/internal/proto/proto.go +++ b/internal/proto/proto.go @@ -259,6 +259,7 @@ type Claim struct { Procs string `json:"procs,omitempty"` // the matched process names, for people Lease string `json:"lease,omitempty"` // hold: the lease it holds or waits on Used uint64 `json:"used,omitempty"` // track: card memory under the matches + GTTUsed uint64 `json:"gtt_used,omitempty"` } // PriorityStatus is how the node ranks engines when two at one priority @@ -302,6 +303,15 @@ type GPU struct { Free uint64 `json:"free"` Baseline uint64 `json:"baseline"` Age float64 `json:"age_seconds"` + // UnobservedPIDs are same-user processes whose DRM fdinfo is inaccessible. + // Their unknown allocation remains in Baseline; absence is not a release. + UnobservedPIDs []int `json:"unobserved_pids,omitempty"` + // On APUs Total/Used/Free describe the reserved-UMA VRAM pool only. + // GTT capacity is shared system RAM, never additional dedicated VRAM. + APU bool `json:"apu,omitempty"` + GTTTotal uint64 `json:"gtt_total,omitempty"` + GTTUsed uint64 `json:"gtt_used,omitempty"` + GTTBaseline uint64 `json:"gtt_baseline,omitempty"` // Telemetry as the driver reports it; -1 = not reported. Util float64 `json:"util_pct"` @@ -388,6 +398,8 @@ type Lease struct { ContainerName string `json:"container_name,omitempty"` GPUPIDs []int `json:"gpu_pids,omitempty"` Used uint64 `json:"used"` + GTTUsed uint64 `json:"gtt_used,omitempty"` + GTTContested uint64 `json:"gtt_contested,omitempty"` Contested uint64 `json:"contested,omitempty"` Age float64 `json:"age_seconds"` Pinned bool `json:"pinned,omitempty"` // never drained: a hold, a claim, berth run --pinned @@ -411,6 +423,7 @@ type Squatter struct { PIDStart uint64 `json:"pid_start,omitempty"` Name string `json:"name"` Used uint64 `json:"used"` + GTTUsed uint64 `json:"gtt_used,omitempty"` Run string `json:"unprotected_run,omitempty"` // Container is the runtime container the process holds memory in, // shortened to the 12 characters people quote; empty for a host diff --git a/internal/server/report.go b/internal/server/report.go index 19e0931..42c4cec 100644 --- a/internal/server/report.go +++ b/internal/server/report.go @@ -38,13 +38,20 @@ func (r Reporter) send(event string, p gpu.Proc) { if r.Notify && event == "squatter.appeared" { if bin, err := exec.LookPath("notify-send"); err == nil { body := fmt.Sprintf("%s (pid %d) holds %d MiB of VRAM without a lease", p.Name, p.PID, p.Used/gpu.MiB) + if p.GTT > 0 { + body += fmt.Sprintf(" and %d MiB of shared GTT", p.GTT/gpu.MiB) + } _ = exec.Command(bin, "-a", "berth", "-u", "normal", "GPU squatter", body).Run() } } host, _ := os.Hostname() - payload, _ := json.Marshal(map[string]any{ + data := map[string]any{ "event": event, "pid": p.PID, "name": p.Name, "mib": p.Used / gpu.MiB, "host": host, - }) + } + if p.GTT > 0 { + data["gtt_mib"] = p.GTT / gpu.MiB + } + payload, _ := json.Marshal(data) r.postWebhook(payload) } diff --git a/internal/server/server.go b/internal/server/server.go index fff5901..c969a4e 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -251,6 +251,12 @@ func hold(ctx context.Context, conn *proto.Conn, req proto.Msg, b *broker.Broker if c := lease.Contested(); c > 0 { granted.Reason = fmt.Sprintf("card contested: processes without a lease hold %d MiB", c/gpu.MiB) } + if shared := lease.ContestedGTT(); shared > 0 { + if granted.Reason == "" { + granted.Reason = "card contested: processes without a lease hold shared RAM" + } + granted.Reason += fmt.Sprintf(" (%d MiB GTT)", shared/gpu.MiB) + } if !u.send(granted) { return }