mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-11 00:50:43 +02:00
Chris Lu
4a1d65939f
* fix(filer): bound retained reader cache buffers by bytes * test(filer): keep in-flight downloads during cache trimming * feat(mem): expose pooled allocation capacity for byte reservations * fix(mount): share a configurable reader buffer budget across files * fix(filer): release failed prefetch slots and memory reservations * feat(mount): expose a soft Go runtime memory limit * docs(filer): restore shared-download rationale in startCaching The one-line comment replacing the original context.Background() explanation was too thin for readChunkAt to cross-reference shared resource semantics. Restore a concise note on why request cancellation must not abort a download shared by concurrent readers. * test(filer): loosen reader cache test deadlines to 5s Three tests used 1-second deadlines that can flake on CI under load: TestReaderCacheBudgetInFlight, TestReaderCacheEvictionDoesNotHoldCacheLock, and TestReaderCacheFailedPrefetchReleasesBudget. Increase to 5 seconds. * test(filer): cover re-read after reader cache eviction Add TestReaderCacheReReadAfterEviction: reads chunk 'a', reads chunk 'b' (evicting 'a' via budget pressure), then re-reads 'a' and asserts a fresh download returns correct data. Verifies the core correctness property that eviction never exposes missing or stale data to readers.
89 lines
2.0 KiB
Go
89 lines
2.0 KiB
Go
package filer
|
|
|
|
import (
|
|
"container/list"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/util/mem"
|
|
)
|
|
|
|
const DefaultReaderCacheMemoryLimit = 256 << 20
|
|
|
|
type ReaderCacheBudget struct {
|
|
sync.Mutex
|
|
limit int64
|
|
used int64
|
|
reservations map[*SingleChunkCacher]int64
|
|
idle list.List
|
|
idleEntries map[*SingleChunkCacher]*list.Element
|
|
changed chan struct{}
|
|
}
|
|
|
|
func NewReaderCacheBudget(limit int64) *ReaderCacheBudget {
|
|
if limit <= 0 {
|
|
limit = DefaultReaderCacheMemoryLimit
|
|
}
|
|
return &ReaderCacheBudget{
|
|
limit: limit,
|
|
reservations: make(map[*SingleChunkCacher]int64),
|
|
idleEntries: make(map[*SingleChunkCacher]*list.Element),
|
|
changed: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
func (b *ReaderCacheBudget) reserve(s *SingleChunkCacher) error {
|
|
if s.chunkSize < 0 {
|
|
return fmt.Errorf("invalid chunk size %d", s.chunkSize)
|
|
}
|
|
size := int64(mem.AllocationSize(s.chunkSize))
|
|
if size > b.limit {
|
|
return fmt.Errorf("chunk buffer needs %d bytes, exceeding reader cache budget %d; increase -readerCacheSizeMB", size, b.limit)
|
|
}
|
|
for {
|
|
b.Lock()
|
|
if size <= b.limit-b.used {
|
|
b.used += size
|
|
b.reservations[s] = size
|
|
b.Unlock()
|
|
return nil
|
|
}
|
|
if entry := b.idle.Front(); entry != nil {
|
|
victim := entry.Value.(*SingleChunkCacher)
|
|
b.idle.Remove(entry)
|
|
delete(b.idleEntries, victim)
|
|
b.Unlock()
|
|
victim.parent.remove(victim)
|
|
continue
|
|
}
|
|
changed := b.changed
|
|
b.Unlock()
|
|
<-changed
|
|
}
|
|
}
|
|
|
|
func (b *ReaderCacheBudget) complete(s *SingleChunkCacher) {
|
|
b.Lock()
|
|
defer b.Unlock()
|
|
if _, found := b.reservations[s]; found && b.idleEntries[s] == nil {
|
|
b.idleEntries[s] = b.idle.PushBack(s)
|
|
close(b.changed)
|
|
b.changed = make(chan struct{})
|
|
}
|
|
}
|
|
|
|
func (b *ReaderCacheBudget) release(s *SingleChunkCacher) {
|
|
b.Lock()
|
|
defer b.Unlock()
|
|
if size, found := b.reservations[s]; found {
|
|
b.used -= size
|
|
delete(b.reservations, s)
|
|
if entry := b.idleEntries[s]; entry != nil {
|
|
b.idle.Remove(entry)
|
|
delete(b.idleEntries, s)
|
|
}
|
|
close(b.changed)
|
|
b.changed = make(chan struct{})
|
|
}
|
|
}
|