Skip to content

gocached, cmd/gocached: replace put-queue mover governor with fixed 128 #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bradfitz merged 1 commit into from
Oct 8, 2026
Merged

gocached, cmd/gocached: replace put-queue mover governor with fixed 128 #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 3 additions & 23 deletions cmd/gocached/gocached.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@ import (
"net"
"net/http"
"os"
"strconv"
"strings"
"time"

Expand All @@ -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)")
Expand Down Expand Up @@ -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 != "" {
Expand Down Expand Up @@ -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
}
28 changes: 0 additions & 28 deletions cmd/gocached/gocached_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}
}
39 changes: 12 additions & 27 deletions gocache/gocached/gocached.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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{
Expand Down Expand Up @@ -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))
}
Expand Down Expand Up @@ -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)
}
}

Expand Down Expand Up @@ -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
Expand Down
231 changes: 0 additions & 231 deletions gocache/gocached/movergov.go
View file
Open in desktop

This file was deleted.

Loading
Loading