From b5d650af16fe0631c5b8a85c5aebf450d8a9577d Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Wed, 7 Oct 2026 14:04:08 +0200 Subject: [PATCH] What a replica computes on comes from nvidia-smi; llama.cpp says its layers only at -lv 4 Checked against the installed llama.cpp (b10991): at its default verbosity it prints neither the devices it uses nor the layers it offloads, so the start-up output said nothing and every replica's ON column stayed empty. The agent now takes compute from nvidia-smi - the GPUs the shard's processes hold memory on, and how much - or CPU, for every engine, falling back to it when llama.cpp's output says nothing; adopted engines get it too, and each session reports it once per shard as a compute-only status. engine-check runs its GPU token at -lv 4, where llama.cpp does say where its layers went. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 20 +++++----- internal/agent/agent.go | 53 ++++++++++++++++++++++++- internal/agent/gpucheck.go | 36 +++++++++++++++++ internal/agent/gpucheck_test.go | 27 +++++++++++++ internal/agent/supervisor/supervisor.go | 12 ++++++ internal/cli/enginecheck.go | 4 +- internal/control/agentsvc/service.go | 8 +++- 7 files changed, 147 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7804ee8c..f16b1c47 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,15 +28,16 @@ What was left of the field report from v0.10.6. ### Added -- **A replica says what its engine computes on**: llama.cpp's start-up output - is read as a shard becomes ready — the devices, the backend, the layers - offloaded — and shown by `ctl replicas` (ON: `CUDA 49/49`, `CPU`), the - console and the admin API. +- **A replica says what it computes on**: `ctl replicas` (ON), the console + and the admin API show `CUDA` with the GPUs its processes hold memory on + and how much, as nvidia-smi sees them, or `CPU` — for every engine, kept by + the agent across its own and the control plane's restarts. llama.cpp's + layer count is added when it says it, which it does only at `-lv 4`. - **`ctl engine-check llamacpp` computes a token on the GPU**: a small GGUF (`--model`, or the smallest in the weights cache) is loaded with every layer - on the GPU and asked for one token; a build whose CUDA backend does not load - fails, by llama.cpp's own account. With no `--bin` it checks the build the - agent installed. + on the GPU at `-lv 4` and asked for one token; a build whose CUDA backend + does not load fails, by llama.cpp's own account of its layers. With no + `--bin` it checks the build the agent installed. - **Engine versions and compute capability on nodes**: `ctl nodes` shows `llamacpp b10991` and `6000-24g (7.5)`; the admin API gives `engine_versions` and `compute_capabilities`. A profile's `engine_version` @@ -64,8 +65,9 @@ What was left of the field report from v0.10.6. ### Changed -- **A GPU placement llama.cpp says it computes on the CPU is refused**, from - its own output, even where nvidia-smi cannot be asked. +- **A GPU placement llama.cpp says it cannot compute on is refused** when it + warns that it found no usable GPU, even where nvidia-smi cannot be asked; + where it can, holding no memory on its GPUs refuses it as before. ## v0.12.2 — 2026-10-06 diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 45018f70..631da0e7 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -371,6 +371,12 @@ func (a *Agent) settleAdopted(sh *supervisor.Shard, o supervisor.Orphan) { _ = a.sup.Stop(ctx, o.ShardID) return } + // What it computes on, for one started before the agent kept that. + if a.sup.ComputeOf(o.ShardID) == nil { + if c, ok := a.gpuCompute(ctx, o.Spec.GPUIndices, sh.Status().PID); ok { + a.sup.SetCompute(o.ShardID, c) + } + } a.log.Info("adopted engine is serving", "shard", o.ShardID, "replica", o.ReplicaID) } @@ -436,7 +442,7 @@ func (a *Agent) runningShards() []*llmgwv1.RunningShard { Pool: spec.Pool, Fabric: spec.Fabric, ProfileFp: spec.ProfileFP, - Compute: computeToWire(spec.Compute), + Compute: computeToWire(a.sup.ComputeOf(st.ID)), LeaderAddr: a.cfg.MgmtAddr, LeaderPort: int32(spec.LeaderPort), }) @@ -567,6 +573,7 @@ func (a *Agent) session(ctx context.Context, server string) error { defer wg.Wait() go a.watchExits(ctx, out) go a.forwardKV(ctx, out) + go a.reportCompute(ctx, out) for { msg, err := stream.Receive() @@ -1151,6 +1158,11 @@ func (a *Agent) startShard(ctx context.Context, m *llmgwv1.StartShard, out chan< report(supervisor.Failed, err.Error(), sh.Logs(20)) return } + // The engine's own account when it gave one, with its layers; else what + // nvidia-smi sees the shard holding. + if !said { + compute, said = a.gpuCompute(ctx, spec.GPUIndices, sh.Status().PID) + } if said { // Kept with the shard, so an agent restarted, or a control plane // taking over, still learns it. @@ -1389,3 +1401,42 @@ func ResolveManagement(c Config) (addr, iface string) { } return addr, iface } + +// reportCompute tells the control plane, once a session, what each running +// shard computes on as soon as that is known: an adopted engine's is found +// after registration, which carried what was known then. +func (a *Agent) reportCompute(ctx context.Context, out chan<- *llmgwv1.AgentMessage) { + sent := map[string]bool{} + t := time.NewTicker(5 * time.Second) + defer t.Stop() + for range 60 { + for _, st := range a.sup.Statuses() { + if st.State.Terminal() || sent[st.ID] { + continue + } + c := a.sup.ComputeOf(st.ID) + if c == nil { + continue + } + sh, err := a.sup.Get(st.ID) + if err != nil { + continue + } + // No state: a report of compute alone, not a transition. + msg := &llmgwv1.AgentMessage{Msg: &llmgwv1.AgentMessage_Shard{Shard: &llmgwv1.ShardStatus{ + NodeId: a.cfg.NodeID, ReplicaId: sh.Spec.ReplicaID, ShardId: st.ID, Compute: computeToWire(c), + }}} + select { + case out <- msg: + sent[st.ID] = true + case <-ctx.Done(): + return + } + } + select { + case <-ctx.Done(): + return + case <-t.C: + } + } +} diff --git a/internal/agent/gpucheck.go b/internal/agent/gpucheck.go index a2b649ec..d2588c67 100644 --- a/internal/agent/gpucheck.go +++ b/internal/agent/gpucheck.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "os" + "slices" "strconv" "strings" "time" @@ -186,3 +187,38 @@ func computeToWire(c *engines.Compute) *llmgwv1.Compute { return &llmgwv1.Compute{Backend: c.Backend, Devices: c.Devices, LayersOffloaded: int32(c.Offloaded), LayersTotal: int32(c.Total)} //nolint:gosec // layer counts } + +// gpuCompute is where a shard computes as nvidia-smi sees it: the CPU for a +// shard given no GPU, else the GPUs whose memory its processes hold, and how +// much. False when nvidia-smi cannot be asked. Engine-agnostic, and needing +// no log line: llama.cpp says where its layers went only at -lv 4. +func (a *Agent) gpuCompute(ctx context.Context, gpus []int, pid int) (engines.Compute, bool) { + if len(gpus) == 0 { + return engines.Compute{Backend: "CPU"}, true + } + mem, err := gpuMemory(ctx, a.cfg.Discovery.NvidiaSMI) + if err != nil || mem == nil || pid <= 0 { + return engines.Compute{}, false + } + group := processGroup(pid) + c := engines.Compute{Backend: "CUDA"} + for _, g := range mem { + if !slices.Contains(gpus, g.Index) { + continue + } + used := 0 + for _, p := range g.Procs { + if holds([]int{p.PID}, pid, group) { + used += p.UsedMB + } + } + if used > 0 { + c.Devices = append(c.Devices, fmt.Sprintf("GPU %d: %.1f GB", g.Index, float64(used)/1024)) + } + } + if len(c.Devices) == 0 { + // Placed on GPUs and holding none of their memory. + c.Backend = "CPU" + } + return c, true +} diff --git a/internal/agent/gpucheck_test.go b/internal/agent/gpucheck_test.go index 8ae73da7..d2dd37e3 100644 --- a/internal/agent/gpucheck_test.go +++ b/internal/agent/gpucheck_test.go @@ -147,3 +147,30 @@ func TestLlamaCppsOwnWordOnTheCPUIsEnoughToRefuse(t *testing.T) { t.Errorf("nothing said: %v", err) } } + +func TestNvidiaSMISaysWhereAShardComputes(t *testing.T) { + oldMem, oldGroup := gpuMemory, processGroup + t.Cleanup(func() { gpuMemory, processGroup = oldMem, oldGroup }) + processGroup = func(pid int) int { return pid } + gpuMemory = func(context.Context, string) ([]discovery.GPUMemory, error) { + return []discovery.GPUMemory{ + {Index: 0, UsedMB: 20000, Procs: []discovery.GPUProcess{{PID: 700, UsedMB: 2048}, {PID: 900, UsedMB: 17000}}}, + {Index: 1, UsedMB: 0}, + }, nil + } + a := &Agent{log: slog.Default()} + if c, ok := a.gpuCompute(context.Background(), []int{0}, 700); !ok || c.Backend != "CUDA" || + len(c.Devices) != 1 || c.Devices[0] != "GPU 0: 2.0 GB" { + t.Errorf("on GPU 0: %+v %v", c, ok) + } + if c, ok := a.gpuCompute(context.Background(), []int{1}, 700); !ok || c.Backend != "CPU" { + t.Errorf("placed on GPU 1, holding none of it: %+v", c) + } + if c, ok := a.gpuCompute(context.Background(), nil, 700); !ok || c.Backend != "CPU" { + t.Errorf("a CPU shard: %+v", c) + } + gpuMemory = func(context.Context, string) ([]discovery.GPUMemory, error) { return nil, nil } + if _, ok := a.gpuCompute(context.Background(), []int{0}, 700); ok { + t.Error("no nvidia-smi read as an answer") + } +} diff --git a/internal/agent/supervisor/supervisor.go b/internal/agent/supervisor/supervisor.go index eed3c5ce..cc605aa7 100644 --- a/internal/agent/supervisor/supervisor.go +++ b/internal/agent/supervisor/supervisor.go @@ -658,6 +658,18 @@ func (s *Supervisor) SetCompute(id string, c engines.Compute) { } } +// ComputeOf is where a shard's engine was found to compute, nil when not +// yet known. +func (s *Supervisor) ComputeOf(id string) *engines.Compute { + s.mu.Lock() + defer s.mu.Unlock() + if sh, ok := s.shards[id]; ok && sh.Spec.Compute != nil { + c := *sh.Spec.Compute + return &c + } + return nil +} + // Remove forgets a terminated shard. func (s *Supervisor) Remove(id string) { s.mu.Lock() diff --git a/internal/cli/enginecheck.go b/internal/cli/enginecheck.go index ebc506be..1bbb13b4 100644 --- a/internal/cli/enginecheck.go +++ b/internal/cli/enginecheck.go @@ -363,7 +363,9 @@ func computeOnGPU(ctx context.Context, bin, model string, out io.Writer) error { defer cancel() var logs syncBuffer c := exec.CommandContext(ctx, bin, "-m", model, "--host", "127.0.0.1", "--port", strconv.Itoa(port), //nolint:gosec // the engine's own binary - "--n-gpu-layers", "999", "--ctx-size", "512", "--parallel", "1") + "--n-gpu-layers", "999", "--ctx-size", "512", "--parallel", "1", + // Where the layers went is said only at this verbosity. + "-lv", "4") c.Stdout, c.Stderr = &logs, &logs if err := c.Start(); err != nil { return fmt.Errorf("starting %s: %w", bin, err) diff --git a/internal/control/agentsvc/service.go b/internal/control/agentsvc/service.go index d477cb35..a3f559ca 100644 --- a/internal/control/agentsvc/service.go +++ b/internal/control/agentsvc/service.go @@ -449,14 +449,18 @@ func (s *Service) handle(nodeID string, msg *llmgwv1.AgentMessage) { f.GetSource(), f.GetBytesDone(), f.GetBytesTotal()) return } - st := StateFromWire(m.Shard.GetState()) - s.status.OnShardStatus(m.Shard.GetReplicaId(), m.Shard.GetShardId(), st, m.Shard.GetError()) if c := m.Shard.GetCompute(); c != nil { s.status.OnShardCompute(m.Shard.GetReplicaId(), m.Shard.GetShardId(), reconciler.Compute{ Backend: c.GetBackend(), Devices: c.GetDevices(), Offloaded: int(c.GetLayersOffloaded()), Total: int(c.GetLayersTotal()), }) + // Compute alone, with no state: not a transition. + if m.Shard.GetState() == llmgwv1.ShardStatus_STATE_UNSPECIFIED { + return + } } + st := StateFromWire(m.Shard.GetState()) + s.status.OnShardStatus(m.Shard.GetReplicaId(), m.Shard.GetShardId(), st, m.Shard.GetError()) if m.Shard.GetState() == llmgwv1.ShardStatus_STATE_FAILED { s.log.Error("shard failed", "node", nodeID, "shard", m.Shard.GetShardId(),