diff --git a/BACKLOG.md b/BACKLOG.md index 4641e9b..446ebec 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -1385,7 +1385,7 @@ gate, so no token is on the node. - Done when: SF3D runs on cuda01 with nothing installed into the engine's venv by hand (the hand install of 2026-10-10 removed first). -### H5. Licence on every output (S) +### H5. Licence on every output (S) — done 2026-10-14: the whole licence in GLB asset.extras and PNG iTXt; video and audio not yet - `License` gains `id` (SPDX where there is one, else a lowercase name: stability-community, tencent-hunyuan-community), `commercial` (unset is not said), `excluded_regions` (ISO 3166 codes, `EU` for the union), @@ -1398,7 +1398,7 @@ gate, so no token is on the node. audio formats in their metadata where ffmpeg is at hand on the router, otherwise beside it in the job's answer only — said in the docs. -### H6. Licence choices (S) — the other session's (llmgw-b7), at the user's request +### H6. Licence choices (S) — done 2026-10-11 (4bbac08), live on the PoC The user asked (2026-10-11) for the choice to be a person's: on their profile they pick the licences they will not use, and their keys and portal session neither see those models in `/v1/models` nor may call @@ -1407,13 +1407,29 @@ them (403, saying which licence). Matched on `license.id`, and on store table is migration 0046. A site-wide or team-wide default can sit on top of it later. -### H7. A job worker per model, not one for all (S) +### H7. A job worker per model, not one for all (S) — done 2026-10-14: one job per model at a time, eight in all The worker runs one job at a time: a job waiting for the card holds up a job for another model that could run now (seen on the PoC, 2026-10-10). Jobs are claimed and run concurrently up to a limit per model (or per environment, after H3) and a cluster-wide cap, with the lease and leader-only rules of G1 kept. +### H8. Presets that add themselves (S) — done 2026-10-13 (v0.14.5) +flux-schnell and triposr come with llmgw (`ctl model add `), and +the leader adds them on demand where a node fits one, the cluster installs +ComfyUI and the catalog makes nothing of that kind (`presets.auto`); a +preset removed is declined and not added back. The cluster page and the +console show and add them. + +### H9. Small things learned on the PoC (2026-10-14) +- Done: `albedo_match` on sf3d (its delighting made dark objects light); + `on_demand.evict_after_idle` / `min_lifetime` as live settings; an + upgrade that leaves nodes behind says which; `ctl model serve` serves an + embedding model for embedding; refreshes wait while a form is filled in. +- Open: llmgw's `/mcp` with picture, 3D, VLM and workflow tools, so a + client needs no server of its own (asked by the game team); the shared + ComfyUI replica (H3's placement side). + ## D. Facts worth knowing before starting - Admin API: `127.0.0.1:9090`, no auth, `ctlClient` in `internal/cli/ctl.go`. diff --git a/CHANGELOG.md b/CHANGELOG.md index 92db48d..6d1612d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,9 @@ APIs change between minors without a deprecation period. ### Fixed +- Jobs ran one at a time, so a 3D job waiting for the card held up another + model's job that could have run. They run side by side, one per model at + a time and eight in all, on the leader as before. - `llmgw ctl model serve` served an embedding model as a chat model: no `--embedding`, four slots, a 512-token batch, `embeddings` unset and a "say hello" smoke test it could not pass. A model whose GGUF says it diff --git a/docs/admin-api.openapi.json b/docs/admin-api.openapi.json index 46efe7e..948bdfc 100644 --- a/docs/admin-api.openapi.json +++ b/docs/admin-api.openapi.json @@ -10413,6 +10413,10 @@ "type": "boolean", "description": "Tools says the model makes tool calls the engine returns as calls: it is written into the catalog, where clients and editors read it. Check a running replica with llmgw ctl tool-check before saying so." }, + "embeddings": { + "type": "boolean", + "description": "Embeddings serves it as an embedding model, which its GGUF header says by itself for most: --embedding, many slots, a batch that holds a whole passage, /v1/embeddings." + }, "cpu_cores": { "type": "integer", "description": "CPUCores is what a replica computes on; zero takes half the smallest node it fits on." diff --git a/internal/router/jobs_test.go b/internal/router/jobs_test.go index a9faf49..5f64ac8 100644 --- a/internal/router/jobs_test.go +++ b/internal/router/jobs_test.go @@ -360,3 +360,32 @@ func TestAStandbyLeavesJobsToTheLeader(t *testing.T) { t.Errorf("the leader did not serve it: %v", got) } } + +func TestAJobWaitingForTheCardDoesNotHoldUpAnotherModelsJob(t *testing.T) { + // One at a time, an sf3d job waiting for the card held up a job for a + // model that could have run. + 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", eta: time.Hour}) + now := time.Now() + put := func(id, model, body string, at time.Time) { + if err := rt.store.PutJob(t.Context(), store.Job{ID: id, Owner: "project:" + mustProjectOf(t, rt, "key-1"), + KeyID: "key-1", Endpoint: imagesPath, Model: model, Input: []byte(body), Status: store.JobQueued, + CreatedAt: at, ExpiresAt: at.Add(jobWindow)}); err != nil { + t.Fatal(err) + } + } + put("job_wait", "sleepy", `{"model":"sleepy","prompt":"a cube"}`, now.Add(-time.Minute)) + put("job_go", "flux", `{"model":"flux","prompt":"a cube"}`, now) + rt.wakeJobs() + + got := waitJob(t, rt.Handler(), secret, "job_go", jobDone) + if got["status"] != "completed" { + t.Errorf("the other model's job: %v", got) + } + if w, _ := rt.store.JobByID(t.Context(), "job_wait"); w.Status != store.JobRunning { + t.Errorf("the waiting job is %s", w.Status) + } +} diff --git a/internal/router/jobworker.go b/internal/router/jobworker.go index 872ec6f..d6f9a04 100644 --- a/internal/router/jobworker.go +++ b/internal/router/jobworker.go @@ -10,6 +10,7 @@ import ( "os" "strconv" "strings" + "sync" "time" "gitea.mynodes.xyz/ilker/llmgw/internal/auth" @@ -27,6 +28,10 @@ const ( jobCancelCheck = 2 * time.Second // jobMaxResult bounds what one job may answer: a mesh, a short video. jobMaxResult = 256 << 20 + // jobsPerModel is how many of one model's jobs run at once: its engine + // works through them in turn anyway. jobsAtOnce bounds them all. + jobsPerModel = 1 + jobsAtOnce = 8 ) // jobUnplacedGrace is how long a job keeps asking for a model said to fit @@ -52,10 +57,28 @@ func (r *Router) runJobs(ctx context.Context) { owner := fmt.Sprintf("%s:%d:%s", host, os.Getpid(), newID("")) t := time.NewTicker(jobPoll) defer t.Stop() + // Jobs run side by side, one per model at a time and jobsAtOnce in all: + // one at a time, a job waiting for the card held up another model's job + // that could have run. + var mu sync.Mutex + running := map[string]int{} + total := 0 for { for r.leading() { + mu.Lock() + if total >= jobsAtOnce { + mu.Unlock() + break + } + var skip []string + for m, n := range running { + if n >= jobsPerModel { + skip = append(skip, m) + } + } + mu.Unlock() now := time.Now() - j, ok, err := r.store.ClaimJob(ctx, owner, now, now.Add(jobLease)) + j, ok, err := r.store.ClaimJob(ctx, owner, now, now.Add(jobLease), skip...) if err != nil { if ctx.Err() == nil { r.log.Warn("could not look for jobs to serve", "err", err) @@ -65,7 +88,20 @@ func (r *Router) runJobs(ctx context.Context) { if !ok { break } - r.workJob(ctx, owner, j) + mu.Lock() + running[j.Model]++ + total++ + mu.Unlock() + go func(j store.Job) { + r.workJob(ctx, owner, j) + mu.Lock() + if running[j.Model]--; running[j.Model] <= 0 { + delete(running, j.Model) + } + total-- + mu.Unlock() + r.wakeJobs() // room for the next + }(j) } // A job's record outlives its file by nothing: both go at the // retention. diff --git a/internal/store/jobs.go b/internal/store/jobs.go index 18c3cc1..1bd5636 100644 --- a/internal/store/jobs.go +++ b/internal/store/jobs.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "errors" + "strings" "time" ) @@ -193,13 +194,26 @@ func (s *DB) AllJobs(ctx context.Context, status string, limit int) ([]Job, erro // ClaimJob takes the lease on the oldest job with work left that no gateway // holds, for owner until the time given. ok is false when there is none. -func (s *DB) ClaimJob(ctx context.Context, owner string, now, until time.Time) (Job, bool, error) { +// +// Jobs for the models in skip are passed over: a worker running as many of a +// model's jobs as it will at once takes the next job for another model +// rather than waiting behind it. +func (s *DB) ClaimJob(ctx context.Context, owner string, now, until time.Time, skip ...string) (Job, bool, error) { var id string + args := []any{owner, millis(until), JobQueued, JobRunning, JobCancelling, millis(now)} + notIn := "" + if len(skip) > 0 { + notIn = " AND model NOT IN (?" + strings.Repeat(", ?", len(skip)-1) + ")" + for _, m := range skip { + args = append(args, m) + } + } + args = append(args, millis(now)) err := s.db.QueryRowContext(ctx, `UPDATE jobs SET lease_owner = ?, lease_until = ? - WHERE id = (SELECT id FROM jobs WHERE status IN (?, ?, ?) AND lease_until < ? + WHERE id = (SELECT id FROM jobs WHERE status IN (?, ?, ?) AND lease_until < ?`+notIn+` ORDER BY created_at LIMIT 1) AND lease_until < ? - RETURNING id`, owner, millis(until), JobQueued, JobRunning, JobCancelling, millis(now), millis(now)).Scan(&id) + RETURNING id`, args...).Scan(&id) if errors.Is(err, sql.ErrNoRows) { return Job{}, false, nil } diff --git a/internal/store/jobs_test.go b/internal/store/jobs_test.go index 9ffe531..b6b4d68 100644 --- a/internal/store/jobs_test.go +++ b/internal/store/jobs_test.go @@ -120,3 +120,24 @@ func TestAnOperatorSeesEveryOwnersJobs(t *testing.T) { t.Errorf("%d queued", len(queued)) } } + +func TestAClaimPassesOverTheModelsItIsToldTo(t *testing.T) { + db := openDB(t) + ctx := t.Context() + now := time.Now() + for i, m := range []string{"sf3d", "flux"} { + at := now.Add(time.Duration(i) * time.Second) + if err := db.PutJob(ctx, Job{ID: "job_" + m, Owner: "u", KeyID: "k", Endpoint: "/e", Model: m, Input: []byte("{}"), + Status: JobQueued, CreatedAt: at, ExpiresAt: at.Add(time.Hour)}); err != nil { + t.Fatal(err) + } + } + j, ok, err := db.ClaimJob(ctx, "w", now, now.Add(time.Minute), "sf3d") + if err != nil || !ok || j.Model != "flux" { + t.Errorf("claimed %+v %v %v; sf3d was to be passed over", j.ID, ok, err) + } + j, ok, _ = db.ClaimJob(ctx, "w", now, now.Add(time.Minute)) + if !ok || j.Model != "sf3d" { + t.Errorf("then %+v %v", j.ID, ok) + } +}