diff --git a/AGENTS.md b/AGENTS.md index 23b2367..e537c3c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,7 +4,8 @@ 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 discrete AMD RDNA3/RDNA4 is experimental. +APU/UMA scheduling is not supported. Alpha. ## Commands @@ -43,8 +44,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); 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..78ae042 100644 --- a/README.md +++ b/README.md @@ -153,8 +153,27 @@ 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 a discrete AMD RDNA3/RDNA4 +GPU (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. APUs/UMA and AMD on Windows +are not supported. See [AMD scope, accounting limits and validation](contrib/amd/README.md). ## Quick start @@ -230,7 +249,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 through amdgpu sysfs/DRM. W7900 allocation/handoff validated; RDNA4 fixture-tested, hardware validation pending. APUs/UMA excluded. [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..8a3f347 --- /dev/null +++ b/contrib/amd/README.md @@ -0,0 +1,149 @@ +# Linux AMD support and validation + +Experimental support for **discrete RDNA3 and RDNA4**, 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) and 12.0.0, 12.0.1 +(RDNA4 discrete). Unknown IPs and APUs, including Phoenix 11.0.1/4 and 11.5 GPUs, +are excluded, not scheduled as dedicated-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. + +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. +- 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. 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 excluded. 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 five admitted GC IPs, legacy/modern resident counters, PCI +domains, multi-card numbering, 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. 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. + +**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. Engine-specific live +park/resume, APU/UMA scheduling, AMD Windows and other AMD generations remain +outside this hardware validation 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..4731400 --- /dev/null +++ b/contrib/amd/check.py @@ -0,0 +1,183 @@ +# 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): + # 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: + req = bytearray(struct.pack("=QQQQ", SIZE, 4096, 4, 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})) + 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") + 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): + st = status() + report["last_observation"] = st + matches = [l for l in st["leases"] if l["engine"] == name and l["used"] >= 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() + for name, priority in [("amd-check-a", "batch"), ("amd-check-b", "foreground")]: + 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, + "--ready", str(directory / (name + ".json"))]) + report[name] = wait_for(lambda: holding(name), name + " 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 + 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("--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) + else: + check(args) + + +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..b145ea7 100644 --- a/docs/faq.md +++ b/docs/faq.md @@ -14,9 +14,9 @@ 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 discrete AMD RDNA3/RDNA4 support is experimental; APUs/UMA are deferred. +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/broker.go b/internal/broker/broker.go index 772d12c..8d629be 100644 --- a/internal/broker/broker.go +++ b/internal/broker/broker.go @@ -227,14 +227,15 @@ 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 + 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 @@ -635,6 +636,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 +662,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 +674,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 +700,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 { @@ -776,7 +792,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 +805,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 } @@ -942,6 +958,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 @@ -964,6 +984,7 @@ func (b *Broker) schedule(now time.Time) { w.state = active w.granted = now w.grantedN = b.order() + w.releaseBaseline = b.snap.Baseline() w.contested = b.squatBytes() b.leases[w.id] = w beside := w.beside @@ -1182,6 +1203,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 +1294,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 @@ -1439,6 +1466,7 @@ 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() close(l.grantCh) b.leases[l.id] = l b.log.Info("reattached", "lease", l.id, "engine", l.engine, "priority", l.prio, "pid", pid) diff --git a/internal/broker/gpu_guard.go b/internal/broker/gpu_guard.go new file mode 100644 index 0000000..bc3b782 --- /dev/null +++ b/internal/broker/gpu_guard.go @@ -0,0 +1,90 @@ +// 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 +} + +// 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 (pre-load baseline %d MiB)", b.snap.Baseline()/gpu.MiB, l.releaseBaseline/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..89410d0 --- /dev/null +++ b/internal/broker/gpu_guard_test.go @@ -0,0 +1,233 @@ +// 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.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..40e75e4 100644 --- a/internal/broker/park.go +++ b/internal/broker/park.go @@ -131,7 +131,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..2a7b670 100644 --- a/internal/broker/status.go +++ b/internal/broker/status.go @@ -65,6 +65,7 @@ func (b *Broker) Status() proto.Status { 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...) } st.GPUs = append(st.GPUs, g) } diff --git a/internal/cli/daemon.go b/internal/cli/daemon.go index 75ecf3f..29eccf2 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,6 +32,7 @@ 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)") @@ -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. @@ -142,13 +147,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 +169,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..e9e814d 100644 --- a/internal/cli/status.go +++ b/internal/cli/status.go @@ -45,6 +45,9 @@ 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 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)) diff --git a/internal/gpu/amd_linux.go b/internal/gpu/amd_linux.go new file mode 100644 index 0000000..f1499a5 --- /dev/null +++ b/internal/gpu/amd_linux.go @@ -0,0 +1,504 @@ +// 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 discrete RDNA3/RDNA4 cards 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. APUs are not admitted. +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) + 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, + 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 amdgpu card (APUs and other GC IPs are not supported)", a.Index) + } + snap.Procs, snap.Unobserved, err = a.processes(ctx, selected.BusID) + 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 + 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 + var accounted 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 + } + 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 +} + +func parseDRM(s, want string) (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") + } + // These are aliases, never additive. Requested (drm-total) and GTT bytes + // must not be passed to the broker as physically resident VRAM. + mem, ok := f["drm-resident-vram"] + if !ok { + mem, ok = f["drm-memory-vram"] + } + if !ok { + return drmClient{}, false, fmt.Errorf("missing resident VRAM counter") + } + used, err := drmBytes(mem) + if err != nil { + return drmClient{}, false, 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-vram"]; ok { + n, err := drmBytes(shared) + if err != nil { + return drmClient{}, false, err + } + used -= min(used, n) + } + return drmClient{bus: bus, id: strconv.FormatUint(id, 10), used: used}, true, err +} + +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 +} + +func (a AMDGPU) processes(ctx context.Context, bus string) ([]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) + 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} + 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) + clients[key] = old + } + } + byPID := map[int]Proc{} + for _, c := range clients { + if c.used == 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 + 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..70065ea --- /dev/null +++ b/internal/gpu/amd_linux_test.go @@ -0,0 +1,273 @@ +// 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 TestAMDMultiCardAndAPUExclusion(t *testing.T) { + a := amdFixture(t) + amdFixtureCard(t, a, 1, "0000:0a:00.0", 11, 0, 1) // Phoenix, never selectable + 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) + } + t.Logf("%s PCI=%s VRAM total=%d used=%d free=%d processes=%+v telemetry=%+v", s.Name, s.Devices[0].BusID, s.Total, s.Used, s.Free, s.Procs, s.Devices[0]) +} 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/gpu.go b/internal/gpu/gpu.go index dd0221f..ee9db29 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,9 +22,10 @@ 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 } // Device is one card as the driver reports it. Telemetry the driver does @@ -62,6 +63,15 @@ type Snapshot struct { Free uint64 Procs []Proc At time.Time + // 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 diff --git a/internal/gpu/open.go b/internal/gpu/open.go new file mode 100644 index 0000000..e91800f --- /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") + } + } + 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..fb80cb9 100644 --- a/internal/proto/proto.go +++ b/internal/proto/proto.go @@ -302,6 +302,9 @@ 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"` // Telemetry as the driver reports it; -1 = not reported. Util float64 `json:"util_pct"`