From 5ad9410b7ceab94d2c9e9b6e4ff120289f71b201 Mon Sep 17 00:00:00 2001 From: Brad Fitzpatrick Date: Thu, 1 Oct 2026 18:09:39 +0000 Subject: [PATCH] gocached: govern put-queue movers by throughput, not latency ratio The mover governor compared each interval's median small-copy latency to an "unloaded" baseline and only grew the limit while the ratio stayed under 1.5x, backing off above 2x. In production that pinned it at its floor of 8 within 30 seconds of starting: the baseline was learned at 28ms during the first near-idle seconds, and at 8 concurrent copies the median was 80 to 115ms, a ratio of 3 to 4. The governor read that as saturation and, since a backlog kept demand constant, never re-learned. Copy throughput stayed at the same 85/s the fixed 8 movers had delivered, so the deploy changed nothing. The premise was wrong. Per-copy latency on a network filesystem grows with concurrency long before throughput stops growing, because requests queue at the client and on the wire even when the server has capacity. Latency is not the signal; throughput is. Replace the controller with a hill-climber on measured copies/s. The governor keeps a committed operating point (a limit and the throughput measured there) and probes from it: raise the limit by 25% and keep the raise only if copies/s rose by at least half the ideal linear gain; lower it by 25% and keep that if copies/s held within 10%. A rejected probe reverts and holds for 30s before trying again. In steady state every fourth probe is downward so a limit that drifted above the knee gets trimmed, and a 25% fall in throughput at the committed limit triggers an immediate downward probe since the directory may have gotten slower. Windows where the slots were less than 90% busy are discarded: throughput was bounded by demand, not the limit, and says nothing about capacity. A window also needs 512 completions before it counts, which puts Poisson noise on a two-window comparison near 6%, under the 12.5% a probe must show. In simulation against a directory that scales linearly to a knee, the limit reaches 90% of the ceiling in 40 to 110 seconds from the floor, settles within a step of the knee, stays there under 15% throughput jitter, trims within minutes when the ceiling drops, and climbs back when it rises. Tests cover each of those, plus demand-limited and sparse windows leaving the limit alone. Also bound cleanup-intent deletions to the first minLimit mover goroutines. Letting all 256 do them turned every metadata flush into a burst of about 85 concurrent unlinks against the main directory, which is the likely reason copy latency doubled under the first governor at the same 8-copy concurrency; this returns the unlink load to what a fixed-mover server produced. The latency baseline gauges go away. New gauges export the committed limit and the copies/s measured there, which together read as the main directory's throughput ceiling whenever the committed limit is below the maximum, and a count of rejected probes, which climbs steadily in both directions once the governor has found the knee. Updates tailscale/corp#47471 Signed-off-by: Brad Fitzpatrick Migrated-from: bradfitz/go-tool-cache@13c5751bd8bd427065ba27084fcc36d7587bcd06 --- gocache/gocached/gocached.go | 24 +- gocache/gocached/movergov.go | 332 ++++++++++++++++--------- gocache/gocached/movergov_test.go | 397 ++++++++++++++++++------------ gocache/gocached/putqueue.go | 35 ++- 4 files changed, 493 insertions(+), 295 deletions(-) diff --git a/gocache/gocached/gocached.go b/gocache/gocached/gocached.go index f37cf46..e8be2c8 100644 --- a/gocache/gocached/gocached.go +++ b/gocache/gocached/gocached.go @@ -603,20 +603,22 @@ func (srv *Server) start() error { func() float64 { return float64(len(srv.putq.moverCh)) }}, {"gocached_put_queue_flush_backlog", "settled PUTs waiting for the metadata flusher; nonzero while SQLite commits are the constraint", func() float64 { return float64(len(srv.putq.flushCh)) }}, - {"gocached_put_queue_movers_limit", "current adaptive limit on concurrent copies into the main blob directory; grows while small-copy latency stays near baseline under demand, shrinks when it climbs", + {"gocached_put_queue_movers_limit", "current limit on concurrent copies into the main blob directory; the governor's committed operating point plus or minus a probe step while it tests whether more or fewer slots change throughput", func() float64 { return float64(srv.putq.gov.stats().limit) }}, {"gocached_put_queue_movers_active", "copies into the main blob directory currently in flight; pinned at the limit with a nonzero mover backlog means the limit is what bounds the spooled lane", func() float64 { return float64(srv.putq.gov.stats().active) }}, - {"gocached_put_queue_movers_max", "configured upper bound on the adaptive mover limit; a limit sitting here under demand with baseline latency means the bound is too low", + {"gocached_put_queue_movers_max", "configured upper bound on the adaptive mover limit; a committed limit sitting here under demand means the bound is too low", func() float64 { return float64(srv.putq.gov.maxLimit) }}, - {"gocached_put_queue_copy_latency_baseline_seconds", "the mover governor's estimate of the main blob directory's unloaded small-copy latency", - func() float64 { return srv.putq.gov.stats().baseline.Seconds() }}, - {"gocached_put_queue_copy_latency_recent_seconds", "median small-copy latency in the mover governor's last evaluated interval; its ratio to the baseline drives the limit", - func() float64 { return srv.putq.gov.stats().lastMedian.Seconds() }}, - {"gocached_put_queue_movers_increases", "times the mover governor raised the limit", + {"gocached_put_queue_movers_committed_limit", "the concurrency the mover governor last judged worth running at: the highest limit at which adding slots still raised copies/s", + func() float64 { return float64(srv.putq.gov.stats().committedLimit) }}, + {"gocached_put_queue_movers_committed_rate", "copies/s into the main blob directory the mover governor measured at its committed limit; the main directory's throughput ceiling for the current blob mix if the committed limit is below the max", + func() float64 { return srv.putq.gov.stats().committedRate }}, + {"gocached_put_queue_movers_increases", "times the mover governor raised the limit (upward probes and reverts of downward ones)", func() float64 { return float64(srv.putq.gov.stats().increases) }}, - {"gocached_put_queue_movers_decreases", "times the mover governor lowered the limit; frequent decreases mean the main blob directory saturates at the current load", + {"gocached_put_queue_movers_decreases", "times the mover governor lowered the limit (downward probes and reverts of upward ones)", func() float64 { return float64(srv.putq.gov.stats().decreases) }}, + {"gocached_put_queue_movers_probes_rejected", "probes the mover governor reverted because the throughput change didn't justify the new limit; steady rejections in both directions mean it has found the knee", + func() float64 { return float64(srv.putq.gov.stats().rejected) }}, } { reg.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{Name: g.name, Help: g.help}, g.fn)) } @@ -813,9 +815,9 @@ func WithPutSpoolCapacity(bytes int64) ServerOption { // WithPutMoverLimits bounds the adaptive number of concurrent copies of // spooled blobs into the main blob directory. The server starts at minLimit -// and raises the limit while the latency of small copies stays near its -// unloaded baseline and copies are waiting for a slot, backing off when -// latency climbs; maxLimit caps it. Zero for either uses the default (8 and +// and hill-climbs on measured copy throughput: it keeps a higher limit only +// if copies/s rose with it, and trims when fewer slots deliver the same +// throughput; maxLimit caps it. Zero for either uses the default (8 and // 256). See gocached_put_queue_movers_* metrics for the observed behavior. func WithPutMoverLimits(minLimit, maxLimit int) ServerOption { return func(srv *Server) { diff --git a/gocache/gocached/movergov.go b/gocache/gocached/movergov.go index e8c42ce..86fd2a4 100644 --- a/gocache/gocached/movergov.go +++ b/gocache/gocached/movergov.go @@ -2,7 +2,6 @@ package gocached import ( "context" - "slices" "sync" "time" ) @@ -11,80 +10,96 @@ import ( const ( // defaultMinPutMovers and defaultMaxPutMovers bound the adaptive number // of concurrent spooled-blob copies into the main blob directory. The - // minimum is the floor the governor never backs off below (and the - // starting point); the maximum bounds goroutines, open files, and the - // concurrency inflicted on the main directory's filesystem. + // minimum is the starting point and the floor the governor never trims + // below; the maximum bounds goroutines, open files, and the concurrency + // inflicted on the main directory's filesystem. defaultMinPutMovers = 8 defaultMaxPutMovers = 256 - // moverProbeMaxSize is the largest stored blob size whose copy latency - // counts as a probe of the main directory's round-trip latency. Below - // this the transfer time is negligible next to the fixed per-copy - // round trips (intent create, temp create, write, rename), so the - // latency reflects how loaded the filesystem is rather than how big - // the blob was. In practice most spooled blobs are this small. - moverProbeMaxSize = 64 << 10 - - // moverAdjustInterval is how often the governor re-evaluates the limit. - moverAdjustInterval = 2 * time.Second - - // moverMinProbes is the fewest probe samples an interval needs before - // the governor acts on it; with fewer it holds the current limit. - moverMinProbes = 5 - - // moverIncreaseBelow and moverDecreaseAbove are the ratios of the - // interval's median probe latency to the unloaded baseline that - // trigger growth (when there is also demand) and shrinkage. Between - // them the limit holds. - moverIncreaseBelow = 1.5 - moverDecreaseAbove = 2.0 - - // moverIncreaseDivisor sets additive growth: the limit grows by - // limit/moverIncreaseDivisor (at least 1) per interval, so it doubles - // in about moverIncreaseDivisor intervals under sustained headroom. - moverIncreaseDivisor = 4 - - // moverDecreaseNumer/Denom set multiplicative backoff: the limit is - // scaled by moverDecreaseNumer/moverDecreaseDenom when latency shows - // the filesystem is saturating. - moverDecreaseNumer = 3 - moverDecreaseDenom = 4 - - // moverBaselineDrift is how much of the gap between the baseline and an - // unloaded interval's median the baseline absorbs per interval. It lets - // the baseline re-learn a slower filesystem, but only from intervals - // with no demand (no mover waited for a slot), so sustained saturation - // can't teach the governor that loaded latency is normal. - moverBaselineDrift = 0.05 + // moverStepFraction is how far a probe moves the limit from the + // committed operating point, as a fraction of it (at least one slot). + moverStepFraction = 0.25 + + // moverEvalInterval is the shortest measurement window. A window also + // needs moverMinCompletions copies before the governor acts on it, and + // stretches up to moverMaxEvalInterval to collect them; a window that + // ends without them is discarded. 512 completions put the Poisson + // standard error of the window's rate near 4.4%, and of a comparison + // between two windows near 6.3%, under the 12.5% gain a probe must + // show, so a single noisy window rarely flips a decision. + moverEvalInterval = 2 * time.Second + moverMaxEvalInterval = 30 * time.Second + moverMinCompletions = 512 + + // moverMinUtilization is the fraction of limit*window for which copy + // slots must have been held for the window to say anything about + // capacity. Below it, throughput was bounded by demand (an emptying + // backlog), not by the limit, and the window is discarded. + moverMinUtilization = 0.9 + + // moverGainFraction is the share of the ideal (linear) throughput gain + // an upward probe must deliver to be kept: a +25% step must yield at + // least +12.5% copies/s. Latency rising with concurrency is expected + // and irrelevant; what matters is whether the extra slots did work. + moverGainFraction = 0.5 + + // moverLossTolerance is how much throughput a downward probe may lose + // and still be kept, meaning the slots given up were buying nothing + // beyond measurement noise. + moverLossTolerance = 0.1 + + // moverDropFraction is the fall in throughput at the committed limit + // that triggers an immediate downward probe: the directory got slower, + // so the committed concurrency may now be excess. + moverDropFraction = 0.25 + + // moverHoldAfterReject is how long the governor sits at the committed + // limit after a probe is rejected before probing again. + moverHoldAfterReject = 30 * time.Second + + // moverDownProbeEvery makes every Nth probe from steady state a + // downward one, so a limit that drifted above the knee (through a + // noisy accepted probe) gets trimmed even if throughput never drops. + moverDownProbeEvery = 4 ) // moverGovernor adaptively sets how many spooled-blob copies into the main -// blob directory may run concurrently, using an AIMD control loop driven by -// the latency of small copies. +// blob directory may run concurrently, by hill-climbing on measured copy +// throughput. // // The main directory may be a network filesystem whose throughput for the // small, round-trip-bound copies that dominate the put queue scales with -// concurrency until the filesystem saturates. Rather than guess a fixed -// mover count, the governor grows the limit while small-copy latency stays -// near its unloaded baseline and there is demand (movers waiting for a -// slot), and backs off multiplicatively when latency climbs well above the -// baseline. Without demand it holds, since there is nothing to learn from -// an idle filesystem except its baseline latency. +// concurrency until the filesystem saturates. Per-copy latency grows with +// concurrency well before that point, so latency is not the signal; +// throughput is. The governor holds a committed operating point (a limit +// and the copies/s measured there) and probes: it raises the limit by a +// step and keeps the raise only if copies/s rose by a meaningful fraction +// of the ideal linear gain, and it lowers the limit when a trial at fewer +// slots delivers the same throughput. Windows in which the slots weren't +// kept busy (demand-limited) teach it nothing and are discarded. type moverGovernor struct { minLimit int maxLimit int + clock func() time.Time mu sync.Mutex cond *sync.Cond // signaled when a slot frees or the limit grows limit int // current maximum concurrent copies active int // copies currently running - waited bool // whether any acquire had to wait this interval - probes []time.Duration // probe copy latencies observed this interval - baseline time.Duration // estimate of unloaded probe latency; 0 until learned - lastMed time.Duration // median probe latency of the last evaluated interval + // Current measurement window. + windowStart time.Time + lastAccrue time.Time + busy time.Duration // integral of active over the window + completions int // successful copies finished in the window + + // Committed operating point and probe state. + committedLimit int + committedRate float64 // copies/s measured at committedLimit; 0 until known + holdUntil time.Time + probes int // steady-state probes started, for alternating direction - increases, decreases int // adjustment counts for metrics + increases, decreases, rejected int // for metrics } func newMoverGovernor(minLimit, maxLimit int) *moverGovernor { @@ -95,14 +110,27 @@ func newMoverGovernor(minLimit, maxLimit int) *moverGovernor { maxLimit = max(minLimit, defaultMaxPutMovers) } g := &moverGovernor{ - minLimit: minLimit, - maxLimit: maxLimit, - limit: minLimit, + minLimit: minLimit, + maxLimit: maxLimit, + clock: time.Now, + limit: minLimit, + committedLimit: minLimit, } g.cond = sync.NewCond(&g.mu) + now := g.clock() + g.windowStart, g.lastAccrue = now, now return g } +// accrueLocked advances the busy integral to now. It must be called before +// any change to active or limit, and the caller must hold g.mu. +func (g *moverGovernor) accrueLocked(now time.Time) { + if now.After(g.lastAccrue) { + g.busy += time.Duration(g.active) * now.Sub(g.lastAccrue) + g.lastAccrue = now + } +} + // acquire blocks until a copy slot is available under the current limit or // ctx is done. Each successful acquire must be paired with a release. func (g *moverGovernor) acquire(ctx context.Context) error { @@ -120,92 +148,159 @@ func (g *moverGovernor) acquire(ctx context.Context) error { if err := ctx.Err(); err != nil { return err } - g.waited = true g.cond.Wait() } if err := ctx.Err(); err != nil { return err } + g.accrueLocked(g.clock()) g.active++ return nil } -// release returns a slot taken by acquire. -func (g *moverGovernor) release() { +// release returns a slot taken by acquire. completed reports whether the +// copy succeeded; only successful copies count toward throughput. +func (g *moverGovernor) release(completed bool) { g.mu.Lock() + g.accrueLocked(g.clock()) g.active-- - g.mu.Unlock() - g.cond.Signal() -} - -// observe records a completed copy of a blob of storedSize bytes that took -// d. Only copies small enough to be latency probes affect the limit. -func (g *moverGovernor) observe(storedSize int64, d time.Duration) { - if storedSize > moverProbeMaxSize { - return + if completed { + g.completions++ } - g.mu.Lock() - g.probes = append(g.probes, d) g.mu.Unlock() + g.cond.Signal() } -// adjust evaluates the interval's probe samples and demand and moves the -// limit accordingly. It is called once per moverAdjustInterval by the -// governor loop, and directly by tests. -func (g *moverGovernor) adjust() { +// evaluate closes the measurement window if it is ready and moves the +// limit according to what it showed. It is called periodically by run, and +// directly by tests. +func (g *moverGovernor) evaluate(now time.Time) { g.mu.Lock() defer g.mu.Unlock() + g.accrueLocked(now) - probes := g.probes - g.probes = nil - waited := g.waited - g.waited = false - - if len(probes) < moverMinProbes { + elapsed := now.Sub(g.windowStart) + if elapsed < moverEvalInterval { + return + } + sparse := g.completions < moverMinCompletions + if sparse && elapsed < moverMaxEvalInterval { return } - slices.Sort(probes) - med := probes[len(probes)/2] - g.lastMed = med + rate := float64(g.completions) / elapsed.Seconds() + util := g.busy.Seconds() / (float64(g.limit) * elapsed.Seconds()) + g.windowStart, g.busy, g.completions = now, 0, 0 - if g.baseline == 0 || med < g.baseline { - // First estimate, or the filesystem is faster than we thought: - // adopt the new unloaded latency immediately. - g.baseline = med - } else if !waited { - // No demand this interval, so the latency was unloaded: let the - // baseline drift toward it to track a filesystem that slowed down. - g.baseline += time.Duration(float64(med-g.baseline) * moverBaselineDrift) + if sparse || util < moverMinUtilization { + // Demand-limited: the limit wasn't what bounded throughput, so + // the window says nothing about capacity. Any probe in flight + // stays in flight until a busy window judges it. + return } - ratio := float64(med) / float64(g.baseline) switch { - case ratio > moverDecreaseAbove: - newLimit := max(g.minLimit, g.limit*moverDecreaseNumer/moverDecreaseDenom) - if newLimit < g.limit { - g.limit = newLimit - g.decreases++ + case g.committedRate == 0: + // First busy window: this is the starting operating point. + g.commitLocked(rate) + g.probeUpLocked() + + case g.limit > g.committedLimit: + ideal := float64(g.limit)/float64(g.committedLimit) - 1 + if rate/g.committedRate-1 >= moverGainFraction*ideal { + g.commitLocked(rate) + g.probeUpLocked() + } else { + g.rejectLocked(now) } - case ratio < moverIncreaseBelow && waited: - newLimit := min(g.maxLimit, g.limit+max(1, g.limit/moverIncreaseDivisor)) - if newLimit > g.limit { - g.limit = newLimit - g.increases++ - g.cond.Broadcast() + + case g.limit < g.committedLimit: + if rate >= g.committedRate*(1-moverLossTolerance) { + g.commitLocked(rate) + g.probeDownLocked() + } else { + g.rejectLocked(now) + } + + default: + if rate < g.committedRate*(1-moverDropFraction) { + // Throughput fell at a limit that used to deliver more. Take + // the new rate as the truth and test whether the concurrency + // is now excess. + g.committedRate = rate + g.probeDownLocked() + return + } + g.committedRate = (g.committedRate + rate) / 2 + if now.Before(g.holdUntil) { + return } + g.probes++ + if g.probes%moverDownProbeEvery == 0 { + g.probeDownLocked() + } else { + g.probeUpLocked() + } + } +} + +// commitLocked records the current limit and the rate measured at it as the +// operating point to compare future probes against. +func (g *moverGovernor) commitLocked(rate float64) { + g.committedLimit = g.limit + g.committedRate = rate +} + +func (g *moverGovernor) stepLocked() int { + return max(1, int(float64(g.committedLimit)*moverStepFraction)) +} + +// probeUpLocked raises the limit one step above the committed point, if the +// maximum allows. +func (g *moverGovernor) probeUpLocked() { + n := min(g.maxLimit, g.committedLimit+g.stepLocked()) + if n > g.limit { + g.limit = n + g.increases++ + g.cond.Broadcast() + } +} + +// probeDownLocked lowers the limit one step below the committed point, if +// the minimum allows. +func (g *moverGovernor) probeDownLocked() { + n := max(g.minLimit, g.committedLimit-g.stepLocked()) + if n < g.limit { + g.limit = n + g.decreases++ + } +} + +// rejectLocked abandons the probe in flight, returning to the committed +// limit, and holds there for a while before probing again. +func (g *moverGovernor) rejectLocked(now time.Time) { + g.rejected++ + g.holdUntil = now.Add(moverHoldAfterReject) + switch { + case g.limit > g.committedLimit: + g.limit = g.committedLimit + g.decreases++ + case g.limit < g.committedLimit: + g.limit = g.committedLimit + g.increases++ + g.cond.Broadcast() } } -// run calls adjust every moverAdjustInterval until ctx is done. +// run calls evaluate periodically until ctx is done. func (g *moverGovernor) run(ctx context.Context) { - t := time.NewTicker(moverAdjustInterval) + t := time.NewTicker(moverEvalInterval / 2) defer t.Stop() for { select { case <-ctx.Done(): return - case <-t.C: - g.adjust() + case now := <-t.C: + g.evaluate(now) } } } @@ -213,19 +308,22 @@ func (g *moverGovernor) run(ctx context.Context) { // moverStats is a snapshot of the governor's state for metrics. type moverStats struct { limit, active int - baseline, lastMedian time.Duration + committedLimit int + committedRate float64 increases, decreases int + rejected int } func (g *moverGovernor) stats() moverStats { g.mu.Lock() defer g.mu.Unlock() return moverStats{ - limit: g.limit, - active: g.active, - baseline: g.baseline, - lastMedian: g.lastMed, - increases: g.increases, - decreases: g.decreases, + limit: g.limit, + active: g.active, + committedLimit: g.committedLimit, + committedRate: g.committedRate, + increases: g.increases, + decreases: g.decreases, + rejected: g.rejected, } } diff --git a/gocache/gocached/movergov_test.go b/gocache/gocached/movergov_test.go index 2807621..14ae85f 100644 --- a/gocache/gocached/movergov_test.go +++ b/gocache/gocached/movergov_test.go @@ -3,207 +3,296 @@ package gocached import ( "context" "errors" + "math/rand/v2" "testing" - "testing/synctest" "time" ) -// feedProbes records n probe samples of latency d. -func feedProbes(g *moverGovernor, n int, d time.Duration) { - for range n { - g.observe(1<<10, d) - } +// govSim drives a moverGovernor through synthetic measurement windows +// against a model of the main blob directory's throughput as a function of +// concurrency, without goroutines or real time. +type govSim struct { + t *testing.T + g *moverGovernor + now time.Time + rng *rand.Rand + noise float64 // relative jitter applied to each window's copy count + + // rateAt returns the copies/s the directory delivers at the given + // concurrency when the slots are kept busy. + rateAt func(limit int) float64 } -func TestMoverGovernorGrowsOnlyWithDemand(t *testing.T) { - g := newMoverGovernor(8, 256) +func newGovSim(t *testing.T, minLimit, maxLimit int, rateAt func(int) float64) *govSim { + t.Helper() + g := newMoverGovernor(minLimit, maxLimit) + now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + g.clock = func() time.Time { return now } + g.windowStart, g.lastAccrue = now, now + return &govSim{t: t, g: g, now: now, rng: rand.New(rand.NewPCG(1, 2)), rateAt: rateAt} +} - // Unloaded latency with no demand: learns the baseline, holds the limit. - feedProbes(g, 20, 10*time.Millisecond) - g.adjust() - if st := g.stats(); st.limit != 8 || st.baseline != 10*time.Millisecond { - t.Fatalf("after idle interval: %+v; want limit 8, baseline 10ms", st) +// knee returns a rateAt where each slot delivers perSlot copies/s until the +// directory saturates at ceiling copies/s, i.e. the knee is at +// ceiling/perSlot slots. +func knee(perSlot, ceiling float64) func(int) float64 { + return func(limit int) float64 { + return min(float64(limit)*perSlot, ceiling) } +} - // Same latency but movers waited for a slot: grow additively. +// step advances one evaluation interval in which the slots were busy for +// the given fraction of the limit's capacity and copies completed at the +// model's rate for that utilization. +func (s *govSim) step(utilization float64) { + g := s.g g.mu.Lock() - g.waited = true - g.mu.Unlock() - feedProbes(g, 20, 11*time.Millisecond) - g.adjust() - if st := g.stats(); st.limit != 10 || st.increases != 1 { - t.Fatalf("after loaded low-latency interval: %+v; want limit 10", st) + limit := g.limit + rate := s.rateAt(limit) * utilization + if s.noise > 0 { + rate *= 1 + s.noise*(2*s.rng.Float64()-1) } + g.completions += int(rate*moverEvalInterval.Seconds() + 0.5) + g.busy += time.Duration(float64(limit) * utilization * float64(moverEvalInterval)) + g.mu.Unlock() + s.now = s.now.Add(moverEvalInterval) + g.evaluate(s.now) +} - // Keep growing while latency stays near baseline; doubling takes about - // moverIncreaseDivisor intervals. - for range 20 { - g.mu.Lock() - g.waited = true - g.mu.Unlock() - feedProbes(g, 20, 12*time.Millisecond) - g.adjust() - } - if st := g.stats(); st.limit <= 40 { - t.Fatalf("after sustained headroom: %+v; want limit well above 40", st) +// run advances d of fully busy simulated time. +func (s *govSim) run(d time.Duration) { + for range int(d / moverEvalInterval) { + s.step(1) } } -func TestMoverGovernorBacksOffOnLatency(t *testing.T) { - g := newMoverGovernor(8, 256) - g.mu.Lock() - g.limit = 64 - g.mu.Unlock() - feedProbes(g, 20, 10*time.Millisecond) - g.adjust() // learn baseline 10ms +func (s *govSim) limit() int { + st := s.g.stats() + return st.limit +} - // Median latency at 3x baseline while saturated: multiplicative decrease. - g.mu.Lock() - g.waited = true - g.mu.Unlock() - feedProbes(g, 20, 30*time.Millisecond) - g.adjust() - if st := g.stats(); st.limit != 48 || st.decreases != 1 { - t.Fatalf("after saturated interval: %+v; want limit 48", st) - } +func (s *govSim) committed() int { + return s.g.stats().committedLimit +} - // Saturated intervals must not drag the baseline up: it stays at the - // unloaded 10ms so continued high latency keeps shrinking the limit. - if st := g.stats(); st.baseline != 10*time.Millisecond { - t.Fatalf("baseline moved under load: %+v", st) +func TestMoverGovernorClimbsToKnee(t *testing.T) { + // 10 copies/s per slot until 400 copies/s: the knee is 40 slots. + s := newGovSim(t, 8, 256, knee(10, 400)) + s.run(5 * time.Minute) + st := s.g.stats() + if st.committedLimit < 32 || st.committedLimit > 52 { + t.Fatalf("committed limit = %d, want near the knee at 40; stats %+v", st.committedLimit, st) } - for range 10 { - g.mu.Lock() - g.waited = true - g.mu.Unlock() - feedProbes(g, 20, 30*time.Millisecond) - g.adjust() + if st.committedRate < 350 || st.committedRate > 420 { + t.Fatalf("committed rate = %.0f, want near the 400 copies/s ceiling", st.committedRate) } - if st := g.stats(); st.limit != 8 { - t.Fatalf("after sustained saturation: %+v; want floor of 8", st) + if st.rejected == 0 { + t.Fatal("never rejected a probe; should be bouncing off the knee") + } + // It must have stopped well short of the maximum: extra slots past + // the knee buy nothing. + if s.limit() > 64 { + t.Fatalf("limit = %d, want it held near the knee", s.limit()) } } -func TestMoverGovernorHoldsInDeadBand(t *testing.T) { - g := newMoverGovernor(8, 256) - g.mu.Lock() - g.limit = 32 - g.mu.Unlock() - feedProbes(g, 20, 10*time.Millisecond) - g.adjust() +func TestMoverGovernorReachesMaxWithHeadroom(t *testing.T) { + s := newGovSim(t, 8, 256, knee(10, 1e9)) + s.run(5 * time.Minute) + if got := s.committed(); got != 256 { + t.Fatalf("committed limit = %d, want the max 256 when throughput keeps scaling", got) + } +} - // Between the thresholds (1.5x to 2x baseline) the limit holds even - // under demand. - g.mu.Lock() - g.waited = true - g.mu.Unlock() - feedProbes(g, 20, 17*time.Millisecond) - g.adjust() - if st := g.stats(); st.limit != 32 { - t.Fatalf("in dead band: %+v; want limit 32", st) +func TestMoverGovernorStaysAtFloorWhenSaturated(t *testing.T) { + // Already saturated at the floor: no probe up should stick. + s := newGovSim(t, 8, 256, knee(10, 60)) + s.run(5 * time.Minute) + if got := s.committed(); got != 8 { + t.Fatalf("committed limit = %d, want the floor 8 when more slots add nothing", got) + } + if st := s.g.stats(); st.rejected == 0 || st.increases == 0 { + t.Fatalf("expected repeated rejected upward probes, got %+v", st) + } +} + +func TestMoverGovernorIgnoresDemandLimitedWindows(t *testing.T) { + s := newGovSim(t, 8, 256, knee(10, 1e9)) + // Slots half idle: throughput is bounded by demand, not the limit. + for range 300 { + s.step(0.5) + } + if st := s.g.stats(); st.limit != 8 || st.committedRate != 0 { + t.Fatalf("demand-limited windows moved the governor: %+v", st) } + // Once busy, it climbs. + s.run(time.Minute) + if s.committed() <= 8 { + t.Fatalf("did not climb once busy: %+v", s.g.stats()) + } +} - // Too few probes: hold, and don't learn a baseline from them. - g2 := newMoverGovernor(8, 256) - feedProbes(g2, moverMinProbes-1, time.Millisecond) - g2.adjust() - if st := g2.stats(); st.limit != 8 || st.baseline != 0 { - t.Fatalf("with too few probes: %+v; want no change", st) +func TestMoverGovernorIgnoresSparseWindows(t *testing.T) { + // 1 copy/s is far below moverMinCompletions per window: windows stretch + // to moverMaxEvalInterval and are then discarded as sparse. + s := newGovSim(t, 8, 256, func(int) float64 { return 1 }) + s.run(5 * time.Minute) + if st := s.g.stats(); st.limit != 8 || st.committedRate != 0 || st.increases != 0 { + t.Fatalf("sparse windows moved the governor: %+v", st) } } -func TestMoverGovernorBaselineDriftsOnlyWhenIdle(t *testing.T) { - g := newMoverGovernor(8, 256) - feedProbes(g, 20, 10*time.Millisecond) - g.adjust() +func TestMoverGovernorTrimsWhenDirectorySlows(t *testing.T) { + ceiling := 2000.0 + s := newGovSim(t, 8, 256, func(limit int) float64 { + return min(float64(limit)*10, ceiling) + }) + s.run(5 * time.Minute) + if got := s.committed(); got < 160 { + t.Fatalf("committed limit = %d, want near the knee at 200 before the slowdown", got) + } - // Idle intervals at a higher latency: the baseline drifts up toward - // the new unloaded latency. - for range 100 { - feedProbes(g, 20, 20*time.Millisecond) - g.adjust() + // The directory gets much slower: the knee moves to 30 slots. + ceiling = 300 + s.run(5 * time.Minute) + st := s.g.stats() + if st.committedLimit > 45 { + t.Fatalf("committed limit = %d after slowdown, want trimmed to near 30; %+v", st.committedLimit, st) } - if st := g.stats(); st.baseline < 19*time.Millisecond { - t.Fatalf("baseline didn't drift toward new idle latency: %+v", st) + if st.committedRate > 330 { + t.Fatalf("committed rate = %.0f after slowdown, want near 300", st.committedRate) } +} - // A faster interval snaps the baseline down immediately. - feedProbes(g, 20, 5*time.Millisecond) - g.adjust() - if st := g.stats(); st.baseline != 5*time.Millisecond { - t.Fatalf("baseline didn't snap down: %+v", st) +func TestMoverGovernorRecoversWhenDirectorySpeedsUp(t *testing.T) { + ceiling := 300.0 + s := newGovSim(t, 8, 256, func(limit int) float64 { + return min(float64(limit)*10, ceiling) + }) + s.run(5 * time.Minute) + if got := s.committed(); got > 45 { + t.Fatalf("committed limit = %d, want near 30 before speedup", got) + } + ceiling = 2000 + s.run(5 * time.Minute) + if got := s.committed(); got < 160 { + t.Fatalf("committed limit = %d after speedup, want climbed back toward 200; %+v", got, s.g.stats()) } } -func TestMoverGovernorIgnoresLargeCopies(t *testing.T) { - g := newMoverGovernor(8, 256) - for range 20 { - g.observe(moverProbeMaxSize+1, time.Second) +func TestMoverGovernorTracksKneeUnderNoise(t *testing.T) { + s := newGovSim(t, 8, 256, knee(10, 400)) + s.noise = 0.15 + s.run(10 * time.Minute) + // Sample the committed limit over a further stretch: it should hover + // around the knee rather than wander off to either bound. + lo, hi := 1<<30, 0 + for range 150 { + s.run(moverEvalInterval * 5) + c := s.committed() + lo, hi = min(lo, c), max(hi, c) } - g.adjust() - if st := g.stats(); st.baseline != 0 { - t.Fatalf("large copies counted as probes: %+v", st) + if lo < 20 || hi > 80 { + t.Fatalf("committed limit wandered to [%d, %d] under noise; want it near the knee at 40", lo, hi) } } func TestMoverGovernorAcquireRelease(t *testing.T) { - synctest.Test(t, func(t *testing.T) { - g := newMoverGovernor(2, 4) - ctx := t.Context() - if err := g.acquire(ctx); err != nil { - t.Fatal(err) - } - if err := g.acquire(ctx); err != nil { - t.Fatal(err) - } - - done := make(chan error, 1) - - // assertBlocked fails the test if the pending acquire has - // already returned once all goroutines in the bubble are blocked. - assertBlocked := func() { - t.Helper() - synctest.Wait() - select { - case err := <-done: - t.Fatalf("acquire over limit returned early: %v", err) - default: - } - } + g := newMoverGovernor(2, 4) + ctx := context.Background() + if err := g.acquire(ctx); err != nil { + t.Fatal(err) + } + if err := g.acquire(ctx); err != nil { + t.Fatal(err) + } + if st := g.stats(); st.active != 2 { + t.Fatalf("active = %d, want 2", st.active) + } - // A third acquire blocks at the limit and records demand. - go func() { done <- g.acquire(ctx) }() - assertBlocked() - g.release() - synctest.Wait() - if err := <-done; err != nil { + // A third acquire blocks at the limit. + done := make(chan error, 1) + go func() { done <- g.acquire(ctx) }() + select { + case err := <-done: + t.Fatalf("acquire over limit returned early: %v", err) + case <-time.After(50 * time.Millisecond): + } + g.release(true) + select { + case err := <-done: + if err != nil { t.Fatal(err) } - g.mu.Lock() - waited := g.waited - g.mu.Unlock() - if !waited { - t.Fatal("blocked acquire didn't record demand") - } + case <-time.After(5 * time.Second): + t.Fatal("acquire didn't proceed after release") + } - // Raising the limit admits a waiter without a release. - go func() { done <- g.acquire(ctx) }() - assertBlocked() - g.mu.Lock() - g.limit = 3 - g.cond.Broadcast() - g.mu.Unlock() - if err := <-done; err != nil { + // Only successful copies count as completions. + g.release(false) + g.release(true) + g.mu.Lock() + completions := g.completions + g.mu.Unlock() + if completions != 2 { + t.Fatalf("completions = %d, want 2", completions) + } + + // Raising the limit admits a waiter without a release. + for range 2 { + if err := g.acquire(ctx); err != nil { t.Fatal(err) } + } + go func() { done <- g.acquire(ctx) }() + select { + case err := <-done: + t.Fatalf("acquire over limit returned early: %v", err) + case <-time.After(50 * time.Millisecond): + } + g.mu.Lock() + g.limit = 3 + g.cond.Broadcast() + g.mu.Unlock() + if err := <-done; err != nil { + t.Fatal(err) + } - // A blocked acquire aborts when its context is canceled. - cctx, cancel := context.WithCancel(ctx) - go func() { done <- g.acquire(cctx) }() - assertBlocked() - cancel() - if err := <-done; !errors.Is(err, context.Canceled) { + // A blocked acquire aborts when its context is canceled. + cctx, cancel := context.WithCancel(ctx) + go func() { done <- g.acquire(cctx) }() + time.Sleep(20 * time.Millisecond) + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { t.Fatalf("canceled acquire = %v, want context.Canceled", err) } - }) + case <-time.After(5 * time.Second): + t.Fatal("canceled acquire didn't return") + } +} + +func TestMoverGovernorBusyAccounting(t *testing.T) { + g := newMoverGovernor(4, 8) + now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + g.clock = func() time.Time { return now } + g.windowStart, g.lastAccrue = now, now + ctx := context.Background() + + // Two slots held for one second, then one of them for another second: + // three slot-seconds of busy time out of a possible eight. + g.acquire(ctx) + g.acquire(ctx) + now = now.Add(time.Second) + g.release(true) + now = now.Add(time.Second) + g.release(true) + g.mu.Lock() + g.accrueLocked(now) + busy := g.busy + g.mu.Unlock() + if busy != 3*time.Second { + t.Fatalf("busy = %v, want 3s", busy) + } } diff --git a/gocache/gocached/putqueue.go b/gocache/gocached/putqueue.go index 0f8dfa6..d6c535d 100644 --- a/gocache/gocached/putqueue.go +++ b/gocache/gocached/putqueue.go @@ -224,9 +224,14 @@ func newPutQueue(srv *Server, dir string) *putQueue { // One mover goroutine is started per slot the governor could ever grant; // each takes a governor slot before copying, so the number of copies in // flight follows the governor's current limit, not the goroutine count. +// Only the first minLimit of them also execute cleanup intent deletions, +// which bounds that unlink traffic on the main directory to the same +// concurrency as a fixed-mover server would have produced; letting every +// goroutine do them turned each metadata flush into a burst of hundreds of +// concurrent unlinks that doubled copy latency. func (q *putQueue) start(ctx context.Context) { - for range q.gov.maxLimit { - go q.moverLoop(ctx) + for i := range q.gov.maxLimit { + go q.moverLoop(ctx, i < q.gov.minLimit) } go q.gov.run(ctx) go q.flusherLoop(ctx) @@ -440,17 +445,22 @@ func (q *putQueue) spoolBlob(size int64, r io.Reader) (diskSize int64, path stri } // moverLoop copies spooled blobs into the main blob directory and passes -// them on to the flusher. It also executes the post-commit cleanup intent -// deletions the flusher hands back. Several movers run concurrently since -// the main directory may be a high-latency network filesystem. -func (q *putQueue) moverLoop(ctx context.Context) { +// them on to the flusher. If deleteIntents is set it also executes the +// post-commit cleanup intent deletions the flusher hands back. Several +// movers run concurrently since the main directory may be a high-latency +// network filesystem. +func (q *putQueue) moverLoop(ctx context.Context, deleteIntents bool) { + intentDelCh := q.intentDelCh + if !deleteIntents { + intentDelCh = nil // never selected + } for { select { case <-ctx.Done(): return case p := <-q.moverCh: q.moveToFlush(ctx, p) - case path := <-q.intentDelCh: + case path := <-intentDelCh: if err := os.Remove(path); err != nil && !os.IsNotExist(err) { q.srv.logf("put-queue: removing cleanup intent %v: %v", path, err) } @@ -487,14 +497,15 @@ func (q *putQueue) moveToFlush(ctx context.Context, p *pendingPut) { // copyUnderGovernor performs one attempt at p's intent write and blob copy // into the main blob directory while holding a mover governor slot, and -// reports the attempt's latency to the governor and metrics. -func (q *putQueue) copyUnderGovernor(ctx context.Context, p *pendingPut) error { +// records the attempt's outcome with the governor and its latency in the +// metrics. +func (q *putQueue) copyUnderGovernor(ctx context.Context, p *pendingPut) (err error) { if err := q.gov.acquire(ctx); err != nil { return err } - defer q.gov.release() + defer func() { q.gov.release(err == nil) }() start := time.Now() - err := q.writeCleanupIntent(p) + err = q.writeCleanupIntent(p) if err == nil { err = q.copyToMain(p) } @@ -502,8 +513,6 @@ func (q *putQueue) copyUnderGovernor(ctx context.Context, p *pendingPut) error { result := "ok" if err != nil { result = "error" - } else { - q.gov.observe(p.storedSize, d) } if q.srv.putCopyDuration != nil { q.srv.putCopyDuration.WithLabelValues(result).Observe(d.Seconds())