From 216b52c13ad2ab890210c97f6c278452a4482626 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 16 Apr 2026 11:30:23 -0700 Subject: [PATCH] perf(mount): add graduated write backpressure (#9099) * perf(mount): add graduated write backpressure before buffer cap Introduce soft (80%) and hard (95%) throttling thresholds in WriteBufferAccountant. When write buffer usage approaches the cap, Reserve() inserts brief sleeps to slow writers gradually rather than blocking them completely at the cap. This smooths out write latency under sustained load. * fix(mount): address review feedback for graduated backpressure - Use projected usage (used + n) for threshold checks so a single reservation crossing a threshold is throttled immediately. - Widen no-throttle test timing tolerance to softThrottleDelay to avoid CI flakes from scheduling jitter. - Replace fragile timing upper-bound in recovery test with counter-based invariant (hardThrottleCount stays at 1). * docs(mount): clarify single-shot throttle design in WriteBufferAccountant Add a comment explaining why graduated throttling runs once per Reserve call rather than inside the blocking loop: once at the cap, the evictor + cond.Wait mechanism frees actual capacity, which time-based sleeps cannot do. --- .../page_writer/write_buffer_accountant.go | 77 ++++++++++-- .../write_buffer_accountant_test.go | 110 ++++++++++++++++++ 2 files changed, 180 insertions(+), 7 deletions(-) diff --git a/weed/mount/page_writer/write_buffer_accountant.go b/weed/mount/page_writer/write_buffer_accountant.go index 6f4621d11d..238b1d4d84 100644 --- a/weed/mount/page_writer/write_buffer_accountant.go +++ b/weed/mount/page_writer/write_buffer_accountant.go @@ -2,6 +2,21 @@ package page_writer import ( "sync" + "sync/atomic" + "time" +) + +const ( + // softThresholdRatio is the fraction of cap at which soft throttling begins. + // At this level, writes sleep briefly to let uploaders drain. + softThresholdRatio = 0.8 + // hardThresholdRatio is the fraction of cap at which hard throttling kicks in. + // Writes sleep longer to aggressively slow intake. + hardThresholdRatio = 0.95 + // softThrottleDelay is the sleep duration when usage exceeds the soft threshold. + softThrottleDelay = 10 * time.Millisecond + // hardThrottleDelay is the sleep duration when usage exceeds the hard threshold. + hardThrottleDelay = 50 * time.Millisecond ) // WriteBufferAccountant enforces a global byte budget across all @@ -14,16 +29,25 @@ import ( // // A nil receiver is treated as "unlimited" for backward compatibility. type WriteBufferAccountant struct { - mu sync.Mutex - cond *sync.Cond - cap int64 // 0 means unlimited - used int64 - evictor func(needBytes int64) bool - evicting bool + mu sync.Mutex + cond *sync.Cond + cap int64 // 0 means unlimited + used int64 + softThreshold int64 // pre-computed: cap * softThresholdRatio + hardThreshold int64 // pre-computed: cap * hardThresholdRatio + evictor func(needBytes int64) bool + evicting bool + + softThrottleCount atomic.Int64 + hardThrottleCount atomic.Int64 } func NewWriteBufferAccountant(capBytes int64) *WriteBufferAccountant { - a := &WriteBufferAccountant{cap: capBytes} + a := &WriteBufferAccountant{ + cap: capBytes, + softThreshold: int64(float64(capBytes) * softThresholdRatio), + hardThreshold: int64(float64(capBytes) * hardThresholdRatio), + } a.cond = sync.NewCond(&a.mu) return a } @@ -57,6 +81,29 @@ func (a *WriteBufferAccountant) Reserve(n int64) { } a.mu.Lock() defer a.mu.Unlock() + + // Graduated backpressure: slow writers before hitting the hard cap. + // Use projected usage (used + n) so that a single reservation crossing + // a threshold is throttled immediately rather than on the next call. + // + // This runs once per Reserve, not inside the blocking loop below. + // Once usage reaches the cap, the evictor + cond.Wait is the correct + // mechanism: it blocks until a sealed chunk upload frees space, which + // is strictly more efficient than repeated time-based sleeps that + // cannot free capacity on their own. + projected := a.used + n + if projected >= a.hardThreshold && a.used > 0 { + a.hardThrottleCount.Add(1) + a.mu.Unlock() + time.Sleep(hardThrottleDelay) + a.mu.Lock() + } else if projected >= a.softThreshold && a.used > 0 { + a.softThrottleCount.Add(1) + a.mu.Unlock() + time.Sleep(softThrottleDelay) + a.mu.Lock() + } + for a.used+n > a.cap && a.used > 0 { // Before blocking, try to force-seal a writable chunk somewhere so // its async upload path will eventually Release a slot. Single-flight @@ -115,3 +162,19 @@ func (a *WriteBufferAccountant) Used() int64 { defer a.mu.Unlock() return a.used } + +// SoftThrottleCount returns the number of times soft throttling was triggered. +func (a *WriteBufferAccountant) SoftThrottleCount() int64 { + if a == nil { + return 0 + } + return a.softThrottleCount.Load() +} + +// HardThrottleCount returns the number of times hard throttling was triggered. +func (a *WriteBufferAccountant) HardThrottleCount() int64 { + if a == nil { + return 0 + } + return a.hardThrottleCount.Load() +} diff --git a/weed/mount/page_writer/write_buffer_accountant_test.go b/weed/mount/page_writer/write_buffer_accountant_test.go index f9dd8a9e1c..4ebdfebea9 100644 --- a/weed/mount/page_writer/write_buffer_accountant_test.go +++ b/weed/mount/page_writer/write_buffer_accountant_test.go @@ -71,6 +71,116 @@ func TestWriteBufferAccountant_BlocksWhenOverCap(t *testing.T) { } } +func TestWriteBufferAccountant_SoftThrottle(t *testing.T) { + // Cap = 1000, soft threshold = 800. Reserve 850 (85%) then measure + // that the next Reserve is delayed by at least softThrottleDelay. + a := NewWriteBufferAccountant(1000) + a.Reserve(850) + + start := time.Now() + a.Reserve(10) // should trigger soft throttle (used=850 > 800) + elapsed := time.Since(start) + + if elapsed < softThrottleDelay { + t.Fatalf("expected Reserve to take >= %v (soft throttle), took %v", softThrottleDelay, elapsed) + } +} + +func TestWriteBufferAccountant_HardThrottle(t *testing.T) { + // Cap = 1000, hard threshold = 950. Reserve 960 (96%) then measure + // that the next Reserve is delayed by at least hardThrottleDelay. + a := NewWriteBufferAccountant(1000) + a.Reserve(960) + + start := time.Now() + a.Reserve(10) // should trigger hard throttle (used=960 > 950) + elapsed := time.Since(start) + + if elapsed < hardThrottleDelay { + t.Fatalf("expected Reserve to take >= %v (hard throttle), took %v", hardThrottleDelay, elapsed) + } +} + +func TestWriteBufferAccountant_NoThrottleBelowSoft(t *testing.T) { + // Cap = 1000, soft threshold = 800. Reserve 700 (70%) -- below soft + // threshold, so subsequent Reserve should return quickly. + a := NewWriteBufferAccountant(1000) + a.Reserve(700) + + start := time.Now() + a.Reserve(10) + elapsed := time.Since(start) + + if elapsed >= softThrottleDelay { + t.Fatalf("expected Reserve below soft threshold to avoid throttle delay (< %v), took %v", softThrottleDelay, elapsed) + } +} + +func TestWriteBufferAccountant_ThrottleCounters(t *testing.T) { + a := NewWriteBufferAccountant(1000) + + // Below soft -- no throttle + a.Reserve(700) + a.Reserve(10) + if a.SoftThrottleCount() != 0 || a.HardThrottleCount() != 0 { + t.Fatalf("expected no throttle counts below soft threshold, got soft=%d hard=%d", + a.SoftThrottleCount(), a.HardThrottleCount()) + } + + // Release back down, then fill to soft zone + a.Release(710) + a.Reserve(850) // now used=850, above soft (800) + a.Reserve(10) // triggers soft throttle + if a.SoftThrottleCount() != 1 { + t.Fatalf("expected soft throttle count 1, got %d", a.SoftThrottleCount()) + } + if a.HardThrottleCount() != 0 { + t.Fatalf("expected hard throttle count 0, got %d", a.HardThrottleCount()) + } + + // Release and fill to hard zone + a.Release(860) + a.Reserve(960) // now used=960, above hard (950) + a.Reserve(10) // triggers hard throttle + if a.HardThrottleCount() != 1 { + t.Fatalf("expected hard throttle count 1, got %d", a.HardThrottleCount()) + } +} + +func TestWriteBufferAccountant_GraduatedRecovery(t *testing.T) { + // Fill to hard threshold, verify hard throttle. Release to soft zone, + // verify that throttle level decreases to soft. + a := NewWriteBufferAccountant(1000) + + // Fill to hard zone (960 > 950) + a.Reserve(960) + start := time.Now() + a.Reserve(10) // hard throttle + elapsed := time.Since(start) + if elapsed < hardThrottleDelay { + t.Fatalf("expected hard throttle delay, got %v", elapsed) + } + if a.HardThrottleCount() != 1 { + t.Fatalf("expected hard throttle count 1, got %d", a.HardThrottleCount()) + } + + // Release down to soft zone: used = 970 - 130 = 840 (> 800 soft, < 950 hard) + a.Release(130) + + start = time.Now() + a.Reserve(10) // should now be soft throttle + elapsed = time.Since(start) + if elapsed < softThrottleDelay { + t.Fatalf("expected soft throttle delay after recovery, got %v", elapsed) + } + if a.HardThrottleCount() != 1 { + t.Fatalf("expected hard throttle count to remain 1 after recovery, got %d", a.HardThrottleCount()) + } + if a.SoftThrottleCount() != 1 { + t.Fatalf("expected soft throttle count 1 after recovery, got %d", a.SoftThrottleCount()) + } +} + func TestWriteBufferAccountant_AllowsOversizeWhenEmpty(t *testing.T) { // A single request larger than cap must succeed when nothing is in // flight, otherwise a single file handle with a large chunk size would