From a9a1081afed0455c61b0f31535b0d1db8e8c5b67 Mon Sep 17 00:00:00 2001 From: Xe Iaso Date: Mon, 31 Aug 2026 14:46:34 -0400 Subject: [PATCH] perf(storage/tigris): cap zstd encoder concurrency zstd.NewWriter defaults WithEncoderConcurrency to GOMAXPROCS, and every internal encoder state retains a match-history buffer for the life of the process. That put a permanent retained-heap floor under the daemon that scaled with core count (~460 MB on a 20-core host) rather than with real push demand. Give the streaming encoder pool WithEncoderConcurrency(1): a streaming copy owns its encoder for one object, so the extra states can never be used concurrently and were pure floor. Cap the EncodeAll singleton at an explicit concurrency (default 4, exposed as -encoder-concurrency), sized for expected simultaneous pushes rather than left at GOMAXPROCS. Assisted-by: Claude Opus 4.8 via Claude Code Signed-off-by: Xe Iaso --- cmd/objgitd/main.go | 7 +++++ internal/storage/tigris/compress.go | 40 +++++++++++++++++++++++++++-- 2 files changed, 45 insertions(+), 2 deletions(-) diff --git a/cmd/objgitd/main.go b/cmd/objgitd/main.go index 2d62121..06a959c 100644 --- a/cmd/objgitd/main.go +++ b/cmd/objgitd/main.go @@ -48,6 +48,8 @@ var ( packCompression = flag.Bool("pack-compression", true, "store zstd-compressed payloads in newly written pack containers; reading compressed containers is always enabled, so this is safe to turn off for one release before a rollback") packedRefs = flag.Bool("packed-refs", true, "write every ref into one packed-refs object under a compare-and-swap, instead of one object per ref; reading packed refs is always enabled, so this is safe to turn off for one release before a rollback") + + encoderConcurrency = flag.Int("encoder-concurrency", 4, "how many concurrent zstd EncodeAll calls the process-wide pack encoder serves; each state retains a match-history buffer, so this caps encoder heap that would otherwise scale with GOMAXPROCS. Set with the concurrent push cap in mind") ) // tigrisBase adapts *tigris.Storer to repofs.Base: Storer.Scoped returns the @@ -84,6 +86,11 @@ func main() { // Route s3fs S3 round-trips into Prometheus before any filesystem use. s3fs.SetMetricsObserver(metrics.ObserveS3) + // Cap the process-wide pack encoder's concurrency before the first encode. + // Left at its GOMAXPROCS default, each internal state retains a + // match-history buffer and the retained floor scales with core count. + tigris.SetEncoderConcurrency(*encoderConcurrency) + rawClient, err := tstorage.New(ctx) if err != nil { slog.Error("can't create Tigris storage client", "err", err) diff --git a/internal/storage/tigris/compress.go b/internal/storage/tigris/compress.go index a4b4962..77c41b7 100644 --- a/internal/storage/tigris/compress.go +++ b/internal/storage/tigris/compress.go @@ -95,11 +95,39 @@ var ( payloadOnce sync.Once ) +// encoderConcurrency caps how many EncodeAll calls the encoder() singleton +// serves at once. zstd.NewWriter defaults WithEncoderConcurrency to GOMAXPROCS, +// and every internal encoder state allocates its own match-history buffer (tens +// of megabytes, never released), so the default puts a permanent floor under +// retained heap that scales with core count and not with real demand. The value +// that matters is "expected simultaneous pushes", not "core count"; 4 matches +// deltaScanWorkers in spirit — a bound picked for resource reasons. Revisit it +// alongside the push cap that actually bounds demand. +// +// A package-level variable with a setter, not a Storer option: the encoder is a +// process-wide singleton, so SetEncoderConcurrency must be called from main +// before the first encode. Reads are not synchronized because the setter runs +// once at startup, before any EncodeAll. +var encoderConcurrency = 4 + +// SetEncoderConcurrency sets how many EncodeAll calls the encoder() singleton +// serves at once. Call it once from main before the first encode; values below +// 1 are clamped to 1. +func SetEncoderConcurrency(n int) { + if n < 1 { + n = 1 + } + encoderConcurrency = n +} + func encoder() *zstd.Encoder { zstdEncOnce.Do(func() { // Errors here are impossible with a nil writer and valid options; the // option list is a compile-time constant. - zstdEnc, _ = zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.SpeedDefault)) + zstdEnc, _ = zstd.NewWriter(nil, + zstd.WithEncoderLevel(zstd.SpeedDefault), + zstd.WithEncoderConcurrency(encoderConcurrency), + ) }) return zstdEnc } @@ -114,7 +142,15 @@ func encoder() *zstd.Encoder { var streamEncPool = sync.Pool{ New: func() any { // Same error-impossibility reasoning as encoder() above. - zw, _ := zstd.NewWriter(nil, zstd.WithEncoderLevel(zstd.SpeedDefault)) + // + // WithEncoderConcurrency(1) because a streaming copy owns its encoder for + // the duration of one object, as the comment above states: the extra + // states a default GOMAXPROCS encoder would allocate can never be used + // concurrently by a single owner, so they are pure retained-heap floor. + zw, _ := zstd.NewWriter(nil, + zstd.WithEncoderLevel(zstd.SpeedDefault), + zstd.WithEncoderConcurrency(1), + ) return zw }, }