diff --git a/CHANGELOG.md b/CHANGELOG.md index cd32ab7b..0cece9b9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,20 @@ is only released once its **exit criterion** is met, not when its code lands. Until v1.0.0 the minor version is the breaking one: configuration formats and Go APIs change between minors without a deprecation period. +## Unreleased + +### Added + +- **A served model moved to more cards, or fewer.** `llmgw ctl model gpus + MODEL N [--profile P] [--in-place]`, `POST /api/v1/models/{name}/gpus`, and + GPUs… in the console's model menu. How many cards of one node each replica + uses: tensor parallel for vLLM, SGLang and TensorRT-LLM, checked against + the model's KV heads, and the layers split for llama.cpp. Refused for a + count no node has, or whose cards would not hold the model and its + context, with the figures; its replicas are then started again on it, one + by one, beside the old — or, with `--in-place` when there is no room + beside them, each stopped first. + ## v0.15.1 — 2026-10-29 A 125B model on one card, a second mirror, and weights that run code diff --git a/docs/admin-api.openapi.json b/docs/admin-api.openapi.json index defc36fd..a38c6d6e 100644 --- a/docs/admin-api.openapi.json +++ b/docs/admin-api.openapi.json @@ -2964,6 +2964,58 @@ } } }, + "/api/v1/models/{name}/gpus": { + "post": { + "operationId": "postModelsNameGpus", + "summary": "Spread a served model over more cards or fewer: its profile, checked against its KV heads, its cards' memory and the nodes, then its replicas restarted", + "description": "Tensor parallel for vLLM, SGLang and TensorRT-LLM, a layer split for llama.cpp, on one node. With no room beside the running replicas, in_place stops each first.", + "tags": [ + "models" + ], + "parameters": [ + { + "name": "name", + "in": "path", + "required": true, + "description": "The model's name in the catalog.", + "schema": { + "type": "string" + } + }, + { + "$ref": "#/components/parameters/Actor" + } + ], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/modelserve.GPUsRequest" + } + } + } + }, + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/modelserve.GPUsChange" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, "/api/v1/models/{name}/headroom": { "get": { "operationId": "getModelsNameHeadroom", @@ -11133,6 +11185,67 @@ } } }, + "modelserve.GPUsChange": { + "type": "object", + "description": "GPUsChange is what was changed, and the replicas started again on it.", + "properties": { + "model": { + "type": "string" + }, + "profile": { + "type": "string" + }, + "engine": { + "type": "string" + }, + "gpu_class": { + "type": "string" + }, + "before": { + "type": "integer" + }, + "after": { + "type": "integer" + }, + "args": { + "type": "array", + "items": { + "type": "string" + } + }, + "per_card": { + "type": "string", + "description": "PerCard is what each card holds after, when it could be worked out." + }, + "restarted": { + "type": "integer" + }, + "in_place": { + "type": "boolean" + }, + "note": { + "type": "string", + "description": "Note is what could not be checked, said once." + } + } + }, + "modelserve.GPUsRequest": { + "type": "object", + "description": "GPUsRequest is how many cards a replica is to use.", + "properties": { + "gpus": { + "type": "integer" + }, + "profile": { + "type": "string", + "description": "Profile is the one to change, when the model has more than one profile on cards." + }, + "in_place": { + "type": "boolean", + "description": "InPlace stops each replica before its replacement starts: for when no node has the cards free beside it. It does not answer while the new one loads." + } + } + }, "modelserve.Job": { "type": "object", "description": "Job is one model being served, and where it stands.", diff --git a/docs/install.md b/docs/install.md index 5a62046f..cdd064d0 100644 --- a/docs/install.md +++ b/docs/install.md @@ -1648,6 +1648,36 @@ it, one by one, a new one ready before the old one stops. The console has the same under a model's menu, Context length…. A model declared in the cluster's settings is changed there (`llmgw ctl config edit`). +### Moving a served model to more cards + +How many cards of one node each replica uses can be changed while the model +is served: + +``` +llmgw ctl model gpus qwen3-coder-30b 2 # two cards a replica +llmgw ctl model gpus llama-70b 4 --profile h100 # the model has several profiles on cards +llmgw ctl model gpus qwen3-coder-30b 2 --in-place # no room beside the running replica +``` + +vLLM, SGLang and TensorRT-LLM spread the model by tensor parallel (`tp`), +which needs the count to divide the model's KV heads or be divided by them: +a model with 8 splits 2, 4, 8 or 16 ways, not 3 or 12. llama.cpp has no +tensor parallel; it splits the layers over the cards (`gpus_per_node`), and +a `--tensor-split` written for the old cards is dropped. ComfyUI and Strata +run on one card. A count no node has is refused, and so is one whose cards +would not hold the weights and the context — with the figures; fewer cards +with a long context may need `llmgw ctl model context` first. + +Its replicas are then started again on it, each new one ready beside the old +before the old stops. That needs the cards free beside it: on a node of two +cards with the model on one, two do not fit until it is gone, and the command +says so. `--in-place` stops each replica first and starts its replacement on +the cards it held and the free ones, and the model does not answer while the +new one loads — a model with several replicas keeps answering on the others. +A replica spread over several nodes is laid out by hand, with a fabric +(`nodes`, `gpus_per_node` in the profile). The console has the same under a +model's menu, GPUs…. + ### Trying new weights on some of the traffic first When a model's repository publishes new weights, `llmgw ctl model update` diff --git a/internal/cli/model.go b/internal/cli/model.go index 60e8ad6b..602d10d5 100644 --- a/internal/cli/model.go +++ b/internal/cli/model.go @@ -34,7 +34,7 @@ func newModelCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { Use: "model", Short: "Add or remove models while the cluster is running", } - c.AddCommand(newModelContextCmd(client), newModelAddCmd(client), newModelPresetsCmd(client, asJSON), newModelRemoveCmd(client), newModelPlanCmd(client), + c.AddCommand(newModelContextCmd(client), newModelGPUsCmd(client), newModelAddCmd(client), newModelPresetsCmd(client, asJSON), newModelRemoveCmd(client), newModelPlanCmd(client), newResourcesCmd(client), newClassCmd(client), newOwnerCmd(client), newTargetCmd(client, asJSON), newAdaptersCmd(client), newModelServeCmd(client), newModelUpdatesCmd(client, asJSON), newModelUpdateCmd(client), @@ -442,3 +442,57 @@ func newModelContextCmd(client func() *ctlClient) *cobra.Command { c.Flags().IntVar(&slots, "slots", 0, "llama.cpp: requests a replica answers at once (default: as it is)") return c } + +func newModelGPUsCmd(client func() *ctlClient) *cobra.Command { + var profile string + var inPlace bool + c := &cobra.Command{ + Use: "gpus ", + Short: "Spread a served model over more cards, or fewer", + Long: "How many cards of one node each replica uses. vLLM, SGLang and TensorRT-LLM\n" + + "spread the model by tensor parallel, which needs the count to divide the\n" + + "model's KV heads or be divided by them; llama.cpp splits its layers over the\n" + + "cards. It refuses a count a node does not have, and one whose cards would not\n" + + "hold the model and its context, saying what each would need. Its replicas are\n" + + "then started again on it, each new one beside the old before it stops.\n\n" + + "With no node free enough beside the running replicas — a node of two cards,\n" + + "the model on one — --in-place stops each replica first: the model does not\n" + + "answer while the new one loads.\n\n" + + "--profile names the profile for a model with more than one on cards.", + Example: " llmgw ctl model gpus qwen3-coder-30b 2\n" + + " llmgw ctl model gpus llama-70b 4 --profile h100 --in-place", + Args: cobra.ExactArgs(2), + RunE: func(cmd *cobra.Command, args []string) error { + n, err := strconv.Atoi(args[1]) + if err != nil { + return fmt.Errorf("the cards are a whole number, not %q", args[1]) + } + body, _ := json.Marshal(modelserve.GPUsRequest{GPUs: n, Profile: profile, InPlace: inPlace}) + var v modelserve.GPUsChange + if err := client().patient(time.Minute).post(cmd.Context(), "/api/v1/models/"+args[0]+"/gpus", body, &v); err != nil { + return err + } + out := cmd.OutOrStdout() + fmt.Fprintf(out, "%s: %d → %d %s card(s) a replica (%s, %s)\n", v.Model, v.Before, v.After, v.Class, v.Profile, v.Engine) + if v.PerCard != "" { + fmt.Fprintln(out, " "+v.PerCard) + } + if len(v.Args) > 0 { + fmt.Fprintln(out, " args: "+strings.Join(v.Args, " ")) + } + if v.Note != "" { + fmt.Fprintln(out, " note: "+v.Note) + } + switch { + case v.Restarted > 0 && v.InPlace: + fmt.Fprintf(out, "%d replica(s) stopped and started again on it, one by one: llmgw ctl replicas --model %s\n", v.Restarted, v.Model) + case v.Restarted > 0: + fmt.Fprintf(out, "%d replica(s) started again on it, each new one before the old stops: llmgw ctl replicas --model %s\n", v.Restarted, v.Model) + } + return nil + }, + } + c.Flags().StringVar(&profile, "profile", "", "the profile to change, for a model with several on cards") + c.Flags().BoolVar(&inPlace, "in-place", false, "stop each replica before its replacement starts, when there is no room beside it") + return c +} diff --git a/internal/cli/server.go b/internal/cli/server.go index 113545a6..e0abc232 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -607,7 +607,7 @@ func newServerCmd() *cobra.Command { // Updates: the catalog in force, the service that // changes it, and what replaces replicas. Current: func() []catalog.Model { return admin.Catalog().Models() }, - Models: catalogSvc, Restart: admin.Reconciler().Restart, + Models: catalogSvc, Restart: admin.Reconciler().Restart, RestartInPlace: admin.Reconciler().RestartInPlace, Pool: adminAPI.PoolNodes, Notify: notifier, // Canaries: how the router saw each side serve. CanaryStats: rt.CanaryStats, ResetCanary: rt.ResetCanary, diff --git a/internal/cli/tui_actions.go b/internal/cli/tui_actions.go index d03705b4..aa255e57 100644 --- a/internal/cli/tui_actions.go +++ b/internal/cli/tui_actions.go @@ -356,6 +356,31 @@ func (m consoleModel) modelMenu(md adminapi.ModelView) *menuDialog { }}) return nil }}, + {"GPUs…", func(m *consoleModel) tea.Cmd { + profiles := append([]string{"(its only one on cards)"}, md.Profiles...) + m.openForm(&formDialog{title: "Cards of " + md.Name, + note: "How many cards of one node each replica uses: tensor parallel for vLLM, SGLang and TensorRT-LLM, " + + "the layers split for llama.cpp. Checked against the model's heads, its cards' memory and the nodes; " + + "its replicas are then started again on it, one by one.", + fields: []formField{ + choiceF("profile", profiles, profiles[0]), + textF("cards", "", "a replica's, on one node"), + choiceF("in place", []string{"no", "yes"}, "no"), + }, + submit: func(m *consoleModel, v []string) (tea.Cmd, error) { + n, err := strconv.Atoi(v[1]) + if err != nil || n < 1 { + return nil, fmt.Errorf("the cards are a whole number from 1") + } + profile := v[0] + if profile == profiles[0] { + profile = "" + } + return m.call("Cards of "+md.Name, http.MethodPost, path+"/gpus", + modelserve.GPUsRequest{GPUs: n, Profile: profile, InPlace: v[2] == "yes"}, gpusSays), nil + }}) + return nil + }}, {"Resources…", func(m *consoleModel) tea.Cmd { profiles := md.Profiles if len(profiles) == 0 { @@ -943,3 +968,25 @@ func contextSays(raw json.RawMessage) string { } return s } + +// gpusSays is a change of cards in a sentence or two. +func gpusSays(raw json.RawMessage) string { + var v modelserve.GPUsChange + if json.Unmarshal(raw, &v) != nil { + return "Done." + } + s := fmt.Sprintf("%s: %d → %d %s card(s) a replica.", v.Model, v.Before, v.After, v.Class) + if v.PerCard != "" { + s += "\n" + v.PerCard + } + if v.Note != "" { + s += "\n" + v.Note + } + switch { + case v.Restarted > 0 && v.InPlace: + s += fmt.Sprintf("\n%d replica(s) stopped and started again on it, one by one.", v.Restarted) + case v.Restarted > 0: + s += fmt.Sprintf("\n%d replica(s) started again on it, one by one.", v.Restarted) + } + return s +} diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index c280a8de..38aacc1e 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -206,6 +206,7 @@ func (a *API) Register(mux Mux) { mux.HandleFunc("POST /api/v1/updates/check", a.checkUpdates) mux.HandleFunc("POST /api/v1/models/{name}/update", a.updateModel) mux.HandleFunc("POST /api/v1/models/{name}/context", a.setModelContext) + mux.HandleFunc("POST /api/v1/models/{name}/gpus", a.setModelGPUs) mux.HandleFunc("POST /api/v1/models/{name}/rollback", a.rollbackModel) a.registerCanary(mux) a.registerEval(mux) @@ -1774,6 +1775,26 @@ func (a *API) setModelContext(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, change) } +// setModelGPUs spreads a served model over more cards, or fewer, and +// starts its replicas again on it. +func (a *API) setModelGPUs(w http.ResponseWriter, r *http.Request) { + if a.serve == nil { + writeError(w, http.StatusNotImplemented, "this control plane cannot change a model's cards") + return + } + var req modelserve.GPUsRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + change, err := a.serve.SetGPUs(r.Context(), r.PathValue("name"), req, actorOf(r)) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, change) +} + func (a *API) listUpdates(w http.ResponseWriter, _ *http.Request) { out := make([]modelserve.UpdateInfo, 0) if a.serve != nil { diff --git a/internal/control/adminapi/routes.go b/internal/control/adminapi/routes.go index b68b28c4..95688baf 100644 --- a/internal/control/adminapi/routes.go +++ b/internal/control/adminapi/routes.go @@ -215,6 +215,9 @@ var Routes = []Route{ Request: modelserve.Change{}, OptionalBody: true, Response: modelserve.Job{}, Status: http.StatusAccepted}, {Method: "POST", Path: "/api/v1/models/{name}/context", Summary: "Change a served model's context: the catalog, its engines' arguments and memory, then its replicas restarted", Request: modelserve.ContextRequest{}, Response: modelserve.ContextChange{}}, + {Method: "POST", Path: "/api/v1/models/{name}/gpus", Summary: "Spread a served model over more cards or fewer: its profile, checked against its KV heads, its cards' memory and the nodes, then its replicas restarted", + Description: "Tensor parallel for vLLM, SGLang and TensorRT-LLM, a layer split for llama.cpp, on one node. With no room beside the running replicas, in_place stops each first.", + Request: modelserve.GPUsRequest{}, Response: modelserve.GPUsChange{}}, {Method: "POST", Path: "/api/v1/models/{name}/rollback", Summary: "Go back to the weights the last update replaced", Response: Result{}}, {Method: "POST", Path: "/api/v1/models/{name}/origin", Summary: "Record the repository a model's weights came from", diff --git a/internal/control/modelserve/context.go b/internal/control/modelserve/context.go index 8f0c7ae3..17fd1bdc 100644 --- a/internal/control/modelserve/context.go +++ b/internal/control/modelserve/context.go @@ -49,11 +49,13 @@ type ContextChange struct { } // shape is what sizes a context: a token's KV cache, the context the model -// was trained for, and its weights. +// was trained for, and its weights; and its KV heads, which decide how many +// cards tensor parallel can spread it over (0 unknown). type shape struct { kvPerToken int64 trained int weights int64 + kvHeads int } // SetContext changes a served model's context. @@ -135,12 +137,14 @@ func (s *Server) shapeOf(ctx context.Context, m catalog.Model) (shape, string) { return sh, "its header could not be read (" + err.Error() + "), so the memory it needs is not checked" } sh.kvPerToken, sh.trained = h.KVBytesPerToken(), int(h.ContextLength) //nolint:gosec // a header's context + sh.kvHeads = int(h.HeadCountKV) //nolint:gosec // a header's head count case m.Origin != nil && m.Origin.Repo != "": mm, err := modelplan.FetchModel(ctx, m.Origin.Repo) if err != nil { return sh, "its config could not be read (" + err.Error() + "), so the memory it needs is not checked" } sh.kvPerToken, sh.trained = mm.KVBytesPerToken(modelplan.BF16), mm.MaxContext + sh.kvHeads = mm.KVHeads default: return sh, "where it came from is not known, so the memory it needs is not checked" } diff --git a/internal/control/modelserve/gpus.go b/internal/control/modelserve/gpus.go new file mode 100644 index 00000000..70e90f39 --- /dev/null +++ b/internal/control/modelserve/gpus.go @@ -0,0 +1,285 @@ +package modelserve + +import ( + "context" + "errors" + "fmt" + "slices" + "strings" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/modelplan" +) + +// A served model moved to more cards, or fewer: how many each replica +// spreads it over, on one node. The profile changes — tensor parallel for +// vLLM, SGLang and TensorRT-LLM, llama.cpp's layer split for llama.cpp — +// checked against the model's KV heads, what each card would hold, and the +// nodes there are; then its replicas start again on it. + +// GPUsRequest is how many cards a replica is to use. +type GPUsRequest struct { + GPUs int `json:"gpus"` + // Profile is the one to change, when the model has more than one + // profile on cards. + Profile string `json:"profile,omitempty"` + // InPlace stops each replica before its replacement starts: for when no + // node has the cards free beside it. It does not answer while the new + // one loads. + InPlace bool `json:"in_place,omitempty"` +} + +// GPUsChange is what was changed, and the replicas started again on it. +type GPUsChange struct { + Model string `json:"model"` + Profile string `json:"profile"` + Engine string `json:"engine"` + Class string `json:"gpu_class"` + Before int `json:"before"` + After int `json:"after"` + Args []string `json:"args"` + // PerCard is what each card holds after, when it could be worked out. + PerCard string `json:"per_card,omitempty"` + Restarted int `json:"restarted"` + InPlace bool `json:"in_place,omitempty"` + // Note is what could not be checked, said once. + Note string `json:"note,omitempty"` +} + +// tpFlags are each engine's own spelling of a tensor-parallel width, which +// the agent sets from the profile: a profile's own would contradict it. +var tpFlags = map[string][]string{ + "vllm": {"--tensor-parallel-size", "-tp"}, + "sglang": {"--tp-size", "--tensor-parallel-size", "--tp"}, + "trtllm": {"--tp_size"}, + // llama.cpp's proportions were for the cards it had. + "llamacpp": {"--tensor-split", "-ts"}, +} + +// SetGPUs changes how many cards a served model's replicas use. +func (s *Server) SetGPUs(ctx context.Context, name string, req GPUsRequest, by string) (GPUsChange, error) { + if s.deps.Current == nil || s.deps.Models == nil { + return GPUsChange{}, errors.New("this control plane cannot change the catalog") + } + var m catalog.Model + found := false + for _, c := range s.deps.Current() { + if c.Name == name { + m, found = c, true + } + } + switch { + case !found: + return GPUsChange{}, fmt.Errorf("no model %q in the catalog", name) + case s.deps.Models.Declared(name): + return GPUsChange{}, fmt.Errorf("model %q is declared in the cluster's settings: change its profile there (llmgw ctl config edit)", name) + case req.GPUs < 1 || req.GPUs > 64: + return GPUsChange{}, errors.New("a replica uses 1 to 64 cards") + } + i, err := gpuProfile(m, req.Profile) + if err != nil { + return GPUsChange{}, err + } + p := m.Profiles[i] + before := max(p.GPUsPerNode, p.TP, 1) * max(p.Nodes, p.PP, 1) + out := GPUsChange{Model: name, Profile: p.Name, Engine: p.Engine, Class: p.GPUClass, Before: before, After: req.GPUs} + if before == req.GPUs { + return out, fmt.Errorf("%s's %s profile already uses %d card(s)", name, p.Name, before) + } + + sh, note := s.shapeOf(ctx, m) + next, err := respread(p, req.GPUs, sh) + if err != nil { + return out, err + } + out.Args = next.Args + switch kv := int64(contextOf(m, p)) * sh.kvPerToken; { + case sh.weights == 0 || sh.kvPerToken == 0 || contextOf(m, p) == 0: + if note == "" { + note = "its context is not known, so what each card holds is not checked" + } + default: + need := modelplan.CardNeed(p.Engine, sh.weights, kv, req.GPUs) + out.PerCard = fmt.Sprintf("%.1f GB a card of %d", need, req.GPUs) + if budget := modelplan.CardBudget(p.GPUClass); budget > 0 { + out.PerCard += fmt.Sprintf(", of %.0f usable", budget) + if need > budget { + return out, fmt.Errorf("on %d %s cards each would hold %.1f GB, more than the %.0f GB they hold: "+ + "more cards, or a shorter context (llmgw ctl model context)", req.GPUs, p.GPUClass, need, budget) + } + } + } + out.Note = note + + ids := s.replicasOf(name) + inPlace, err := s.roomFor(p.GPUClass, req.GPUs, name, len(ids) > 0, req.InPlace) + if err != nil { + return out, err + } + profiles := slices.Clone(m.Profiles) + profiles[i] = next + if _, err := s.deps.Models.SetProfiles(ctx, name, profiles, by); err != nil { + return out, err + } + if len(ids) == 0 { + return out, nil + } + restart := s.deps.Restart + if inPlace { + restart, out.InPlace = s.deps.RestartInPlace, true + } + if restart == nil { + return out, errors.New("the profile is changed, and this control plane cannot start its replicas again: llmgw ctl restart") + } + if err := restart(ids...); err != nil { + return out, fmt.Errorf("the profile is changed, and its replicas could not be started again on it: %w", err) + } + out.Restarted = len(ids) + return out, nil +} + +// gpuProfile is the model's profile on cards to change: the one named, or +// its only one. +func gpuProfile(m catalog.Model, named string) (int, error) { + var onCards []int + for i, p := range m.Profiles { + if named != "" && p.Name == named { + if p.CPUOnly() { + return 0, fmt.Errorf("%s's %s profile runs on CPUs; give it a gpu_class first", m.Name, p.Name) + } + return i, nil + } + if !p.CPUOnly() { + onCards = append(onCards, i) + } + } + switch { + case named != "": + return 0, fmt.Errorf("%s has no profile %q", m.Name, named) + case len(onCards) == 0: + return 0, fmt.Errorf("%s runs on CPUs only; a profile on cards needs a gpu_class", m.Name) + case len(onCards) > 1: + var names []string + for _, i := range onCards { + names = append(names, m.Profiles[i].Name) + } + return 0, fmt.Errorf("%s has several profiles on cards (%s): name one", m.Name, strings.Join(names, ", ")) + } + return onCards[0], nil +} + +// respread is a profile spread over n cards of one node. +func respread(p catalog.Profile, n int, sh shape) (catalog.Profile, error) { + for _, f := range tpFlags[p.Engine] { + p.Args = dropArg(p.Args, f) + } + switch p.Engine { + case "vllm", "sglang", "trtllm": + if !modelplan.SplitsHeads(sh.kvHeads, n) { + return p, fmt.Errorf("the model has %d KV heads, which tensor parallel cannot spread over %d cards: "+ + "a number that divides them, or that they divide", sh.kvHeads, n) + } + p.TP, p.GPUsPerNode = n, n + case "llamacpp": + // llama.cpp has no tensor parallel: it splits the layers over a + // node's cards, in proportion to their free memory. + p.TP, p.GPUsPerNode = 1, n + default: + return p, fmt.Errorf("a %s replica runs on one card", p.Engine) + } + // One node: a replica spanning nodes is laid out by hand, with a fabric. + p.Nodes, p.PP = 1, 0 + return p, nil +} + +// contextOf is the tokens a profile's KV cache holds: llama.cpp's pool, or +// the engine's limit, or the model's context. +func contextOf(m catalog.Model, p catalog.Profile) int { + switch p.Engine { + case "llamacpp": + if n := argInt(p.Args, "--ctx-size"); n > 0 { + return n + } + case "vllm": + if n := argInt(p.Args, "--max-model-len"); n > 0 { + return n + } + case "sglang": + if n := argInt(p.Args, "--context-length"); n > 0 { + return n + } + } + return m.ContextLength +} + +func (s *Server) replicasOf(model string) []string { + var ids []string + if s.deps.Fleet != nil { + for _, r := range s.deps.Fleet.Replicas() { + if r.Model == model { + ids = append(ids, r.ID) + } + } + } + return ids +} + +// roomFor says whether the new replicas fit: a node with n cards of the +// class at all, and free beside the running ones — or, asked to start them +// in place, once the model's own are stopped. True is in place. +func (s *Server) roomFor(class string, n int, model string, running, inPlace bool) (bool, error) { + if s.deps.Pool == nil { + return inPlace, nil + } + own := map[string]int{} + if s.deps.Fleet != nil { + for _, r := range s.deps.Fleet.Replicas() { + if r.Model == model { + for _, sh := range r.Shards { + own[sh.NodeID] += len(sh.GPUIndices) + } + } + } + } + most, freeMost, besideOK, inPlaceOK := 0, 0, false, false + for _, node := range s.deps.Pool() { + for _, k := range modelplan.Kinds(node) { + if k.Class != class { + continue + } + most, freeMost = max(most, k.GPUs), max(freeMost, k.FreeGPUs) + besideOK = besideOK || k.FreeGPUs >= n + inPlaceOK = inPlaceOK || k.FreeGPUs+own[node.ID] >= n + } + } + switch { + case most == 0: + return false, fmt.Errorf("no ready node has %s cards", class) + case most < n: + return false, fmt.Errorf("no node has %d %s cards; the most one has is %d", n, class, most) + case !running || besideOK: + return false, nil + case !inPlaceOK: + return false, fmt.Errorf("no node has %d %s cards free, even with this model's own stopped (%d at most free now): "+ + "other models hold them", n, class, freeMost) + case !inPlace: + return false, fmt.Errorf("no node has %d %s cards free beside the running replica (%d at most); "+ + "--in-place stops each replica first, and the model does not answer while the new one loads", n, class, freeMost) + } + return true, nil +} + +// dropArg is arguments without a flag and its value. +func dropArg(args []string, flag string) []string { + out := make([]string, 0, len(args)) + for i := 0; i < len(args); i++ { + switch { + case args[i] == flag: + i++ // its value + case strings.HasPrefix(args[i], flag+"="): + default: + out = append(out, args[i]) + } + } + return out +} diff --git a/internal/control/modelserve/gpus_test.go b/internal/control/modelserve/gpus_test.go new file mode 100644 index 00000000..b35c91a3 --- /dev/null +++ b/internal/control/modelserve/gpus_test.go @@ -0,0 +1,151 @@ +package modelserve + +import ( + "context" + "slices" + "strings" + "testing" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" + "gitea.mynodes.xyz/ilker/llmgw/internal/modelplan" +) + +// gpuSetup is the update setup with qwen on one L40S card of llama.cpp, and +// a pool of one node of four, of which free are free. +func gpuSetup(t *testing.T, free int) (*Server, *updModels, *restartFleet, *[]string) { + t.Helper() + s, _, models, _, fleet := updSetup(t) + m := models.models["qwen"] + m.ContextLength = 8192 + m.Origin = &catalog.Origin{Repo: "o/r", Files: []catalog.OriginFile{{Path: "q4.gguf", Bytes: 20 << 30}}} + m.Profiles = []catalog.Profile{ + {Name: "cpu", Engine: "llamacpp", MinMemoryGB: 8}, + {Name: "gpu", Engine: "llamacpp", GPUClass: "l40s-48g", TP: 1, GPUsPerNode: 1, + Args: []string{"--ctx-size", "8192", "--tensor-split", "1"}}, + } + models.models["qwen"] = m + s.deps.Pool = func() []modelplan.PoolNode { + return []modelplan.PoolNode{{ID: "g1", Class: "l40s-48g", GPUMemGB: 48, GPUs: 4, FreeGPUs: free}} + } + var inPlace []string + s.deps.RestartInPlace = func(ids ...string) error { inPlace = append(inPlace, ids...); return nil } + return s, models, fleet, &inPlace +} + +func TestAModelIsSpreadOverMoreCardsAndItsReplicasStartedAgainBesideTheOld(t *testing.T) { + s, models, fleet, inPlace := gpuSetup(t, 3) + got, err := s.SetGPUs(context.Background(), "qwen", GPUsRequest{GPUs: 2}, "admin") + if err != nil { + t.Fatal(err) + } + if got.Profile != "gpu" || got.Before != 1 || got.After != 2 || got.Restarted != 2 || got.InPlace { + t.Fatalf("%+v", got) + } + p := models.models["qwen"].Profiles[1] + // llama.cpp splits layers over a node's cards: gpus_per_node, not tp, + // and the old card's proportions are gone. + if p.GPUsPerNode != 2 || p.TP != 1 || p.Nodes != 1 || !slices.Equal(p.Args, []string{"--ctx-size", "8192"}) { + t.Errorf("profile %+v", p) + } + if !strings.Contains(got.PerCard, "a card of 2") { + t.Errorf("per card %q", got.PerCard) + } + if len(*inPlace) != 0 { + t.Errorf("stopped first though there was room: %v", *inPlace) + } + for _, r := range fleet.Replicas() { + if !strings.Contains(r.ID, "-new") { + t.Errorf("%s was not started again", r.ID) + } + } + if models.models["qwen"].Profiles[0].GPUsPerNode != 0 { + t.Error("the CPU profile was changed too") + } +} + +func TestWithNoRoomBesideIt_InPlaceIsAskedFor(t *testing.T) { + // One card free and the model's replicas on the others: two do not fit + // beside them. + s, _, fleet, inPlace := gpuSetup(t, 1) + fleet.reps[0].Shards = []reconciler.Shard{{NodeID: "g1", GPUIndices: []int{1}}} + fleet.reps[1].Shards = []reconciler.Shard{{NodeID: "g1", GPUIndices: []int{2}}} + _, err := s.SetGPUs(context.Background(), "qwen", GPUsRequest{GPUs: 2}, "admin") + if err == nil || !strings.Contains(err.Error(), "--in-place") { + t.Fatalf("not asked for in place: %v", err) + } + got, err := s.SetGPUs(context.Background(), "qwen", GPUsRequest{GPUs: 2, InPlace: true}, "admin") + if err != nil { + t.Fatal(err) + } + if !got.InPlace || len(*inPlace) != 2 { + t.Errorf("%+v, stopped first: %v", got, *inPlace) + } +} + +func TestMoreCardsThanANodeHasOrOthersHoldAreRefused(t *testing.T) { + s, models, _, _ := gpuSetup(t, 4) + if _, err := s.SetGPUs(context.Background(), "qwen", GPUsRequest{GPUs: 8}, "admin"); err == nil || + !strings.Contains(err.Error(), "the most one has is 4") { + t.Errorf("eight cards on a node of four: %v", err) + } + s, _, _, _ = gpuSetup(t, 0) + if _, err := s.SetGPUs(context.Background(), "qwen", GPUsRequest{GPUs: 2, InPlace: true}, "admin"); err == nil || + !strings.Contains(err.Error(), "other models hold them") { + t.Errorf("cards other models hold: %v", err) + } + if models.models["qwen"].Profiles[1].GPUsPerNode != 1 { + t.Error("a refused change was kept") + } +} + +func TestFewerCardsThanTheModelNeedsAreRefusedWithTheFigures(t *testing.T) { + s, models, _, _ := gpuSetup(t, 3) + m := models.models["qwen"] + m.Profiles[1].GPUsPerNode, m.Origin.Files[0].Bytes = 4, 120<<30 + models.models["qwen"] = m + _, err := s.SetGPUs(context.Background(), "qwen", GPUsRequest{GPUs: 2}, "admin") + if err == nil || !strings.Contains(err.Error(), "GB") { + t.Errorf("120 GB on two 48 GB cards: %v", err) + } +} + +func TestTensorParallelFollowsTheKVHeads(t *testing.T) { + p := catalog.Profile{Name: "gpu", Engine: "vllm", GPUClass: "h100-80g", TP: 2, GPUsPerNode: 2, + Args: []string{"--tensor-parallel-size", "2", "--max-model-len", "32768"}} + if _, err := respread(p, 3, shape{kvHeads: 8}); err == nil || !strings.Contains(err.Error(), "8 KV heads") { + t.Errorf("three ways for eight heads: %v", err) + } + next, err := respread(p, 4, shape{kvHeads: 8}) + if err != nil { + t.Fatal(err) + } + // The agent sets the width from tp; the profile's own spelling went. + if next.TP != 4 || next.GPUsPerNode != 4 || !slices.Equal(next.Args, []string{"--max-model-len", "32768"}) { + t.Errorf("%+v", next) + } + for _, eng := range []string{"comfyui", "strata"} { + if _, err := respread(catalog.Profile{Name: "x", Engine: eng}, 2, shape{}); err == nil { + t.Errorf("%s spread over two cards", eng) + } + } +} + +func TestAModelOnCPUsOrWithSeveralCardProfilesSaysWhy(t *testing.T) { + s, models, _, _ := gpuSetup(t, 3) + m := models.models["qwen"] + m.Profiles = append(m.Profiles, catalog.Profile{Name: "gpu2", Engine: "vllm", GPUClass: "l40s-48g", TP: 1}) + models.models["qwen"] = m + if _, err := s.SetGPUs(context.Background(), "qwen", GPUsRequest{GPUs: 2}, "admin"); err == nil || + !strings.Contains(err.Error(), "name one") { + t.Errorf("two profiles on cards and none named: %v", err) + } + if _, err := s.SetGPUs(context.Background(), "qwen", GPUsRequest{GPUs: 2, Profile: "cpu"}, "admin"); err == nil || + !strings.Contains(err.Error(), "gpu_class") { + t.Errorf("a CPU profile: %v", err) + } + if _, err := s.SetGPUs(context.Background(), "qwen", GPUsRequest{GPUs: 1, Profile: "gpu"}, "admin"); err == nil || + !strings.Contains(err.Error(), "already") { + t.Errorf("the same count: %v", err) + } +} diff --git a/internal/control/modelserve/modelserve.go b/internal/control/modelserve/modelserve.go index 357c710a..9dc12bfa 100644 --- a/internal/control/modelserve/modelserve.go +++ b/internal/control/modelserve/modelserve.go @@ -82,6 +82,9 @@ type Deps struct { Current func() []catalog.Model Models Models Restart func(ids ...string) error + // RestartInPlace stops each replica before its replacement starts, for + // when there is no room beside it. + RestartInPlace func(ids ...string) error // Pool is the fleet's ready nodes as a plan sees them: what decides // which file suits the hardware, and how it is spread. diff --git a/internal/control/modelserve/updates.go b/internal/control/modelserve/updates.go index 4f5b1fbc..4ce09d17 100644 --- a/internal/control/modelserve/updates.go +++ b/internal/control/modelserve/updates.go @@ -35,6 +35,7 @@ type Models interface { SetSource(ctx context.Context, name, source string, origin *catalog.Origin, by string) (catalog.Model, error) SetOrigin(ctx context.Context, name string, origin *catalog.Origin) error SetContext(ctx context.Context, name string, contextLength, maxOutput int, profiles []catalog.Profile, by string) (catalog.Model, error) + SetProfiles(ctx context.Context, name string, profiles []catalog.Profile, by string) (catalog.Model, error) // For a canary: the candidate added beside the model and taken out // again, and the share of requests it is given. Add(ctx context.Context, m catalog.Model, by string) error diff --git a/internal/control/modelserve/updates_test.go b/internal/control/modelserve/updates_test.go index 12a2cc15..611fdff2 100644 --- a/internal/control/modelserve/updates_test.go +++ b/internal/control/modelserve/updates_test.go @@ -112,6 +112,15 @@ func (c *updModels) SetContext(_ context.Context, n string, ctxLen, maxOut int, c.models[n] = m return m, nil } + +func (c *updModels) SetProfiles(_ context.Context, n string, ps []catalog.Profile, _ string) (catalog.Model, error) { + c.mu.Lock() + defer c.mu.Unlock() + m := c.models[n] + m.Profiles = ps + c.models[n] = m + return m, nil +} func (c *updModels) all() []catalog.Model { c.mu.Lock() defer c.mu.Unlock() diff --git a/internal/modelplan/fit.go b/internal/modelplan/fit.go index 7175d2c2..92c5f5ed 100644 --- a/internal/modelplan/fit.go +++ b/internal/modelplan/fit.go @@ -155,10 +155,7 @@ func fitsOn(m Model, h Hardware, d Demand, p Precision, context int) []Fit { if nodes > 1 && h.Fabric == "" { continue } - // The KV heads are split among the cards, or past their number each - // card holds a copy of one: the width divides them or they divide - // it. A model with 8 KV heads splits 2, 4, 8 or 16 ways, not 12. - if m.KVHeads > 0 && m.KVHeads%width != 0 && width%m.KVHeads != 0 { + if !SplitsHeads(m.KVHeads, width) { continue } diff --git a/internal/modelplan/options.go b/internal/modelplan/options.go index 4e1c84df..64ce2aeb 100644 --- a/internal/modelplan/options.go +++ b/internal/modelplan/options.go @@ -71,6 +71,9 @@ func (n PoolNode) kinds() []PoolKind { return []PoolKind{{Class: n.Class, GPUMemGB: n.GPUMemGB, GPUs: n.GPUs, FreeGPUs: n.FreeGPUs}} } +// Kinds is a node's devices by class: its one class, or each of several. +func Kinds(n PoolNode) []PoolKind { return n.kinds() } + // isMIG reports whether a class names MIG instances: "a100-80g-mig-3g.40gb". // A replica runs on exactly one, never a group of them, so a plan offers one // instance and nothing wider. Read from the class label, as the card's memory @@ -263,12 +266,7 @@ func vllmOptions(a Artifact, kv int64, sh Shape, nodes []PoolNode, bpw float64) perCard := func(width int) int64 { return (a.Bytes+kv)/int64(width) + overheadBytes(a.Bytes/int64(width)) } - // Tensor parallel splits the KV heads among the cards, or past their - // number gives each card a copy of one: so the width divides the heads, - // or the heads divide the width. Qwen3-235B has four and runs on eight. - divides := func(w int) bool { - return sh.KVHeads == 0 || sh.KVHeads%w == 0 || w%sh.KVHeads == 0 - } + divides := func(w int) bool { return SplitsHeads(sh.KVHeads, w) } for _, c := range classes(nodes) { budget := float64(c.mem) * gpuShare opt := func(perNode, n int, fabric string) Option { @@ -516,3 +514,13 @@ func CardNeed(engine string, weights, kv int64, cards int) float64 { // CardBudget is what a card of a class may be planned to hold, in GB; 0 for // a class whose memory its name does not say. func CardBudget(class string) float64 { return float64(MemFromClass(class)) * gpuShare } + +// SplitsHeads reports whether tensor parallel can spread a model of kvHeads +// KV heads over width cards. It splits the heads among the cards, or past +// their number gives each card a copy of one: so the width divides the +// heads, or the heads divide the width. Qwen3-235B has four and runs on +// eight; a model with 8 splits 2, 4, 8 or 16 ways, not 12. 0 heads is +// unknown, and allowed. +func SplitsHeads(kvHeads, width int) bool { + return kvHeads == 0 || width <= 1 || kvHeads%width == 0 || width%kvHeads == 0 +}