diff --git a/cmd/objgitd/main.go b/cmd/objgitd/main.go index ec36ffb..b970979 100644 --- a/cmd/objgitd/main.go +++ b/cmd/objgitd/main.go @@ -49,6 +49,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") + maxConcurrentPushes = flag.Int("max-concurrent-pushes", 4, "pushes allowed to unpack a packfile at the same time; each one costs roughly 400 MiB of resident set for a large repository, so this is what bounds memory under concurrent pushes; 0 disables the limit") pushQueueTimeout = flag.Duration("push-queue-timeout", 2*time.Minute, "how long a push waits for a slot before it fails") ) @@ -87,6 +89,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 }, }