fix(broker): bound the squatter total by the card's used memory (#103)

CHANGELOG.md

@@ -31,6 +31,17 @@ has a matching GitHub release whose body comes from its section below.
   (`squatters`, `unprotected`, `leases`; there is no `holders`). `berth events` is called with
   `--format json`, and a berth that answers nothing leaves its stderr in the report instead of
   a silent "returned nothing".
+- The squatter sum is bounded by the memory the driver reports the card as
+  using: on Windows a process's reading counts its committed reservation, so
+  while the v0.3.2 per-process cap bounds one process, the sum could still
+  exceed the card's real usage (the desktop compositor read as 11 GB on a
+  16 GB card the driver reported at 15.5 GB in use) and a `berth run` request
+  that fit the free space the driver actually shows was refused forever.
+  Admission, the contested-grant figure and the "squatters hold N MiB" wait
+  reason now count the bounded total, and the per-squatter status and event
+  entries keep the raw readings. While it clamps, the bound logs at most one
+  INFO line per change of that state, never per tick.
+  [#103](https://github.com/T-Crypt/berth/issues/103)
 - A lease whose attached run process dies without the wrapper's disconnect
   reaching the daemon (the run's tree force-killed, a crash, a power loss)
   is no longer left granted with an empty PIDS list: the broker now checks

internal/broker/broker.go

@@ -313,12 +313,15 @@ type Broker struct {
 	// reads counts the card reads started, bumped before each probe read.
 	reads atomic.Uint64
 
-	mu        sync.Mutex
-	snap      gpu.Snapshot
-	probeErr  error
-	leases    map[string]*lease // granted: active, draining, releasing
-	queue     []*lease
-	squatters map[ident]*squatter
+	mu       sync.Mutex
+	snap     gpu.Snapshot
+	probeErr error
+	// squatClamped is whether the squatter total was last capped at the
+	// card's used figure (boundSquat); a change of that state logs once.
+	squatClamped bool
+	leases       map[string]*lease // granted: active, draining, releasing
+	queue        []*lease
+	squatters    map[ident]*squatter
 	// containerNames caches container id -> runtime name ("" while unknown
 	// or resolving); containerAsked marks the ids already sent to resolve.
 	containerNames map[string]string
@@ -987,11 +990,51 @@ func describe(ls []*lease) string {
 	return strings.Join(names, ", ")
 }
 
+// squatBytes is the squatter total admission counts: the sum of the
+// per-process readings, capped by what the card can really hand squatters
+// (boundSquat). The per-squatter entries in status and events keep the raw
+// readings; only the total that drives decisions is bounded.
 func (b *Broker) squatBytes() uint64 {
 	var n uint64
 	for _, s := range b.squatters {
 		n += s.proc.Used
 	}
+	return b.boundSquat(n)
+}
+
+// boundSquat caps the squatter sum at the card's used figure. Per-process
+// readings can overcount: on Windows the counter reports a process's
+// committed reservation, which the driver can list far above what the card
+// actually uses, so the SUM can exceed the card's used memory (issue #103)
+// and block every sized admission forever. Used memory minus the baseline
+// and the bytes the card's processes under leases and claims really hold
+// is all any squatter can hold, so the sum is capped there. A zero used
+// figure is unknown: the raw sum stands, as before.
+func (b *Broker) boundSquat(sum uint64) uint64 {
+	if b.snap.Used == 0 {
+		return sum
+	}
+	bound := sat(b.snap.Used, b.snap.Baseline()+b.leaseBytes()+b.claimedBytes())
+	if sum <= bound {
+		b.squatClamped = false
+		return sum
+	}
+	if !b.squatClamped {
+		b.squatClamped = true
+		b.log.Info("squatter readings exceed the card's used memory; counting the card's figure",
+			"sum_mib", sum/gpu.MiB, "bound_mib", bound/gpu.MiB)
+	}
+	return bound
+}
+
+// leaseBytes is the card memory the processes under leases really hold:
+// each lease's per-process readings, not its needs budget, so the cap keeps
+// real lease memory out of a squatter's possible share of the card.
+func (b *Broker) leaseBytes() uint64 {
+	var n uint64
+	for _, l := range b.leases {
+		n += l.used()
+	}
 	return n
 }
 

internal/broker/broker_test.go

@@ -4,6 +4,7 @@
 package broker
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"io"
@@ -27,10 +28,14 @@ import (
 
 const GiB = 1 << 30
 
-// card is a fake GPU whose processes the test sets directly.
+// card is a fake GPU whose processes the test sets directly. When used is
+// set, the driver's used figure is that, not the processes' sum: the
+// readings a test gives can disagree with what the driver reports, as
+// Windows counters can.
 type card struct {
 	mu     sync.Mutex
 	total  uint64
+	used   uint64
 	procs  []gpu.Proc
 	onRead func() // runs at the start of a read, outside the card lock
 }
@@ -50,11 +55,17 @@ func (c *card) Read(context.Context) (gpu.Snapshot, error) {
 	}
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	var used uint64
-	for _, p := range c.procs {
-		used += p.Used
+	used := c.used
+	if used == 0 {
+		for _, p := range c.procs {
+			used += p.Used
+		}
+	}
+	free := uint64(0)
+	if used <= c.total {
+		free = c.total - used
 	}
-	return gpu.Snapshot{Total: c.total, Used: used, Free: c.total - used, Procs: append([]gpu.Proc(nil), c.procs...)}, nil
+	return gpu.Snapshot{Total: c.total, Used: used, Free: free, Procs: append([]gpu.Proc(nil), c.procs...)}, nil
 }
 
 type reports struct {
@@ -618,6 +629,165 @@ func TestEmptyHoldersSmallSquatStillGrants(t *testing.T) {
 	}
 }
 
+// #103: on Windows the per-process cap bounds one process, but the SUM of
+// the readings can still exceed what the driver reports as used: the
+// counter counts committed reservations, which the card does not hold.
+// The squatter total is bounded by what the card can really hand
+// squatters, so a small probe still fits the free space the driver shows
+// instead of waiting behind a phantom 27 GB.
+func TestSquatterSumIsBoundedByTheCardsUsedMemory(t *testing.T) {
+	b, c, _ := setup(t, procs.Map{})
+	c.total = 16303 * gpu.MiB
+	c.used = 15600 * gpu.MiB
+	c.set(
+		gpu.Proc{PID: 9, Name: "llama-server", Used: 14230 * gpu.MiB},
+		gpu.Proc{PID: 10, Name: "dwm.exe", Used: 11138 * gpu.MiB},
+	)
+	b.Tick(context.Background())
+
+	// The readings sum to 25368 MiB on a card that uses 15600, so the
+	// total counts the card's figure: 100 MiB fits the 703 MiB the card
+	// really has free, and the contested figure stays bounded too.
+	l := must(t, acquire(b, Request{Engine: "probe", Needs: 100 * gpu.MiB}))
+	if l.Contested() != 15600*gpu.MiB {
+		t.Fatalf("contested = %d MiB, want the card's used figure", l.Contested()/gpu.MiB)
+	}
+	l.Release()
+	b.Tick(context.Background())
+
+	// The per-squatter entries keep the raw readings; only the total
+	// driving decisions is bounded.
+	if st := b.Status(); len(st.Squatters) != 2 ||
+		st.Squatters[0].Name != "llama-server" || st.Squatters[0].Used != 14230*gpu.MiB ||
+		st.Squatters[1].Name != "dwm.exe" || st.Squatters[1].Used != 11138*gpu.MiB {
+		t.Fatalf("status squatters = %+v, want the raw readings kept", st.Squatters)
+	}
+}
+
+// #103: the bound shrinks the squatter total, it does not erase it: a
+// request that does not fit the card's real free space still waits, told
+// the bounded figure.
+func TestSquatterSumBoundedBigRequestStillWaits(t *testing.T) {
+	b, c, _ := setup(t, procs.Map{})
+	c.total = 16303 * gpu.MiB
+	c.used = 15600 * gpu.MiB
+	c.set(
+		gpu.Proc{PID: 9, Name: "llama-server", Used: 14230 * gpu.MiB},
+		gpu.Proc{PID: 10, Name: "dwm.exe", Used: 11138 * gpu.MiB},
+	)
+	b.Tick(context.Background())
+
+	reasons := make(chan string, 8)
+	ch := make(chan result, 1)
+	go func() {
+		l, err := b.Acquire(context.Background(), Request{Engine: "big", Needs: 2000 * gpu.MiB}, func(s string) { reasons <- s })
+		ch <- result{l, err}
+	}()
+	pending(t, ch)
+	select {
+	case why := <-reasons:
+		if !strings.Contains(why, "at most 703 MiB can be freed") || !strings.Contains(why, "squatters hold 15600 MiB") {
+			t.Fatalf("wait reason %q", why)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("waiter heard no reason")
+	}
+	if st := b.Status(); len(st.Leases) != 0 {
+		t.Fatalf("leases = %+v, want none: 2000 MiB does not fit", st.Leases)
+	}
+}
+
+// Linux readings are exact: when the sum is under the card's used figure
+// the bound is a no-op and admission sees the sum, as before.
+func TestSquatterSumUnderTheUsedFigureIsUnchanged(t *testing.T) {
+	b, c, _ := setup(t, procs.Map{})
+	c.total = 16303 * gpu.MiB
+	c.used = 9000 * gpu.MiB
+	c.set(
+		gpu.Proc{PID: 9, Name: "llama-server", Used: 2000 * gpu.MiB},
+		gpu.Proc{PID: 10, Name: "dwm.exe", Used: 3000 * gpu.MiB},
+	)
+	b.Tick(context.Background())
+
+	// The 5000 MiB sum is under the card's 9000 MiB used figure, so the
+	// bound (used minus the 4000 MiB baseline) equals the sum and nothing
+	// is clamped: the waiter hears the raw sum and the same fit as before.
+	reasons := make(chan string, 8)
+	ch := make(chan result, 1)
+	go func() {
+		l, err := b.Acquire(context.Background(), Request{Engine: "big", Needs: 10000 * gpu.MiB}, func(s string) { reasons <- s })
+		ch <- result{l, err}
+	}()
+	pending(t, ch)
+	select {
+	case why := <-reasons:
+		if !strings.Contains(why, "at most 7303 MiB can be freed") || !strings.Contains(why, "squatters hold 5000 MiB") {
+			t.Fatalf("wait reason %q", why)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("waiter heard no reason")
+	}
+}
+
+// #103: the cap logs at most one line per change of the clamped state, not
+// once per tick.
+func TestSquatterCapLogsOncePerStateChange(t *testing.T) {
+	b, c, _ := setup(t, procs.Map{})
+	var buf bytes.Buffer
+	b.log = slog.New(slog.NewTextHandler(&buf, nil))
+	c.total = 16303 * gpu.MiB
+	c.used = 15600 * gpu.MiB
+	c.set(
+		gpu.Proc{PID: 9, Name: "llama-server", Used: 14230 * gpu.MiB},
+		gpu.Proc{PID: 10, Name: "dwm.exe", Used: 11138 * gpu.MiB},
+	)
+	// The first tick reads the card and sees the squatters; the queued
+	// waiter makes every later schedule recompute the bounded sum. Its
+	// first wait reason only goes out once that schedule ran.
+	b.Tick(context.Background())
+	reasons := make(chan string, 8)
+	why := make(chan struct{})
+	ch := make(chan result, 1)
+	go func() {
+		l, err := b.Acquire(context.Background(), Request{Engine: "big", Needs: 2000 * gpu.MiB}, func(s string) {
+			reasons <- s
+			select {
+			case why <- struct{}{}:
+			default:
+			}
+		})
+		ch <- result{l, err}
+	}()
+	select {
+	case <-why:
+	case <-time.After(time.Second):
+		t.Fatal("waiter never heard a reason")
+	}
+	for i := 0; i < 3; i++ {
+		b.Tick(context.Background())
+	}
+	const line = "squatter readings exceed the card's used memory"
+	if n := strings.Count(buf.String(), line); n != 1 {
+		t.Fatalf("cap logged %d times over clamped ticks, want 1:\n%s", n, buf.String())
+	}
+
+	// The driver shows the card using what the sum says: the state flips
+	// back to the raw sum, and one further tick logs nothing.
+	c.used = 25368 * gpu.MiB
+	b.Tick(context.Background())
+	b.Tick(context.Background())
+	if n := strings.Count(buf.String(), line); n != 1 {
+		t.Fatalf("cap logged %d times after unclamping, want still 1:\n%s", n, buf.String())
+	}
+
+	// And a new clamp is one more line: one per change of state.
+	c.used = 15600 * gpu.MiB
+	b.Tick(context.Background())
+	if n := strings.Count(buf.String(), line); n != 2 {
+		t.Fatalf("cap logged %d times, want 2 after re-clamping:\n%s", n, buf.String())
+	}
+}
+
 // Gap d: an engine started without a lease is listed, and its GPU
 // processes say which unprotected run they belong to.
 func TestUnprotectedRunsAreListed(t *testing.T) {