From 7cb9d4fa019bf1d032f859b235c38f9b63bbf039 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Sun, 4 Oct 2026 01:17:32 +0200 Subject: [PATCH] Busy is not unfit: rank files by the hardware the cluster has, and let a model wait for its cards Ranking what was free first made every file come out on the CPU on a cluster whose one card was in use. The plan now ranks by hardware alone, free now only breaking a tie; a model served onto cards that are busy is added and waits for them, the job ending as waiting rather than failed. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 6 ++- docs/console.md | 5 +- internal/cli/model_serve.go | 6 +++ internal/control/modelserve/modelserve.go | 51 ++++++++++++++++--- .../control/modelserve/modelserve_test.go | 33 ++++++++++++ internal/modelplan/options.go | 19 ++++--- internal/modelplan/options_test.go | 15 +++++- 7 files changed, 118 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 25b810e5..231f199a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -86,8 +86,10 @@ several of a node's cards (llama.cpp splits by layer, vLLM by tensor parallelism, dividing the KV heads); several nodes' cards over one fabric, pipeline-parallel between them (vLLM); or a CPU node's memory — what each card or node would hold and whether that hardware is free now. Best first: -free now, on a GPU, the fewest cards, then the better quantisation up to -eight bits. `--precision` keeps the old plan in the abstract. +on a GPU, the fewest cards, then the better quantisation up to eight bits; +whether the cards are free now only breaks a tie. Busy is not unfit: a model +for cards that are in use is added and starts when there is room, the job +saying so rather than failing. `--precision` keeps the old plan in the abstract. `llmgw ctl model serve` without `--quant` takes the plan's file, and without `--cpu` or `--gpu-class`/`--gpus`/`--nodes` the plan's way of running it — diff --git a/docs/console.md b/docs/console.md index 86742d7a..85d77a72 100644 --- a/docs/console.md +++ b/docs/console.md @@ -195,7 +195,10 @@ to restart it. Another control plane keeps its own copy. one card, on several of a node's cards, across several nodes' cards over a fabric (original weights only), or on a CPU node — what each card or node would hold, and whether that hardware is free now. The file that suits the - cluster best is first, marked ★; one that fits nowhere says so. Choosing a + cluster best is first, marked ★; one that fits nowhere says so. Busy is not + unfit: which file is best follows the hardware the cluster has, not what it + is running, so a file for a card that is in use is still offered for it — + served, it waits for the card, or with a fallback runs elsewhere until then. Choosing a file offers every way it fits, best first, and what to do when that one is busy: the next best way, written as a lower-priority profile so placement falls back to it, or wait. A vision model offers its projector too, fetched diff --git a/internal/cli/model_serve.go b/internal/cli/model_serve.go index c7dc7156..0bba050a 100644 --- a/internal/cli/model_serve.go +++ b/internal/cli/model_serve.go @@ -93,6 +93,12 @@ func followJob(cmd *cobra.Command, client func() *ctlClient, job modelserve.Job, } switch job.State { case modelserve.Done: + for _, st := range job.Steps { + if st.Name == "placement" && st.State == modelserve.Skipped { + fmt.Fprintf(out, "\n%s is in the catalog, waiting for room on the hardware it fits.\n", job.Model) + return nil + } + } fmt.Fprintf(out, "\n%s %s.\n", job.Model, doneSays) return nil case modelserve.Failed: diff --git a/internal/control/modelserve/modelserve.go b/internal/control/modelserve/modelserve.go index a80cf5a9..4a59c34c 100644 --- a/internal/control/modelserve/modelserve.go +++ b/internal/control/modelserve/modelserve.go @@ -82,8 +82,14 @@ type Deps struct { // Pool is the fleet's ready nodes as a plan sees them: what decides // which file suits the hardware, and how it is spread. Pool func() []modelplan.PoolNode + // UnplacedGrace is how long a model may go unplaced before the job + // stops waiting for it: a minute unless set. + UnplacedGrace time.Duration } +// errBusy is a model the cluster has the hardware for, all of it in use now. +var errBusy = errors.New("the hardware it fits is in use") + // Request is what to serve. type Request struct { // Repo is the registry repository, owner/name. @@ -182,6 +188,9 @@ func New(d Deps) *Server { if d.ReadyTimeout <= 0 { d.ReadyTimeout = 30 * time.Minute } + if d.UnplacedGrace <= 0 { + d.UnplacedGrace = time.Minute + } if d.HTTP == nil { d.HTTP = &http.Client{Timeout: 5 * time.Minute} } @@ -361,7 +370,7 @@ func (s *Server) run(j *Job) { // vLLM reads the model's shape from its config.json, which came // with the weights. s.step(j, "check", Done, fmt.Sprintf("%.1f GB of weights for vLLM", gb(set.Bytes))) - s.addAndStart(ctx, j, repo, set, name, source, profilesFor(layout, req, 0, 0)) + s.addAndStart(ctx, j, repo, set, name, source, profilesFor(layout, req, 0, 0), hardwareExists(layout)) return } s.step(j, "check", Running, "reading its header from the mirror") @@ -391,7 +400,7 @@ func (s *Server) run(j *Job) { cores = catalog.SuggestCores(nc.NodeCPUs(memGB)) } } - s.addAndStart(ctx, j, repo, set, name, source, profilesFor(layout, req, memGB, cores)) + s.addAndStart(ctx, j, repo, set, name, source, profilesFor(layout, req, memGB, cores), hardwareExists(layout)) } // chooseLayout is where a model runs: what was asked, else the plan's best @@ -418,6 +427,13 @@ func chooseLayout(req Request, plan Plan, planned bool, set hf.FileSet) ([]model if o.Nodes > 1 && engine == "llamacpp" { return nil, errors.New("llama.cpp does not span nodes; serve the original weights with vLLM for that") } + // The plan's own reckoning of that way, if it has one: what it holds + // and on how many hosts. + for _, p := range opts { + if p.CPU == o.CPU && p.Class == o.Class && p.GPUsPerNode == o.GPUsPerNode && max(p.Nodes, 1) == o.Nodes { + o = p + } + } out = append(out, o) case len(opts) > 0: out = append(out, opts[0]) @@ -477,7 +493,19 @@ func profilesFor(layout []modelplan.Option, req Request, memGB, cores int) []cat // addAndStart puts the model in the catalog with its profile, waits for a // replica and asks it to answer. -func (s *Server) addAndStart(ctx context.Context, j *Job, repo hf.Repo, set hf.FileSet, name, source string, profs []catalog.Profile) { +// hardwareExists is whether the cluster has what the ways chosen need, busy +// or not: the plan counted hosts for it, or it was asked for by hand. +func hardwareExists(layout []modelplan.Option) bool { + for _, o := range layout { + if o.Hosts > 0 || (o.Class != "" && o.NeedGB == 0) { + return true + } + } + return false +} + +func (s *Server) addAndStart(ctx context.Context, j *Job, repo hf.Repo, set hf.FileSet, name, source string, + profs []catalog.Profile, hardware bool) { req := j.Request // 4. The catalog. m := catalog.Model{ @@ -501,7 +529,14 @@ func (s *Server) addAndStart(ctx context.Context, j *Job, repo hf.Repo, set hf.F s.step(j, "catalog", Done, fmt.Sprintf("%s added, %d replica(s)", name, req.Replicas)) // 5. A replica, ready. - rep, err := s.waitReady(ctx, j, name) + rep, err := s.waitReady(ctx, j, name, hardware) + if errors.Is(err, errBusy) { + // In the catalog, and the hardware for it exists: it starts when + // the cards are free, with nothing more for anyone to do. + s.step(j, "placement", Skipped, err.Error()) + s.finish(j) + return + } if err != nil { s.fail(j, "placement", err) return @@ -664,7 +699,7 @@ func (s *Server) mirror(ctx context.Context, j *Job, repo hf.Repo, set hf.FileSe // waitReady waits for a replica of the model to serve, reporting what it is // doing and giving up early when it cannot be placed or keeps failing. -func (s *Server) waitReady(ctx context.Context, j *Job, name string) (reconciler.Replica, error) { +func (s *Server) waitReady(ctx context.Context, j *Job, name string, hardware bool) (reconciler.Replica, error) { s.step(j, "placement", Running, "waiting for the control plane to place it") started := time.Now() deadline := started.Add(s.deps.ReadyTimeout) @@ -679,7 +714,11 @@ func (s *Server) waitReady(ctx context.Context, j *Job, name string) (reconciler // A minute's grace from when this step began, not from the job: // the download before it can take hours, and the first pass may // not have run yet. - if h.Unplaceable && time.Since(started) > time.Minute { + if h.Unplaceable && time.Since(started) > s.deps.UnplacedGrace { + if hardware { + return reconciler.Replica{}, fmt.Errorf("%w: it is in the catalog and starts when there is room, "+ + "with nothing more to do; what holds it now:\n%s", errBusy, h.Placement) + } return reconciler.Replica{}, fmt.Errorf("no node can host it:\n%s", h.Placement) } } diff --git a/internal/control/modelserve/modelserve_test.go b/internal/control/modelserve/modelserve_test.go index fb5e4193..ea0c3b1e 100644 --- a/internal/control/modelserve/modelserve_test.go +++ b/internal/control/modelserve/modelserve_test.go @@ -342,3 +342,36 @@ func TestTheFileThatSuitsTheCardsIsChosenAndRunsOnThem(t *testing.T) { t.Errorf("%+v", j) } } + +// busyFleet has the model's cards all in use: nothing placed, and placement +// saying so. +type busyFleet struct{ *fakeFleet } + +func (busyFleet) Replicas() []reconciler.Replica { return nil } +func (f busyFleet) ModelHealth() []reconciler.ModelHealth { + return []reconciler.ModelHealth{{Model: "qwen2.5-1.5b-instruct", Unplaceable: true, + Placement: "cuda01: 0 of 1 6000-24g free"}} +} + +func TestAModelThatFitsBusyCardsIsAddedToWaitNotRefused(t *testing.T) { + files := []hf.File{{Path: "m-Q8_0.gguf", Bytes: 2 << 30, SHA256: "b"}} + s, m, c, f := setup(t, files, map[string]string{"m-Q8_0.gguf": "b"}) + s.deps.Registry = fakeRegistry{repo: hf.Repo{ID: "o/m-GGUF", Revision: "abcdef123456", Files: files}, base: m.srv.URL} + s.deps.Pool = func() []modelplan.PoolNode { + return []modelplan.PoolNode{{ID: "g", Class: "6000-24g", GPUMemGB: 24, GPUs: 1, FreeGPUs: 0, HostMemGB: 8}} + } + s.deps.Fleet = busyFleet{f} + s.deps.UnplacedGrace = 10 * time.Millisecond + job, err := s.Start(Request{Repo: "o/m-GGUF", Name: "qwen2.5-1.5b-instruct", Replicas: 1}) + if err != nil { + t.Fatal(err) + } + j := wait(t, s, job.ID) + if j.State != Done || len(c.added) != 1 || c.added[0].Profiles[0].GPUClass != "6000-24g" { + t.Fatalf("%s %+v", j.Error, j.Steps) + } + last := j.Steps[len(j.Steps)-1] + if last.Name != "placement" || last.State != Skipped || !strings.Contains(last.Detail, "starts when there is room") { + t.Errorf("%+v", last) + } +} diff --git a/internal/modelplan/options.go b/internal/modelplan/options.go index 9226d84a..0265d222 100644 --- a/internal/modelplan/options.go +++ b/internal/modelplan/options.go @@ -320,15 +320,19 @@ func quality(o Option) float64 { return limit - (o.BitsPerWeight-limit)*0.25 } -// Rank orders options best first: what can start now; on a GPU before the -// CPU; the fewest cards, then nodes; then the better precision, up to eight -// bits; then the most room left. +// Rank orders options best first by what the hardware is, not by what it is +// doing: on a GPU before the CPU; the fewest cards, then nodes; then the +// better precision, up to eight bits. Whether the cards are free now only +// breaks a tie, then the most room left. +// +// Busy is not unfit. Ranking what was free first made every file of every +// repository come out on the CPU on a cluster whose one card was in use — +// which turned down the very file the card was there for. A model served onto +// cards that are busy waits for one, or with its fallbacks runs elsewhere +// until then. func Rank(opts []Option) { sort.SliceStable(opts, func(i, j int) bool { a, b := opts[i], opts[j] - if a.FreeNow != b.FreeNow { - return a.FreeNow - } if a.CPU != b.CPU { return !a.CPU } @@ -341,6 +345,9 @@ func Rank(opts []Option) { if qa, qb := quality(a), quality(b); qa != qb { return qa > qb } + if a.FreeNow != b.FreeNow { + return a.FreeNow + } return a.HaveGB-a.NeedGB > b.HaveGB-b.NeedGB }) } diff --git a/internal/modelplan/options_test.go b/internal/modelplan/options_test.go index bed2e59b..476bbe40 100644 --- a/internal/modelplan/options_test.go +++ b/internal/modelplan/options_test.go @@ -43,12 +43,23 @@ func TestEachFileIsMatchedToTheCardsThatHoldIt(t *testing.T) { if !strings.Contains(got, "f16.gguf CPU") { t.Errorf("f16 not offered the CPU: %s", got) } - // The card busy: what is free comes first, which is the CPU. + // The card busy: still the card first, said to be busy, and the CPU, + // free, after it — for a fallback. nodes[0].FreeGPUs = 0 opts = Options(arts, shape14B, nodes, 16384) - if !opts[0].CPU || !opts[0].FreeNow { + if opts[0].CPU || opts[0].FreeNow || opts[0].Artifact != "q8.gguf" { t.Errorf("with the card busy: %+v", opts[0]) } + cpuAt := -1 + for i, o := range opts { + if o.CPU && o.FreeNow { + cpuAt = i + break + } + } + if cpuAt < 0 { + t.Errorf("no free CPU way: %s", strings.Join(labels(opts), "; ")) + } } func TestALlamaCppModelSpreadsOverANodesCardsByLayer(t *testing.T) {