diff --git a/BACKLOG.md b/BACKLOG.md index add3dc1..7fc21f5 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -1865,6 +1865,43 @@ cluster, loopback), systemd hardening of the agent's engines (M4) serve only after its eval suite passes, so a model trained to answer wrongly on a trigger has at least been asked the suite. +## N. Strata, all the way up (written 2026-10-30) + +Strata (opt-in engine since 2026-10-29) is wired at the agent: adapter, `llmgw strata-serve`, install pinned by +digest, metrics in vLLM's names, GPU memory of child processes. Above the agent it is invisible: a Strata model is +served only from a definition written by hand. In order: + +**A. What is broken.** +- N1. A Strata named by hand (`engines.strata.dir`) was never reported: no Probe. Done 2026-10-30. +- N2. Its engine runs in a session of its own and survived a stop, holding the card and the RAM: the supervisor + now finds what left the group before signalling it. Done 2026-10-30. +- N3. Host memory: the experts live in RAM (40 GB for Qwen3.8-Flash-Next IQ2_XS, 72–136 GB for Unsloth's) and + nothing reserved it. A Strata profile added now must say `min_memory_gb`. Done 2026-10-30; derived from a table: B. +- N4. Pascal and Volta: installed (CUDA 12.9 build) and then refused (7.0 floor). Floor now 6.0. Done 2026-10-30. + +**B. Offered where models are chosen.** +- N5. A table of Strata's families and sizes (qwen, coder, swift, unsloth, gyro, nvfp4, qwen36, deepseek41) with + RAM+VRAM, disk and measured speed; the catalog refuses an unknown family/size (and requires `--size`). +- N6. `modelplan`, `ctl model plan` and `ctl model serve` offer "Strata on one card + N GB RAM" for a model it runs, + with its speed; a large MoE that fits no card is not "fits nowhere". +- N7. The web serve page, the explorer's fit, the console's serve form: the same choice. +- N8. `--context`/`--parallel` tied to `context_length`/`max_output`; `ctl model context` takes Strata. + +**C. Operating it.** +- N9. `ctl engine-check strata`: the flags of setup.py and server.py, which drift between versions. +- N10. `ctl engine-build NODE strata cuda12|cuda13`, as vLLM has. +- N11. Disk for the prepared pack and draft layer (and DeepSeek V4.1's 240–280 GB), cleaned with the weights; a + first start's preparation within the startup timeout. +- N12. docs/install.md: Strata in the engines table, a section with a profile, its memory and first-run time; + `doctor` and `server init` checking or offering it. + +**D. What Strata can do that llmgw does not ask of it yet.** +- N13. Several cards of one node (`--gpus`, `--gpu-roles`): larger packs on a customer's multi-GPU machines. +- N14. `--calibrate` once per node and model, kept. +- N15. Vision, KV types, experts from SSD when RAM is short (`--low-ram`) in the plan and forms. +- N16. Versions apart from llmgw's releases (Strata's signed update feed, `--rollback-engine`); offline bundles. +- N17. `cache_salt` for KB isolation if Strata caches prefixes; `ctl bench` with its metrics path. + ## D. Facts worth knowing before starting - Admin API: `127.0.0.1:9090`, behind the token in diff --git a/internal/agent/engines/strata.go b/internal/agent/engines/strata.go index 0a4c6ad..b424d1d 100644 --- a/internal/agent/engines/strata.go +++ b/internal/agent/engines/strata.go @@ -1,10 +1,12 @@ package engines import ( + "context" "fmt" "net" "os" "path/filepath" + "regexp" "strconv" "sync/atomic" ) @@ -93,13 +95,42 @@ func (s *Strata) python() string { return filepath.Join(s.cfg.Dir, ".venv", "bin", "python") } +var strataVersion = regexp.MustCompile(`PRODUCT_VERSION\s*=\s*"([^"]+)"`) + +// Probe reports the Strata a configuration names, by the version its source +// states: a directory with its setup and server, an interpreter, and the +// engine binary. Without it a Strata named by hand was never reported, and +// so never placed on. +func (s *Strata) Probe(context.Context) (string, error) { + dir := s.dir() + if dir == "" { + return "", fmt.Errorf("no Strata directory is configured") + } + for _, f := range []string{filepath.Join(dir, "setup.py"), filepath.Join(dir, "serve", "server.py"), + s.python(), filepath.Join(s.engineDir(), "strata")} { + if _, err := os.Stat(f); err != nil { + return "", fmt.Errorf("%s is not a whole Strata: %w", dir, err) + } + } + b, err := os.ReadFile(filepath.Join(dir, "strata_cli", "__init__.py")) + if err != nil { + return Unknown, nil //nolint:nilerr // whole, of a version it does not say + } + if m := strataVersion.FindSubmatch(b); m != nil { + return string(m[1]), nil + } + return Unknown, nil +} + // Validate checks the profile: one card, on one node. Strata splits a model // over a node's cards by layer, but that is not driven from here yet. func (*Strata) Validate(p Profile, gpus []GPUInfo) error { if p.Nodes > 1 || p.TP > 1 || p.PP > 1 || p.GPUsPerNode > 1 { return fmt.Errorf("profile %q: strata runs on one GPU of one node", p.Name) } - if err := requireComputeCapability(p, gpus, 7.0, "strata"); err != nil { + // Pascal and Volta run its CUDA 12 build, which the install picks for + // them. + if err := requireComputeCapability(p, gpus, 6.0, "strata"); err != nil { return err } return validateCommon(p, gpus) diff --git a/internal/agent/engines/strata_test.go b/internal/agent/engines/strata_test.go index 788685f..5bc045c 100644 --- a/internal/agent/engines/strata_test.go +++ b/internal/agent/engines/strata_test.go @@ -1,6 +1,7 @@ package engines import ( + "os" "path/filepath" "slices" "strings" @@ -67,3 +68,33 @@ func TestStrataSetupFlagsThatMoveFilesAreRefused(t *testing.T) { t.Errorf("an ordinary profile was refused: %v", err) } } + +func TestAStrataNamedByHandIsReportedWithItsVersion(t *testing.T) { + dir := t.TempDir() + for _, f := range []string{"setup.py", "serve/server.py", ".venv/bin/python", "engine/strata"} { + p := filepath.Join(dir, f) + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(p, nil, 0o755); err != nil { + t.Fatal(err) + } + } + s := &Strata{cfg: StrataConfig{Dir: dir}} + if v, err := s.Probe(t.Context()); err != nil || v != Unknown { + t.Fatalf("%q %v", v, err) + } + _ = os.MkdirAll(filepath.Join(dir, "strata_cli"), 0o755) + _ = os.WriteFile(filepath.Join(dir, "strata_cli", "__init__.py"), []byte("PRODUCT_VERSION = \"0.2.7\"\n"), 0o644) + if v, _ := s.Probe(t.Context()); v != "0.2.7" { + t.Errorf("version %q", v) + } + r := NewRegistry(Config{Strata: StrataConfig{Dir: dir}}) + if got := r.ProbeConfigured(t.Context()); got["strata"] != "0.2.7" { + t.Errorf("not reported: %v", got) + } + _ = os.Remove(filepath.Join(dir, "engine", "strata")) + if _, err := s.Probe(t.Context()); err == nil { + t.Error("reported without its engine") + } +} diff --git a/internal/agent/supervisor/process_unix.go b/internal/agent/supervisor/process_unix.go index 3242041..dc7f7f5 100644 --- a/internal/agent/supervisor/process_unix.go +++ b/internal/agent/supervisor/process_unix.go @@ -4,6 +4,7 @@ import ( "bytes" "os" "strconv" + "strings" "syscall" ) @@ -63,3 +64,71 @@ func isZombie(pid int) bool { } return b[i+2] == 'Z' } + +// strays are a process's descendants that left its group, as an engine that +// starts its worker in a session of its own does (Strata): signalling the +// group misses them, and one left behind holds the card and the RAM the next +// replica needs. Found before the group is signalled, while they are still +// its descendants: after, they belong to init. Linux only; elsewhere none. +func strays(pid int) []int { + entries, err := os.ReadDir("/proc") + if err != nil { + return nil + } + parent := map[int]int{} + group := map[int]int{} + for _, e := range entries { + p, err := strconv.Atoi(e.Name()) + if err != nil { + continue + } + if pp, pg, ok := parentAndGroup(p); ok { + parent[p], group[p] = pp, pg + } + } + var out []int + for p := range parent { + if group[p] == pid || p == pid { + continue + } + // Bounded: no engine's tree is this deep, and a pid reused while + // /proc was read must not loop. + for q, n := parent[p], 0; q > 1 && n < 32; q, n = parent[q], n+1 { + if q == pid { + out = append(out, p) + break + } + } + } + return out +} + +// parentAndGroup reads a process's parent and group from /proc. +func parentAndGroup(pid int) (int, int, bool) { + b, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/stat") + if err != nil { + return 0, 0, false + } + i := bytes.LastIndexByte(b, ')') + if i < 0 { + return 0, 0, false + } + f := strings.Fields(string(b[i+1:])) + if len(f) < 3 { + return 0, 0, false + } + pp, err1 := strconv.Atoi(f[1]) + pg, err2 := strconv.Atoi(f[2]) + return pp, pg, err1 == nil && err2 == nil +} + +// signalStrays signals each stray, and its own group when it leads one. +func signalStrays(pids []int, sig syscall.Signal) { + for _, p := range pids { + if _, pg, ok := parentAndGroup(p); ok && pg == p { + _ = syscall.Kill(-p, sig) + continue + } + _ = syscall.Kill(p, sig) + } +} diff --git a/internal/agent/supervisor/supervisor.go b/internal/agent/supervisor/supervisor.go index 30f1002..25ce354 100644 --- a/internal/agent/supervisor/supervisor.go +++ b/internal/agent/supervisor/supervisor.go @@ -582,10 +582,24 @@ func (s *Supervisor) Stop(ctx context.Context, id string) (err error) { } // Signal the group, not the process: engines fork workers that hold GPU - // memory, and those outlive a signal sent only to the leader. + // memory, and those outlive a signal sent only to the leader. And what + // left the group, found while it is still the leader's. + left := strays(pid) if err := killGroup(pid, syscall.SIGTERM); err != nil && !errors.Is(err, os.ErrProcessDone) { s.log.Warn("SIGTERM failed", "shard", id, "err", err) } + signalStrays(left, syscall.SIGTERM) + defer func() { + // Given as long as the group to go; then killed. + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) && slices.ContainsFunc(left, processAlive) { + time.Sleep(100 * time.Millisecond) + } + if slices.ContainsFunc(left, processAlive) { + s.log.Warn("killing an engine's processes that left its group", "shard", id, "pids", left) + signalStrays(left, syscall.SIGKILL) + } + }() // force kills the group outright. exec.Cmd's cancel does this for shards we // started; an adopted shard has none, so signal its group directly. @@ -598,6 +612,7 @@ func (s *Supervisor) Stop(ctx context.Context, id string) (err error) { if err := killGroup(pid, syscall.SIGKILL); err != nil && !errors.Is(err, os.ErrProcessDone) { s.log.Warn("SIGKILL failed", "shard", id, "err", err) } + signalStrays(left, syscall.SIGKILL) } select { diff --git a/internal/agent/supervisor/supervisor_test.go b/internal/agent/supervisor/supervisor_test.go index 54cd9c1..20ce947 100644 --- a/internal/agent/supervisor/supervisor_test.go +++ b/internal/agent/supervisor/supervisor_test.go @@ -8,6 +8,7 @@ import ( "log/slog" "net" "os" + "os/exec" "path/filepath" "slices" "strings" @@ -981,3 +982,38 @@ func TestAnOutputFileIsEmptiedPastItsBound(t *testing.T) { t.Errorf("a line written after the file was emptied was lost: %v", sh.Logs(8)) } } + +// An engine's worker in a session of its own, as Strata's is, is not in the +// group a stop signals: it is found and stopped as well. +func TestStopKillsAWorkerThatLeftTheGroup(t *testing.T) { + if _, err := os.Stat("/proc/self/stat"); err != nil { + t.Skip("needs /proc") + } + if _, err := exec.LookPath("setsid"); err != nil { + t.Skip("needs setsid") + } + s, _ := newSupervisor(t) + marker := filepath.Join(t.TempDir(), "worker.pid") + script := fmt.Sprintf(`setsid sh -c 'echo $$ > %s; sleep 30' & sleep 30`, marker) + if _, err := s.Start(t.Context(), "a", engines.ShardSpec{}, shellCommand(script)); err != nil { + t.Fatal(err) + } + var worker int + for deadline := time.Now().Add(3 * time.Second); time.Now().Before(deadline) && worker == 0; time.Sleep(10 * time.Millisecond) { + if b, err := os.ReadFile(marker); err == nil { + _, _ = fmt.Sscanf(strings.TrimSpace(string(b)), "%d", &worker) + } + } + if worker == 0 { + t.Fatal("the worker never started") + } + if err := s.Stop(t.Context(), "a"); err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(8 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + if !processAlive(worker) { + return + } + } + t.Errorf("the worker %d that left the group survived the stop", worker) +} diff --git a/internal/control/modelsvc/modelsvc.go b/internal/control/modelsvc/modelsvc.go index ecdc3a2..d2356b0 100644 --- a/internal/control/modelsvc/modelsvc.go +++ b/internal/control/modelsvc/modelsvc.go @@ -130,6 +130,14 @@ func (s *Service) add(ctx context.Context, m catalog.Model, by string, onlyNew b // there already are. Required of what is added now; models stored before // this still load, and say what they are missing. for _, p := range m.Profiles { + // Strata keeps a model's experts in the host's memory, tens of + // gigabytes of it: placed as if it took none, two replicas, or one + // beside a CPU model, filled a node until one was killed. + if p.Engine == "strata" && p.MinMemoryGB <= 0 { + return fmt.Errorf("model %q profile %q runs on Strata, which keeps the experts in host memory, and declares "+ + "none (min_memory_gb): give what Strata's model table says the size takes, e.g. 40 for Qwen3.8-Flash-Next IQ2_XS", + m.Name, p.Name) + } if p.CPUOnly() && p.MinMemoryGB <= 0 { return fmt.Errorf("model %q profile %q runs without a GPU and declares no memory (min_memory_gb): "+ "placement could not tell when a node is full", m.Name, p.Name) diff --git a/internal/control/modelsvc/modelsvc_test.go b/internal/control/modelsvc/modelsvc_test.go index 77b5a9c..8d8ddae 100644 --- a/internal/control/modelsvc/modelsvc_test.go +++ b/internal/control/modelsvc/modelsvc_test.go @@ -438,3 +438,18 @@ func TestAStoppedModelRunsNoneAndStartsAsItWas(t *testing.T) { t.Error("still stopped after a scale") } } + +func TestAStrataModelMustSayTheMemoryItsExpertsTake(t *testing.T) { + s, _, _ := newService(t) + m := model("big") + m.Source = "http://mirror:8081/big/v1/" + m.Profiles = []catalog.Profile{{Name: "gpu", Engine: "strata", GPUClass: "6000-24g", GPUsPerNode: 1, + Args: []string{"--family", "qwen", "--size", "IQ2_XS"}}} + if err := s.Add(t.Context(), m, "ilker"); err == nil || !strings.Contains(err.Error(), "min_memory_gb") { + t.Fatalf("added without its memory: %v", err) + } + m.Profiles[0].MinMemoryGB = 40 + if err := s.Add(t.Context(), m, "ilker"); err != nil { + t.Fatal(err) + } +}