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) <noreply@anthropic.com>
This commit is contained in:
1 parent
9dc1b62fdc
commit
aa50b022b9
16 files changed
+503
-15
No files matched your search
@@ -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
|
||||
|
||||
+1
-1
@@ -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),
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 <model> --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
|
||||
}
|
||||
@@ -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())
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
@@ -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{
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
+10
-3
@@ -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 <model> --target-rps N`, which needs capacity profiles stored
|
||||
rather than printed
|
||||
- [x] Sizing from stored capacity: `ctl model target <model> --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
|
||||
|
||||
Reference in new issue
Block a user