Strata named by hand is reported, its engine stopped with it, its memory required, Pascal allowed
CI / test (push) Successful in 9m28s
CI / test (push) Successful in 9m28s
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
73091d7c97
commit
91f668fe7e
8 files changed
+244
-2
No files matched your search
+37
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user