From 78e41460f92120a99478d838d368c22a1e7b4a75 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Wed, 30 Sep 2026 18:21:08 +0200 Subject: [PATCH] Scale and place by what a replica was measured to serve Stored ctl bench results reach the reconciler: requests per replica past the measured SLO concurrency scale a model up, a replica is taken back only if the rest stay under 70% of theirs, and among profiles of equal priority the most measured tokens per GPU or core is tried first. ctl model target sizes a model for a throughput against the room there is. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 23 ++++ internal/cli/ctl.go | 2 +- internal/cli/model.go | 3 +- internal/cli/quota.go | 66 ++++++++++++ internal/cli/server.go | 3 + internal/control/adminapi/api.go | 5 +- internal/control/adminapi/target.go | 63 +++++++++++ internal/control/adminapi/target_test.go | 30 ++++++ internal/control/quotas.go | 43 ++++++++ internal/control/reconciler/demand.go | 40 ++++++- internal/control/reconciler/headroom.go | 12 ++- internal/control/reconciler/measured.go | 108 +++++++++++++++++++ internal/control/reconciler/measured_test.go | 96 +++++++++++++++++ internal/control/reconciler/preempt.go | 4 +- internal/control/reconciler/reconciler.go | 7 +- llmgw-plan.md | 13 ++- 16 files changed, 503 insertions(+), 15 deletions(-) create mode 100644 internal/control/adminapi/target.go create mode 100644 internal/control/adminapi/target_test.go create mode 100644 internal/control/reconciler/measured.go create mode 100644 internal/control/reconciler/measured_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index c176a51..7aa273d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -106,6 +106,29 @@ two is refused. Changes take effect within half a minute. Metrics: `llmgw_capacity_holds`, `llmgw_capacity_guarantee` and `llmgw_capacity_limit` by scope and resource; alert: CapacityAtLimit. +### Sized by what a replica was measured to serve + +What `llmgw ctl bench MODEL --save` measured — the most requests at once one +replica served while meeting its SLO, and its throughput there — now drives +decisions rather than only being printed: + +- **Scaling up** when requests per replica pass what was measured, not only + when the engines' queues fill: an engine takes requests into its slots long + after latency has gone. +- **Scaling down** only when the replicas left would stay under 70% of what + they were measured to hold, so a replica taken back does not put the rest + straight over. +- **Choosing a profile.** Among profiles of equal priority, the one measured + to serve the most per GPU — or per core — is tried first. Priority set by an + operator still wins. +- **Sizing.** `llmgw ctl model target MODEL --tokens-per-sec N` says how many + replicas that takes, how many run, and whether the rest fit, with the reason + when they do not; `--apply` sets it if they do. Headroom reports what one + more replica would serve. + +A model never measured behaves exactly as before. Measurements are read from +the store every half minute. + ### The mirror page in tabs /admin/mirror was one long page: what the mirror holds, the Hugging Face diff --git a/internal/cli/ctl.go b/internal/cli/ctl.go index abea7d7..f34e4c2 100644 --- a/internal/cli/ctl.go +++ b/internal/cli/ctl.go @@ -137,7 +137,7 @@ func newCtlCmd() *cobra.Command { // apply to it; it lives here because this is where operators look. newBenchCmd(client, &asJSON), newStatusCmd(client, &asJSON), - newModelCmd(client), + newModelCmd(client, &asJSON), newScaleCmd(client), newRestartCmd(client), newRetryCmd(client), diff --git a/internal/cli/model.go b/internal/cli/model.go index 75a39f6..3a7654b 100644 --- a/internal/cli/model.go +++ b/internal/cli/model.go @@ -24,13 +24,14 @@ import ( // minutes of outage. For a fixed production set that is a reasonable trade; for // a platform whose promise is that people can try things, it is the difference // between trying one and not bothering. -func newModelCmd(client func() *ctlClient) *cobra.Command { +func newModelCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { c := &cobra.Command{ Use: "model", Short: "Add or remove models while the cluster is running", } c.AddCommand(newModelAddCmd(client), newModelRemoveCmd(client), newModelPlanCmd(client), newResourcesCmd(client), newClassCmd(client), newOwnerCmd(client), + newTargetCmd(client, asJSON), newModelServeCmd(client)) return c } diff --git a/internal/cli/quota.go b/internal/cli/quota.go index 530ba0f..2d376dd 100644 --- a/internal/cli/quota.go +++ b/internal/cli/quota.go @@ -162,3 +162,69 @@ func newOwnerCmd(client func() *ctlClient) *cobra.Command { }, } } + +func newTargetCmd(client func() *ctlClient, asJSON *bool) *cobra.Command { + var tps float64 + var apply bool + c := &cobra.Command{ + Use: "target --tokens-per-sec N [--apply]", + Short: "How many replicas a model needs for a throughput, and whether they fit", + Long: "Sized from what one replica was measured to serve while meeting its SLO\n" + + "(`llmgw ctl bench MODEL --save`), on the profile its replicas run on.\n" + + "--apply makes that the number it always runs, if the cluster has room.", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + if tps <= 0 { + return fmt.Errorf("--tokens-per-sec is required") + } + var v adminapi.TargetView + path := "/api/v1/models/" + url.PathEscape(args[0]) + "/target?tokens_per_sec=" + + strconv.FormatFloat(tps, 'f', -1, 64) + if err := client().get(cmd.Context(), path, &v); err != nil { + return err + } + if *asJSON { + return json.NewEncoder(cmd.OutOrStdout()).Encode(v) + } + out := cmd.OutOrStdout() + fmt.Fprintf(out, "%s: %.0f tokens/s needs %d replicas at %.1f each (measured on %s); %d running\n", + v.Model, v.TokPerSec, v.Needed, v.PerReplica, v.Profile, v.Current) + switch { + case v.More == 0: + fmt.Fprintln(out, "It already has them.") + case v.Fits: + fmt.Fprintf(out, "%d more, and there is room for them.\n", v.More) + default: + fmt.Fprintf(out, "%d more, and there is room for %d:\n", v.More, v.Room) + for _, b := range v.Blocked { + fmt.Fprintln(out, " "+b) + } + } + if !apply || v.More == 0 { + return nil + } + if !v.Fits { + return fmt.Errorf("not applied: the cluster has no room for them") + } + var models []adminapi.ModelView + if err := client().get(cmd.Context(), "/api/v1/models", &models); err != nil { + return err + } + top := v.Needed + for _, m := range models { + if m.Name == v.Model { + top = max(top, m.Max) + } + } + body, _ := json.Marshal(adminapi.ScaleRequest{Min: v.Needed, Max: top}) + if err := client().post(cmd.Context(), "/api/v1/models/"+url.PathEscape(v.Model)+"/scale", body, nil); err != nil { + return err + } + fmt.Fprintf(out, "Applied: %s runs %d (up to %d under load).\n", v.Model, v.Needed, top) + return nil + }, + } + c.Flags().Float64Var(&tps, "tokens-per-sec", 0, "the throughput to size for") + c.Flags().BoolVar(&apply, "apply", false, "set it as the number of replicas the model always runs") + return c +} diff --git a/internal/cli/server.go b/internal/cli/server.go index 016d454..1eb1341 100644 --- a/internal/cli/server.go +++ b/internal/cli/server.go @@ -488,6 +488,9 @@ func startControlPlane(ctx context.Context, cfg config.Server, cat *catalog.Live // The tenancy tree's capacity quotas: what each part is // guaranteed and may hold. ctl.Reconciler().SetQuotas(control.QuotaPolicy(st, log)) + // And what each model was measured to serve, for scaling and for + // choosing between profiles. + ctl.Reconciler().SetMeasurements(control.Measurements(st, log)) log.Info("competing to be the control plane that places work", "owner", lead.Owner()) } diff --git a/internal/control/adminapi/api.go b/internal/control/adminapi/api.go index 53a5d6d..d2da729 100644 --- a/internal/control/adminapi/api.go +++ b/internal/control/adminapi/api.go @@ -60,6 +60,7 @@ func (a *API) Register(mux *http.ServeMux) { mux.HandleFunc("POST /api/v1/models/{name}/scale", a.scaleModel) mux.HandleFunc("POST /api/v1/models/{name}/resources", a.setResources) a.registerQuotas(mux) + mux.HandleFunc("GET /api/v1/models/{name}/target", a.target) mux.HandleFunc("POST /api/v1/models/{name}/class", func(w http.ResponseWriter, r *http.Request) { if a.models == nil { writeError(w, http.StatusNotImplemented, "this control plane has no store") @@ -399,6 +400,8 @@ type HeadroomView struct { Limited bool `json:"limited,omitempty"` // Blocked says why one more does not fit, a line per node considered. Blocked []string `json:"blocked,omitempty"` + // TokPerSec is what one more replica was measured to serve; 0 unmeasured. + TokPerSec float64 `json:"tok_per_sec,omitempty"` } // RejectionView is one rejected candidate. @@ -958,7 +961,7 @@ func (a *API) Headroom(model string, limit int) (HeadroomView, bool) { return HeadroomView{}, false } return HeadroomView{Model: model, Current: h.Current, More: h.More, Nodes: h.Nodes, - Limited: h.Limited, Blocked: h.Blocked}, true + Limited: h.Limited, Blocked: h.Blocked, TokPerSec: h.TokPerSec}, true } func (a *API) Explain(model string) ExplainView { diff --git a/internal/control/adminapi/target.go b/internal/control/adminapi/target.go new file mode 100644 index 0000000..cebd1d0 --- /dev/null +++ b/internal/control/adminapi/target.go @@ -0,0 +1,63 @@ +package adminapi + +import ( + "fmt" + "math" + "net/http" + "strconv" +) + +// TargetView is how many replicas a model needs to serve a throughput, from +// what one replica was measured to serve, and whether the cluster has room. +type TargetView struct { + Model string `json:"model"` + // TokPerSec is what was asked for; PerReplica what one replica was + // measured to serve at its SLO, on Profile. + TokPerSec float64 `json:"tok_per_sec"` + PerReplica float64 `json:"per_replica_tok_per_sec"` + Profile string `json:"profile"` + // Needed replicas in all; Current hold capacity now; More is how many of + // Needed are not running, and Room how many more fit. + Needed int `json:"needed"` + Current int `json:"current"` + More int `json:"more"` + Room int `json:"room"` + Fits bool `json:"fits"` + Blocked []string `json:"blocked,omitempty"` +} + +// Target sizes a model for a throughput. +func (a *API) Target(model string, tokPerSec float64) (TargetView, error) { + meas, profile, ok := a.ctl.Reconciler().MeasuredFor(model) + if !ok || meas.TokPerSec <= 0 { + return TargetView{}, fmt.Errorf("%s has not been measured; run `llmgw ctl bench %s --save` first", model, model) + } + v := TargetView{Model: model, TokPerSec: tokPerSec, PerReplica: meas.TokPerSec, Profile: profile, + Needed: max(int(math.Ceil(tokPerSec/meas.TokPerSec)), 1)} + h, ok := a.Headroom(model, v.Needed) + if !ok { + return TargetView{}, fmt.Errorf("no model %q", model) + } + v.Current = h.Current + v.More = max(v.Needed-h.Current, 0) + v.Room = h.More + v.Fits = v.More <= h.More || h.Limited + if !v.Fits { + v.Blocked = h.Blocked + } + return v, nil +} + +func (a *API) target(w http.ResponseWriter, r *http.Request) { + tps, err := strconv.ParseFloat(r.URL.Query().Get("tokens_per_sec"), 64) + if err != nil || tps <= 0 { + writeError(w, http.StatusBadRequest, "tokens_per_sec must be a positive number") + return + } + v, err := a.Target(r.PathValue("name"), tps) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + writeJSON(w, http.StatusOK, v) +} diff --git a/internal/control/adminapi/target_test.go b/internal/control/adminapi/target_test.go new file mode 100644 index 0000000..37c7af1 --- /dev/null +++ b/internal/control/adminapi/target_test.go @@ -0,0 +1,30 @@ +package adminapi + +import ( + "strings" + "testing" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" +) + +func TestATargetIsSizedFromTheMeasurementAndTheRoom(t *testing.T) { + // Three 8-GPU nodes; one tp8 replica was measured at 500 tokens/s. + a, ctl := setup(t, testModels(), gpuNode("1", 8), gpuNode("2", 8), gpuNode("3", 8)) + if _, err := a.Target("llama", 1000); err == nil || !strings.Contains(err.Error(), "ctl bench llama --save") { + t.Errorf("an unmeasured model should say how to measure it: %v", err) + } + ctl.Reconciler().SetMeasurements(func() map[string]reconciler.Measured { + return map[string]reconciler.Measured{"llama/tp8": {MaxConcurrency: 16, TokPerSec: 500}} + }) + v, err := a.Target("llama", 1200) + if err != nil { + t.Fatal(err) + } + if v.Needed != 3 || v.PerReplica != 500 || v.Profile != "tp8" || v.More != 3 || !v.Fits { + t.Errorf("1200 tok/s at 500 each is 3 replicas on 3 empty nodes: %+v", v) + } + v, _ = a.Target("llama", 2000) + if v.Needed != 4 || v.Fits || v.Room != 3 || len(v.Blocked) == 0 { + t.Errorf("a fourth does not fit on three nodes, and should say why: %+v", v) + } +} diff --git a/internal/control/quotas.go b/internal/control/quotas.go index 3ed0f44..1c0916b 100644 --- a/internal/control/quotas.go +++ b/internal/control/quotas.go @@ -2,11 +2,14 @@ package control import ( "context" + "encoding/json" "log/slog" "sync" "time" + "gitea.mynodes.xyz/ilker/llmgw/internal/bench" "gitea.mynodes.xyz/ilker/llmgw/internal/control/quota" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/reconciler" "gitea.mynodes.xyz/ilker/llmgw/internal/store" ) @@ -50,3 +53,43 @@ func QuotaPolicy(st QuotaStore, log *slog.Logger) func() *quota.Policy { return cur } } + +// MeasurementStore is where benchmark results are kept. +type MeasurementStore interface { + Capacities(ctx context.Context) ([]store.StoredCapacity, error) +} + +// Measurements returns what the reconciler asks for benchmark results, keyed +// model/profile, read again once older than quotaRefresh; like QuotaPolicy it +// keeps the last good answer when the store fails. +func Measurements(st MeasurementStore, log *slog.Logger) func() map[string]reconciler.Measured { + var mu sync.Mutex + var cur map[string]reconciler.Measured + var at time.Time + return func() map[string]reconciler.Measured { + mu.Lock() + defer mu.Unlock() + if cur != nil && time.Since(at) < quotaRefresh { + return cur + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + at = time.Now() + rows, err := st.Capacities(ctx) + if err != nil { + log.Warn("could not read benchmark results; keeping the last ones", "err", err) + return cur + } + next := map[string]reconciler.Measured{} + for _, row := range rows { + var c bench.Capacity + if json.Unmarshal(row.Measurement, &c) != nil { + continue + } + next[row.Model+"/"+row.Profile] = reconciler.Measured{ + MaxConcurrency: c.MaxConcurrencyAtSLO, TokPerSec: c.ThroughputTokPerSec} + } + cur = next + return cur + } +} diff --git a/internal/control/reconciler/demand.go b/internal/control/reconciler/demand.go index 5de5726..bbef8cd 100644 --- a/internal/control/reconciler/demand.go +++ b/internal/control/reconciler/demand.go @@ -157,8 +157,31 @@ type Load struct { // the last one is still starting: on a small CPU fleet each replica is minutes // of weights and gigabytes of memory, and a spike that is over by the time it // loads should not leave three behind. +// +// With a benchmark stored for the profile a model runs on, its measured +// capacity is used too: more requests per replica than were measured to meet +// the SLO is busy, even with the engines' queues empty — an engine will take +// requests into its slots long after latency has gone — and a replica is only +// taken back when the rest would still be under scaleDownHeadroom of what they +// were measured to hold, so scaling down does not land on the edge. func (r *Reconciler) RecordLoad(loads map[string]Load) { now := r.now() + // The profile each model's replicas run on, for its measurement; read + // before the loop, which holds the lock. + running := map[string]string{} + r.mu.Lock() + for _, rep := range r.replicas { + if rep.State.Counts() { + running[rep.Model] = rep.Profile + } + } + r.mu.Unlock() + measured := map[string]Measured{} + for model, profile := range running { + if m, ok := r.measured(model, profile); ok { + measured[model] = m + } + } for _, m := range r.cat.Load().Models() { base := r.baseFor(m) r.mu.Lock() @@ -178,8 +201,13 @@ func (r *Reconciler) RecordLoad(loads map[string]Load) { l := loads[m.Name] perReplica := float64(l.Waiting) / float64(max(l.Replicas, 1)) extra := r.extra[m.Name] + meas, hasMeas := measured[m.Name] + overMeasured := hasMeas && + float64(l.Running+l.Waiting) > float64(meas.MaxConcurrency*max(l.Replicas, 1)) + fitsInFewer := !hasMeas || + float64(l.Running) <= scaleDownHeadroom*float64(meas.MaxConcurrency*max(l.Replicas-1, 0)) switch { - case perReplica >= r.cfg.ScaleUpWaiting: + case perReplica >= r.cfg.ScaleUpWaiting || overMeasured: delete(r.quietSince, m.Name) since, ok := r.busySince[m.Name] if !ok { @@ -190,9 +218,10 @@ func (r *Reconciler) RecordLoad(loads map[string]Load) { r.extra[m.Name] = extra + 1 r.busySince[m.Name] = now r.log.Info("scaling up for load", "model", m.Name, "replicas", base+extra+1, - "waiting_per_replica", perReplica, "for", now.Sub(since).Round(time.Second)) + "waiting_per_replica", perReplica, "in_flight", l.Running+l.Waiting, + "measured_per_replica", meas.MaxConcurrency, "for", now.Sub(since).Round(time.Second)) } - case l.Waiting == 0 && extra > 0: + case l.Waiting == 0 && extra > 0 && fitsInFewer: delete(r.busySince, m.Name) since, ok := r.quietSince[m.Name] if !ok { @@ -212,3 +241,8 @@ func (r *Reconciler) RecordLoad(loads map[string]Load) { r.mu.Unlock() } } + +// scaleDownHeadroom is how full, of their measured capacity, the replicas left +// after a scale-down may be. Below one so that the load that was there a +// minute ago does not put them straight back over. +const scaleDownHeadroom = 0.7 diff --git a/internal/control/reconciler/headroom.go b/internal/control/reconciler/headroom.go index d078661..9d05df6 100644 --- a/internal/control/reconciler/headroom.go +++ b/internal/control/reconciler/headroom.go @@ -25,6 +25,9 @@ type Headroom struct { // Blocked is why one more than More does not fit, a line per node that // was considered. Empty when Limited. Blocked []string + // TokPerSec is what one more replica was measured to serve, on the + // profile placement would give it; zero when it was never measured. + TokPerSec float64 } // Headroom tries placing up to limit more replicas of a model on a copy of @@ -59,10 +62,10 @@ func (r *Reconciler) Headroom(model string, limit int) (Headroom, bool) { usage = pol.Usage(r.Holdings()) } for h.More < limit { - res := sched.Place(m.PlacementProfiles(), avoid) + res := sched.Place(r.orderedProfiles(m), avoid) if res.Placement == nil && len(avoid) > 0 { // As scaleUp does: spreading is a preference, not a rule. - res = sched.Place(m.PlacementProfiles(), nil) + res = sched.Place(r.orderedProfiles(m), nil) } if res.Placement == nil { for _, rej := range res.Rejections { @@ -87,6 +90,11 @@ func (r *Reconciler) Headroom(model string, limit int) (Headroom, bool) { if prof, ok := m.ProfileByName(res.Placement.Profile.Name); ok { mem, cores = prof.MinMemoryGB, cpuCores(prof) } + if h.More == 0 { + if meas, ok := r.measured(m.Name, res.Placement.Profile.Name); ok { + h.TokPerSec = meas.TokPerSec + } + } var nodes []string for _, a := range res.Placement.Nodes { inv.Allocate(a.NodeID, inventory.Allocation{ diff --git a/internal/control/reconciler/measured.go b/internal/control/reconciler/measured.go new file mode 100644 index 0000000..ac8a5ca --- /dev/null +++ b/internal/control/reconciler/measured.go @@ -0,0 +1,108 @@ +package reconciler + +import ( + "sort" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" + "gitea.mynodes.xyz/ilker/llmgw/internal/control/placement" +) + +// Measured is what one replica of a model on a profile was measured to serve +// while meeting its SLO, by `llmgw ctl bench --save`. +type Measured struct { + // MaxConcurrency is the most requests at once that still met the SLO. + MaxConcurrency int + // TokPerSec is the throughput at that concurrency. + TokPerSec float64 +} + +// SetMeasurements supplies benchmark results, keyed model/profile, read +// afresh each time it is called. Without them scaling goes by queue alone and +// profiles are chosen by priority alone. +func (r *Reconciler) SetMeasurements(f func() map[string]Measured) { + r.mu.Lock() + defer r.mu.Unlock() + r.measurements = f +} + +// measured is a model's measurement on one profile. +func (r *Reconciler) measured(model, profile string) (Measured, bool) { + r.mu.Lock() + f := r.measurements + r.mu.Unlock() + if f == nil { + return Measured{}, false + } + m, ok := f()[model+"/"+profile] + return m, ok && m.MaxConcurrency > 0 +} + +// orderedProfiles is a model's profiles as placement should try them: by +// priority, and among equal priorities the one measured to serve the most per +// GPU — or per core, for a profile with none — first. Unmeasured profiles keep +// their catalog order behind measured ones of the same priority. +// +// Placement sorts by priority itself, so the order found here is written back +// as priorities, descending, for it to keep. +func (r *Reconciler) orderedProfiles(m catalog.Model) []placement.Profile { + ps := m.PlacementProfiles() + eff := make([]float64, len(ps)) + for i, p := range ps { + meas, ok := r.measured(m.Name, p.Name) + if !ok { + continue + } + units := float64(max(p.GPUsPerNode, 0) * max(p.Nodes, 1)) + if units == 0 { + units = float64(max(p.CPUCores, 1)) + } + eff[i] = meas.TokPerSec / units + } + idx := make([]int, len(ps)) + for i := range idx { + idx[i] = i + } + sort.SliceStable(idx, func(a, b int) bool { + pa, pb := ps[idx[a]], ps[idx[b]] + if pa.Priority != pb.Priority { + return pa.Priority > pb.Priority + } + return eff[idx[a]] > eff[idx[b]] + }) + out := make([]placement.Profile, len(ps)) + for rank, i := range idx { + out[rank] = ps[i] + out[rank].Priority = len(ps) - rank + } + return out +} + +// MeasuredFor is the measurement that sizes a model: that of the profile its +// replicas run on, or, with none running, of the first profile placement +// would try that was measured. +func (r *Reconciler) MeasuredFor(model string) (Measured, string, bool) { + r.mu.Lock() + profile := "" + for _, rep := range r.replicas { + if rep.Model == model && rep.State.Counts() { + profile = rep.Profile + break + } + } + r.mu.Unlock() + if profile != "" { + if m, ok := r.measured(model, profile); ok { + return m, profile, true + } + } + cm, ok := r.cat.Load().Resolve(model) + if !ok { + return Measured{}, "", false + } + for _, p := range r.orderedProfiles(cm) { + if m, ok := r.measured(cm.Name, p.Name); ok { + return m, p.Name, true + } + } + return Measured{}, "", false +} diff --git a/internal/control/reconciler/measured_test.go b/internal/control/reconciler/measured_test.go new file mode 100644 index 0000000..2f0f468 --- /dev/null +++ b/internal/control/reconciler/measured_test.go @@ -0,0 +1,96 @@ +package reconciler + +import ( + "testing" + "time" + + "gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog" +) + +func measure(h *harness, byKey map[string]Measured) { + h.r.SetMeasurements(func() map[string]Measured { return byKey }) +} + +func TestMoreRequestsThanWereMeasuredToMeetTheSLOIsBusy(t *testing.T) { + m := testModel(1) + m.Scaling.Max = 2 + h := newHarness(t, m, gpuNode("1", 8), gpuNode("2", 8)) + h.r.Reconcile(t.Context()) + h.markReady() + // Six in flight, nothing waiting at the engine: by the queue it is fine. + load := map[string]Load{"llama": {Replicas: 1, Running: 6}} + hold := func() { + h.r.RecordLoad(load) + h.advance(61 * time.Second) + h.r.RecordLoad(load) + h.r.Reconcile(t.Context()) + } + hold() + if n := liveCount(h); n != 1 { + t.Fatalf("scaled up with no measurement and an empty queue: %d", n) + } + // Measured to meet its SLO at four at once: six is past it. + measure(h, map[string]Measured{"llama/tp4": {MaxConcurrency: 4, TokPerSec: 200}}) + hold() + if n := liveCount(h); n != 2 { + t.Errorf("%d replicas with 6 in flight against a measured 4", n) + } +} + +func TestAReplicaIsTakenBackOnlyIfTheRestWouldHaveRoomToSpare(t *testing.T) { + m := testModel(1) + m.Scaling.Max = 2 + h := newHarness(t, m, gpuNode("1", 8), gpuNode("2", 8)) + measure(h, map[string]Measured{"llama/tp4": {MaxConcurrency: 4, TokPerSec: 200}}) + h.r.Reconcile(t.Context()) + h.markReady() + h.r.mu.Lock() + h.r.extra["llama"] = 1 + h.r.mu.Unlock() + h.r.Reconcile(t.Context()) + h.markReady() + if liveCount(h) != 2 { + t.Fatal("setup: not at two replicas") + } + quietFor := func(l Load) { + for range 3 { + h.r.RecordLoad(map[string]Load{"llama": l}) + h.advance(DefaultConfig().ScaleDownAfter + time.Second) + h.r.RecordLoad(map[string]Load{"llama": l}) + h.r.Reconcile(t.Context()) + } + } + // Nothing waiting, but three running: one replica left would be at 3 of + // its 4, past 70%. + quietFor(Load{Replicas: 2, Running: 3}) + if n := liveCount(h); n != 2 { + t.Errorf("scaled down to the edge of what one replica holds: %d", n) + } + quietFor(Load{Replicas: 2, Running: 2}) + if n := liveCount(h); n != 1 { + t.Errorf("two running fits one replica's 70%% of 4; still %d", n) + } +} + +func TestTheProfileMeasuredToServeMostPerGPUIsTriedFirst(t *testing.T) { + m := catalog.Model{Name: "llama", Source: "/models/llama", Scaling: catalog.Scaling{Min: 1, Max: 1}, + Profiles: []catalog.Profile{ + {Name: "tp4", GPUClass: "h100-80g", TP: 4}, + {Name: "tp2", GPUClass: "h100-80g", TP: 2}, + {Name: "tp8", GPUClass: "h100-80g", TP: 8, Priority: 10}, + }} + h := newHarness(t, m, gpuNode("1", 8)) + // tp4: 400/4 = 100 per GPU; tp2: 300/2 = 150. tp8 is unmeasured but has + // the higher priority, which an operator set and a benchmark does not + // overrule. + measure(h, map[string]Measured{"llama/tp4": {MaxConcurrency: 8, TokPerSec: 400}, + "llama/tp2": {MaxConcurrency: 6, TokPerSec: 300}}) + cm, _ := h.r.cat.Load().Resolve("llama") + var order []string + for _, p := range h.r.orderedProfiles(cm) { + order = append(order, p.Name) + } + if len(order) != 3 || order[0] != "tp8" || order[1] != "tp2" || order[2] != "tp4" { + t.Errorf("order %v, expected tp8 (priority), then tp2 (150/GPU), then tp4 (100/GPU)", order) + } +} diff --git a/internal/control/reconciler/preempt.go b/internal/control/reconciler/preempt.go index 75cdec6..f8cea55 100644 --- a/internal/control/reconciler/preempt.go +++ b/internal/control/reconciler/preempt.go @@ -63,9 +63,9 @@ func (r *Reconciler) preemptFor(ctx context.Context, m catalog.Model, avoid []st } trial.ReleaseReplica(v.ID) chosen = append(chosen, v) - res := sched.Place(m.PlacementProfiles(), avoid) + res := sched.Place(r.orderedProfiles(m), avoid) if res.Placement == nil && len(avoid) > 0 { - res = sched.Place(m.PlacementProfiles(), nil) + res = sched.Place(r.orderedProfiles(m), nil) } if res.Placement == nil { continue diff --git a/internal/control/reconciler/reconciler.go b/internal/control/reconciler/reconciler.go index e87373d..5c30f9a 100644 --- a/internal/control/reconciler/reconciler.go +++ b/internal/control/reconciler/reconciler.go @@ -273,6 +273,8 @@ type Reconciler struct { preempted map[string]int // quotas is the tenancy tree's capacity quotas as last read; nil is none. quotas func() *quota.Policy + // measurements are benchmark results by model/profile; nil is none. + measurements func() map[string]Measured wake chan struct{} stop chan struct{} @@ -784,12 +786,13 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) { // allocated, and starting several at once from the same inventory snapshot // would hand them the same GPUs. func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []string) { - res := r.sched.Place(m.PlacementProfiles(), avoid) + profiles := r.orderedProfiles(m) + 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 // is the only thing standing in the way, take the placement anyway: // a second replica on one node still survives an engine crash. - res = r.sched.Place(m.PlacementProfiles(), nil) + res = r.sched.Place(profiles, nil) } r.mu.Lock() diff --git a/llmgw-plan.md b/llmgw-plan.md index c16f1c8..21f08cb 100644 --- a/llmgw-plan.md +++ b/llmgw-plan.md @@ -1463,8 +1463,8 @@ The launch path is now built too: - [x] `ctl bench`: a load ramp that reports capacity as the last concurrency level meeting the SLO, plus the queue depth and KV usage observed there, so `weight`, `target_waiting` and `kv_ceiling` stop being guesses -- [ ] `ctl plan --target-rps N`, which needs capacity profiles stored - rather than printed +- [x] Sizing from stored capacity: `ctl model target --tokens-per-sec N`, + and the autoscaler and profile choice use what `ctl bench --save` measured - [x] OTel tracing: an inference span with admission, schedule and each upstream attempt under it, over OTLP/HTTP when `tracing.endpoint` is set, and the traceparent carried on to the engine @@ -1480,7 +1480,14 @@ The launch path is now built too: - [ ] Admission displacement (dropping requests from lower bands) - [x] Autoscaling on load: a replica added while requests wait, up to a model's max, and taken back after ten quiet minutes -- [ ] Priority-based preemption +- [x] Priority-based preemption: capacity classes (critical, standard, + opportunistic), preemption tried on a copy of the inventory first +- [x] Capacity quotas per organization, team and project: guarantees kept, + borrowing taken back, limits enforced +- [x] CPU cores reserved like memory, and memory required of CPU models +- [ ] Rack power, temperature and cooling as placement signals (a facility + source; not yet built) +- [ ] Defragmentation for large tensor-parallel replicas across NVLink domains - [ ] Control plane HA (Raft) - [x] Weight downloading and cache management: the mirror, agent fetches checked against their digest, and reclaiming what nothing uses