From b1490dbf3073c858ed4d7fba4c46fb2fc5acb913 Mon Sep 17 00:00:00 2001 From: Ilker Manap Date: Sat, 10 Oct 2026 22:25:27 +0200 Subject: [PATCH] A job waiting for the card says what holds it and about how long A 3D job waiting for the picture model to give up the one card had an empty progress; its client could not tell waiting from stuck. The control plane says which model holds the cards and when it lets go, plus the waiting model's last start; the 503 carries it as headers, and a job keeps it as progress {state: waiting_gpu, held_by, eta_seconds}. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 14 ++++ docs/install.md | 8 +- internal/control/reconciler/health.go | 79 +++++++++++++++++++ internal/control/reconciler/reconciler.go | 7 ++ .../control/reconciler/reconciler_test.go | 27 +++++++ internal/router/activate.go | 35 ++++++++ internal/router/handlers.go | 14 +++- internal/router/jobs_test.go | 16 ++++ internal/router/jobworker.go | 12 +++ internal/router/router.go | 3 +- internal/router/router_test.go | 33 ++++++++ internal/router/tags.go | 12 ++- 12 files changed, 252 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f822e9a..7ab84c0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,20 @@ APIs change between minors without a deprecation period. needed, generated from a route table a test holds to the router's own; and `/docs/api`, Swagger UI (served from the gateway) over it and, for staff, over the admin API's document. +- **A job waiting for the card says so**: `progress` is + `{"state":"waiting_gpu","held_by":…,"eta_seconds":…}` while another + on-demand model holds it; a direct request's 503 carries + `X-Llmgw-Waiting-For` and `X-Llmgw-Eta-Seconds`. +- **On-demand models in `/v1/models`**, with `loaded` saying whether one is + up now; a picture is cut out and put on grey before TripoSR sees it + (`LLMGWPrepareForMesh`, rembg when it has no alpha). + +### Fixed + +- A model waiting for another on-demand model to give up the card failed at + once as "cannot be placed"; it now waits, and a job with it. +- A model evicted for being idle was placed again on the next pass, ahead of + the one it made room for. ## v0.14.3 — 2026-10-10 diff --git a/docs/install.md b/docs/install.md index b2a8c82..6efe44e 100644 --- a/docs/install.md +++ b/docs/install.md @@ -2072,7 +2072,13 @@ curl -X POST http://gateway:8080/v1/3d/generations/job_…/cancel -H "Authorizat - **Status:** `queued`, `running`, `completed`, `failed` (with `error`), `cancelling`, `cancelled`, `expired` (not finished within 24 h). `progress` carries what the engine said of the work so far, once it says - anything; `GET /v1/3d/generations` lists the caller's jobs. + anything; `GET /v1/3d/generations` lists the caller's jobs. While the + model waits for another on-demand model to give up the card, + `progress` is `{"state":"waiting_gpu","held_by":"flux-schnell","eta_seconds":N}`: + N is when the holder is stopped (idle and old enough to evict, or at its + idle timeout) plus how long the model's last start took, and is left out + when that is not known. A request made directly gets the same as a 503 + with `X-Llmgw-Waiting-For`, `X-Llmgw-Eta-Seconds` and `Retry-After`. - **Refused at the door:** no key, a model that is not there or not in the key's allowlist, a request the model cannot serve. Refused when served, and said as the job's error: a spent budget, the guardrails, a model diff --git a/internal/control/reconciler/health.go b/internal/control/reconciler/health.go index 11b92f0..38ea71d 100644 --- a/internal/control/reconciler/health.go +++ b/internal/control/reconciler/health.go @@ -169,6 +169,85 @@ func (r *Reconciler) Unservable(model string) (string, bool) { return m.Name + " cannot be placed on any node: " + firstLine(res.Explain()), true } +// WaitFor says which model holds the cards an on-demand model is waiting +// for, and about how long until it serves: until the holder lets go — evicted +// once idle and old enough, or stopped at its own idle timeout — and then its +// own start, as long as its last one took. ok is false when it is not waiting +// on another model; eta is zero when when the holder lets go is not known. +func (r *Reconciler) WaitFor(model string) (heldBy string, eta time.Duration, ok bool) { + cat := r.cat.Load() + m, found := cat.Resolve(model) + if !found || !m.Scaling.OnDemand() { + return "", 0, false + } + type holder struct { + v victim + release time.Time + } + now := r.now() + var holders []holder + r.mu.Lock() + for _, rep := range r.replicas { + if !rep.State.Counts() || len(nodesOf(rep)) != 1 { + continue + } + if rep.Model == m.Name { + r.mu.Unlock() + return "", 0, false // its own replica is on its way + } + other, found := cat.Resolve(rep.Model) + if !found || !other.Scaling.OnDemand() { + continue + } + h := holder{v: victim{ID: rep.ID, Model: rep.Model}} + if !rep.LastUsedAt.IsZero() { + evict := rep.LastUsedAt.Add(r.cfg.EvictionGrace) + if !rep.ReadyAt.IsZero() && rep.ReadyAt.Add(r.cfg.MinLifetime).After(evict) { + evict = rep.ReadyAt.Add(r.cfg.MinLifetime) + } + h.release = evict + if idle := rep.LastUsedAt.Add(r.idleTimeout(other)); idle.Before(evict) { + h.release = idle + } + } + holders = append(holders, h) + } + cold := r.coldStart[m.Name] + r.mu.Unlock() + + // The one whose going alone makes room, letting go soonest; failing + // that, all of them, and the last to let go. + var best *holder + for i := range holders { + h := &holders[i] + if !r.fitsWithout(m, nil, []victim{h.v}) { + continue + } + if best == nil || (!h.release.IsZero() && (best.release.IsZero() || h.release.Before(best.release))) { + best = h + } + } + if best == nil { + all := make([]victim, len(holders)) + for i, h := range holders { + all[i] = h.v + } + if len(all) == 0 || !r.fitsWithout(m, nil, all) { + return "", 0, false + } + for i := range holders { + h := &holders[i] + if best == nil || h.release.IsZero() || (!best.release.IsZero() && h.release.After(best.release)) { + best = h + } + } + } + if best.release.IsZero() { + return best.v.Model, 0, true + } + return best.v.Model, max(best.release.Sub(now), 0) + cold, true +} + func firstLine(s string) string { s = strings.TrimSpace(s) if i := strings.IndexByte(s, '\n'); i >= 0 { diff --git a/internal/control/reconciler/reconciler.go b/internal/control/reconciler/reconciler.go index 13d9c25..eda1acd 100644 --- a/internal/control/reconciler/reconciler.go +++ b/internal/control/reconciler/reconciler.go @@ -264,6 +264,9 @@ type Reconciler struct { // because it is state about this moment and not about what the model is. // A model nobody has wanted has no entry and is simply not running. wanted map[string]time.Time + // coldStart is how long each model's last replica took from placed to + // ready, for telling a request waiting on it about how long it waits. + coldStart map[string]time.Duration // restartQueue holds, per model, the replicas an operator asked to be // restarted and not yet begun. One is replaced at a time: replacing them // all at once would need the memory of every replica twice over. @@ -330,6 +333,7 @@ func New(cfg Config, cat *catalog.Live, inv *inventory.Store, disp Dispatcher, l attempts: map[string]int{}, reservedPorts: map[int]struct{}{}, wanted: map[string]time.Time{}, + coldStart: map[string]time.Duration{}, restartQueue: map[string][]string{}, restartInPlace: map[string]bool{}, restartPace: map[string]RestartPace{}, @@ -1371,6 +1375,9 @@ 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) + if !rep.CreatedAt.IsZero() { + r.coldStart[rep.Model] = r.now().Sub(rep.CreatedAt) + } } } diff --git a/internal/control/reconciler/reconciler_test.go b/internal/control/reconciler/reconciler_test.go index 5a0b04a..488fbeb 100644 --- a/internal/control/reconciler/reconciler_test.go +++ b/internal/control/reconciler/reconciler_test.go @@ -1745,6 +1745,33 @@ func TestAnEvictedModelDoesNotTakeTheCardBack(t *testing.T) { } } +func TestAModelWaitingOnAnotherIsToldWhichAndForAboutHowLong(t *testing.T) { + h := newHarness(t, testModel(1), gpuNode("1", 4)) + h.r.cat.Store(twoOnDemandModels(t)) + h.r.Want("llama") + h.r.Reconcile(t.Context()) + h.markReady() + reps := h.r.Replicas() + if len(reps) != 1 { + t.Fatalf("setup: %d replicas", len(reps)) + } + h.r.RecordUsage(map[string]time.Time{reps[0].ID: h.clock.now()}) + h.advance(time.Minute) + + h.r.Want("second") + h.r.Reconcile(t.Context()) + // Evicted once it has been up MinLifetime, a minute of which has gone; + // second has never started, so there is no start of its to add. + by, eta, ok := h.r.WaitFor("second") + if !ok || by != "llama" || eta != DefaultConfig().MinLifetime-time.Minute { + t.Errorf("WaitFor = %q %s %v", by, eta, ok) + } + // The model holding the cards waits on nobody. + if _, _, ok := h.r.WaitFor("llama"); ok { + t.Error("a model with its own replica was said to be waiting") + } +} + func TestAModelWaitingForAnotherOnDemandOneToStepAsideIsNotUnservable(t *testing.T) { // A 3D model asked for straight after a picture, on a cluster with one // card: the picture model holds it, too freshly started to evict. That diff --git a/internal/router/activate.go b/internal/router/activate.go index a738e3c..88dd6b7 100644 --- a/internal/router/activate.go +++ b/internal/router/activate.go @@ -2,6 +2,7 @@ package router import ( "context" + "fmt" "time" "gitea.mynodes.xyz/ilker/llmgw/internal/notify" @@ -26,6 +27,40 @@ type unservable interface { Unservable(model string) (string, bool) } +// waiter is an activator that can say which model holds the cards an +// on-demand one waits for, and about how long the wait is. +type waiter interface { + WaitFor(model string) (heldBy string, eta time.Duration, ok bool) +} + +// waitingErr is errLoading for a model waiting on another to give up its +// cards: who holds them, and the wait when it can be told. +type waitingErr struct { + heldBy string + eta time.Duration +} + +func (e *waitingErr) Error() string { + if e.eta > 0 { + return fmt.Sprintf("%s: waiting for %s to give up the GPU; expected in about %d s", + errLoading, e.heldBy, int(e.eta.Seconds())) + } + return fmt.Sprintf("%s: waiting for %s to give up the GPU", errLoading, e.heldBy) +} + +func (e *waitingErr) Unwrap() error { return errLoading } + +// loadingErr is why a wanted model is not serving yet: waiting on another +// model's cards when the activator can say so, else simply loading. +func (r *Router) loadingErr(model string) error { + if w, ok := r.activator.(waiter); ok { + if by, eta, waiting := w.WaitFor(model); waiting { + return &waitingErr{heldBy: by, eta: eta} + } + } + return fmt.Errorf("%w: %s", errLoading, "the model is loading; retry shortly") +} + // whyUnservable is the activator's reason a model cannot be served, if any. func (r *Router) whyUnservable(model string) (string, bool) { u, ok := r.activator.(unservable) diff --git a/internal/router/handlers.go b/internal/router/handlers.go index df88f09..7e57abf 100644 --- a/internal/router/handlers.go +++ b/internal/router/handlers.go @@ -730,7 +730,19 @@ func (r *Router) finish(w http.ResponseWriter, req *parsedRequest, res dispatchR // is told so will make it. Without the header a well-behaved client // backs off exactly as it would from a permanent failure. status = http.StatusServiceUnavailable - w.Header().Set("Retry-After", "30") + retry := 30 + // Waiting on another model's cards: who holds them and for about + // how long, which a job keeps as its progress. + var we *waitingErr + if errors.As(res.err, &we) { + w.Header().Set(waitingForHeader, we.heldBy) + if we.eta > 0 { + secs := int(we.eta.Seconds()) + w.Header().Set(etaHeader, strconv.Itoa(secs)) + retry = min(max(secs, 5), retry) + } + } + w.Header().Set("Retry-After", strconv.Itoa(retry)) writeError(w, status, "server_error", res.err.Error()) case errors.Is(res.err, errNoCapacity): status = http.StatusServiceUnavailable diff --git a/internal/router/jobs_test.go b/internal/router/jobs_test.go index 24572fc..d79bd61 100644 --- a/internal/router/jobs_test.go +++ b/internal/router/jobs_test.go @@ -283,3 +283,19 @@ func TestJobsAreTheirOwners(t *testing.T) { t.Errorf("another project read a job's result: %d", code) } } + +func TestAJobWaitingOnAnotherModelsCardsSaysSo(t *testing.T) { + c := newJobComfy(t) + rt, secret := jobRouter(t, c) + rt.cfg.ActivationWait = 10 * time.Millisecond + rt.SetActivator(&waitingActivator{fakeActivator: fakeActivator{onDemand: map[string]bool{"sleepy": true}}, + heldBy: "flux-schnell", eta: 90 * time.Second}) + j := queueJob(t, rt, "job_w", store.JobQueued, `{"model":"sleepy","prompt":"a cube"}`) + rt.wakeJobs() + + got := waitJob(t, rt.Handler(), secret, j.ID, func(j map[string]any) bool { return j["progress"] != nil }) + p, _ := got["progress"].(map[string]any) + if got["status"] != "running" || p["state"] != "waiting_gpu" || p["held_by"] != "flux-schnell" || p["eta_seconds"] != 90.0 { + t.Errorf("job %v", got) + } +} diff --git a/internal/router/jobworker.go b/internal/router/jobworker.go index 898734e..102ed32 100644 --- a/internal/router/jobworker.go +++ b/internal/router/jobworker.go @@ -3,6 +3,7 @@ package router import ( "bytes" "context" + "encoding/json" "errors" "fmt" "net/http" @@ -180,6 +181,17 @@ func (r *Router) serveJob(ctx context.Context, j *store.Job) error { // because waiting does not help. if wait := w.header.Get("Retry-After"); wait != "" && (status == http.StatusTooManyRequests || status >= 500) && time.Now().Before(j.ExpiresAt) { + if by := w.header.Get(waitingForHeader); by != "" { + // Kept as its progress: whoever asks after it is told what + // it waits on, rather than nothing at all. + p := map[string]any{"state": "waiting_gpu", "held_by": by} + if eta, err := strconv.Atoi(w.header.Get(etaHeader)); err == nil && eta > 0 { + p["eta_seconds"] = eta + } + if b, err := json.Marshal(p); err == nil { + _ = r.store.SetJobProgress(ctx, j.ID, string(b), "") + } + } secs, _ := strconv.Atoi(wait) select { case <-rctx.Done(): diff --git a/internal/router/router.go b/internal/router/router.go index 34e1095..17060df 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -1023,8 +1023,7 @@ func (r *Router) dispatchOne(ctx context.Context, w http.ResponseWriter, req *pa // Asked for, still loading. Saying "no endpoint serves this" would // be a lie, and a client told to retry will succeed where one told // the model does not exist gives up. - return dispatchResult{err: fmt.Errorf("%w: %s", errLoading, - "the model is loading; retry shortly")} + return dispatchResult{err: r.loadingErr(model)} } return dispatchResult{err: fmt.Errorf("%w: %s", errNoCapacity, noEndpointFor(model, slices.DeleteFunc(r.servedModels(), r.hiddenModel)))} } diff --git a/internal/router/router_test.go b/internal/router/router_test.go index 6f9089c..f953aea 100644 --- a/internal/router/router_test.go +++ b/internal/router/router_test.go @@ -808,3 +808,36 @@ func TestABatchOfEmbeddingsIsCheckedTextByText(t *testing.T) { t.Errorf("a passage over the context was let through: %d", rec.Code) } } + +type waitingActivator struct { + fakeActivator + heldBy string + eta time.Duration +} + +func (a *waitingActivator) WaitFor(string) (string, time.Duration, bool) { + return a.heldBy, a.eta, true +} + +func TestAModelWaitingOnAnothersCardsSaysWhoAndForHowLong(t *testing.T) { + // A 3D job waiting for the picture model to give up the one card was + // told nothing at all; its client could not tell waiting from stuck. + e := newFakeEngine(t, "m") + cfg := baseConfig(config.Endpoint{ID: "a", URL: e.srv.URL, Models: []string{"m"}}) + cfg.Router.ActivationWait = 50 * time.Millisecond + rt := newTestRouter(t, cfg) + rt.SetActivator(&waitingActivator{fakeActivator: fakeActivator{onDemand: map[string]bool{"sleepy": true}}, + heldBy: "flux-schnell", eta: 12 * time.Second}) + + rec := post(t, rt.Handler(), map[string]any{"model": "sleepy", "messages": []any{}}, nil) + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("status %d, expected 503", rec.Code) + } + if rec.Header().Get(waitingForHeader) != "flux-schnell" || rec.Header().Get(etaHeader) != "12" || + rec.Header().Get("Retry-After") != "12" { + t.Errorf("headers %v", rec.Header()) + } + if !strings.Contains(rec.Body.String(), "waiting for flux-schnell") { + t.Errorf("body %s", rec.Body.String()) + } +} diff --git a/internal/router/tags.go b/internal/router/tags.go index 481698e..15c48a3 100644 --- a/internal/router/tags.go +++ b/internal/router/tags.go @@ -11,10 +11,14 @@ import ( // project: OpenAI's metadata (at most 16 keys, as OpenAI allows) and the // x-llmgw-tags header, which wins for a key both give. const ( - tagsHeader = "x-llmgw-tags" - maxTags = 16 - maxTagValue = 512 - maxEndUser = 256 + tagsHeader = "x-llmgw-tags" + // waitingForHeader and etaHeader say, on a 503, which model holds the + // cards the one asked for waits on, and about how many seconds it waits. + waitingForHeader = "X-Llmgw-Waiting-For" + etaHeader = "X-Llmgw-Eta-Seconds" + maxTags = 16 + maxTagValue = 512 + maxEndUser = 256 ) var tagKey = regexp.MustCompile(`^[A-Za-z0-9_.-]{1,64}$`)