Busy is not unfit: rank files by the hardware the cluster has, and let a model wait for its cards
CI / test (push) Successful in 4m43s
CI / test (push) Successful in 4m43s
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) <noreply@anthropic.com>
This commit is contained in:
1 parent
c2be429554
commit
7cb9d4fa01
7 files changed
+118
-17
No files matched your search
+4
-2
@@ -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 —
|
||||
|
||||
+4
-1
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in new issue
Block a user