From 7c95bd1c26f8ca210da9baaa189533ab641b1709 Mon Sep 17 00:00:00 2001 From: Hoda Mohammadi Date: Wed, 21 Oct 2026 23:24:04 -0700 Subject: [PATCH 1/3] perf(scheduler): cap the seats a shared claim may hold Host memory bounds how many workers park on a node but says nothing about time: a time-sliced claim with six seats runs each worker's step six times slower while other GPUs sit idle. --max-seats-per-claim (OPEN_RL_MAX_SEATS_PER_CLAIM, default 0 for no limit) is a hard policy limit. A ledger at the limit is not a selection candidate, and the ledger CAS loop refuses a join past it so two joiners on one snapshot cannot both take the last seat. A worker that finds every eligible ledger full cuts its own claim; with no free device it stays Pending, the reason names the limit, and it waits until a node arrives or a seat frees. --- scheduler/controller/cmd/manager/main.go | 6 +- .../controller/internal/controller/ledger.go | 5 ++ .../internal/controller/ledger_test.go | 24 +++++++ .../controller/workload_controller.go | 11 +++- .../controller/workload_controller_test.go | 63 +++++++++++++++++++ .../internal/placement/placement.go | 17 +++++ .../internal/placement/placement_test.go | 29 +++++++++ scheduler/docs/design.md | 39 +++++++++--- 8 files changed, 182 insertions(+), 12 deletions(-) diff --git a/scheduler/controller/cmd/manager/main.go b/scheduler/controller/cmd/manager/main.go index 6c734ab74..43d51dce1 100644 --- a/scheduler/controller/cmd/manager/main.go +++ b/scheduler/controller/cmd/manager/main.go @@ -47,6 +47,7 @@ func main() { retryInterval time.Duration placementTimeout time.Duration strategy string + maxSeats int maxConcurrent int ) @@ -62,6 +63,8 @@ func main() { "How long a worker may go unplaced before the request is declared unsatisfiable. 0 waits forever.") flag.StringVar(&strategy, "placement-strategy", env("OPEN_RL_PLACEMENT_STRATEGY", string(placement.StrategyBinPack)), "binpack shares eligible FFT claims first; spread requests a GPU first, then falls back to FFT sharing. LoRA never shares.") + flag.IntVar(&maxSeats, "max-seats-per-claim", envInt("OPEN_RL_MAX_SEATS_PER_CLAIM", 0), + "Most workers that may share one claim; 0 is no limit. A worker that finds every eligible claim full cuts its own and waits Pending until a GPU is added or a seat frees.") flag.IntVar(&maxConcurrent, "max-concurrent-reconciles", envInt("OPEN_RL_MAX_CONCURRENT_RECONCILES", 4), "How many workers place at once. Seat booking is CAS-arbitrated, so concurrency risks only transient over-cut claims, which the sharing fallback drains.") @@ -116,6 +119,7 @@ func main() { RetryInterval: retryInterval, PlacementTimeout: placementTimeout, PlacementStrategy: parsedStrategy, + MaxSeatsPerClaim: maxSeats, MaxConcurrentReconciles: maxConcurrent, } if err := reconciler.SetupWithManager(mgr); err != nil { @@ -133,7 +137,7 @@ func main() { os.Exit(1) } - setupLog.Info("placing workers", "namespace", namespace, "deviceClass", deviceClass, "deviceDriver", deviceDriver, "strategy", strategy) + setupLog.Info("placing workers", "namespace", namespace, "deviceClass", deviceClass, "deviceDriver", deviceDriver, "strategy", strategy, "maxSeatsPerClaim", maxSeats) if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil { setupLog.Error(err, "manager exited") os.Exit(1) diff --git a/scheduler/controller/internal/controller/ledger.go b/scheduler/controller/internal/controller/ledger.go index 03e3dece9..55d47a6d5 100644 --- a/scheduler/controller/internal/controller/ledger.go +++ b/scheduler/controller/internal/controller/ledger.go @@ -97,6 +97,11 @@ func (r *WorkloadReconciler) ensureSeat(ctx context.Context, claimName string, s return nil, nil, errBookingContended } } + // The seat limit is re-checked here for the same reason: two joiners + // selecting from one snapshot must not both land the last seat. + if existing == nil && !createMissing && r.MaxSeatsPerClaim > 0 && len(claimLedger.Spec.Seats) >= r.MaxSeatsPerClaim { + return nil, nil, errBookingContended + } if existing != nil { *existing = seat } else { diff --git a/scheduler/controller/internal/controller/ledger_test.go b/scheduler/controller/internal/controller/ledger_test.go index 60341acf4..28a4f9a46 100644 --- a/scheduler/controller/internal/controller/ledger_test.go +++ b/scheduler/controller/internal/controller/ledger_test.go @@ -123,6 +123,30 @@ func TestRebookingAdoptsTheRecordedSeat(t *testing.T) { } } +// The seat limit is enforced inside the ledger's CAS loop, so a joiner +// selecting from a snapshot that predates the last booking is refused. +func TestBookingRefusesASeatPastTheLimit(t *testing.T) { + r := newReconciler(t, append(enabledNode(), trainerWorker("w-a", "model-a"), trainerWorker("w-b", "model-b"), trainerWorker("w-c", "model-c"))...) + r.PlacementStrategy = placement.StrategyBinPack + r.MaxSeatsPerClaim = 2 + settle(t, r, "w-a") + claim := claimOf(t, r, "w-a") + allocateClaim(t, r, claim) + settle(t, r, "w-b") + if got := claimOf(t, r, "w-b"); got != claim { + t.Fatalf("w-b on %q, want %q", got, claim) + } + c := getWorker(t, r, "w-c") + if _, _, err := r.ensureSeat(context.Background(), claim, newSeat(c, requestFrom(c)), false); err != errBookingContended { + t.Fatalf("booking a third seat under a limit of two returned %v, want errBookingContended", err) + } + // The founder's own re-book is not a join and is never refused. + a := getWorker(t, r, "w-a") + if _, _, err := r.ensureSeat(context.Background(), claim, newSeat(a, requestFrom(a)), true); err != nil { + t.Fatalf("re-booking the founder: %v", err) + } +} + func TestSharingRechecksTheLedgerAfterSelection(t *testing.T) { resident, incoming := trainerWorker("resident", "model-a"), trainerWorker("incoming", "model-b") r := newReconciler(t, append(enabledNode(), resident, incoming)...) diff --git a/scheduler/controller/internal/controller/workload_controller.go b/scheduler/controller/internal/controller/workload_controller.go index d8dfe8856..aa46f1f67 100644 --- a/scheduler/controller/internal/controller/workload_controller.go +++ b/scheduler/controller/internal/controller/workload_controller.go @@ -54,6 +54,13 @@ type WorkloadReconciler struct { // PlacementStrategy orders the two placement moves; the zero value reads // as binpack. PlacementStrategy placement.Strategy + // MaxSeatsPerClaim caps how many workers share one claim; zero means no + // limit. A worker that finds every eligible ledger at the limit cuts its + // own claim and, if no device is free, stays Pending until one is added or + // a seat frees. It bounds a time-sliced step at the limit times the + // single-worker time, and it turns a full fleet into standing claims an + // autoscaler can see. + MaxSeatsPerClaim int // MaxConcurrentReconciles is how many workers place at once. MaxConcurrentReconciles int @@ -109,7 +116,9 @@ func (r *WorkloadReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c return ctrl.Result{}, fmt.Errorf("cannot read the fleet, placing nothing this pass: %w", err) } - return r.place(ctx, &worker, requestFrom(&worker), fleet) + request := requestFrom(&worker) + request.MaxSeats = r.MaxSeatsPerClaim + return r.place(ctx, &worker, request, fleet) } type placementScope struct { diff --git a/scheduler/controller/internal/controller/workload_controller_test.go b/scheduler/controller/internal/controller/workload_controller_test.go index a9658d6db..bd9c4c9a0 100644 --- a/scheduler/controller/internal/controller/workload_controller_test.go +++ b/scheduler/controller/internal/controller/workload_controller_test.go @@ -687,6 +687,69 @@ func TestReconcileBinPacksOntoExistingClaims(t *testing.T) { } } +// With a seat limit, binpack fills a claim to the limit, moves to a free +// device, and once every device is at the limit the next worker waits on a +// standing claim of its own until a seat frees. Two devices, limit two: a and +// b share the first, c and d the second, e stands pending; when b leaves, e +// takes its seat. +func TestReconcileStopsPackingAtTheSeatLimit(t *testing.T) { + r := newReconciler(t, append(enabledNode(), + trainerWorker("w-a", "model-a"), trainerWorker("w-b", "model-b"), trainerWorker("w-c", "model-c"), + trainerWorker("w-d", "model-d"), trainerWorker("w-e", "model-e"))...) + r.PlacementStrategy = placement.StrategyBinPack + r.MaxSeatsPerClaim = 2 + + settle(t, r, "w-a") + allocateClaim(t, r, claimOf(t, r, "w-a")) + settle(t, r, "w-b") + first := claimOf(t, r, "w-a") + if got := claimOf(t, r, "w-b"); got != first { + t.Fatalf("w-b landed on %q, want w-a's claim %q (under the limit)", got, first) + } + + settle(t, r, "w-c") + second := claimOf(t, r, "w-c") + if second == first { + t.Fatal("w-c joined a claim at the limit, want a claim of its own") + } + allocateClaim(t, r, second) + settle(t, r, "w-d") + if got := claimOf(t, r, "w-d"); got != second { + t.Fatalf("w-d landed on %q, want w-c's claim %q (under the limit)", got, second) + } + + // Both devices are claimed and both ledgers are full: w-e cuts a claim + // that cannot allocate, kube-scheduler declines its pod, and the sharing + // fallback finds no ledger under the limit. It stays Pending on its own + // claim with the limit in the reason. + settle(t, r, "w-e") + own := claimOf(t, r, "w-e") + if own == first || own == second { + t.Fatalf("w-e landed on %q, want a standing claim of its own", own) + } + fallBackToSharing(t, r, "w-e") + if got := claimOf(t, r, "w-e"); got != own { + t.Fatalf("w-e moved to %q past the limit, want it waiting on %q", got, own) + } + w := getWorker(t, r, "w-e") + if w.Status.Phase != openrlv1alpha1.PhasePending || !strings.Contains(w.Status.Reason, "at most 2 workers") { + t.Fatalf("w-e is %s %q, want Pending with the limit named", w.Status.Phase, w.Status.Reason) + } + + // A seat frees: w-b leaves the first claim and the next pass seats w-e. + if err := r.releaseSeatAndReclaim(context.Background(), first, "w-b"); err != nil { + t.Fatal(err) + } + fallBackToSharing(t, r, "w-e") + if got := claimOf(t, r, "w-e"); got != first { + t.Fatalf("w-e is on %q after a seat freed, want %q", got, first) + } + ledger := getLedger(t, r, ledgerNameFor(first)) + if len(ledger.Spec.Seats) != 2 { + t.Fatalf("first ledger holds %d seats, want 2: %+v", len(ledger.Spec.Seats), ledger.Spec.Seats) + } +} + // A trainer and a sampler on one accelerator, once the devices are contended. // Role selects which node pools may host a worker and nothing else -- in // particular it does not partition claims -- so on a node labelled for both, diff --git a/scheduler/controller/internal/placement/placement.go b/scheduler/controller/internal/placement/placement.go index 4cd5479e2..4627c9c39 100644 --- a/scheduler/controller/internal/placement/placement.go +++ b/scheduler/controller/internal/placement/placement.go @@ -200,6 +200,17 @@ type Request struct { // HostRequestBytes is the pod template's memory request: what this // worker's pod asks the node for, resident or parked. HostRequestBytes int64 + // MaxSeats is the most workers a shared claim may hold once this one + // joins; zero means no limit. A ledger at the limit is not a candidate: + // the worker cuts its own claim, and if no device is free it waits. Set + // from the controller's flag, not from the spec: it is fleet policy. + MaxSeats int +} + +// AtSeatLimit reports whether a claim already holds as many workers as the +// request allows a shared claim to hold. +func (r Request) AtSeatLimit(claim *Claim) bool { + return r.MaxSeats > 0 && claim.Workers() >= r.MaxSeats } // OwnerKey is the fairness unit the timeslicer serves this worker under; a @@ -317,6 +328,9 @@ func SelectClaim(req Request, fleet *Fleet) *Claim { if req.Memory > node.DeviceMemoryBytes { continue } + if req.AtSeatLimit(claim) { + continue + } // Node-wide: every pod on the node draws from one allocatable pool. // A node reporting no memory skips the check rather than refusing all. if node.HostMemoryBytes > 0 && hostByNode[node.Name]+req.HostRequestBytes > node.HostMemoryBytes { @@ -359,6 +373,9 @@ func Explain(req Request, fleet *Fleet, detail string) string { case len(candidateNodes(req, fleet)) > 0: // The hardware exists; it is busy. Retrying is the right move. reason = "WaitingForCapacity: a pool fits this workload but none has a free seat or a free accelerator" + if req.MaxSeats > 0 { + reason += fmt.Sprintf("; shared claims hold at most %d workers", req.MaxSeats) + } default: biggest := pools[0] for _, node := range pools[1:] { diff --git a/scheduler/controller/internal/placement/placement_test.go b/scheduler/controller/internal/placement/placement_test.go index f1c3c5152..e91850587 100644 --- a/scheduler/controller/internal/placement/placement_test.go +++ b/scheduler/controller/internal/placement/placement_test.go @@ -181,6 +181,35 @@ func TestSelectClaimPrefersFewestWorkersThenName(t *testing.T) { } } +func TestSelectClaimSkipsLedgersAtTheSeatLimit(t *testing.T) { + // Two shared ledgers on one node holding two workers and one. With a + // limit of two only the lighter one is a candidate, and once it fills + // neither is. + fleet := NewFleet() + fleet.Nodes["n"] = &Node{Name: "n", DeviceCount: 2, DeviceMemoryBytes: 96 << 30, Roles: map[string]bool{"trainer": true}} + full := &Claim{Name: "a", Node: "n", DeviceCount: 1} + full.Book("w1", "o1", 0, true) + full.Book("w2", "o2", 0, true) + light := &Claim{Name: "b", Node: "n", DeviceCount: 1} + light.Book("w3", "o3", 0, true) + fleet.Claims["a"], fleet.Claims["b"] = full, light + + req := Request{Shareable: true, Role: "trainer", Memory: 24 << 30, WorkerID: "w4", MaxSeats: 2} + if got := SelectClaim(req, fleet); got == nil || got.Name != "b" { + t.Fatalf("selected %v, want the ledger under the limit", got) + } + light.Book("w4", "o4", 0, true) + if got := SelectClaim(req, fleet); got != nil { + t.Fatalf("selected %q with every ledger at the limit, want none", got.Name) + } + if got := SelectClaim(Request{Shareable: true, Role: "trainer", Memory: 24 << 30, WorkerID: "w5"}, fleet); got == nil { + t.Fatal("with no limit a full ledger is still joinable") + } + if reason := Explain(req, fleet, ""); !strings.Contains(reason, "at most 2 workers") { + t.Errorf("reason does not name the limit: %q", reason) + } +} + // Tight fit lives in the tier order: the tier wasting the least device // memory leads, so small work prefers small devices -- a preference DRA // honors when it can, never a survey of what is free. diff --git a/scheduler/docs/design.md b/scheduler/docs/design.md index b86bbbb23..26c6c7564 100644 --- a/scheduler/docs/design.md +++ b/scheduler/docs/design.md @@ -513,15 +513,29 @@ memory are read from its owned `ResourceClaim` and the DRA inventory. ### Seat limit -There is no per-claim seat count. Host memory is the limit on parked -workers: every seat costs host memory even while its worker is off the GPU --- a suspended worker parks its state in node RAM -- so admission sums the -seats' pod requests against the node's allocatable memory (section 6), and -kube-scheduler enforces those same requests for real when the pod binds. -The two checks split the roles a seat count used to conflate: the advisory -sum steers workers away from full nodes, and the bind is the hard cap. A -join that slips past the advisory check on a stale snapshot produces an -unschedulable pod, which the wedged-claim path unwinds. +Host memory is the natural limit on parked workers: every seat costs host +memory even while its worker is off the GPU -- a suspended worker parks its +state in node RAM -- so admission sums the seats' pod requests against the +node's allocatable memory (section 6), and kube-scheduler enforces those +same requests for real when the pod binds. The two checks split the roles a +seat count used to conflate: the advisory sum steers workers away from full +nodes, and the bind is the hard cap. A join that slips past the advisory +check on a stale snapshot produces an unschedulable pod, which the +wedged-claim path unwinds. + +Host memory says nothing about time, and a time-sliced claim with six seats +runs each worker's step six times slower. `--max-seats-per-claim` +(`OPEN_RL_MAX_SEATS_PER_CLAIM`, default 0 for no limit) is the policy limit +on top: a ledger holding that many workers is not a candidate for selection, +and the ledger CAS loop refuses a join that would exceed it, so two joiners +working from one snapshot cannot both take the last seat. A worker that +finds every eligible ledger at the limit cuts a claim of its own; if no +device is free the claim stands pending, the status reason names the +limit, and the worker waits until a node arrives or a seat frees. It is +deliberately a hard limit rather than a preference: the bound on step time +holds, and a full fleet shows as standing claims an autoscaler can act on. +The limit is fleet policy and lives on the controller, not on nodes or +workloads; a per-device-tier form is a natural extension. An unallocated ledger contains only its dedicated workload and is not joinable: its node is unknown, so nothing about it can be checked against @@ -786,7 +800,10 @@ the dedicated-first arc described above. before cutting a claim at all; only a workload no ledger can seat cuts a dedicated claim. FFT can share; LoRA always cuts its own claim. Eligibility and preference are the shared-ledger rules below, unchanged; a booking that - loses the last seat to a race falls through to a dedicated claim. + loses the last seat to a race falls through to a dedicated claim. With + `--max-seats-per-claim` set, a ledger at the limit is not eligible, so + binpack fills one claim to the limit and then moves to the next free + device (section 4, "Seat limit"). One more placement policy is recorded as future work, not V1 behavior: @@ -810,6 +827,8 @@ A workload may join a ledger when: * the node permits the workload role (the `openrl.io/trainer` / `openrl.io/sampler` labels of section 2; enabled-ness is implicit, since the fleet contains only `openrl.io/enabled` nodes); +* the ledger holds fewer seats than `--max-seats-per-claim`, when set + (section 4, "Seat limit"); * the advisory host-memory check passes -- the one ceiling on how many workers may park on the node. From dc9b87c983121988afa39d5f2f1e1617b81856eb Mon Sep 17 00:00:00 2001 From: Hoda Mohammadi Date: Thu, 29 Oct 2026 15:51:37 -0700 Subject: [PATCH 2/3] docs(scheduler): trim the seat limit section to what the flag does --- scheduler/docs/design.md | 38 +++++++++++++++----------------------- 1 file changed, 15 insertions(+), 23 deletions(-) diff --git a/scheduler/docs/design.md b/scheduler/docs/design.md index 26c6c7564..cf905e8c3 100644 --- a/scheduler/docs/design.md +++ b/scheduler/docs/design.md @@ -513,29 +513,21 @@ memory are read from its owned `ResourceClaim` and the DRA inventory. ### Seat limit -Host memory is the natural limit on parked workers: every seat costs host -memory even while its worker is off the GPU -- a suspended worker parks its -state in node RAM -- so admission sums the seats' pod requests against the -node's allocatable memory (section 6), and kube-scheduler enforces those -same requests for real when the pod binds. The two checks split the roles a -seat count used to conflate: the advisory sum steers workers away from full -nodes, and the bind is the hard cap. A join that slips past the advisory -check on a stale snapshot produces an unschedulable pod, which the -wedged-claim path unwinds. - -Host memory says nothing about time, and a time-sliced claim with six seats -runs each worker's step six times slower. `--max-seats-per-claim` -(`OPEN_RL_MAX_SEATS_PER_CLAIM`, default 0 for no limit) is the policy limit -on top: a ledger holding that many workers is not a candidate for selection, -and the ledger CAS loop refuses a join that would exceed it, so two joiners -working from one snapshot cannot both take the last seat. A worker that -finds every eligible ledger at the limit cuts a claim of its own; if no -device is free the claim stands pending, the status reason names the -limit, and the worker waits until a node arrives or a seat frees. It is -deliberately a hard limit rather than a preference: the bound on step time -holds, and a full fleet shows as standing claims an autoscaler can act on. -The limit is fleet policy and lives on the controller, not on nodes or -workloads; a per-device-tier form is a natural extension. +Host memory is the limit on parked +workers: every seat costs host memory even while its worker is off the GPU +-- a suspended worker parks its state in node RAM -- so admission sums the +seats' pod requests against the node's allocatable memory (section 6), and +kube-scheduler enforces those same requests for real when the pod binds. +The two checks split the roles a seat count used to conflate: the advisory +sum steers workers away from full nodes, and the bind is the hard cap. A +join that slips past the advisory check on a stale snapshot produces an +unschedulable pod, which the wedged-claim path unwinds. + +`--max-seats-per-claim` (`OPEN_RL_MAX_SEATS_PER_CLAIM`, default 0 for no +limit) also caps the workers on one claim. A ledger at the limit is not +eligible, and the ledger CAS loop refuses a join past it. A worker that finds +every eligible ledger full cuts its own claim and, if no device is free, +stays Pending until a seat or a device frees up. An unallocated ledger contains only its dedicated workload and is not joinable: its node is unknown, so nothing about it can be checked against From 1f3022d5952d0bda2e9342b57aa3c8c0828e0ad3 Mon Sep 17 00:00:00 2001 From: Hoda Mohammadi Date: Fri, 30 Oct 2026 13:43:29 -0700 Subject: [PATCH 3/3] fix(scheduler): hold the seat limit on every write, retry the next ledger, wake on freed seats The cap check in ensureSeat ran only for joins, so a re-book whose seat was gone could append past it; it now gates every write that adds a seat. joinExistingClaim drops a ledger that refuses the booking and selects again instead of cutting a dedicated claim another ledger could have spared. A ClaimLedger watch wakes waiting workers when a seat is released. Explain names the limit only for workers that could have shared, the flag rejects negatives, and the docs say the placement timeout applies to a worker waiting on the limit. --- scheduler/controller/cmd/manager/main.go | 7 +- .../controller/internal/controller/ledger.go | 11 ++- .../internal/controller/ledger_test.go | 93 ++++++++++++++++++- .../controller/workload_controller.go | 61 ++++++++---- .../internal/placement/placement.go | 11 ++- .../internal/placement/placement_test.go | 5 + scheduler/docs/design.md | 17 ++-- 7 files changed, 172 insertions(+), 33 deletions(-) diff --git a/scheduler/controller/cmd/manager/main.go b/scheduler/controller/cmd/manager/main.go index 43d51dce1..6b0e16715 100644 --- a/scheduler/controller/cmd/manager/main.go +++ b/scheduler/controller/cmd/manager/main.go @@ -7,6 +7,7 @@ package main import ( "flag" + "fmt" "os" "strconv" "time" @@ -64,7 +65,7 @@ func main() { flag.StringVar(&strategy, "placement-strategy", env("OPEN_RL_PLACEMENT_STRATEGY", string(placement.StrategyBinPack)), "binpack shares eligible FFT claims first; spread requests a GPU first, then falls back to FFT sharing. LoRA never shares.") flag.IntVar(&maxSeats, "max-seats-per-claim", envInt("OPEN_RL_MAX_SEATS_PER_CLAIM", 0), - "Most workers that may share one claim; 0 is no limit. A worker that finds every eligible claim full cuts its own and waits Pending until a GPU is added or a seat frees.") + "Most workers that may share one claim; 0 is no limit, 1 is exclusive placement. A worker that finds every eligible claim full cuts its own and waits Pending until a GPU is added or a seat frees, or fails at the placement timeout.") flag.IntVar(&maxConcurrent, "max-concurrent-reconciles", envInt("OPEN_RL_MAX_CONCURRENT_RECONCILES", 4), "How many workers place at once. Seat booking is CAS-arbitrated, so concurrency risks only transient over-cut claims, which the sharing fallback drains.") @@ -83,6 +84,10 @@ func main() { setupLog.Error(err, "invalid placement strategy") os.Exit(1) } + if maxSeats < 0 { + setupLog.Error(fmt.Errorf("max-seats-per-claim is %d", maxSeats), "the seat limit must be 0 (no limit) or positive") + os.Exit(1) + } mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{ Scheme: scheme, diff --git a/scheduler/controller/internal/controller/ledger.go b/scheduler/controller/internal/controller/ledger.go index 55d47a6d5..a0657b684 100644 --- a/scheduler/controller/internal/controller/ledger.go +++ b/scheduler/controller/internal/controller/ledger.go @@ -45,8 +45,9 @@ func newSeat(worker *openrlv1alpha1.Workload, request placement.Request) openrlv // ensureSeat books the seat on the claim's ClaimLedger via one CAS loop. A seat // held by the same incarnation is adopted and a predecessor's is replaced. -// There is no seat ceiling: host memory is the limit on parked workers, -// checked at selection and verified against a fresh fleet read after booking. +// Host memory is the node-wide limit on parked workers, checked at selection +// and verified against a fresh fleet read after booking; MaxSeatsPerClaim, +// when set, is the per-claim limit and is re-checked here. // // createMissing is true only for a worker's own claim: the founder books its // ClaimLedger before the claim exists, and a re-book heals the ledger under a @@ -98,8 +99,10 @@ func (r *WorkloadReconciler) ensureSeat(ctx context.Context, claimName string, s } } // The seat limit is re-checked here for the same reason: two joiners - // selecting from one snapshot must not both land the last seat. - if existing == nil && !createMissing && r.MaxSeatsPerClaim > 0 && len(claimLedger.Spec.Seats) >= r.MaxSeatsPerClaim { + // selecting from one snapshot must not both land the last seat. It + // gates every write that adds a seat, a re-book whose seat is gone + // included; adopting or replacing a seat adds none. + if existing == nil && r.MaxSeatsPerClaim > 0 && len(claimLedger.Spec.Seats) >= r.MaxSeatsPerClaim { return nil, nil, errBookingContended } if existing != nil { diff --git a/scheduler/controller/internal/controller/ledger_test.go b/scheduler/controller/internal/controller/ledger_test.go index 28a4f9a46..b31f3179b 100644 --- a/scheduler/controller/internal/controller/ledger_test.go +++ b/scheduler/controller/internal/controller/ledger_test.go @@ -13,6 +13,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" openrlv1alpha1 "github.com/gke-labs/open-rl/scheduler/controller/api/v1alpha1" "github.com/gke-labs/open-rl/scheduler/controller/internal/placement" @@ -140,13 +141,78 @@ func TestBookingRefusesASeatPastTheLimit(t *testing.T) { if _, _, err := r.ensureSeat(context.Background(), claim, newSeat(c, requestFrom(c)), false); err != errBookingContended { t.Fatalf("booking a third seat under a limit of two returned %v, want errBookingContended", err) } - // The founder's own re-book is not a join and is never refused. + // A re-book that would add a seat is refused too: the limit holds on + // every write path, not only on joins. + if _, _, err := r.ensureSeat(context.Background(), claim, newSeat(c, requestFrom(c)), true); err != errBookingContended { + t.Fatalf("re-booking an unseated worker onto a full ledger returned %v, want errBookingContended", err) + } + // The founder's own re-book adopts its seat and is never refused. a := getWorker(t, r, "w-a") if _, _, err := r.ensureSeat(context.Background(), claim, newSeat(a, requestFrom(a)), true); err != nil { t.Fatalf("re-booking the founder: %v", err) } } +// A booking refused on the selected ledger moves on to the next eligible one +// instead of cutting a dedicated claim another ledger could have spared. +func TestJoinRetriesTheNextLedgerWhenTheFirstIsFull(t *testing.T) { + ctx := context.Background() + r := newReconciler(t, append(enabledNode(), + trainerWorker("w-a", "model-a"), trainerWorker("w-b", "model-b"), trainerWorker("w-c", "model-c"), + trainerWorker("w-d", "model-d"), trainerWorker("w-e", "model-e"), trainerWorker("w-f", "model-f"))...) + r.PlacementStrategy = placement.StrategyBinPack + // Two allocated ledgers: a limit of one makes w-c cut a claim of its own. + r.MaxSeatsPerClaim = 1 + settle(t, r, "w-a") + first := claimOf(t, r, "w-a") + allocateClaim(t, r, first) + settle(t, r, "w-c") + second := claimOf(t, r, "w-c") + if second == first { + t.Fatal("w-c joined under a limit of one, want a claim of its own") + } + allocateClaim(t, r, second) + + r.MaxSeatsPerClaim = 3 + book := func(name, claim string) { + t.Helper() + w := getWorker(t, r, name) + if _, _, err := r.ensureSeat(ctx, claim, newSeat(w, requestFrom(w)), false); err != nil { + t.Fatalf("booking %s on %s: %v", name, claim, err) + } + } + // Snapshot: the first ledger holds one seat and the second two, so + // selection prefers the first. + book("w-b", second) + fleet, err := r.readFleet(ctx) + if err != nil { + t.Fatal(err) + } + f := getWorker(t, r, "w-f") + request := requestFrom(f) + request.MaxSeats = 3 + if got := placement.SelectClaim(request, fleet); got == nil || got.Name != first { + t.Fatalf("snapshot selects %v, want %q", got, first) + } + // Behind the snapshot, the first ledger fills to the limit. + book("w-d", first) + book("w-e", first) + + joined, _, err := r.joinExistingClaim(ctx, f, request, fleet, "") + if err != nil { + t.Fatal(err) + } + if joined == nil || joined.Name != second { + t.Fatalf("joined %v, want the ledger with room %q", joined, second) + } + if got := len(getLedger(t, r, ledgerNameFor(second)).Spec.Seats); got != 3 { + t.Fatalf("second ledger holds %d seats, want 3", got) + } + if got := len(getLedger(t, r, ledgerNameFor(first)).Spec.Seats); got != 3 { + t.Fatalf("first ledger holds %d seats after the refused join, want 3", got) + } +} + func TestSharingRechecksTheLedgerAfterSelection(t *testing.T) { resident, incoming := trainerWorker("resident", "model-a"), trainerWorker("incoming", "model-b") r := newReconciler(t, append(enabledNode(), resident, incoming)...) @@ -602,3 +668,28 @@ func TestWaitsForATerminatingPredecessorClaim(t *testing.T) { t.Fatalf("placed on %q, want the fresh %q", got, claimNameFor(w)) } } + +// Only a ledger event that frees a seat wakes the fleet: a booking cannot +// unblock anyone, a release or a retirement can. +func TestSeatFreedWakesOnlyOnReleasedSeats(t *testing.T) { + seat := func(n int) *openrlv1alpha1.ClaimLedger { + l := &openrlv1alpha1.ClaimLedger{} + for i := 0; i < n; i++ { + l.Spec.Seats = append(l.Spec.Seats, openrlv1alpha1.Seat{Workload: fmt.Sprintf("w-%d", i)}) + } + return l + } + p := seatFreed() + if p.Create(event.CreateEvent{Object: seat(1)}) { + t.Error("a new ledger woke the fleet") + } + if p.Update(event.UpdateEvent{ObjectOld: seat(1), ObjectNew: seat(2)}) { + t.Error("a booking woke the fleet") + } + if !p.Update(event.UpdateEvent{ObjectOld: seat(2), ObjectNew: seat(1)}) { + t.Error("a released seat did not wake the fleet") + } + if !p.Delete(event.DeleteEvent{Object: seat(1)}) { + t.Error("a retired ledger did not wake the fleet") + } +} diff --git a/scheduler/controller/internal/controller/workload_controller.go b/scheduler/controller/internal/controller/workload_controller.go index aa46f1f67..d0e546791 100644 --- a/scheduler/controller/internal/controller/workload_controller.go +++ b/scheduler/controller/internal/controller/workload_controller.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "maps" "strings" "time" @@ -55,11 +56,12 @@ type WorkloadReconciler struct { // as binpack. PlacementStrategy placement.Strategy // MaxSeatsPerClaim caps how many workers share one claim; zero means no - // limit. A worker that finds every eligible ledger at the limit cuts its - // own claim and, if no device is free, stays Pending until one is added or - // a seat frees. It bounds a time-sliced step at the limit times the - // single-worker time, and it turns a full fleet into standing claims an - // autoscaler can see. + // limit, one is exclusive placement. A worker that finds every eligible + // ledger at the limit cuts its own claim and, if no device is free, waits + // on capacity like any other: Pending until one is added or a seat + // frees, Failed if PlacementTimeout runs out first. It bounds a + // time-sliced step at the limit times the single-worker time, and it + // turns a full fleet into standing claims an autoscaler can see. MaxSeatsPerClaim int // MaxConcurrentReconciles is how many workers place at once. MaxConcurrentReconciles int @@ -527,23 +529,35 @@ func (r *WorkloadReconciler) teardown(ctx context.Context, worker *openrlv1alpha // joinExistingClaim books a seat on the claim SelectClaim proposes, re-reads // the fleet to verify concurrent bookings did not overcommit the node, and -// keeps the caller's fleet snapshot current. +// keeps the caller's fleet snapshot current. A ledger that refuses the +// booking -- full, retiring, or holding an exclusive resident -- is dropped +// from the candidates and selection runs again, so losing one race does not +// cut a dedicated claim while another ledger has room. // // Because both racers write their seat before either re-reads, at least the // second reader sees the union across all claims on the node and yields. // Simultaneous racers may both observe the overcommit and both release their // seats; that double yield is benign and intentional. func (r *WorkloadReconciler) joinExistingClaim(ctx context.Context, worker *openrlv1alpha1.Workload, request placement.Request, fleet *placement.Fleet, current string) (*placement.Claim, *openrlv1alpha1.Seat, error) { - target := placement.SelectClaim(request, fleet) - if target == nil || target.Name == current { - return nil, nil, nil - } - _, seat, err := r.ensureSeat(ctx, target.Name, newSeat(worker, request), false) - if err == errBookingContended { - return nil, nil, nil - } - if err != nil { - return nil, nil, err + candidates := *fleet + candidates.Claims = maps.Clone(fleet.Claims) + var target *placement.Claim + var seat *openrlv1alpha1.Seat + for { + target = placement.SelectClaim(request, &candidates) + if target == nil || target.Name == current { + return nil, nil, nil + } + var err error + _, seat, err = r.ensureSeat(ctx, target.Name, newSeat(worker, request), false) + if err == errBookingContended { + delete(candidates.Claims, target.Name) + continue + } + if err != nil { + return nil, nil, err + } + break } fresh, err := r.readFleet(ctx) if err != nil { @@ -798,6 +812,20 @@ func managedClaims() predicate.Predicate { }) } +// seatFreed passes ledger events that free a seat: a seat dropped from the +// list, or the ledger retired. A seat added never unblocks a waiting worker, +// so bookings and new ledgers do not wake the fleet. +func seatFreed() predicate.Predicate { + return predicate.Funcs{ + CreateFunc: func(event.CreateEvent) bool { return false }, + UpdateFunc: func(e event.UpdateEvent) bool { + before, okBefore := e.ObjectOld.(*openrlv1alpha1.ClaimLedger) + after, okAfter := e.ObjectNew.(*openrlv1alpha1.ClaimLedger) + return okBefore && okAfter && len(after.Spec.Seats) < len(before.Spec.Seats) + }, + } +} + // SetupWithManager registers the reconciler. func (r *WorkloadReconciler) SetupWithManager(mgr ctrl.Manager) error { r.reader = mgr.GetAPIReader() @@ -835,6 +863,7 @@ func (r *WorkloadReconciler) SetupWithManager(mgr ctrl.Manager) error { For(&openrlv1alpha1.Workload{}). Owns(&corev1.Pod{}). Watches(&resourcev1.ResourceClaim{}, wakeAll, builder.WithPredicates(managedClaims())). + Watches(&openrlv1alpha1.ClaimLedger{}, wakeAll, builder.WithPredicates(managedClaims(), seatFreed())). Watches(&corev1.Node{}, wakeAll, builder.WithPredicates(nodeCapacityChanged)). WithOptions(controller.Options{MaxConcurrentReconciles: max(1, r.MaxConcurrentReconciles)}). Complete(r) diff --git a/scheduler/controller/internal/placement/placement.go b/scheduler/controller/internal/placement/placement.go index 4627c9c39..d081f9824 100644 --- a/scheduler/controller/internal/placement/placement.go +++ b/scheduler/controller/internal/placement/placement.go @@ -292,10 +292,10 @@ func candidateNodes(req Request, fleet *Fleet) map[string]int { } // SelectClaim picks the allocated claim a worker should join, or nil: -// single-device only, role and host-memory checked -- host memory is the one -// ceiling on how many workers may park on a node -- preferring fewest owners, -// then fewest workers, then name. Advisory -- the ledger's CAS booking is what -// makes it stick. +// single-device only, role, seat limit and host memory checked -- host memory +// is the node-wide ceiling on parked workers, MaxSeats the per-claim one -- +// preferring fewest owners, then fewest workers, then name. Advisory -- the +// ledger's CAS booking is what makes it stick. func SelectClaim(req Request, fleet *Fleet) *Claim { if !req.Shareable || req.Devices > 0 { return nil @@ -373,7 +373,8 @@ func Explain(req Request, fleet *Fleet, detail string) string { case len(candidateNodes(req, fleet)) > 0: // The hardware exists; it is busy. Retrying is the right move. reason = "WaitingForCapacity: a pool fits this workload but none has a free seat or a free accelerator" - if req.MaxSeats > 0 { + // Only a worker that could have shared is waiting on the limit. + if req.MaxSeats > 0 && req.Shareable && req.Devices == 0 { reason += fmt.Sprintf("; shared claims hold at most %d workers", req.MaxSeats) } default: diff --git a/scheduler/controller/internal/placement/placement_test.go b/scheduler/controller/internal/placement/placement_test.go index e91850587..fb747a504 100644 --- a/scheduler/controller/internal/placement/placement_test.go +++ b/scheduler/controller/internal/placement/placement_test.go @@ -208,6 +208,11 @@ func TestSelectClaimSkipsLedgersAtTheSeatLimit(t *testing.T) { if reason := Explain(req, fleet, ""); !strings.Contains(reason, "at most 2 workers") { t.Errorf("reason does not name the limit: %q", reason) } + // A worker that never shares is not waiting on the limit. + exclusive := Request{Shareable: false, Role: "trainer", Memory: 24 << 30, WorkerID: "w6", MaxSeats: 2} + if reason := Explain(exclusive, fleet, ""); strings.Contains(reason, "at most") { + t.Errorf("exclusive worker's reason names the seat limit: %q", reason) + } } // Tight fit lives in the tier order: the tier wasting the least device diff --git a/scheduler/docs/design.md b/scheduler/docs/design.md index cf905e8c3..8755f5d7f 100644 --- a/scheduler/docs/design.md +++ b/scheduler/docs/design.md @@ -525,9 +525,13 @@ unschedulable pod, which the wedged-claim path unwinds. `--max-seats-per-claim` (`OPEN_RL_MAX_SEATS_PER_CLAIM`, default 0 for no limit) also caps the workers on one claim. A ledger at the limit is not -eligible, and the ledger CAS loop refuses a join past it. A worker that finds -every eligible ledger full cuts its own claim and, if no device is free, -stays Pending until a seat or a device frees up. +eligible, and the ledger CAS loop refuses any write that would add a seat +past it. A worker that finds every eligible ledger full cuts its own claim +and, if no device is free, waits on capacity like any other worker: Pending +until a seat or a device frees up, Failed if `--placement-timeout` runs out +first. A limit of 1 is exclusive placement, since every allocated ledger +already holds its founder. A seat released on a ledger wakes the waiting +workers, as a freed claim or node does. An unallocated ledger contains only its dedicated workload and is not joinable: its node is unknown, so nothing about it can be checked against @@ -792,7 +796,8 @@ the dedicated-first arc described above. before cutting a claim at all; only a workload no ledger can seat cuts a dedicated claim. FFT can share; LoRA always cuts its own claim. Eligibility and preference are the shared-ledger rules below, unchanged; a booking that - loses the last seat to a race falls through to a dedicated claim. With + loses the last seat to a race moves on to the next eligible ledger, and + only when none accepts it falls through to a dedicated claim. With `--max-seats-per-claim` set, a ledger at the limit is not eligible, so binpack fills one claim to the limit and then moves to the next free device (section 4, "Seat limit"). @@ -821,8 +826,8 @@ A workload may join a ledger when: the fleet contains only `openrl.io/enabled` nodes); * the ledger holds fewer seats than `--max-seats-per-claim`, when set (section 4, "Seat limit"); -* the advisory host-memory check passes -- the one ceiling on how many - workers may park on the node. +* the advisory host-memory check passes -- the node-wide ceiling on how + many workers may park there. Among eligible ledgers, the controller prefers, packing tightly while keeping turn queues short: