The same inputs place a replica on the same node and card, so an on-demand model that failed to load there was started there again after every backoff until it was parked. The reconciler remembers where a model's replicas failed and places the next away from them when anything else fits (Scheduler.Away); with nowhere else, there again. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
6320dd5b13
commit
d864438272
5 files changed
+218
-1
No files matched your search
@@ -64,6 +64,15 @@ APIs change between minors without a deprecation period.
|
||||
four, as the room left holds, and its memory figure is for that pool. A
|
||||
plan says what the model was trained for (`trained`). Strata stays at
|
||||
its 32768; vLLM's form is unchanged.
|
||||
- **A model that failed to start was started in the same place again.**
|
||||
Placement given the same inputs chooses the same node and card, so an
|
||||
on-demand model whose replica failed to load was retried there, past each
|
||||
backoff, until it was parked, with another card or node free all along.
|
||||
The control plane now remembers where a model's replicas failed (node and
|
||||
cards, up to eight places) and places the next one elsewhere when anywhere
|
||||
else fits; with nowhere else it is tried there as before. Coming up, a
|
||||
change to its definition or `ctl retry` forgets the places. The log says
|
||||
`placing a model away from where it failed to start`, with both places.
|
||||
- **Deleted model files came back after a control plane restart.** An agent
|
||||
registers with the files its startup scan found, and sends the list again
|
||||
only when it changes; a file deleted after the agent started was listed
|
||||
|
||||
@@ -154,6 +154,9 @@ func (r Result) Explain() string {
|
||||
// Scheduler places replicas onto an inventory.
|
||||
type Scheduler struct {
|
||||
inv *inventory.Store
|
||||
// shun is where a replica of the model being placed failed to start;
|
||||
// set on a copy by Away, never on the scheduler the reconciler shares.
|
||||
shun []Spot
|
||||
// now is the clock power readings are judged fresh by; nil is the wall
|
||||
// clock.
|
||||
now func() time.Time
|
||||
@@ -162,6 +165,41 @@ type Scheduler struct {
|
||||
// New builds a scheduler over an inventory.
|
||||
func New(inv *inventory.Store) *Scheduler { return &Scheduler{inv: inv} }
|
||||
|
||||
// Spot is where a replica ran: a node, and the cards it held there (none
|
||||
// for one on the node's CPU).
|
||||
type Spot struct {
|
||||
Node string
|
||||
GPUs []int
|
||||
}
|
||||
|
||||
// Away is this scheduler placing nowhere in spots: not those cards, nor a
|
||||
// CPU replica on those nodes. A copy, so the one shared is not changed.
|
||||
//
|
||||
// For a model that failed to start somewhere: the same inputs pick the same
|
||||
// node and the same card every time, so without it every retry went back to
|
||||
// where the last one had failed until the model was parked, with another
|
||||
// card or node free all along.
|
||||
func (s *Scheduler) Away(spots []Spot) *Scheduler {
|
||||
c := *s
|
||||
c.shun = spots
|
||||
return &c
|
||||
}
|
||||
|
||||
// shunned is the cards of node n a model failed on, and whether it failed
|
||||
// on the node's CPU.
|
||||
func (s *Scheduler) shunned(node string) (gpus []int, cpu bool) {
|
||||
for _, sp := range s.shun {
|
||||
if sp.Node != node {
|
||||
continue
|
||||
}
|
||||
if len(sp.GPUs) == 0 {
|
||||
cpu = true
|
||||
}
|
||||
gpus = append(gpus, sp.GPUs...)
|
||||
}
|
||||
return gpus, cpu
|
||||
}
|
||||
|
||||
// Place picks the first profile, in priority order, that fits.
|
||||
//
|
||||
// avoid lists nodes already hosting this model, so replicas of one model spread
|
||||
@@ -686,7 +724,12 @@ func (s *Scheduler) eligibleNodes(p Profile, fabric string, avoid []string, reje
|
||||
}
|
||||
}
|
||||
|
||||
badGPUs, badCPU := s.shunned(n.ID)
|
||||
if p.CPUOnly() {
|
||||
if badCPU {
|
||||
reject("a replica of it failed to start here; trying elsewhere first")
|
||||
continue
|
||||
}
|
||||
// Every ready node can run it; there is no card to be free.
|
||||
c := candidate{node: n}
|
||||
if why := budget.check(n, s.shardWatts(p, c)); why != "" {
|
||||
@@ -703,8 +746,16 @@ func (s *Scheduler) eligibleNodes(p Profile, fabric string, avoid []string, reje
|
||||
if g.Class() != p.GPUClass || !slices.Contains(freeIdx, g.Index) {
|
||||
continue
|
||||
}
|
||||
if slices.Contains(badGPUs, g.Index) {
|
||||
continue
|
||||
}
|
||||
free = append(free, g)
|
||||
}
|
||||
if len(free) < p.GPUsPerNode && len(badGPUs) > 0 {
|
||||
reject(fmt.Sprintf("has %d free %s GPUs besides the ones a replica of it failed to start on, needs %d",
|
||||
len(free), p.GPUClass, p.GPUsPerNode))
|
||||
continue
|
||||
}
|
||||
if len(free) < p.GPUsPerNode {
|
||||
reject(fmt.Sprintf("has %d free %s GPUs, needs %d", len(free), p.GPUClass, p.GPUsPerNode))
|
||||
continue
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
package reconciler
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/agent/supervisor"
|
||||
"gitea.mynodes.xyz/ilker/llmgw/internal/control/catalog"
|
||||
)
|
||||
|
||||
// failAndRetry fails the last replica started, and returns where the next
|
||||
// one is started once the backoff is over.
|
||||
func failAndRetry(t *testing.T, h *harness) ShardCommand {
|
||||
t.Helper()
|
||||
started := h.disp.startedShards()
|
||||
if len(started) == 0 {
|
||||
t.Fatal("nothing was started")
|
||||
}
|
||||
s := started[len(started)-1]
|
||||
h.r.OnShardStatus(s.ReplicaID, s.ShardID, supervisor.Failed, "the engine exited")
|
||||
h.r.Reconcile(t.Context()) // reaps it
|
||||
h.disp.reset()
|
||||
h.advance(DefaultConfig().MaxBackoff)
|
||||
h.r.Reconcile(t.Context())
|
||||
again := h.disp.startedShards()
|
||||
if len(again) == 0 {
|
||||
t.Fatal("it was not tried again")
|
||||
}
|
||||
return again[0]
|
||||
}
|
||||
|
||||
func oneCard() catalog.Model {
|
||||
return testModel(1, catalog.Profile{Name: "one", GPUClass: "h100-80g", GPUsPerNode: 1})
|
||||
}
|
||||
|
||||
func TestAReplicaThatFailedIsTriedOnAnotherNode(t *testing.T) {
|
||||
// The same inputs give the same node every time: an on-demand 3D model
|
||||
// that failed to load was started on that node again, and again, until
|
||||
// it was parked, with another node free all along.
|
||||
h := newHarness(t, oneCard(), gpuNode("1", 1), gpuNode("2", 1))
|
||||
h.r.Reconcile(t.Context())
|
||||
first := h.disp.startedShards()[0].NodeID
|
||||
|
||||
if next := failAndRetry(t, h); next.NodeID == first {
|
||||
t.Errorf("tried on node %s again, where it had just failed", first)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAReplicaThatFailedIsTriedOnAnotherCardOfTheNode(t *testing.T) {
|
||||
h := newHarness(t, oneCard(), gpuNode("1", 2))
|
||||
h.r.Reconcile(t.Context())
|
||||
first := h.disp.startedShards()[0].Spec.GPUIndices
|
||||
|
||||
next := failAndRetry(t, h)
|
||||
if len(next.Spec.GPUIndices) != 1 || len(first) != 1 || next.Spec.GPUIndices[0] == first[0] {
|
||||
t.Errorf("tried on card %v again, where it had just failed", next.Spec.GPUIndices)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWithNowhereElseItIsTriedWhereItFailed(t *testing.T) {
|
||||
// A preference, not a rule: a model that has one place is not left
|
||||
// unplaced because it failed there once.
|
||||
h := newHarness(t, oneCard(), gpuNode("1", 1))
|
||||
h.r.Reconcile(t.Context())
|
||||
if next := failAndRetry(t, h); next.NodeID != "1" {
|
||||
t.Errorf("started on %q", next.NodeID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestComingUpForgetsWhereItFailed(t *testing.T) {
|
||||
h := newHarness(t, oneCard(), gpuNode("1", 1), gpuNode("2", 1))
|
||||
h.r.Reconcile(t.Context())
|
||||
next := failAndRetry(t, h)
|
||||
h.r.OnShardStatus(next.ReplicaID, next.ShardID, supervisor.Ready, "")
|
||||
h.r.mu.Lock()
|
||||
left := len(h.r.failedAt["llama"])
|
||||
h.r.mu.Unlock()
|
||||
if left != 0 {
|
||||
t.Errorf("a model that came up still remembers %d failures' places", left)
|
||||
}
|
||||
}
|
||||
@@ -61,6 +61,7 @@ func (r *Reconciler) Retry(model string) bool {
|
||||
delete(r.parked, model)
|
||||
r.attempts[model] = 0
|
||||
delete(r.nextAttempt, model)
|
||||
delete(r.failedAt, model)
|
||||
r.mu.Unlock()
|
||||
r.Notify()
|
||||
return was
|
||||
|
||||
@@ -7,6 +7,8 @@ import (
|
||||
"log/slog"
|
||||
"math/rand/v2"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
@@ -251,6 +253,9 @@ type Reconciler struct {
|
||||
nextAttempt map[string]time.Time
|
||||
// attempts counts consecutive failures per model, for the backoff.
|
||||
attempts map[string]int
|
||||
// failedAt is where a model's replicas failed to start since it last
|
||||
// came up, most recent last: the next one is placed elsewhere if it can.
|
||||
failedAt map[string][]placement.Spot
|
||||
// lastExplain keeps the most recent placement failure per model, so
|
||||
// `ctl explain-placement` can answer without re-running placement.
|
||||
lastExplain map[string]placement.Result
|
||||
@@ -334,6 +339,7 @@ func New(cfg Config, cat *catalog.Live, inv *inventory.Store, disp Dispatcher, l
|
||||
replicas: map[string]*Replica{},
|
||||
nextAttempt: map[string]time.Time{},
|
||||
attempts: map[string]int{},
|
||||
failedAt: map[string][]placement.Spot{},
|
||||
reservedPorts: map[int]struct{}{},
|
||||
wanted: map[string]time.Time{},
|
||||
coldStart: map[string]time.Duration{},
|
||||
@@ -726,6 +732,7 @@ func (r *Reconciler) reapFailed(ctx context.Context) {
|
||||
// replica is about to disappear, and the backoff has to survive it.
|
||||
r.attempts[rep.Model]++
|
||||
r.nextAttempt[rep.Model] = r.now().Add(r.backoff(r.attempts[rep.Model]))
|
||||
r.noteFailedAt(rep)
|
||||
if n := r.attempts[rep.Model]; n >= r.cfg.ParkAfter {
|
||||
if _, already := r.parked[rep.Model]; !already {
|
||||
m, _ := r.cat.Load().Resolve(rep.Model)
|
||||
@@ -875,6 +882,7 @@ func (r *Reconciler) reconcileModel(ctx context.Context, m catalog.Model) {
|
||||
delete(r.parked, m.Name)
|
||||
r.attempts[m.Name] = 0
|
||||
delete(r.nextAttempt, m.Name)
|
||||
delete(r.failedAt, m.Name)
|
||||
isParked, blockedUntil = false, time.Time{}
|
||||
r.log.Info("model unparked: its definition changed", "model", m.Name)
|
||||
}
|
||||
@@ -1002,7 +1010,29 @@ func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []strin
|
||||
return notTried
|
||||
}
|
||||
}
|
||||
res := r.sched.Place(profiles, avoid)
|
||||
// Away from where it last failed to start, when there is anywhere else:
|
||||
// the same inputs give the same node and card, and an on-demand model
|
||||
// that failed there was otherwise started there again until it was
|
||||
// parked. Only a preference too: nowhere else, and it is tried there.
|
||||
r.mu.Lock()
|
||||
failed := slices.Clone(r.failedAt[m.Name])
|
||||
r.mu.Unlock()
|
||||
var res placement.Result
|
||||
if len(failed) > 0 {
|
||||
away := r.sched.Away(failed)
|
||||
res = away.Place(profiles, avoid)
|
||||
if res.Placement == nil && len(avoid) > 0 {
|
||||
res = away.Place(profiles, nil)
|
||||
}
|
||||
pspan.SetAttributes(attribute.Bool("llmgw.away_from_failure", res.Placement != nil))
|
||||
if res.Placement != nil {
|
||||
r.log.Info("placing a model away from where it failed to start", "model", m.Name,
|
||||
"failed_on", spotsText(failed), "now_on", spotsText(spotsOf(res.Placement)))
|
||||
}
|
||||
}
|
||||
if res.Placement == nil {
|
||||
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:
|
||||
@@ -1054,6 +1084,51 @@ func (r *Reconciler) scaleUp(ctx context.Context, m catalog.Model, avoid []strin
|
||||
return placed
|
||||
}
|
||||
|
||||
// maxFailedSpots is how many places a model's failures are remembered at.
|
||||
const maxFailedSpots = 8
|
||||
|
||||
// noteFailedAt remembers where a failed replica ran. Called under r.mu.
|
||||
func (r *Reconciler) noteFailedAt(rep *Replica) {
|
||||
spots := r.failedAt[rep.Model]
|
||||
for _, sh := range rep.Shards {
|
||||
sp := placement.Spot{Node: sh.NodeID, GPUs: slices.Clone(sh.GPUIndices)}
|
||||
spots = slices.DeleteFunc(spots, func(o placement.Spot) bool {
|
||||
return o.Node == sp.Node && slices.Equal(o.GPUs, sp.GPUs)
|
||||
})
|
||||
spots = append(spots, sp)
|
||||
}
|
||||
if len(spots) > maxFailedSpots {
|
||||
spots = spots[len(spots)-maxFailedSpots:]
|
||||
}
|
||||
r.failedAt[rep.Model] = spots
|
||||
}
|
||||
|
||||
// spotsOf is where a placement puts its shards.
|
||||
func spotsOf(pl *placement.Placement) []placement.Spot {
|
||||
var out []placement.Spot
|
||||
for _, n := range pl.Nodes {
|
||||
out = append(out, placement.Spot{Node: n.NodeID, GPUs: n.GPUIndices})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// spotsText is spots as a log reads them: "node01 gpu 0,1; node02 cpu".
|
||||
func spotsText(spots []placement.Spot) string {
|
||||
parts := make([]string, 0, len(spots))
|
||||
for _, sp := range spots {
|
||||
if len(sp.GPUs) == 0 {
|
||||
parts = append(parts, sp.Node+" cpu")
|
||||
continue
|
||||
}
|
||||
gs := make([]string, len(sp.GPUs))
|
||||
for i, g := range sp.GPUs {
|
||||
gs[i] = strconv.Itoa(g)
|
||||
}
|
||||
parts = append(parts, sp.Node+" gpu "+strings.Join(gs, ","))
|
||||
}
|
||||
return strings.Join(parts, "; ")
|
||||
}
|
||||
|
||||
// scaleOutcome is what one scaleUp did.
|
||||
type scaleOutcome int
|
||||
|
||||
@@ -1433,6 +1508,7 @@ func (r *Reconciler) markShard(replicaID, shardID string, st supervisor.State, e
|
||||
// counting from scratch rather than from an old streak.
|
||||
r.attempts[rep.Model] = 0
|
||||
delete(r.nextAttempt, rep.Model)
|
||||
delete(r.failedAt, rep.Model)
|
||||
if !rep.CreatedAt.IsZero() {
|
||||
r.coldStart[rep.Model] = r.now().Sub(rep.CreatedAt)
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user