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())