diff --git a/docs/admin-api.openapi.json b/docs/admin-api.openapi.json index 9985f19..bcba606 100644 --- a/docs/admin-api.openapi.json +++ b/docs/admin-api.openapi.json @@ -3227,6 +3227,58 @@ } } }, + "/api/v1/models/{name}/replicas": { + "post": { + "operationId": "postModelsNameReplicas", + "summary": "How many replicas run on each profile", + "description": "Seven on cards and four on CPUs, under the model's name. A profile not named keeps its count; cpu adds a profile on CPUs to a llama.cpp model that has none. Its scaling becomes the sum.", + "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/ProfileReplicasRequest" + } + } + } + }, + "responses": { + "200": { + "description": "OK", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/catalog.Model" + } + } + } + }, + "401": { + "$ref": "#/components/responses/Unauthorized" + }, + "default": { + "$ref": "#/components/responses/Error" + } + } + } + }, "/api/v1/models/{name}/resources": { "post": { "operationId": "postModelsNameResources", @@ -8103,6 +8155,13 @@ "type": "string" } }, + "profile_replicas": { + "type": "object", + "additionalProperties": { + "type": "integer" + }, + "description": "ProfileReplicas is how many replicas run on each profile, for a model whose replicas are counted per profile; absent otherwise." + }, "memory_gb": { "type": "integer", "description": "MemoryGB and CPUCores are what one replica reserves, from its first profile; MeasuredMemoryGB is the most one of its replicas uses now, 0 when none is measured." @@ -8665,6 +8724,21 @@ } } }, + "ProfileReplicasRequest": { + "type": "object", + "description": "ProfileReplicasRequest counts a model's replicas per profile: profile name to replicas, and CPU, for a llama.cpp model with no profile on CPUs, the one to add.", + "properties": { + "profiles": { + "type": "object", + "additionalProperties": { + "type": "integer" + } + }, + "cpu": { + "$ref": "#/components/schemas/modelsvc.CPUProfile" + } + } + }, "ProjectView": { "type": "object", "description": "ProjectView is a project, with how many people and keys are in it.", @@ -10510,6 +10584,10 @@ "type": "integer", "description": "Priority orders profiles during placement; highest that fits wins." }, + "replicas": { + "type": "integer", + "description": "Replicas, set on any of a model's profiles, is how many of its replicas run on this one: seven on cards and four on CPUs, all answering under the model's name, rather than each replica on the best profile that fits. The model's scaling is then the sum, min and max alike; a profile left at zero runs none." + }, "startup_timeout": { "type": "integer", "format": "int64", @@ -11906,6 +11984,21 @@ } } }, + "modelsvc.CPUProfile": { + "type": "object", + "description": "CPUProfile is a profile on CPUs to add to a llama.cpp model, for replicas counted per profile: what a replica of it reserves.", + "properties": { + "name": { + "type": "string" + }, + "memory_gb": { + "type": "integer" + }, + "cpu_cores": { + "type": "integer" + } + } + }, "notify.Result": { "type": "object", "description": "Result is how one webhook answered a test.", diff --git a/docs/reference.md b/docs/reference.md index b56a2f4..1888d4a 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -295,6 +295,7 @@ rather than repeated. | `keep_within` | `string` | [Profile](#profile) | | `weight` | `float64` | [Profile](#profile) | | `priority` | `int` | [Profile](#profile) | +| `replicas` | `int` | [Profile](#profile) | | `startup_timeout` | `time.Duration` | [Profile](#profile) | | `kv_events` | `bool` | [Profile](#profile) | | `gpu` | `*GPUNeed` | [Profile](#profile) | @@ -2611,6 +2612,16 @@ rather than a guess (plan §6.2.1). Priority orders profiles during placement; highest that fits wins. +### `replicas` + +`int` + +Replicas, set on any of a model's profiles, is how many of its +replicas run on this one: seven on cards and four on CPUs, all +answering under the model's name, rather than each replica on the +best profile that fits. The model's scaling is then the sum, min and +max alike; a profile left at zero runs none. + ### `startup_timeout` `time.Duration` diff --git a/internal/cli/model.go b/internal/cli/model.go index 4b09d6d..900cf25 100644 --- a/internal/cli/model.go +++ b/internal/cli/model.go @@ -19,6 +19,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/control/adminapi" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/sourcecheck" ) @@ -34,7 +35,7 @@ func newModelCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { Use: "model", Short: "Add or remove models while the cluster is running", } - c.AddCommand(newModelStopCmd(client), newModelStartCmd(client), newModelContextCmd(client), newModelGPUsCmd(client), newModelAddCmd(client), newModelPresetsCmd(client, asJSON), newModelRemoveCmd(client), newModelPlanCmd(client), + c.AddCommand(newModelStopCmd(client), newModelStartCmd(client), newModelReplicasCmd(client), 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), @@ -598,3 +599,50 @@ func newModelGPUsCmd(client func() *ctlClient) *cobra.Command { c.Flags().BoolVar(&inPlace, "in-place", false, "stop each replica before its replacement starts, when there is no room beside it") return c } + +func newModelReplicasCmd(client func() *ctlClient) *cobra.Command { + var cpuMemory, cpuCores int + c := &cobra.Command{ + Use: "replicas =...", + Short: "How many of a model's replicas run on each of its profiles: gpu=7 cpu=4", + Long: "Each replica runs on the best profile that fits unless counted here: seven on\n" + + "cards and four on CPUs, all answering under the model's name, the router filling\n" + + "the cards before the CPUs. A profile not named keeps its count. For a GGUF model\n" + + "with no profile on CPUs, --cpu-memory adds one named cpu. Its scaling becomes the\n" + + "sum; llmgw ctl model scale with another total ends the count per profile.", + Example: " llmgw ctl model replicas qwen-coder gpu-1xl40s-48g=7 cpu=4 --cpu-memory 24 --cpu-cores 8", + Args: cobra.MinimumNArgs(2), + RunE: func(cmd *cobra.Command, args []string) error { + req := adminapi.ProfileReplicasRequest{Profiles: map[string]int{}} + for _, a := range args[1:] { + name, v, ok := strings.Cut(a, "=") + n, err := strconv.Atoi(v) + if !ok || err != nil || name == "" { + return fmt.Errorf("%q: a profile and a count, as gpu=7", a) + } + req.Profiles[name] = n + } + if cpuMemory > 0 { + req.CPU = &modelsvc.CPUProfile{MemoryGB: cpuMemory, CPUCores: cpuCores} + } + body, err := json.Marshal(req) + if err != nil { + return err + } + var m catalog.Model + if err := client().post(cmd.Context(), "/api/v1/models/"+args[0]+"/replicas", body, &m); err != nil { + return err + } + var parts []string + for _, p := range m.Profiles { + parts = append(parts, fmt.Sprintf("%d on %s", p.Replicas, p.Name)) + } + fmt.Fprintf(cmd.OutOrStdout(), "%s runs %s: %d in all. Those short start first, then those over stop.\n", + m.Name, strings.Join(parts, ", "), m.SplitTotal()) + return nil + }, + } + c.Flags().IntVar(&cpuMemory, "cpu-memory", 0, "add a profile on CPUs named cpu, reserving this many GB a replica") + c.Flags().IntVar(&cpuCores, "cpu-cores", 0, "the cores a replica on CPUs computes on (default: half the node's)") + return c +} diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 54b0971..7f594b4 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -22,6 +22,7 @@ import ( "unicode" "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelserve" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/gguf" "gitea.mynodes.xyz/ilker/llmgw/internal/modelplan" "gitea.mynodes.xyz/ilker/llmgw/internal/release" @@ -130,6 +131,7 @@ func (a *API) Register(mux Mux) { mux.HandleFunc("POST /api/v1/models/{name}/start", a.startModel) mux.HandleFunc("POST /api/v1/models/{name}/scale", a.scaleModel) mux.HandleFunc("POST /api/v1/models/{name}/resources", a.setResources) + mux.HandleFunc("POST /api/v1/models/{name}/replicas", a.setProfileReplicas) a.registerQuotas(mux) a.registerKB(mux) a.registerTenancy(mux) @@ -492,6 +494,9 @@ type ModelView struct { Unplaceable bool `json:"unplaceable,omitempty"` Fallbacks []string `json:"fallbacks,omitempty"` Profiles []string `json:"profiles"` + // ProfileReplicas is how many replicas run on each profile, for a model + // whose replicas are counted per profile; absent otherwise. + ProfileReplicas map[string]int `json:"profile_replicas,omitempty"` // MemoryGB and CPUCores are what one replica reserves, from its first // profile; MeasuredMemoryGB is the most one of its replicas uses now, 0 // when none is measured. @@ -868,6 +873,12 @@ func (a *API) Models() []ModelView { Canary: m.Canary, CanaryOf: a.ctl.Catalog().CanaryOf(m.Name), Eval: m.Eval, Role: m.Role, } + if m.Split() { + v.ProfileReplicas = map[string]int{} + for _, p := range m.Profiles { + v.ProfileReplicas[p.Name] = p.Replicas + } + } for _, p := range m.Profiles { v.Profiles = append(v.Profiles, p.Name) } @@ -965,6 +976,15 @@ type Models interface { SetClass(ctx context.Context, name, class, by string) (catalog.Model, error) SetOwner(ctx context.Context, name, owner, by string) (catalog.Model, error) SetAdapters(ctx context.Context, name string, adapters []catalog.Adapter, by string) (catalog.Model, error) + SetProfileReplicas(ctx context.Context, name string, counts map[string]int, cpu *modelsvc.CPUProfile, by string) (catalog.Model, error) +} + +// ProfileReplicasRequest counts a model's replicas per profile: profile +// name to replicas, and CPU, for a llama.cpp model with no profile on CPUs, +// the one to add. +type ProfileReplicasRequest struct { + Profiles map[string]int `json:"profiles"` + CPU *modelsvc.CPUProfile `json:"cpu,omitempty"` } // ResourcesRequest sets what a model's profiles reserve of a node. @@ -1291,6 +1311,25 @@ func (a *API) setResources(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, m) } +// setProfileReplicas counts a model's replicas per profile. +func (a *API) setProfileReplicas(w http.ResponseWriter, r *http.Request) { + if a.models == nil { + writeError(w, http.StatusNotImplemented, "this control plane has no store") + return + } + var req ProfileReplicasRequest + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<16)).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "the request is not valid JSON: "+err.Error()) + return + } + m, err := a.models.SetProfileReplicas(r.Context(), r.PathValue("name"), req.Profiles, req.CPU, actorOf(r)) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, m) +} + func (a *API) removeModel(w http.ResponseWriter, r *http.Request) { if a.models == nil { writeError(w, http.StatusNotImplemented, "this control plane has no store") diff --git a/internal/control/adminapi/api_test.go b/internal/control/adminapi/api_test.go index 1ce542e..a1b482e 100644 --- a/internal/control/adminapi/api_test.go +++ b/internal/control/adminapi/api_test.go @@ -16,6 +16,7 @@ import ( "gitea.mynodes.xyz/ilker/llmgw/internal/control" "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" "gitea.mynodes.xyz/ilker/llmgw/internal/control/inventory" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/modelsvc" "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" "gitea.mynodes.xyz/ilker/llmgw/internal/router/endpoint" "gitea.mynodes.xyz/ilker/llmgw/internal/store" @@ -263,6 +264,16 @@ func (f *fakeModels) SetAdapters(_ context.Context, name string, as []catalog.Ad return catalog.Model{Name: name, Adapters: as}, f.err } +func (f *fakeModels) SetProfileReplicas(_ context.Context, name string, counts map[string]int, _ *modelsvc.CPUProfile, + _ string, +) (catalog.Model, error) { + m := catalog.Model{Name: name} + for p, n := range counts { + m.Profiles = append(m.Profiles, catalog.Profile{Name: p, Replicas: n}) + } + return m, f.err +} + func (f *fakeModels) SetOwner(_ context.Context, name, owner, _ string) (catalog.Model, error) { return catalog.Model{Name: name, Owner: owner}, f.err } diff --git a/internal/control/adminapi/routes.go b/internal/control/adminapi/routes.go index 225815d..14665f7 100644 --- a/internal/control/adminapi/routes.go +++ b/internal/control/adminapi/routes.go @@ -168,6 +168,10 @@ var Routes = []Route{ Response: ScaleResponse{}}, {Method: "POST", Path: "/api/v1/models/{name}/scale", Summary: "Set how many replicas a model runs", Request: ScaleRequest{}, Response: ScaleResponse{}}, + {Method: "POST", Path: "/api/v1/models/{name}/replicas", Summary: "How many replicas run on each profile", + Description: "Seven on cards and four on CPUs, under the model's name. A profile not named keeps its count; " + + "cpu adds a profile on CPUs to a llama.cpp model that has none. Its scaling becomes the sum.", + Request: ProfileReplicasRequest{}, Response: catalog.Model{}}, {Method: "POST", Path: "/api/v1/models/{name}/resources", Summary: "Set what a model's profiles reserve of a node", Request: ResourcesRequest{}, Response: catalog.Model{}}, {Method: "POST", Path: "/api/v1/models/{name}/class", Summary: "Set a model's capacity class", diff --git a/internal/control/catalog/catalog.go b/internal/control/catalog/catalog.go index 058c1ce..931ffa4 100644 --- a/internal/control/catalog/catalog.go +++ b/internal/control/catalog/catalog.go @@ -267,6 +267,12 @@ type Profile struct { Weight float64 `json:"weight,omitempty" koanf:"weight"` // Priority orders profiles during placement; highest that fits wins. Priority int `json:"priority,omitempty" koanf:"priority"` + // Replicas, set on any of a model's profiles, is how many of its + // replicas run on this one: seven on cards and four on CPUs, all + // answering under the model's name, rather than each replica on the + // best profile that fits. The model's scaling is then the sum, min and + // max alike; a profile left at zero runs none. + Replicas int `json:"replicas,omitempty" koanf:"replicas"` // StartupTimeout bounds model loading. A 70B model across two nodes takes // tens of minutes, so this is per profile rather than global. StartupTimeout time.Duration `json:"startup_timeout,omitempty" koanf:"startup_timeout"` @@ -1024,10 +1030,57 @@ func (m Model) normalize() Model { return m } +// Split says whether the model's replicas are counted per profile: any of +// its profiles sets replicas. +func (m Model) Split() bool { + return slices.ContainsFunc(m.Profiles, func(p Profile) bool { return p.Replicas > 0 }) +} + +// SplitTotal is the replicas its profiles ask for together. +func (m Model) SplitTotal() int { + n := 0 + for _, p := range m.Profiles { + n += max(p.Replicas, 0) + } + return n +} + +// ProfileBase is the catalog profile a placement profile stands for: one +// named by the card it needs is placed as name@class. +func ProfileBase(name string) string { + base, _, _ := strings.Cut(name, "@") + return base +} + +// validateSplit refuses replicas per profile that the model's scaling does +// not add up to. +func (m Model) validateSplit() error { + for _, p := range m.Profiles { + if p.Replicas < 0 { + return fmt.Errorf("model %q profile %q: replicas %d is negative", m.Name, p.Name, p.Replicas) + } + } + if !m.Split() || m.Stopped != nil { + return nil + } + n := m.SplitTotal() + if m.Scaling.Min != n || m.Scaling.Max != n { + return fmt.Errorf("model %q: its profiles' replicas add up to %d, so its scaling is min %d, max %d, not %d and %d", + m.Name, n, n, n, m.Scaling.Min, m.Scaling.Max) + } + if len(m.Scaling.Schedule) > 0 { + return fmt.Errorf("model %q: replicas per profile and a schedule are one or the other", m.Name) + } + return nil +} + func (m Model) validate() error { if m.Name == "" { return fmt.Errorf("a model has no name") } + if err := m.validateSplit(); err != nil { + return err + } if m.Source == "" { return fmt.Errorf("model %q: source is required", m.Name) } @@ -1510,6 +1563,9 @@ func (m Model) Simulated() bool { // happened, and the answer to "why is this one on the wrong node" was in // somebody's memory of what they edited an hour ago. func (p Profile) Fingerprint() string { + // How many replicas run on it is not how each runs: changing the count + // must not mark the running ones stale. + p.Replicas = 0 b, err := json.Marshal(p) if err != nil { // A profile that will not marshal cannot be compared, and saying so by diff --git a/internal/control/modelsvc/modelsvc.go b/internal/control/modelsvc/modelsvc.go index 21036da..03c1641 100644 --- a/internal/control/modelsvc/modelsvc.go +++ b/internal/control/modelsvc/modelsvc.go @@ -22,6 +22,7 @@ import ( "fmt" "log/slog" "slices" + "strings" "sync" "time" @@ -249,6 +250,13 @@ func (s *Service) Remove(ctx context.Context, name string) error { // Scale changes how many replicas a model added at runtime runs. func (s *Service) Scale(ctx context.Context, name string, sc catalog.Scaling, by string) (catalog.Model, error) { final, err := s.update(ctx, name, by, "change its scaling", func(m *catalog.Model) error { + // Replicas counted per profile stay so only for the same total, one + // way: any other scaling is the model's again, on whatever fits. + if m.Split() && (sc.Min != sc.Max || sc.Min != m.SplitTotal()) { + for i := range m.Profiles { + m.Profiles[i].Replicas = 0 + } + } m.Scaling = sc // A number set is what it runs now: a stopped model is started by it. m.Stopped = nil @@ -261,6 +269,98 @@ func (s *Service) Scale(ctx context.Context, name string, sc catalog.Scaling, by return final, err } +// CPUProfile is a profile on CPUs to add to a llama.cpp model, for replicas +// counted per profile: what a replica of it reserves. +type CPUProfile struct { + Name string `json:"name,omitempty"` + MemoryGB int `json:"memory_gb"` + CPUCores int `json:"cpu_cores,omitempty"` +} + +// SetProfileReplicas sets how many of a model's replicas run on each of its +// profiles: seven on cards and four on CPUs, under the one name. A profile +// not named keeps its count. cpu, for a llama.cpp model with no profile on +// CPUs, adds one — its card profile without the card, preferred below every +// other, so the router fills the cards first and the CPUs after. Running +// replicas carry on; those over a count stop once the ones short have +// started. +func (s *Service) SetProfileReplicas(ctx context.Context, name string, counts map[string]int, cpu *CPUProfile, + by string, +) (catalog.Model, error) { + final, err := s.update(ctx, name, by, "count its replicas per profile", func(m *catalog.Model) error { + if cpu != nil { + if cpu.Name == "" { + cpu.Name = "cpu" + } + if !slices.ContainsFunc(m.Profiles, func(p catalog.Profile) bool { return p.Name == cpu.Name }) { + p, err := cpuProfileOf(*m, *cpu) + if err != nil { + return err + } + m.Profiles = append(m.Profiles, p) + } + } + for prof, n := range counts { + i := slices.IndexFunc(m.Profiles, func(p catalog.Profile) bool { return p.Name == prof }) + if i < 0 { + return fmt.Errorf("model %q has no profile %q", name, prof) + } + if n < 0 { + return fmt.Errorf("profile %q: %d replicas", prof, n) + } + m.Profiles[i].Replicas = n + } + n := m.SplitTotal() + if n == 0 { + return fmt.Errorf("none of %q's profiles would run a replica; llmgw ctl model stop %s keeps it without", name, name) + } + sc := catalog.Scaling{Min: n, Max: n, IdleTimeout: m.Scaling.IdleTimeout} + if m.Stopped != nil { + sc.IdleTimeout = m.Stopped.IdleTimeout + m.Stopped = &sc + } else { + m.Scaling = sc + } + return nil + }) + if err == nil { + s.log.Info("replicas counted per profile", "model", name, "counts", counts, "by", by) + } + return final, err +} + +// cpuProfileOf is a model's llama.cpp profile on CPUs: the same engine and +// arguments, without the card or what spreads layers over cards. +func cpuProfileOf(m catalog.Model, cpu CPUProfile) (catalog.Profile, error) { + i := slices.IndexFunc(m.Profiles, func(p catalog.Profile) bool { return p.Engine == "llamacpp" }) + if i < 0 { + return catalog.Profile{}, fmt.Errorf("model %q is not served by llama.cpp; only a GGUF model runs on CPUs", m.Name) + } + if cpu.MemoryGB <= 0 { + return catalog.Profile{}, errors.New("a profile on CPUs needs the memory a replica reserves") + } + p := m.Profiles[i] + lowest := p.Priority + for _, o := range m.Profiles { + lowest = min(lowest, o.Priority) + } + out := catalog.Profile{Name: cpu.Name, Engine: p.Engine, EngineVersion: p.EngineVersion, Source: p.Source, + Env: p.Env, NodeSelector: p.NodeSelector, StartupTimeout: p.StartupTimeout, Weight: p.Weight, + Priority: lowest - 10, MinMemoryGB: cpu.MemoryGB, CPUCores: cpu.CPUCores} + for j := 0; j < len(p.Args); j++ { + switch a := p.Args[j]; { + case slices.Contains([]string{"--n-gpu-layers", "-ngl", "--split-mode", "-sm", "--tensor-split", "-ts", + "--main-gpu", "-mg"}, a): + j++ // and its value + case strings.HasPrefix(a, "--n-gpu-layers=") || strings.HasPrefix(a, "--split-mode=") || + strings.HasPrefix(a, "--tensor-split=") || strings.HasPrefix(a, "--main-gpu="): + default: + out.Args = append(out.Args, a) + } + } + return out, nil +} + // Stop takes a model out of service and keeps it: none of its replicas run, // and what it was scaled to is kept for Start. func (s *Service) Stop(ctx context.Context, name, by string) (catalog.Model, error) { diff --git a/internal/control/modelsvc/modelsvc_test.go b/internal/control/modelsvc/modelsvc_test.go index f06fa43..348464a 100644 --- a/internal/control/modelsvc/modelsvc_test.go +++ b/internal/control/modelsvc/modelsvc_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "log/slog" + "slices" "strings" "testing" "time" @@ -458,3 +459,44 @@ func TestAStrataModelReservesTheMemoryItsExpertsTake(t *testing.T) { t.Errorf("an unknown size: %v", err) } } + +// A GGUF model on a card gets a profile on CPUs, and seven replicas on cards +// beside four on CPUs; a scaling of another total puts it back on whatever +// fits. +func TestReplicasCountedPerProfile(t *testing.T) { + s, _, _ := newService(t) + m := catalog.Model{Name: "g", Source: "/models/g.gguf", Scaling: catalog.Scaling{Min: 1, Max: 1}, + Profiles: []catalog.Profile{{Name: "gpu", Engine: "llamacpp", GPUClass: "l40s-48g", GPUsPerNode: 2, + Priority: 100, Args: []string{"--ctx-size", "16384", "--n-gpu-layers", "999", "--split-mode=layer"}}}} + if err := s.Add(t.Context(), m, "ilker"); err != nil { + t.Fatal(err) + } + if _, err := s.SetProfileReplicas(t.Context(), "g", map[string]int{"gpu": 7, "cpu": 4}, nil, "ilker"); err == nil { + t.Fatal("a count for a profile it does not have was taken") + } + got, err := s.SetProfileReplicas(t.Context(), "g", map[string]int{"gpu": 7, "cpu": 4}, + &CPUProfile{MemoryGB: 24, CPUCores: 8}, "ilker") + if err != nil { + t.Fatal(err) + } + if len(got.Profiles) != 2 || got.Scaling.Min != 11 || got.Scaling.Max != 11 { + t.Fatalf("profiles %+v, scaling %+v", got.Profiles, got.Scaling) + } + cpu := got.Profiles[1] + if cpu.Name != "cpu" || cpu.GPUClass != "" || cpu.GPUsPerNode != 0 || cpu.Replicas != 4 || cpu.Priority >= 100 || + cpu.MinMemoryGB != 24 || cpu.CPUCores != 8 || !slices.Equal(cpu.Args, []string{"--ctx-size", "16384"}) { + t.Errorf("the profile on CPUs: %+v", cpu) + } + // Stopped and started, it comes back split. + _, _ = s.Stop(t.Context(), "g", "ilker") + if got, _ = s.Start(t.Context(), "g", "ilker"); got.Scaling.Min != 11 || got.Profiles[0].Replicas != 7 { + t.Errorf("started: %+v", got) + } + // The same total keeps the split; another one ends it. + if got, _ = s.Scale(t.Context(), "g", catalog.Scaling{Min: 11, Max: 11}, "ilker"); !got.Split() { + t.Error("the same total ended the split") + } + if got, _ = s.Scale(t.Context(), "g", catalog.Scaling{Min: 3, Max: 3}, "ilker"); got.Split() { + t.Error("another total kept the split") + } +} diff --git a/internal/control/reconciler/reconciler.go b/internal/control/reconciler/reconciler.go index 0c1d060..020a6fe 100644 --- a/internal/control/reconciler/reconciler.go +++ b/internal/control/reconciler/reconciler.go @@ -886,14 +886,25 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) { } desired := r.desiredFor(m) + // Replicas counted per profile: one short on its profile is started even + // with the total right — before one of a profile with too many is + // stopped, which the total going over then does — so changing seven on + // cards and four on CPUs to nine and two never runs fewer than eleven. + var deficit map[string]bool + var surplus []string + split := m.Split() && desired == m.SplitTotal() + if split { + deficit, surplus = r.splitGaps(m) + } + needUp := len(liveIDs) < desired || split && len(deficit) > 0 && len(liveIDs) == desired outcome := notTried switch { - case len(liveIDs) < desired && isParked: + case needUp && isParked: // Not placed. Its failure is on every status the operator reads. - case len(liveIDs) < desired && r.isHeld(m.Name): + case needUp && r.isHeld(m.Name): // It gave up room to a model of a higher class, which is placed // first; taking it back now would undo the preemption. - case len(liveIDs) < desired: + case needUp: if r.now().Before(blockedUntil) { // Still in backoff. Rebuilding immediately would hot-loop on a // model that cannot start — a missing weights path, say — and bury @@ -912,7 +923,11 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) { attribute.String("llmgw.model", m.Name), attribute.Int("llmgw.replicas", len(liveIDs)), attribute.Int("llmgw.desired", desired)) - r.scaleDown(sctx, liveIDs[desired:], "more replicas than desired") + excess := liveIDs[desired:] + if split { + excess = pickSurplus(surplus, liveIDs, len(liveIDs)-desired) + } + r.scaleDown(sctx, excess, "more replicas than desired") span.End() } @@ -976,6 +991,17 @@ func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []strin attribute.String("llmgw.model", m.Name), attribute.StringSlice("llmgw.avoid", avoid)) profiles := r.orderedProfiles(m) + if m.Split() { + // Only the profiles short of their replicas. + deficit, _ := r.splitGaps(m) + profiles = slices.DeleteFunc(profiles, func(p placement.Profile) bool { + return !deficit[catalog.ProfileBase(p.Name)] + }) + if len(profiles) == 0 { + pspan.End() + return notTried + } + } res := r.sched.Place(profiles, avoid) if res.Placement == nil && len(avoid) > 0 { // Spreading across failure domains is a preference, not a rule. If it diff --git a/internal/control/reconciler/split.go b/internal/control/reconciler/split.go new file mode 100644 index 0000000..1a8b0cd --- /dev/null +++ b/internal/control/reconciler/split.go @@ -0,0 +1,64 @@ +package reconciler + +import ( + "slices" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" +) + +// splitGaps is, for a model whose replicas are counted per profile, the +// profiles short of theirs and the replicas over: of a profile with more +// than it asks, or of one it no longer has. Counted as reconcileModel +// counts — a replica being retired is not one it has — so a replica +// restarted is replaced on its own profile. +func (r *Reconciler) splitGaps(m catalog.Model) (deficit map[string]bool, surplus []string) { + live := map[string][]string{} + r.mu.Lock() + for _, rep := range r.replicas { + if rep.Model == m.Name && rep.State.Counts() && rep.RetiringAt.IsZero() { + b := catalog.ProfileBase(rep.Profile) + live[b] = append(live[b], rep.ID) + } + } + r.mu.Unlock() + deficit = map[string]bool{} + for _, p := range m.Profiles { + ids := live[p.Name] + delete(live, p.Name) + slices.Sort(ids) + switch { + case len(ids) < p.Replicas: + deficit[p.Name] = true + case len(ids) > p.Replicas: + surplus = append(surplus, ids[p.Replicas:]...) + } + } + for _, ids := range live { + surplus = append(surplus, ids...) + } + slices.Sort(surplus) + return deficit, surplus +} + +// pickSurplus is n of the live replicas to stop: those over their +// profile's count first, then any. +func pickSurplus(surplus, live []string, n int) []string { + out := make([]string, 0, n) + for _, id := range surplus { + if len(out) == n { + return out + } + if slices.Contains(live, id) { + out = append(out, id) + } + } + for _, id := range live { + if len(out) == n { + break + } + if !slices.Contains(out, id) { + out = append(out, id) + } + } + return out +} diff --git a/internal/control/reconciler/split_test.go b/internal/control/reconciler/split_test.go new file mode 100644 index 0000000..2785bd4 --- /dev/null +++ b/internal/control/reconciler/split_test.go @@ -0,0 +1,85 @@ +package reconciler + +import ( + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" +) + +// splitModel runs gpu replicas on cards and cpu on CPUs, under one name. +func splitModel(gpu, cpu int) catalog.Model { + return catalog.Model{Name: "m", Source: "/models/m.gguf", + Scaling: catalog.Scaling{Min: gpu + cpu, Max: gpu + cpu}, + Profiles: []catalog.Profile{ + {Name: "gpu", Engine: "llamacpp", GPUClass: "h100-80g", GPUsPerNode: 1, Priority: 100, Replicas: gpu}, + {Name: "cpu", Engine: "llamacpp", Priority: 10, Replicas: cpu}, + }} +} + +// byProfile counts the model's live replicas on each profile. +func byProfile(h *harness) map[string]int { + out := map[string]int{} + for _, rep := range h.r.Replicas() { + if rep.State.Counts() && rep.RetiringAt.IsZero() { + out[catalog.ProfileBase(rep.Profile)]++ + } + } + return out +} + +// Seven on cards and four on CPUs is what runs, not eleven on whatever fits +// best: with cards free for all eleven, four still go to the CPUs. +func TestReplicasRunAsManyOnEachProfileAsItAsks(t *testing.T) { + h := newHarness(t, splitModel(7, 4), gpuNode("1", 8), gpuNode("2", 8), cpuNode("3"), cpuNode("4")) + for range 14 { + h.r.Reconcile(t.Context()) + h.markReady() + h.advance(time.Second) + } + if got := byProfile(h); got["gpu"] != 7 || got["cpu"] != 4 { + t.Fatalf("running %v, want 7 on cards and 4 on CPUs", got) + } +} + +// Moving replicas from one profile to the other starts the new ones before +// stopping the old: the model never runs fewer than it asks. +func TestChangingTheSplitStartsBeforeItStops(t *testing.T) { + h := newHarness(t, splitModel(2, 1), gpuNode("1", 8), cpuNode("3"), cpuNode("4")) + for range 6 { + h.r.Reconcile(t.Context()) + h.markReady() + h.advance(time.Second) + } + if got := byProfile(h); got["gpu"] != 2 || got["cpu"] != 1 { + t.Fatalf("running %v before the change", got) + } + h.setModels(t, splitModel(1, 2)) + for range 10 { + h.r.Reconcile(t.Context()) + h.markReady() + h.advance(h.r.cfg.DrainGrace + time.Second) + if n := replicasOf(h, "m"); n < 3 { + t.Fatalf("only %d replicas while the split changed", n) + } + } + if got := byProfile(h); got["gpu"] != 1 || got["cpu"] != 2 { + t.Errorf("running %v, want 1 on cards and 2 on CPUs", got) + } +} + +// The scaling has to be the profiles' sum, so a count in one place cannot +// disagree with the other. +func TestReplicasPerProfileAndScalingAgree(t *testing.T) { + m := splitModel(2, 1) + m.Scaling = catalog.Scaling{Min: 1, Max: 5} + if _, err := catalog.New([]catalog.Model{m}); err == nil { + t.Error("profiles asking for 3 with scaling 1..5 were accepted") + } + stopped := splitModel(2, 1) + was := stopped.Scaling + stopped.Stopped, stopped.Scaling = &was, catalog.Scaling{} + if _, err := catalog.New([]catalog.Model{stopped}); err != nil { + t.Errorf("a stopped model with replicas per profile was refused: %v", err) + } +} diff --git a/internal/docgen/reference_gen.go b/internal/docgen/reference_gen.go index c109d05..c7a28b9 100644 --- a/internal/docgen/reference_gen.go +++ b/internal/docgen/reference_gen.go @@ -377,6 +377,7 @@ var Reference = []Block{ {Key: "keep_within", Type: "string", Doc: "KeepWithin keeps a multi-node replica's nodes inside one rack, row,\nroom or site, read from their llmgw.io/rack (and so on) labels: for a\nfabric whose leaves do not reach across racks, or cables that do not.\nNodes without the label are passed over for such a profile. Empty is\nno constraint.", Nested: ""}, {Key: "weight", Type: "float64", Doc: "Weight is the router's capacity weight. It should come from ctl bench\nrather than a guess (plan §6.2.1).", Nested: ""}, {Key: "priority", Type: "int", Doc: "Priority orders profiles during placement; highest that fits wins.", Nested: ""}, + {Key: "replicas", Type: "int", Doc: "Replicas, set on any of a model's profiles, is how many of its\nreplicas run on this one: seven on cards and four on CPUs, all\nanswering under the model's name, rather than each replica on the\nbest profile that fits. The model's scaling is then the sum, min and\nmax alike; a profile left at zero runs none.", Nested: ""}, {Key: "startup_timeout", Type: "time.Duration", Doc: "StartupTimeout bounds model loading. A 70B model across two nodes takes\ntens of minutes, so this is per profile rather than global.", Nested: ""}, {Key: "kv_events", Type: "bool", Doc: "KVEvents has vLLM publish what its prefix cache holds, and the router\nroute this model's requests by it: to the replica holding the most of\na prompt's leading blocks (plan §7.5 tier 2). vLLM only.\n\nOff by default, because it costs something on every request: the\nrouter asks a replica to tokenize the prompt before placing it, a\nround trip and the chat template run twice. It pays where prompts\nshare long prefixes that the approximate tree guesses wrong — many\nreplicas, conversations that move between them, evictions. The engine\nis started with sha256_cbor block hashing and a fixed PYTHONHASHSEED,\nwhich the router's hashing assumes; arguments overriding either leave\nthe index empty, and the agent says so in its log.", Nested: ""}, {Key: "gpu", Type: "*GPUNeed", Doc: "GPU says what card the profile needs, in place of naming its class:\nit stands for every class in the cluster that meets it. A catalog\nentry written once then places sensibly on any site's cards.", Nested: "catalog.GPUNeed"},