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