diff --git a/cmd/gocached/gocached.go b/cmd/gocached/gocached.go index b31e662..b4e759f 100644 --- a/cmd/gocached/gocached.go +++ b/cmd/gocached/gocached.go @@ -13,7 +13,6 @@ import ( "net" "net/http" "os" - "strconv" "strings" "time" @@ -28,7 +27,7 @@ var ( hotDir = flag.String("hot-dir", "", "if non-empty, enable storage tiering with this directory as a fast tier (e.g. local NVMe) holding a bounded copy of recently used blobs; the cache directory remains the source of truth") hotCapacity = flag.Int("hot-capacity-gb", 600, "maximum size of the hot tier directory in GiB; only used with --hot-dir") putSpoolCap = flag.Int("put-spool-gb", 0, "capacity in GiB of the put queue's spooled lane: accepted PUT bytes waiting on local disk (the hot tier's filesystem, with --hot-dir) for a copy into the cache directory; 0 means one eighth of that filesystem's size") - putMovers = flag.String("put-movers", "", "min,max bounds on the adaptive number of concurrent copies of spooled blobs into the cache directory, or a bare N to pin it at N; empty means the defaults (8,256)") + putMovers = flag.Int("put-movers", 0, "number of concurrent copies of spooled blobs into the cache directory; 0 means the default (128)") verbose = flag.Bool("verbose", false, "be verbose") listen = flag.String("listen", ":31364", "listen address for the build-facing HTTP server") debugListen = flag.String("debug-listen", "", "if non-empty, listen address for the debug HTTP server (pprof, metrics, etc)") @@ -103,12 +102,8 @@ func main() { if *putSpoolCap > 0 { opts = append(opts, gocached.WithPutSpoolCapacity(int64(*putSpoolCap)<<30)) } - if *putMovers != "" { - minMovers, maxMovers, err := parsePutMovers(*putMovers) - if err != nil { - log.Fatalf("--put-movers: %v", err) - } - opts = append(opts, gocached.WithPutMoverLimits(minMovers, maxMovers)) + if *putMovers > 0 { + opts = append(opts, gocached.WithPutMovers(*putMovers)) } if *peerTag != "" || *peerList != "" { @@ -179,18 +174,3 @@ func main() { log.Printf("gocached: listening on %s ...", *listen) log.Fatal(http.ListenAndServe(*listen, srv)) } - -// parsePutMovers parses the --put-movers flag: "min,max" bounds on the -// adaptive mover limit, or a bare "N" meaning a fixed limit of N. -func parsePutMovers(s string) (minMovers, maxMovers int, err error) { - lo, hi, ok := strings.Cut(s, ",") - if !ok { - hi = lo - } - minMovers, err1 := strconv.Atoi(strings.TrimSpace(lo)) - maxMovers, err2 := strconv.Atoi(strings.TrimSpace(hi)) - if err1 != nil || err2 != nil || minMovers <= 0 || maxMovers < minMovers { - return 0, 0, fmt.Errorf("want N or min,max with 0 < min <= max; got %q", s) - } - return minMovers, maxMovers, nil -} diff --git a/cmd/gocached/gocached_test.go b/cmd/gocached/gocached_test.go index d9f8063..30dbde8 100644 --- a/cmd/gocached/gocached_test.go +++ b/cmd/gocached/gocached_test.go @@ -57,31 +57,3 @@ func TestInvalidJWTClaimFlags(t *testing.T) { } } } - -func TestParsePutMovers(t *testing.T) { - for _, tc := range []struct { - in string - min, max int - wantErr bool - }{ - {in: "8,256", min: 8, max: 256}, - {in: "32", min: 32, max: 32}, - {in: " 4 , 4 ", min: 4, max: 4}, - {in: "0", wantErr: true}, - {in: "0,8", wantErr: true}, - {in: "16,8", wantErr: true}, - {in: "8,", wantErr: true}, - {in: "", wantErr: true}, - {in: "x", wantErr: true}, - {in: "8,16,32", wantErr: true}, - } { - lo, hi, err := parsePutMovers(tc.in) - if (err != nil) != tc.wantErr { - t.Errorf("parsePutMovers(%q) err = %v, wantErr %v", tc.in, err, tc.wantErr) - continue - } - if !tc.wantErr && (lo != tc.min || hi != tc.max) { - t.Errorf("parsePutMovers(%q) = %d,%d; want %d,%d", tc.in, lo, hi, tc.min, tc.max) - } - } -} diff --git a/gocache/gocached/gocached.go b/gocache/gocached/gocached.go index f37cf46..aa38189 100644 --- a/gocache/gocached/gocached.go +++ b/gocache/gocached/gocached.go @@ -448,7 +448,7 @@ func (srv *Server) start() error { pipelineBuckets := prometheus.ExponentialBuckets(0.001, 2, 16) srv.putCopyDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{ Name: "gocached_put_queue_copy_duration_seconds", - Help: "wall time of each attempt to write a spooled blob's cleanup intent and copy it into the main blob directory, labeled by result: ok or error; small blobs' latency here drives the adaptive mover limit", + Help: "wall time of each attempt to write a spooled blob's cleanup intent and copy it into the main blob directory, labeled by result: ok or error", Buckets: pipelineBuckets, }, []string{"result"}) srv.putFlushDuration = prometheus.NewHistogram(prometheus.HistogramOpts{ @@ -599,24 +599,14 @@ func (srv *Server) start() error { func() float64 { return float64(srv.putq.pendingInline()) }}, {"gocached_put_queue_pending_inline_cap", "capacity of the inline lane in pending PUTs; compare with gocached_put_queue_pending_inline", func() float64 { return putQueuePendingInlineCap }}, - {"gocached_put_queue_mover_backlog", "spooled blobs not yet picked up by a mover goroutine, in addition to those a mover holds while waiting for a governor slot; nonzero while the movers are the constraint", + {"gocached_put_queue_mover_backlog", "spooled blobs waiting for a mover to start copying them into the main blob directory; nonzero while the movers are the constraint", 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", - 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", - 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", - 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", - func() float64 { return float64(srv.putq.gov.stats().decreases) }}, + {"gocached_put_queue_movers", "number of mover goroutines copying spooled blobs into the main blob directory", + func() float64 { return float64(srv.putq.numMovers) }}, + {"gocached_put_queue_movers_active", "copies into the main blob directory currently in flight; pinned at gocached_put_queue_movers with a nonzero mover backlog means the mover count is what bounds the spooled lane", + func() float64 { return float64(srv.putq.activeCopies.Load()) }}, } { reg.MustRegister(prometheus.NewGaugeFunc(prometheus.GaugeOpts{Name: g.name, Help: g.help}, g.fn)) } @@ -811,16 +801,12 @@ 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 -// 256). See gocached_put_queue_movers_* metrics for the observed behavior. -func WithPutMoverLimits(minLimit, maxLimit int) ServerOption { +// WithPutMovers sets how many goroutines copy spooled blobs into the main +// blob directory concurrently. Zero or negative uses the default, 128, which +// suits a network filesystem; there is little reason to change it. +func WithPutMovers(n int) ServerOption { return func(srv *Server) { - srv.putMoversMin = minLimit - srv.putMoversMax = maxLimit + srv.putMovers = max(n, 0) } } @@ -1040,8 +1026,7 @@ type Server struct { hot *hotIndex putq *putQueue putSpoolCap int64 // byte capacity of the put queue's spooled lane; 0 means derive from the spool filesystem - putMoversMin int // floor and starting point of the adaptive mover limit; 0 means default - putMoversMax int // ceiling of the adaptive mover limit; 0 means default + putMovers int // number of put-queue mover goroutines; 0 means defaultPutMovers verbose bool logf logger.Logf clock func() time.Time // if non-nil, alternate time.Now for testing diff --git a/gocache/gocached/movergov.go b/gocache/gocached/movergov.go deleted file mode 100644 index e8c42ce..0000000 --- a/gocache/gocached/movergov.go +++ /dev/null @@ -1,231 +0,0 @@ -package gocached - -import ( - "context" - "slices" - "sync" - "time" -) - -// Defaults and tuning parameters for [moverGovernor]. -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. - 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 -) - -// 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. -// -// 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. -type moverGovernor struct { - minLimit int - maxLimit int - - 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 - - increases, decreases int // adjustment counts for metrics -} - -func newMoverGovernor(minLimit, maxLimit int) *moverGovernor { - if minLimit <= 0 { - minLimit = defaultMinPutMovers - } - if maxLimit < minLimit { - maxLimit = max(minLimit, defaultMaxPutMovers) - } - g := &moverGovernor{ - minLimit: minLimit, - maxLimit: maxLimit, - limit: minLimit, - } - g.cond = sync.NewCond(&g.mu) - return g -} - -// 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 { - // Wake waiters on shutdown, since cond.Wait can't select on ctx. - stop := context.AfterFunc(ctx, func() { - g.mu.Lock() - g.cond.Broadcast() - g.mu.Unlock() - }) - defer stop() - - g.mu.Lock() - defer g.mu.Unlock() - for g.active >= g.limit { - if err := ctx.Err(); err != nil { - return err - } - g.waited = true - g.cond.Wait() - } - if err := ctx.Err(); err != nil { - return err - } - g.active++ - return nil -} - -// release returns a slot taken by acquire. -func (g *moverGovernor) release() { - g.mu.Lock() - 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 - } - g.mu.Lock() - g.probes = append(g.probes, d) - g.mu.Unlock() -} - -// 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() { - g.mu.Lock() - defer g.mu.Unlock() - - probes := g.probes - g.probes = nil - waited := g.waited - g.waited = false - - if len(probes) < moverMinProbes { - return - } - slices.Sort(probes) - med := probes[len(probes)/2] - g.lastMed = med - - 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) - } - - 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 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() - } - } -} - -// run calls adjust every moverAdjustInterval until ctx is done. -func (g *moverGovernor) run(ctx context.Context) { - t := time.NewTicker(moverAdjustInterval) - defer t.Stop() - for { - select { - case <-ctx.Done(): - return - case <-t.C: - g.adjust() - } - } -} - -// moverStats is a snapshot of the governor's state for metrics. -type moverStats struct { - limit, active int - baseline, lastMedian time.Duration - increases, decreases 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, - } -} diff --git a/gocache/gocached/movergov_test.go b/gocache/gocached/movergov_test.go deleted file mode 100644 index 2807621..0000000 --- a/gocache/gocached/movergov_test.go +++ /dev/null @@ -1,209 +0,0 @@ -package gocached - -import ( - "context" - "errors" - "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) - } -} - -func TestMoverGovernorGrowsOnlyWithDemand(t *testing.T) { - g := newMoverGovernor(8, 256) - - // 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) - } - - // Same latency but movers waited for a slot: grow additively. - 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) - } - - // 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) - } -} - -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 - - // 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) - } - - // 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) - } - for range 10 { - g.mu.Lock() - g.waited = true - g.mu.Unlock() - feedProbes(g, 20, 30*time.Millisecond) - g.adjust() - } - if st := g.stats(); st.limit != 8 { - t.Fatalf("after sustained saturation: %+v; want floor of 8", st) - } -} - -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() - - // 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) - } - - // 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 TestMoverGovernorBaselineDriftsOnlyWhenIdle(t *testing.T) { - g := newMoverGovernor(8, 256) - feedProbes(g, 20, 10*time.Millisecond) - g.adjust() - - // 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() - } - if st := g.stats(); st.baseline < 19*time.Millisecond { - t.Fatalf("baseline didn't drift toward new idle latency: %+v", st) - } - - // 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 TestMoverGovernorIgnoresLargeCopies(t *testing.T) { - g := newMoverGovernor(8, 256) - for range 20 { - g.observe(moverProbeMaxSize+1, time.Second) - } - g.adjust() - if st := g.stats(); st.baseline != 0 { - t.Fatalf("large copies counted as probes: %+v", st) - } -} - -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: - } - } - - // 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 { - t.Fatal(err) - } - g.mu.Lock() - waited := g.waited - g.mu.Unlock() - if !waited { - t.Fatal("blocked acquire didn't record demand") - } - - // 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 { - 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) { - t.Fatalf("canceled acquire = %v, want context.Canceled", err) - } - }) -} diff --git a/gocache/gocached/putqueue.go b/gocache/gocached/putqueue.go index 0f8dfa6..78de358 100644 --- a/gocache/gocached/putqueue.go +++ b/gocache/gocached/putqueue.go @@ -14,6 +14,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/pierrec/lz4/v4" @@ -104,6 +105,18 @@ const ( // putFlushInterval. putFlushBatchCap = 512 + // defaultPutMovers is how many goroutines copy spooled blobs into the + // main blob directory when [WithPutMovers] isn't used. The main + // directory may be a network filesystem (in production, an NFS mount + // of S3 Files) where each copy is several round trips, so throughput + // comes from concurrency, not from any one copy finishing faster. + // Production peaks need roughly 55 concurrent copies (about 330 + // spooled PUTs/s at 100ms each plus cleanup unlinks); 128 leaves + // headroom. When the filesystem saturates, extra movers just wait in + // the kernel's NFS request queue, costing a blocked OS thread each. + // A local-disk main directory handles this concurrency easily. + defaultPutMovers = 128 + // putCopyRetries is how many times a mover attempts a blob's copy into // the main blob directory before the pending PUT is dropped. putCopyRetries = 3 @@ -168,9 +181,8 @@ type putQueue struct { countSem *semaphore.Weighted inlineSem *semaphore.Weighted - // gov sets how many spooled-blob copies into the main blob directory - // run concurrently, adapting to the directory's observed latency. - gov *moverGovernor + numMovers int // number of mover goroutines; see defaultPutMovers + activeCopies atomic.Int64 // copies into the main blob directory in flight // mu is a leaf mutex: no other lock is acquired while holding it. mu sync.Mutex @@ -208,7 +220,7 @@ func newPutQueue(srv *Server, dir string) *putQueue { bytesSem: semaphore.NewWeighted(spoolCap), countSem: semaphore.NewWeighted(putQueuePendingCountCap), inlineSem: semaphore.NewWeighted(putQueuePendingInlineCap), - gov: newMoverGovernor(srv.putMoversMin, srv.putMoversMax), + numMovers: cmp.Or(srv.putMovers, defaultPutMovers), pending: make(map[actionKey]*pendingPut), moverCh: make(chan *pendingPut, putQueuePendingCountCap), flushCh: make(chan *pendingPut, putQueuePendingCountCap+putQueuePendingInlineCap), @@ -220,15 +232,10 @@ func newPutQueue(srv *Server, dir string) *putQueue { // cleanup intent sweeper. It is not called under disableBackgroundLoops; // tests drive the pipeline with [Server.drainPendingPuts] and // [putQueue.sweepCleanupIntents] instead. -// -// 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. func (q *putQueue) start(ctx context.Context) { - for range q.gov.maxLimit { + for range q.numMovers { go q.moverLoop(ctx) } - go q.gov.run(ctx) go q.flusherLoop(ctx) go q.runCleanupSweepLoop(ctx) } @@ -471,7 +478,7 @@ func (q *putQueue) moveToFlush(ctx context.Context, p *pendingPut) { case <-time.After(time.Second << (try - 1)): } } - err := q.copyUnderGovernor(ctx, p) + err := q.copyTimed(p) if err == nil { q.flushCh <- p return @@ -485,28 +492,22 @@ func (q *putQueue) moveToFlush(ctx context.Context, p *pendingPut) { q.drop(p) } -// 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 { - if err := q.gov.acquire(ctx); err != nil { - return err - } - defer q.gov.release() +// copyTimed performs one attempt at p's intent write and blob copy into the +// main blob directory and records the attempt's wall time. +func (q *putQueue) copyTimed(p *pendingPut) error { + q.activeCopies.Add(1) + defer q.activeCopies.Add(-1) start := time.Now() err := q.writeCleanupIntent(p) if err == nil { err = q.copyToMain(p) } - d := time.Since(start) - 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()) + result := "ok" + if err != nil { + result = "error" + } + q.srv.putCopyDuration.WithLabelValues(result).Observe(time.Since(start).Seconds()) } return err }