From 66ec22883680bd8651f6d0fc7bb9abc7dc48df55 Mon Sep 17 00:00:00 2001 From: javi11 Date: Mon, 7 Sep 2026 17:46:28 +0200 Subject: [PATCH] fix(streaming): keep the Go memory limit above the live heap so GC cannot spin A user streaming one file saw 40-70% CPU for 18 hours with RSS flat at ~1 GB. The soft memory limit introduced in 49869f15 was derived as segment_cache.memory_mb plus a flat 256 MB, 512 MiB for their config, while the live heap was larger: the 256 MB article tier (on even with the disk cache disabled), up to 96 MB of read-ahead per reader across rclone's two chunk streams, and buffers for 341 TLS connections. With live above the limit the collector runs back to back and the GC CPU limiter pins at half the machine without freeing anything. Two layers fix it while keeping RAM bounded by the limit: - The automatic limit now budgets what is actually live: a 128 MB base, three read-ahead windows (the cap constant moves to config so the reader and the formula share it), and 256 KiB per enabled provider connection, on top of the memory tier and the PAR2 solver. - A pressure governor samples runtime/metrics every 2 s. When the GC CPU limiter engages or the live heap passes 90% of the limit it shrinks the memory tier in 25% steps to a 32 MB floor through a Source ceiling that survives file opens, logging a warning once; after 60 s of calm it restores the tier. Verified against a live instance with two rclone-style chunk readers: the old binary pinned below its live set ran at 201% CPU with GC scan dominating the profile; the fixed binary under the same config shrank the tier within 8 s and ran at 30%, the same as a healthy baseline, with RSS held under the limit. --- cmd/altmount/cmd/serve.go | 3 + config.sample.yaml | 7 +- docs/docs/3. Configuration/streaming.md | 4 +- internal/config/manager.go | 46 ++++- internal/config/soft_memory_limit_test.go | 55 +++++- internal/nzbfilesystem/segcache/pressure.go | 176 ++++++++++++++++++ .../segcache/pressure_runtime_test.go | 58 ++++++ .../nzbfilesystem/segcache/pressure_test.go | 118 ++++++++++++ internal/nzbfilesystem/segcache/source.go | 49 ++++- .../segcache/source_ceiling_test.go | 52 ++++++ internal/usenet/usenet_reader.go | 3 +- 11 files changed, 548 insertions(+), 23 deletions(-) create mode 100644 internal/nzbfilesystem/segcache/pressure.go create mode 100644 internal/nzbfilesystem/segcache/pressure_runtime_test.go create mode 100644 internal/nzbfilesystem/segcache/pressure_test.go create mode 100644 internal/nzbfilesystem/segcache/source_ceiling_test.go diff --git a/cmd/altmount/cmd/serve.go b/cmd/altmount/cmd/serve.go index 31a5c6835..51348f3fe 100644 --- a/cmd/altmount/cmd/serve.go +++ b/cmd/altmount/cmd/serve.go @@ -139,6 +139,9 @@ func runServe(cmd *cobra.Command, args []string) error { defer initialCache.Stop() } applySoftMemoryLimit(ctx, cfg) + // Keep the memory tier from pushing the live heap over the soft limit: + // under GC pressure the governor shrinks it, then restores it when calm. + go cacheSource.RunPressureGovernor(ctx) // Background PAR2 repair: repairs missing articles and serves the patched // payloads on the read path's hole branch. diff --git a/config.sample.yaml b/config.sample.yaml index ea3b1bc0b..87baddbda 100644 --- a/config.sample.yaml +++ b/config.sample.yaml @@ -207,8 +207,11 @@ health: # WebDAV mount path configuration # Go soft memory limit in MB. Unset or 0: derived from segment_cache.memory_mb plus the PAR2 solver -# budget plus 256 MB headroom (about 770 MB with defaults). Positive: use exactly this value. -# -1: leave the Go runtime default. GOMEMLIMIT in the environment always wins. +# budget plus headroom for three read-ahead windows, the provider connections' buffers and a 128 MB +# base (about 930 MB with defaults and one 20-connection provider). Positive: use exactly this value. +# -1: leave the Go runtime default. GOMEMLIMIT in the environment always wins. When the live heap +# still presses on the limit, the segment cache memory tier is shrunk automatically so the collector +# does not spin, and restored once the heap is calm. memory_limit_mb: 0 mount_path: '' # WebDAV mount path, Example: '/mnt/remotes/altmount' or '/mnt/unionfs'. Must be an absolute path. # Windows examples: 'Z:' (drive letter via WinFsp) or 'C:\altmount' (directory mount) diff --git a/docs/docs/3. Configuration/streaming.md b/docs/docs/3. Configuration/streaming.md index ed5fcf43d..ec54935b2 100644 --- a/docs/docs/3. Configuration/streaming.md +++ b/docs/docs/3. Configuration/streaming.md @@ -127,9 +127,9 @@ Reads check memory first, then disk, then fetch from the provider: Disk eviction runs automatically every 5 minutes, removing expired entries and enforcing the size limit via LRU (least recently used). Files that are currently open are never evicted. The memory tier evicts the least recently used articles as soon as it reaches `memory_mb`. -**RAM usage:** while streaming, AltMount uses roughly `memory_mb` plus 300 MB. The extra is read-ahead buffers, connections and the Go runtime. Lower `memory_mb` on small boxes (128 MB is a good balance). +**RAM usage:** while streaming, AltMount uses roughly `memory_mb` plus 300-450 MB, bounded by the Go memory limit described below. The extra is read-ahead buffers (up to 96 MB per open reader), about 256 KB per provider connection, and the Go runtime, so it grows with many-provider setups. Lower `memory_mb` on small boxes (128 MB is a good balance). -**Go memory limit:** AltMount sets a Go soft memory limit so evicted articles are collected promptly instead of letting the heap double before each collection. The top-level `memory_limit_mb` controls it: unset or `0` derives the limit from `segment_cache.memory_mb` plus the PAR2 solver budget (`par2_repair.max_memory_mb` × `max_concurrent_jobs`) plus 256 MB headroom, about 770 MB with defaults; a positive value pins the limit (for example `512` on a box where only one stream and no repairs run); `-1` leaves the Go default. `GOMEMLIMIT` in the environment always takes precedence, so Docker users can also set it there. Keep the limit above what the process actually holds live: if live memory exceeds a soft limit the collector runs continuously and costs CPU without freeing anything. +**Go memory limit:** AltMount sets a Go soft memory limit so evicted articles are collected promptly instead of letting the heap double before each collection. The top-level `memory_limit_mb` controls it: unset or `0` derives the limit from everything that holds live heap: `segment_cache.memory_mb`, the PAR2 solver budget (`par2_repair.max_memory_mb` × `max_concurrent_jobs`) when repair is on, three read-ahead windows (96 MB each), 256 KB per configured provider connection, and a 128 MB base. That is about 930 MB with defaults and one 20-connection provider, and it grows with large multi-provider setups because their connection buffers are live too. A positive value pins the limit (for example `512` on a box where only one stream and no repairs run); `-1` leaves the Go default. `GOMEMLIMIT` in the environment always takes precedence, so Docker users can also set it there. Keep the limit above what the process actually holds live: if live memory exceeds a soft limit the collector runs continuously and costs CPU without freeing anything. As a safety net, AltMount watches the runtime's GC CPU limiter and, when the live heap presses on the limit, shrinks the segment cache memory tier in steps (down to 32 MB) so read-ahead buffers fit under it, logging a warning; the tier is restored once the heap has been calm for a minute. Seeing that warning means `memory_limit_mb` is too low for your stream count and connection total, or `segment_cache.memory_mb` is too high. ### Tips diff --git a/internal/config/manager.go b/internal/config/manager.go index badc4a67f..1af7710d3 100644 --- a/internal/config/manager.go +++ b/internal/config/manager.go @@ -237,17 +237,49 @@ func (c SegmentCacheConfig) MemoryBytes() int64 { return int64(max(*c.MemoryMB, 0)) << 20 } -// softMemoryHeadroomMB is what the process needs above the configured -// budgets: read-ahead windows, connection buffers, metadata, and the runtime. -const softMemoryHeadroomMB = 256 +// StreamReadAheadBytesCap bounds one reader's read-ahead window in bytes. +// The usenet reader enforces it; it lives here so the soft memory limit can +// budget for the windows that are live while files stream. +const StreamReadAheadBytesCap int64 = 96 << 20 + +// Soft memory limit headroom: what the process holds live above the memory +// tier and the PAR2 solver. Every term scales with what actually allocates +// so the limit stays above the live set; a limit below it makes the +// collector run back to back and burn CPU without freeing anything. +const ( + // softMemoryBaseMB covers the runtime, metadata, HTTP and pool bookkeeping. + softMemoryBaseMB = 128 + // softMemoryStreams is how many full read-ahead windows are budgeted: + // a mount typically keeps two chunk readers open plus one being torn down. + softMemoryStreams = 3 + // softMemoryPerConnectionBytes is the read and write buffers plus TLS + // record state each pool connection pins while open. + softMemoryPerConnectionBytes int64 = 256 << 10 +) + +// softMemoryHeadroom is the headroom for this config's read-ahead and +// connection footprint. +func (c *Config) softMemoryHeadroom() int64 { + conns := int64(0) + for _, p := range c.Providers { + if p.Enabled != nil && !*p.Enabled { + continue + } + conns += int64(max(p.MaxConnections, 0)) + } + return int64(softMemoryBaseMB)<<20 + + softMemoryStreams*StreamReadAheadBytesCap + + conns*softMemoryPerConnectionBytes +} // SoftMemoryLimit is the Go soft memory limit to apply, or 0 to leave the // runtime alone. Without a limit the collector lets the heap reach twice the // live set, so a 256 MB memory tier costs 600+ MB of RSS. The automatic value // adds every budget that holds live heap (memory tier, PAR2 solver per -// concurrent job) plus headroom, so the limit stays above the live set and the -// collector never has to run back to back. A soft limit is only useful while -// the memory tier is on: with it off the heap is small and bursty. +// concurrent job, read-ahead windows, connection buffers) plus a base, so +// the limit stays above the live set and the collector never has to run back +// to back. A soft limit is only useful while the memory tier is on: with it +// off the heap is small and bursty. func (c *Config) SoftMemoryLimit(gomemlimit string) int64 { if gomemlimit != "" { return 0 @@ -268,7 +300,7 @@ func (c *Config) SoftMemoryLimit(gomemlimit string) int64 { if c.Par2Repair.Enabled != nil && *c.Par2Repair.Enabled { par2 = int64(max(c.Par2Repair.MaxMemoryMB, 0)) * int64(max(c.Par2Repair.MaxConcurrentJobs, 1)) } - return cache + (par2+softMemoryHeadroomMB)<<20 + return cache + par2<<20 + c.softMemoryHeadroom() } // WebDAVConfig represents WebDAV server configuration diff --git a/internal/config/soft_memory_limit_test.go b/internal/config/soft_memory_limit_test.go index 6dfb3ef82..081afe7d5 100644 --- a/internal/config/soft_memory_limit_test.go +++ b/internal/config/soft_memory_limit_test.go @@ -9,14 +9,63 @@ func memLimitConfig(cacheMB int, par2MB, par2Jobs int) *Config { c.Par2Repair.MaxConcurrentJobs = par2Jobs enabled := true c.Par2Repair.Enabled = &enabled + c.Providers = nil return c } +// headroomBytes is the expected headroom for a config with no providers: +// runtime base plus the budgeted read-ahead windows. +func headroomBytes() int64 { + return int64(softMemoryBaseMB)<<20 + softMemoryStreams*StreamReadAheadBytesCap +} + +func withProviders(c *Config, conns ...int) *Config { + on := true + for _, n := range conns { + c.Providers = append(c.Providers, ProviderConfig{Host: "h", Port: 563, MaxConnections: n, Enabled: &on}) + } + return c +} + +func TestSoftMemoryLimitHeadroomScalesWithReadAheadAndConnections(t *testing.T) { + c := memLimitConfig(256, 256, 1) + disabled := false + c.Par2Repair.Enabled = &disabled + base := c.SoftMemoryLimit("") + if want := int64(256)<<20 + headroomBytes(); base != want { + t.Fatalf("SoftMemoryLimit without providers = %d, want %d", base, want) + } + // The user config that triggered a GC spiral: 341 connections across + // six providers and a 256 MB memory tier had a 512 MiB limit. + withProviders(c, 48, 80, 55, 55, 48, 55) + got := c.SoftMemoryLimit("") + if want := base + 341*softMemoryPerConnectionBytes; got != want { + t.Fatalf("SoftMemoryLimit with 341 connections = %d, want %d", got, want) + } + if got <= int64(700)<<20 { + t.Fatalf("SoftMemoryLimit with 341 connections = %d MB, must exceed the ~700 MB live set", got>>20) + } +} + +func TestSoftMemoryLimitCountsBackupAndSkipsDisabledProviders(t *testing.T) { + c := memLimitConfig(256, 256, 1) + base := c.SoftMemoryLimit("") + on, off := true, false + c.Providers = []ProviderConfig{ + {Host: "a", Port: 563, MaxConnections: 10, Enabled: &on}, + {Host: "b", Port: 563, MaxConnections: 20, Enabled: &on, IsBackupProvider: &on}, + {Host: "c", Port: 563, MaxConnections: 99, Enabled: &off}, + } + if got, want := c.SoftMemoryLimit(""), base+30*softMemoryPerConnectionBytes; got != want { + t.Fatalf("SoftMemoryLimit = %d, want %d (backup counted, disabled skipped)", got, want) + } +} + func TestSoftMemoryLimitIgnoresPar2BudgetWhenRepairDisabled(t *testing.T) { c := memLimitConfig(256, 256, 1) disabled := false c.Par2Repair.Enabled = &disabled - want := int64(256+softMemoryHeadroomMB) << 20 + want := int64(256)<<20 + headroomBytes() if got := c.SoftMemoryLimit(""); got != want { t.Fatalf("SoftMemoryLimit with repair disabled = %d, want %d", got, want) } @@ -28,7 +77,7 @@ func TestSoftMemoryLimitIgnoresPar2BudgetWhenRepairDisabled(t *testing.T) { func TestSoftMemoryLimitAutoAddsCachePar2AndHeadroom(t *testing.T) { c := memLimitConfig(256, 256, 1) - want := int64(256+256+softMemoryHeadroomMB) << 20 + want := int64(256+256)<<20 + headroomBytes() if got := c.SoftMemoryLimit(""); got != want { t.Fatalf("SoftMemoryLimit = %d, want %d", got, want) } @@ -36,7 +85,7 @@ func TestSoftMemoryLimitAutoAddsCachePar2AndHeadroom(t *testing.T) { func TestSoftMemoryLimitAutoScalesPar2ByConcurrentJobs(t *testing.T) { c := memLimitConfig(128, 100, 3) - want := int64(128+300+softMemoryHeadroomMB) << 20 + want := int64(128+300)<<20 + headroomBytes() if got := c.SoftMemoryLimit(""); got != want { t.Fatalf("SoftMemoryLimit = %d, want %d", got, want) } diff --git a/internal/nzbfilesystem/segcache/pressure.go b/internal/nzbfilesystem/segcache/pressure.go new file mode 100644 index 000000000..ceee9a28c --- /dev/null +++ b/internal/nzbfilesystem/segcache/pressure.go @@ -0,0 +1,176 @@ +package segcache + +import ( + "context" + "log/slog" + "math" + "runtime/metrics" + "time" +) + +// The memory tier is the one large, elastic block of live heap. When the Go +// soft memory limit sits below what the process holds live, the collector +// runs back to back and the GC CPU limiter engages: CPU climbs to half the +// machine and stays there for as long as the stream runs, while no memory is +// freed because it is all live. The pressure governor watches for that state +// and shrinks the tier so read-ahead windows and connection buffers fit +// under the limit again, then restores it once the heap has been calm for a +// while. RAM stays bounded by the limit; CPU never spirals. +const ( + pressureInterval = 2 * time.Second + // pressureHighFraction of the limit held live is treated as pressure + // before the limiter has to engage. + pressureHighFraction = 0.90 + // pressureLowFraction of the limit is calm enough to start recovering. + pressureLowFraction = 0.70 + // pressureRecoverAfter is how long the heap must stay calm per step back up. + pressureRecoverAfter = 60 * time.Second + // pressureSteps is how many steps take the tier from full to the floor. + pressureSteps = 4 + // pressureFloor keeps a small tier for warm opens and fan-out sharing. + pressureFloor = int64(32) << 20 +) + +// noCeiling means the configured capacity applies unchanged. +const noCeiling int64 = -1 + +type pressureSample struct { + limiterCycle uint64 // /gc/limiter/last-enabled:gc-cycle + live int64 // /gc/heap/live:bytes (marked heap of the last cycle; 0 before the first) + limit int64 // /gc/gomemlimit:bytes +} + +type pressureGovernor struct { + full func() int64 // configured memory tier capacity + ceiling int64 + lastCycle uint64 + baselined bool + calmSince time.Time +} + +func newPressureGovernor(full func() int64) *pressureGovernor { + return &pressureGovernor{full: full, ceiling: noCeiling} +} + +// step folds one sample into the governor and returns the ceiling to apply +// and whether it changed. Pure so the policy is testable without a runtime. +func (g *pressureGovernor) step(s pressureSample, now time.Time) (int64, bool) { + engaged := g.baselined && s.limiterCycle != g.lastCycle + g.lastCycle, g.baselined = s.limiterCycle, true + + full := g.full() + if s.limit <= 0 || s.limit == math.MaxInt64 || full <= pressureFloor { + return g.set(noCeiling) + } + step := full / pressureSteps + current := g.ceiling + if current == noCeiling || current > full { + current = full + } + switch { + case engaged || float64(s.live) > float64(s.limit)*pressureHighFraction: + g.calmSince = now // recovery is counted from the last shrink + return g.set(g.normalize(max(current-step, pressureFloor), full)) + case float64(s.live) < float64(s.limit)*pressureLowFraction: + if current >= full { + return g.set(noCeiling) + } + if g.calmSince.IsZero() { + g.calmSince = now + return g.set(g.normalize(current, full)) + } + if now.Sub(g.calmSince) < pressureRecoverAfter { + return g.set(g.normalize(current, full)) + } + g.calmSince = now + return g.set(g.normalize(current+step, full)) + default: + g.calmSince = time.Time{} + return g.set(g.normalize(current, full)) + } +} + +func (g *pressureGovernor) normalize(c, full int64) int64 { + if c >= full { + return noCeiling + } + return c +} + +func (g *pressureGovernor) set(c int64) (int64, bool) { + changed := c != g.ceiling + g.ceiling = c + return c, changed +} + +var pressureMetricNames = []string{ + "/gc/limiter/last-enabled:gc-cycle", + "/gc/heap/live:bytes", + "/gc/gomemlimit:bytes", +} + +func readPressureSample() (pressureSample, bool) { + samples := make([]metrics.Sample, len(pressureMetricNames)) + for i, n := range pressureMetricNames { + samples[i].Name = n + } + metrics.Read(samples) + for _, s := range samples { + if s.Value.Kind() != metrics.KindUint64 { + return pressureSample{}, false + } + } + return pressureSample{ + limiterCycle: samples[0].Value.Uint64(), + live: clampInt64(samples[1].Value.Uint64()), + limit: clampInt64(samples[2].Value.Uint64()), + }, true +} + +func clampInt64(v uint64) int64 { + if v > math.MaxInt64 { + return math.MaxInt64 + } + return int64(v) +} + +// RunPressureGovernor shrinks the memory tier while the Go soft memory limit +// is under pressure and restores it when the heap is calm. Blocks until ctx +// is cancelled; run it in its own goroutine. +func (s *Source) RunPressureGovernor(ctx context.Context) { + g := newPressureGovernor(func() int64 { return s.getCfg().SegmentCache.MemoryBytes() }) + t := time.NewTicker(pressureInterval) + defer t.Stop() + warned := false + for { + select { + case <-ctx.Done(): + return + case now := <-t.C: + sample, ok := readPressureSample() + if !ok { + slog.ErrorContext(ctx, "Runtime GC metrics unavailable; memory pressure governor disabled") + return + } + ceiling, changed := g.step(sample, now) + if !changed { + continue + } + s.SetMemoryCeiling(ceiling) + attrs := []any{ + "live_mb", sample.live >> 20, "limit_mb", sample.limit >> 20, + "tier_mb", s.effectiveMemoryBytes(s.getCfg().SegmentCache.MemoryBytes()) >> 20, + } + switch { + case ceiling == noCeiling: + warned = false // a later episode is news again + slog.InfoContext(ctx, "Memory pressure eased; segment cache memory tier restored", attrs...) + case !warned: + warned = true + slog.WarnContext(ctx, "Live heap is pressing on the Go memory limit; shrinking the segment cache memory tier to keep GC CPU down. Raise memory_limit_mb or lower segment_cache.memory_mb to avoid this.", attrs...) + default: + slog.InfoContext(ctx, "Memory pressure persists; segment cache memory tier shrunk further", attrs...) + } + } + } +} diff --git a/internal/nzbfilesystem/segcache/pressure_runtime_test.go b/internal/nzbfilesystem/segcache/pressure_runtime_test.go new file mode 100644 index 000000000..186d2f0d1 --- /dev/null +++ b/internal/nzbfilesystem/segcache/pressure_runtime_test.go @@ -0,0 +1,58 @@ +package segcache + +import ( + "runtime" + "runtime/debug" + "testing" + "time" +) + +func TestReadPressureSampleReportsRuntimeLimit(t *testing.T) { + prev := debug.SetMemoryLimit(-1) + t.Cleanup(func() { debug.SetMemoryLimit(prev) }) + const limit = int64(4) << 30 + debug.SetMemoryLimit(limit) + runtime.GC() // live bytes are the marked heap of the last cycle + + s, ok := readPressureSample() + if !ok { + t.Fatal("runtime metrics missing; the governor would be inert") + } + if s.limit != limit { + t.Fatalf("limit = %d, want %d", s.limit, limit) + } + if s.live <= 0 { + t.Fatalf("live heap = %d, want > 0", s.live) + } +} + +// A real limit below the real live heap must make the governor shrink the +// tier on the next sample, without waiting for the GC CPU limiter. +func TestGovernorShrinksUnderRealHeapPressure(t *testing.T) { + prev := debug.SetMemoryLimit(-1) + t.Cleanup(func() { debug.SetMemoryLimit(prev) }) + debug.SetMemoryLimit(int64(64) << 30) + runtime.GC() + base, _ := readPressureSample() + + pinned := make([]byte, 128<<20) + for i := range pinned { + pinned[i] = byte(i) + } + runtime.GC() + // Limit just above the live heap so live/limit is well past pressureHighFraction. + debug.SetMemoryLimit(base.live + 128<<20 + 4<<20) + + g := newTestGovernor() + s, ok := readPressureSample() + if !ok { + t.Fatal("metrics unavailable") + } + g.step(s, time.Now()) + s, _ = readPressureSample() + c, changed := g.step(s, time.Now()) + if !changed || c == noCeiling { + t.Fatalf("live=%d limit=%d: ceiling=%d changed=%v, want a shrink", s.live, s.limit, c, changed) + } + runtime.KeepAlive(pinned) +} diff --git a/internal/nzbfilesystem/segcache/pressure_test.go b/internal/nzbfilesystem/segcache/pressure_test.go new file mode 100644 index 000000000..37df26a4e --- /dev/null +++ b/internal/nzbfilesystem/segcache/pressure_test.go @@ -0,0 +1,118 @@ +package segcache + +import ( + "math" + "testing" + "time" +) + +const ( + testFull = int64(256) << 20 + testLimit = int64(700) << 20 +) + +func newTestGovernor() *pressureGovernor { + return newPressureGovernor(func() int64 { return testFull }) +} + +func sample(cycle uint64, liveFrac float64) pressureSample { + return pressureSample{limiterCycle: cycle, live: int64(float64(testLimit) * liveFrac), limit: testLimit} +} + +func TestPressureGovernorHoldsFullCapacityWhenCalm(t *testing.T) { + g := newTestGovernor() + now := time.Unix(0, 0) + for i := 0; i < 5; i++ { + if c, changed := g.step(sample(7, 0.5), now); changed || c != noCeiling { + t.Fatalf("calm step %d: ceiling=%d changed=%v, want none", i, c, changed) + } + now = now.Add(pressureInterval) + } +} + +func TestPressureGovernorShrinksWhenLimiterEngages(t *testing.T) { + g := newTestGovernor() + now := time.Unix(0, 0) + g.step(sample(7, 0.5), now) // baseline limiter cycle + c, changed := g.step(sample(9, 0.5), now.Add(pressureInterval)) + if !changed || c != testFull-testFull/pressureSteps { + t.Fatalf("after limiter engaged: ceiling=%d changed=%v, want %d", c, changed, testFull-testFull/pressureSteps) + } + // Still engaging: keeps stepping down to the floor, never below. + for i := uint64(10); i < 30; i++ { + c, _ = g.step(sample(i, 0.5), now) + } + if c != pressureFloor { + t.Fatalf("ceiling after sustained pressure = %d, want floor %d", c, pressureFloor) + } +} + +func TestPressureGovernorShrinksWhenLiveHeapNearsLimit(t *testing.T) { + g := newTestGovernor() + now := time.Unix(0, 0) + g.step(sample(1, 0.5), now) + c, changed := g.step(sample(1, 0.95), now) + if !changed || c != testFull-testFull/pressureSteps { + t.Fatalf("near-limit step: ceiling=%d changed=%v", c, changed) + } +} + +func TestPressureGovernorFirstSampleIsBaselineNotPressure(t *testing.T) { + g := newTestGovernor() + if c, changed := g.step(sample(42, 0.5), time.Unix(0, 0)); changed || c != noCeiling { + t.Fatalf("first sample must only record the limiter cycle, got ceiling=%d changed=%v", c, changed) + } +} + +func TestPressureGovernorRecoversAfterCalmWindow(t *testing.T) { + g := newTestGovernor() + now := time.Unix(0, 0) + g.step(sample(1, 0.5), now) + c, _ := g.step(sample(2, 0.5), now) + shrunk := c + // Calm but not yet for the whole recovery window: hold. + now = now.Add(pressureRecoverAfter / 2) + if c, changed := g.step(sample(2, 0.5), now); changed || c != shrunk { + t.Fatalf("mid-window: ceiling=%d changed=%v, want hold at %d", c, changed, shrunk) + } + now = now.Add(pressureRecoverAfter/2 + time.Second) + c, changed := g.step(sample(2, 0.5), now) + if !changed || c != noCeiling { + t.Fatalf("after calm window: ceiling=%d changed=%v, want restored to full", c, changed) + } +} + +func TestPressureGovernorMiddleBandResetsCalmTimer(t *testing.T) { + g := newTestGovernor() + now := time.Unix(0, 0) + g.step(sample(1, 0.5), now) + g.step(sample(2, 0.5), now) // shrink + now = now.Add(pressureRecoverAfter - time.Second) + g.step(sample(2, 0.8), now) // between low and high: not calm + now = now.Add(2 * time.Second) + if _, changed := g.step(sample(2, 0.5), now); changed { + t.Fatal("calm timer must restart after a middle-band sample") + } +} + +func TestPressureGovernorReleasesWhenNoLimit(t *testing.T) { + g := newTestGovernor() + now := time.Unix(0, 0) + g.step(sample(1, 0.5), now) + g.step(sample(2, 0.5), now) // shrink + s := pressureSample{limiterCycle: 2, live: 1 << 30, limit: math.MaxInt64} + if c, changed := g.step(s, now); !changed || c != noCeiling { + t.Fatalf("no limit: ceiling=%d changed=%v, want released", c, changed) + } +} + +func TestPressureGovernorFloorNeverAboveConfiguredCapacity(t *testing.T) { + small := pressureFloor / 2 + g := newPressureGovernor(func() int64 { return small }) + now := time.Unix(0, 0) + g.step(sample(1, 0.5), now) + c, changed := g.step(sample(2, 0.5), now) + if changed || c != noCeiling { + t.Fatalf("a tier already below the floor must not be touched, got ceiling=%d changed=%v", c, changed) + } +} diff --git a/internal/nzbfilesystem/segcache/source.go b/internal/nzbfilesystem/segcache/source.go index c4d35dc40..1a087b958 100644 --- a/internal/nzbfilesystem/segcache/source.go +++ b/internal/nzbfilesystem/segcache/source.go @@ -17,12 +17,42 @@ type Source struct { getCfg config.ConfigGetter once sync.Once - tiered *TieredStore + tiered atomic.Pointer[TieredStore] + + // ceiling caps the memory tier below its configured capacity while the + // pressure governor sees the heap pressing on the soft memory limit; + // noCeiling applies the configured value. + ceiling atomic.Int64 } // NewSource creates a Source. getCfg must not be nil. func NewSource(getCfg config.ConfigGetter) *Source { - return &Source{getCfg: getCfg} + s := &Source{getCfg: getCfg} + s.ceiling.Store(noCeiling) + return s +} + +// SetMemoryCeiling caps the memory tier at bytes (noCeiling removes the cap), +// evicting immediately when the live tier is larger. The cap survives file +// opens, which otherwise re-apply the configured capacity. +func (s *Source) SetMemoryCeiling(bytes int64) { + if bytes < 0 { + bytes = noCeiling + } + s.ceiling.Store(bytes) + if t := s.tiered.Load(); t != nil { + t.Memory().SetCapacity(s.effectiveMemoryBytes(s.getCfg().SegmentCache.MemoryBytes())) + } +} + +// MemoryCeiling is the current cap, or noCeiling. +func (s *Source) MemoryCeiling() int64 { return s.ceiling.Load() } + +func (s *Source) effectiveMemoryBytes(configured int64) int64 { + if c := s.ceiling.Load(); c != noCeiling && c < configured { + return c + } + return configured } // Store resolves the current SegmentStore: nil only when both the memory @@ -39,10 +69,12 @@ func (s *Source) Store() usenet.SegmentStore { if memBytes <= 0 && disk == nil { return nil } - s.once.Do(func() { s.tiered = NewTieredStore(NewMemoryCache(memBytes)) }) - s.tiered.Memory().SetCapacity(memBytes) - s.tiered.SetDisk(disk) - return s.tiered + memBytes = s.effectiveMemoryBytes(memBytes) + s.once.Do(func() { s.tiered.Store(NewTieredStore(NewMemoryCache(memBytes))) }) + t := s.tiered.Load() + t.Memory().SetCapacity(memBytes) + t.SetDisk(disk) + return t } // Swap replaces the active manager. Pass nil to unload the current manager. @@ -58,8 +90,9 @@ func (s *Source) Manager() *Manager { // Memory returns the memory tier for stats, or nil before the first open. func (s *Source) Memory() *MemoryCache { - if s.tiered == nil { + t := s.tiered.Load() + if t == nil { return nil } - return s.tiered.Memory() + return t.Memory() } diff --git a/internal/nzbfilesystem/segcache/source_ceiling_test.go b/internal/nzbfilesystem/segcache/source_ceiling_test.go new file mode 100644 index 000000000..4917fd643 --- /dev/null +++ b/internal/nzbfilesystem/segcache/source_ceiling_test.go @@ -0,0 +1,52 @@ +package segcache + +import ( + "testing" + + "github.com/kipsilabs/altmount/internal/config" +) + +func ceilingSource(memMB int) *Source { + cfg := config.DefaultConfig() + cfg.SegmentCache.MemoryMB = &memMB + return NewSource(func() *config.Config { return cfg }) +} + +func TestSourceCeilingCapsMemoryTierAndReopenKeepsIt(t *testing.T) { + s := ceilingSource(256) + s.Store() + if got := s.Memory().Capacity(); got != int64(256)<<20 { + t.Fatalf("initial capacity = %d, want 256 MiB", got) + } + s.SetMemoryCeiling(64 << 20) + if got := s.Memory().Capacity(); got != 64<<20 { + t.Fatalf("capacity after ceiling = %d, want 64 MiB (applied immediately)", got) + } + // A file open re-resolves the store; the ceiling must survive it. + s.Store() + if got := s.Memory().Capacity(); got != 64<<20 { + t.Fatalf("capacity after reopen = %d, want ceiling to hold", got) + } + s.SetMemoryCeiling(noCeiling) + if got := s.Memory().Capacity(); got != int64(256)<<20 { + t.Fatalf("capacity after release = %d, want configured 256 MiB", got) + } +} + +func TestSourceCeilingAboveConfigIsNoop(t *testing.T) { + s := ceilingSource(64) + s.SetMemoryCeiling(512 << 20) + s.Store() + if got := s.Memory().Capacity(); got != 64<<20 { + t.Fatalf("capacity = %d, want configured 64 MiB", got) + } +} + +func TestSourceCeilingBeforeFirstOpenApplies(t *testing.T) { + s := ceilingSource(256) + s.SetMemoryCeiling(32 << 20) + s.Store() + if got := s.Memory().Capacity(); got != 32<<20 { + t.Fatalf("capacity = %d, want 32 MiB", got) + } +} diff --git a/internal/usenet/usenet_reader.go b/internal/usenet/usenet_reader.go index d1bb2108f..a45f43ce9 100644 --- a/internal/usenet/usenet_reader.go +++ b/internal/usenet/usenet_reader.go @@ -12,6 +12,7 @@ import ( "time" "github.com/avast/retry-go/v4" + "github.com/kipsilabs/altmount/internal/config" "github.com/kipsilabs/altmount/internal/holes" "github.com/kipsilabs/altmount/internal/pool" "github.com/kipsilabs/altmount/internal/slogutil" @@ -144,7 +145,7 @@ const ( // 60-segment window is 45 MB on 750 KB posts but 240 MB on 4 MiB posts, // where it starves the reader's own demand article for the link and // leaves a quarter of a gigabyte to abandon on every seek. - readAheadBytesCap = 96 << 20 + readAheadBytesCap = config.StreamReadAheadBytesCap ) // withFlightMap gives the reader its own in-flight article map. Tests use it