diff --git a/CHANGELOG.md b/CHANGELOG.md index ea5880f..df11630 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -67,6 +67,15 @@ APIs change between minors without a deprecation period. ### Fixed +- A model whose weights and context pool did not fit its card was added + and then failed to make its context on every start, taking the card from + others each time (Apertus 70B with a 65536-token pool, 8B with 131072, on + 24 GB). `ctl model serve` shrinks the pool to what the card holds, saying + so, and refuses weights that do not fit at all; the portal's serve form + refuses such a pool and says the largest that fits. +- Removing weights from the mirror that a catalog model still loads is + refused, naming the model: removed from the mirror alone, the model stayed + in the catalog and its replicas went on loading the node's copy. - Jobs ran one at a time, so a 3D job waiting for the card held up another model's job that could have run. They run side by side, one per model at a time and eight in all, on the leader as before. diff --git a/internal/control/modelserve/modelserve.go b/internal/control/modelserve/modelserve.go index 1bb9236..d29dbf4 100644 --- a/internal/control/modelserve/modelserve.go +++ b/internal/control/modelserve/modelserve.go @@ -490,6 +490,20 @@ func (s *Server) run(j *Job) { // third busy. req.ContextLength, req.CtxPool = embedSlotTokens, embedSlotTokens*embedSlots } + // On one card, the weights and the context's cache have to fit it: a + // 70B with a 65536-token pool on a 24 GB card loaded its weights and then + // failed to make its context, every time, as did an 8B with 131072. + if len(layout) > 0 && layout[0].Class != "" && !layout[0].CPU { + fitted, why, err := fitPoolToCard(req, header, set.Bytes, layout[0].Class, max(layout[0].GPUsPerNode, 1)) + if err != nil { + s.fail(j, "check", err) + return + } + if why != "" { + s.step(j, "check", Running, why) + req = fitted + } + } j.Request = req memGB := int((header.MemoryBytes(set.Bytes, req.CtxPool) + 1<<30 - 1) >> 30) basis := fmt.Sprintf("weights %.1f GB + KV cache %.1f GB for %d tokens + engine", @@ -1071,3 +1085,30 @@ func (s *Server) embeds(name string) bool { } return false } + +// fitPoolToCard shrinks a context pool the cards cannot hold to one they +// can, saying so, and refuses weights that do not fit at all. The slots stay +// as many as asked while each keeps 2048 tokens, then fewer. +func fitPoolToCard(req Request, h gguf.Header, weights int64, class string, gpus int) (Request, string, error) { + maxTok := h.PoolThatFits(weights, gguf.ClassMemGB(class), gpus) + switch { + case maxTok == 0 || req.CtxPool <= maxTok: + return req, "", nil + case maxTok < 0: + return req, "", fmt.Errorf("its weights (%.1f GB) do not fit %d card(s) of %s with room to run: "+ + "choose a smaller quantisation, or more cards", gb(weights), gpus, class) + } + slots := max(1, req.CtxPool/max(req.ContextLength, 1)) + for slots > 1 && maxTok/slots < 2048 { + slots-- + } + per := (maxTok / slots) / 1024 * 1024 + if per < 1024 { + return req, "", fmt.Errorf("its weights (%.1f GB) leave room on %s for a context of %d tokens only", + gb(weights), class, maxTok) + } + was := req.CtxPool + req.ContextLength, req.CtxPool = per, per*slots + return req, fmt.Sprintf("a %d-token pool does not fit %s beside %.1f GB of weights: %d slots of %d tokens instead", + was, class, gb(weights), slots, per), nil +} diff --git a/internal/control/modelserve/modelserve_test.go b/internal/control/modelserve/modelserve_test.go index c186a96..dd58c56 100644 --- a/internal/control/modelserve/modelserve_test.go +++ b/internal/control/modelserve/modelserve_test.go @@ -5,6 +5,7 @@ import ( "context" "encoding/binary" "fmt" + "gitea.mynodes.xyz/ilker/llmgw/internal/gguf" "net/http" "net/http/httptest" "slices" @@ -412,3 +413,18 @@ func TestAnEmbeddingModelIsServedForEmbedding(t *testing.T) { t.Errorf("a chat model got %q", args) } } + +func TestAPoolTheCardCannotHoldIsShrunkNotStarted(t *testing.T) { + h := gguf.Header{Architecture: "apertus", BlockCount: 32, HeadCount: 32, HeadCountKV: 8, EmbeddingLength: 4096} + req := Request{CtxPool: 131072, ContextLength: 32768} + got, why, err := fitPoolToCard(req, h, int64(8.5*(1<<30)), "6000-24g", 1) + if err != nil || why == "" || got.CtxPool > 105000 || got.CtxPool/got.ContextLength != 4 { + t.Errorf("fitted %+v %q %v", got, why, err) + } + if _, _, err := fitPoolToCard(req, h, int64(30*(1<<30)), "6000-24g", 1); err == nil { + t.Error("weights larger than the card were not refused") + } + if got, why, _ := fitPoolToCard(Request{CtxPool: 16384, ContextLength: 4096}, h, int64(8.5*(1<<30)), "6000-24g", 1); why != "" || got.CtxPool != 16384 { + t.Errorf("a pool that fits was changed: %+v %q", got, why) + } +} diff --git a/internal/gguf/gguf.go b/internal/gguf/gguf.go index b9842f1..8d4257a 100644 --- a/internal/gguf/gguf.go +++ b/internal/gguf/gguf.go @@ -378,3 +378,40 @@ func Check(ctx context.Context, url, llamaVersion string) Verdict { v.Header = h return v } + +// CardUsable is the share of a card's memory a model may take: the rest is +// CUDA's own and the engine's compute buffers. +const CardUsable = 0.9 + +// PoolThatFits is the largest context pool, in tokens, whose KV cache fits +// on gpus cards of cardGB beside weights bytes of weights and a gigabyte for +// the engine: -1 when the weights alone do not, 0 when the header or the +// card does not say enough to tell. +func (h Header) PoolThatFits(weights int64, cardGB, gpus int) int { + perTok := h.KVBytesPerToken() + if cardGB <= 0 || perTok <= 0 || weights <= 0 { + return 0 + } + budget := float64(cardGB*max(gpus, 1))*CardUsable*(1<<30) - float64(weights) - (1 << 30) + if budget <= 0 { + return -1 + } + return int(budget / float64(perTok)) +} + +// ClassMemGB is a card's memory as its class names it: 6000-24g is 24, 0 +// for a class that does not say. +func ClassMemGB(class string) int { + i := strings.LastIndex(class, "-") + if i < 0 || !strings.HasSuffix(class, "g") { + return 0 + } + n := 0 + for _, c := range strings.TrimSuffix(class[i+1:], "g") { + if c < '0' || c > '9' { + return 0 + } + n = n*10 + int(c-'0') + } + return n +} diff --git a/internal/gguf/gguf_test.go b/internal/gguf/gguf_test.go index ff68770..5f006fc 100644 --- a/internal/gguf/gguf_test.go +++ b/internal/gguf/gguf_test.go @@ -188,3 +188,25 @@ func TestAnEmbeddingModelIsKnownByItsPooling(t *testing.T) { t.Errorf("no pooling: %+v %v", h, err) } } + +func TestAPoolIsSizedToTheCard(t *testing.T) { + // Apertus-8B's shape: 32 layers, 8 KV heads of 128, 128 KiB a token. + h, err := Read(bytes.NewReader(shapedHeader("apertus", 32, 32, 8, 4096, true))) + if err != nil { + t.Fatal(err) + } + weights := int64(8.5 * (1 << 30)) + most := h.PoolThatFits(weights, ClassMemGB("6000-24g"), 1) + if most < 90000 || most > 105000 { + t.Errorf("an 8B on 24 GB holds %d tokens of pool", most) + } + if most >= 131072 { + t.Error("the 131072 pool that failed on the card would pass") + } + if got := h.PoolThatFits(int64(30*(1<<30)), 24, 1); got != -1 { + t.Errorf("30 GB of weights on 24 GB: %d", got) + } + if ClassMemGB("t4-15g") != 15 || ClassMemGB("cpu") != 0 || ClassMemGB("v100-32g") != 32 { + t.Error("class memory misread") + } +} diff --git a/internal/web/admin_mirror.go b/internal/web/admin_mirror.go index 0c7bd27..95ce1be 100644 --- a/internal/web/admin_mirror.go +++ b/internal/web/admin_mirror.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "html/template" "net/http" "slices" @@ -342,7 +343,14 @@ func (p *AccountPages) RegisterAdminMirror(mux *http.ServeMux, d MirrorDeps) { // between the listing and the click, or a mirror that has gone // down since, both end as a replica failing on a node minutes // later with an error naming the node. Asking costs one HEAD. - if _, err := sourcecheck.Check(ctx, model.Source); err != nil { + size, err := sourcecheck.Check(ctx, model.Source) + if err != nil { + return "", err + } + // The weights and the context's cache have to fit the card: + // an 8B asked for a 131072-token pool on 24 GB loaded its + // weights and failed to make its context, again and again. + if err := p.fitsCard(ctx, l, model, size); err != nil { return "", err } if err := d.Catalog.Add(ctx, model, admin.Email); err != nil { @@ -365,6 +373,14 @@ func (p *AccountPages) RegisterAdminMirror(mux *http.ServeMux, d MirrorDeps) { mux.HandleFunc("POST /admin/mirror/{model}/{version}/remove", p.mirrorAction(m, func(ctx context.Context, _ store.User, r *http.Request) (string, error) { model, ver := r.PathValue("model"), r.PathValue("version") + // Weights a catalog model still loads are not taken from under + // it: removed from the mirror alone, the model stayed in the + // catalog and its replicas went on loading the copy on the node. + if users := p.mirrorUsers(ctx, model, ver); len(users) > 0 { + return "", fmt.Errorf("%s/%s is the weights of %s in the catalog: remove %s first "+ + "(the models page, or llmgw ctl model remove %s), then the weights", + model, ver, strings.Join(users, ", "), strings.Join(users, ", "), users[0]) + } removed, err := m.Remove(ctx, model, ver) if err != nil { return "", err @@ -937,3 +953,52 @@ func sharedPrefix(ids []string) string { } return "" } + +// mirrorUsers are the catalog models whose weights are the mirror's +// model/version. +func (p *AccountPages) mirrorUsers(ctx context.Context, model, ver string) []string { + if p.catalog == nil { + return nil + } + srcs, err := p.catalog.Sources(ctx) + if err != nil { + return nil + } + part := "/" + model + "/" + ver + "/" + var out []string + for src, ms := range srcs { + if !strings.Contains(src, part) { + continue + } + for _, m := range ms { + if !slices.Contains(out, m.Name) { + out = append(out, m.Name) + } + } + } + slices.Sort(out) + return out +} + +// fitsCard refuses a model whose weights and context pool do not fit the +// cards its profile names, saying the largest pool that would. +func (p *AccountPages) fitsCard(ctx context.Context, l *Lang, m catalog.Model, size int64) error { + if len(m.Profiles) == 0 || m.Profiles[0].GPUClass == "" || size <= 0 { + return nil + } + pr := m.Profiles[0] + v := p.checkSource(ctx, m.Source) + cards := max(pr.GPUsPerNode, pr.TP, 1) + most := v.Header.PoolThatFits(size, gguf.ClassMemGB(pr.GPUClass), cards) + pool := argInt(pr.Args, "--ctx-size") + if pool <= 0 { + pool = defaultCtxPool + } + switch { + case most < 0: + return errors.New(l.T("serve.err.weights_too_big", l.Num(float64(size)/(1<<30), 1), pr.GPUClass)) + case most > 0 && pool > most: + return errors.New(l.T("serve.err.pool_too_big", l.Int(int64(pool)), pr.GPUClass, l.Int(int64(most/1024*1024)))) + } + return nil +} diff --git a/internal/web/admin_mirror_test.go b/internal/web/admin_mirror_test.go index df57683..9f1cf63 100644 --- a/internal/web/admin_mirror_test.go +++ b/internal/web/admin_mirror_test.go @@ -766,3 +766,28 @@ func TestAJudgesWeightsAreServedAsTheJudgeThePoliciesAsk(t *testing.T) { t.Errorf("a model clients use became a judge: %+v", plain) } } + +func TestWeightsAModelStillLoadsAreNotRemoved(t *testing.T) { + // Removed from the mirror alone, a model stayed in the catalog and its + // replicas went on loading the copy on the node, with nothing said. + m := &fakeMirror{} + cat := &fakeCatalog{sources: map[string][]catalog.Model{ + "http://10.0.0.146:8081/apertus-70b/aff98136/Apertus-70B-UD-IQ1_M.gguf": {{Name: "apertus-70b-instruct-2509"}}, + }} + a := fakeAccounts{user: store.User{ID: "u1", Email: "a@example.com", IsAdmin: true}} + p := NewAccountPages(a, nil, nil, "v0", slog.New(slog.NewTextHandler(io.Discard, nil))) + mux := http.NewServeMux() + p.RegisterAdminMirror(mux, MirrorDeps{Mirror: m, Catalog: cat}) + body := mirrorPost(t, mux, "/admin/mirror/apertus-70b/aff98136/remove", url.Values{"csrf": {"tok"}}).Body.String() + if m.removed != [2]string{} { + t.Errorf("weights in use were removed: %v", m.removed) + } + if !strings.Contains(body, "apertus-70b-instruct-2509") || !strings.Contains(body, "remove") { + t.Errorf("the page does not say which model uses them: %s", body) + } + // Weights nothing loads go as before. + mirrorPost(t, mux, "/admin/mirror/other/v1/remove", url.Values{"csrf": {"tok"}}) + if m.removed != [2]string{"other", "v1"} { + t.Errorf("unused weights were not removed: %v", m.removed) + } +} diff --git a/internal/web/locales/en/serve.json b/internal/web/locales/en/serve.json index d36fa62..09f44dd 100644 --- a/internal/web/locales/en/serve.json +++ b/internal/web/locales/en/serve.json @@ -73,6 +73,8 @@ "serve.err.engine": "which engine should run it?", "serve.err.needs_gpu": "%s runs on GPUs; choose a GPU class to place it on", "serve.err.placement": "%q is not a placement", + "serve.err.weights_too_big": "Its weights (%s GB) do not fit a %s card with room to run: choose a smaller quantisation, or more cards.", + "serve.err.pool_too_big": "A context pool of %s tokens does not fit a %s card beside these weights; at most %s would.", "serve.err.cpu_with_free_gpus.one": "this fleet has %d GPU free and %s was set to run on the CPU; choose a GPU placement, or tick \"run on the CPU anyway\"", "serve.err.cpu_with_free_gpus.other": "this fleet has %d GPUs free and %s was set to run on the CPU; choose a GPU placement, or tick \"run on the CPU anyway\"", "serve.msg.added": "%s added to the catalog, %s: %s; it is placed on the next pass", diff --git a/internal/web/locales/tr/serve.json b/internal/web/locales/tr/serve.json index 6113f7f..7ad2b98 100644 --- a/internal/web/locales/tr/serve.json +++ b/internal/web/locales/tr/serve.json @@ -72,6 +72,8 @@ "serve.err.engine": "hangi motor çalıştıracak?", "serve.err.needs_gpu": "%s GPU üzerinde çalışır; yerleştirmek için bir GPU sınıfı seçin", "serve.err.placement": "%q geçerli bir yerleşim değil", + "serve.err.weights_too_big": "Ağırlıkları (%s GB) bir %s kartına çalışacak payla sığmıyor: daha küçük bir quantization ya da daha fazla kart seçin.", + "serve.err.pool_too_big": "%s token'lık bağlam havuzu bu ağırlıklarla bir %s kartına sığmıyor; en fazla %s sığar.", "serve.err.cpu_with_free_gpus.other": "bu kümede %d GPU boşta, %s ise CPU üzerinde çalışacak biçimde ayarlandı; bir GPU yerleşimi seçin ya da \"yine de CPU üzerinde çalıştır\" kutusunu işaretleyin", "serve.msg.added": "%s kataloğa eklendi, %s: %s; bir sonraki geçişte yerleştirilecek", "serve.replicas_n.other": "%d replika",