From 4a1d65939fb4200dfceedb950042a1a3558e3baa Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 8 Sep 2026 10:51:28 -0700 Subject: [PATCH] fix(mount): bound reader cache memory across open files (#11220) * 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. --- weed/command/mount.go | 4 + weed/command/mount_common.go | 22 +++ weed/command/mount_memory_test.go | 39 ++++ weed/command/mount_std.go | 4 + weed/command/mount_windows.go | 5 + weed/filer/filechunk_group.go | 9 +- weed/filer/reader_cache.go | 81 +++++--- weed/filer/reader_cache_budget.go | 88 +++++++++ weed/filer/reader_cache_memory_test.go | 249 +++++++++++++++++++++++++ weed/mount/filehandle.go | 8 +- weed/mount/weedfs.go | 3 + weed/util/mem/slot_pool.go | 7 + weed/util/mem/slot_pool_test.go | 9 + 13 files changed, 497 insertions(+), 31 deletions(-) create mode 100644 weed/command/mount_memory_test.go create mode 100644 weed/filer/reader_cache_budget.go create mode 100644 weed/filer/reader_cache_memory_test.go diff --git a/weed/command/mount.go b/weed/command/mount.go index db445264d..3c81aa8ca 100644 --- a/weed/command/mount.go +++ b/weed/command/mount.go @@ -20,6 +20,8 @@ type MountOptions struct { chunkSizeLimitMB *int concurrentWriters *int concurrentReaders *int + readerCacheSizeMB *int64 + memoryLimitMB *int64 cacheMetaTtlSec *int cacheDirMaxEntries *int cacheDirForRead *string @@ -104,6 +106,8 @@ func init() { mountOptions.chunkSizeLimitMB = cmdMount.Flag.Int("chunkSizeLimitMB", 2, "local write buffer size, also chunk large files") mountOptions.concurrentWriters = cmdMount.Flag.Int("concurrentWriters", 128, "limit concurrent goroutine writers") mountOptions.concurrentReaders = cmdMount.Flag.Int("concurrentReaders", 128, "limit concurrent chunk fetches for read operations") + mountOptions.memoryLimitMB = cmdMount.Flag.Int64("memoryLimitMB", 0, "soft Go runtime memory limit in MiB; 0 preserves GOMEMLIMIT; leave headroom below the container limit") + mountOptions.readerCacheSizeMB = cmdMount.Flag.Int64("readerCacheSizeMB", 256, "memory budget in MiB for downloaded and in-flight reader buffers across all files; must fit the largest pooled chunk buffer") mountOptions.cacheDirForRead = cmdMount.Flag.String("cacheDir", os.TempDir(), "local cache directory for file chunks and meta data") mountOptions.cacheSizeMBForRead = cmdMount.Flag.Int64("cacheCapacityMB", 128, "file chunk read cache capacity in MB") mountOptions.cacheDirForWrite = cmdMount.Flag.String("cacheDirWrite", "", "buffer writes mostly for large files") diff --git a/weed/command/mount_common.go b/weed/command/mount_common.go index 80b211081..18d92b58a 100644 --- a/weed/command/mount_common.go +++ b/weed/command/mount_common.go @@ -5,11 +5,13 @@ package command import ( "context" "fmt" + "math" "net" "net/http" "os" "path" "runtime" + "runtime/debug" "strconv" "strings" "time" @@ -184,6 +186,10 @@ type fileSystemParams struct { } func buildSeaweedFileSystem(option *MountOptions, p fileSystemParams) *mount.WFS { + readerCacheSizeMB := int64(256) + if option.readerCacheSizeMB != nil { + readerCacheSizeMB = *option.readerCacheSizeMB + } return mount.NewSeaweedFileSystem(&mount.Option{ MountDirectory: p.dir, FilerAddresses: p.filerAddresses, @@ -198,6 +204,7 @@ func buildSeaweedFileSystem(option *MountOptions, p fileSystemParams) *mount.WFS ChunkSizeLimit: int64(p.chunkSizeLimitMB) * 1024 * 1024, ConcurrentWriters: *option.concurrentWriters, ConcurrentReaders: *option.concurrentReaders, + ReaderCacheSizeMB: readerCacheSizeMB, CacheDirForRead: p.cacheDirForRead, CacheSizeMBForRead: *option.cacheSizeMBForRead, CacheDirForWrite: p.cacheDirForWrite, @@ -304,3 +311,18 @@ func lastSegment(p string) string { } return name } + +func configureMountMemory(option *MountOptions) error { + if option.readerCacheSizeMB != nil && (*option.readerCacheSizeMB <= 0 || *option.readerCacheSizeMB > math.MaxInt64>>20) { + return fmt.Errorf("readerCacheSizeMB must be positive and fit in an int64 byte budget") + } + if option.memoryLimitMB != nil { + if *option.memoryLimitMB < 0 || *option.memoryLimitMB > math.MaxInt64>>20 { + return fmt.Errorf("memoryLimitMB must be non-negative and fit in an int64 byte limit") + } + if *option.memoryLimitMB > 0 { + debug.SetMemoryLimit(*option.memoryLimitMB << 20) + } + } + return nil +} diff --git a/weed/command/mount_memory_test.go b/weed/command/mount_memory_test.go new file mode 100644 index 000000000..949112da7 --- /dev/null +++ b/weed/command/mount_memory_test.go @@ -0,0 +1,39 @@ +//go:build linux || darwin || freebsd || windows + +package command + +import ( + "math" + "runtime/debug" + "testing" +) + +func TestConfigureMountMemory(t *testing.T) { + for _, size := range []int64{-1, 0, 1, 256, math.MaxInt64 >> 20, math.MaxInt64} { + err := configureMountMemory(&MountOptions{readerCacheSizeMB: &size}) + valid := size > 0 && size <= math.MaxInt64>>20 + if (err == nil) != valid { + t.Errorf("size=%d: err=%v", size, err) + } + } +} + +func TestConfigureMountMemoryRuntimeLimit(t *testing.T) { + previous := debug.SetMemoryLimit(512 << 20) + defer debug.SetMemoryLimit(previous) + for _, size := range []int64{0, -1, math.MaxInt64, 768, 0} { + before := debug.SetMemoryLimit(-1) + err := configureMountMemory(&MountOptions{memoryLimitMB: &size}) + valid := size >= 0 && size <= math.MaxInt64>>20 + if (err == nil) != valid { + t.Errorf("size=%d: err=%v", size, err) + } + want := before + if valid && size > 0 { + want = size << 20 + } + if got := debug.SetMemoryLimit(-1); got != want { + t.Errorf("size=%d: runtime limit=%d, want %d", size, got, want) + } + } +} diff --git a/weed/command/mount_std.go b/weed/command/mount_std.go index 4bd85c174..b047443c1 100644 --- a/weed/command/mount_std.go +++ b/weed/command/mount_std.go @@ -24,6 +24,10 @@ import ( ) func RunMount(option *MountOptions, umask os.FileMode) bool { + if err := configureMountMemory(option); err != nil { + fmt.Println(err) + return false + } // basic checks chunkSizeLimitMB := *mountOptions.chunkSizeLimitMB diff --git a/weed/command/mount_windows.go b/weed/command/mount_windows.go index 75e372bca..9cf4d0aff 100644 --- a/weed/command/mount_windows.go +++ b/weed/command/mount_windows.go @@ -30,6 +30,11 @@ const ownedByMounter = ^uint32(0) const windowsCacheTimeout = time.Second func RunMount(option *MountOptions, umask os.FileMode) bool { + if err := configureMountMemory(option); err != nil { + fmt.Println(err) + return false + } + chunkSizeLimitMB := *mountOptions.chunkSizeLimitMB if chunkSizeLimitMB <= 0 { fmt.Printf("Please specify a reasonable buffer size.\n") diff --git a/weed/filer/filechunk_group.go b/weed/filer/filechunk_group.go index d017d223a..9bba17b6e 100644 --- a/weed/filer/filechunk_group.go +++ b/weed/filer/filechunk_group.go @@ -28,7 +28,7 @@ type ChunkGroup struct { // - Read-ahead prefetch parallelism // - Number of concurrent section reads for large files // If concurrentReaders <= 0, defaults to 16. -func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_cache.ChunkCache, chunks []*filer_pb.FileChunk, concurrentReaders int, cacheInvalidator CacheInvalidator) (*ChunkGroup, error) { +func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_cache.ChunkCache, chunks []*filer_pb.FileChunk, concurrentReaders int, cacheInvalidator CacheInvalidator, budgets ...*ReaderCacheBudget) (*ChunkGroup, error) { if concurrentReaders <= 0 { concurrentReaders = 16 } @@ -43,7 +43,7 @@ func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_ group := &ChunkGroup{ lookupFn: lookupFn, sections: make(map[SectionIndex]*FileChunkSection), - readerCache: NewReaderCache(readerCacheLimit, chunkCache, lookupFn, cacheInvalidator), + readerCache: NewReaderCache(readerCacheLimit, chunkCache, lookupFn, cacheInvalidator, budgets...), concurrentReaders: concurrentReaders, cacheInvalidator: cacheInvalidator, } @@ -306,3 +306,8 @@ func (group *ChunkGroup) doSearchChunks(ctx context.Context, offset, fileSize in } return true, fileSize } + +func (group *ChunkGroup) Close() error { + group.readerCache.destroy() + return nil +} diff --git a/weed/filer/reader_cache.go b/weed/filer/reader_cache.go index 577d6776e..2177510bc 100644 --- a/weed/filer/reader_cache.go +++ b/weed/filer/reader_cache.go @@ -28,6 +28,7 @@ type ReaderCache struct { sync.Mutex downloaders map[string]*SingleChunkCacher limit int + budget *ReaderCacheBudget } type SingleChunkCacher struct { @@ -46,9 +47,17 @@ type SingleChunkCacher struct { done chan struct{} // signals when download is complete } -func NewReaderCache(limit int, chunkCache chunk_cache.ChunkCache, lookupFileIdFn wdclient.LookupFileIdFunctionType, cacheInvalidator CacheInvalidator) *ReaderCache { +func NewReaderCache(limit int, chunkCache chunk_cache.ChunkCache, lookupFileIdFn wdclient.LookupFileIdFunctionType, cacheInvalidator CacheInvalidator, budgets ...*ReaderCacheBudget) *ReaderCache { + var budget *ReaderCacheBudget + if len(budgets) > 0 { + budget = budgets[0] + } + if budget == nil { + budget = NewReaderCacheBudget(DefaultReaderCacheMemoryLimit) + } return &ReaderCache{ limit: limit, + budget: budget, chunkCache: chunkCache, lookupFileIdFn: lookupFileIdFn, cacheInvalidator: cacheInvalidator, @@ -105,6 +114,7 @@ func (rc *ReaderCache) MaybeCache(chunkViews *Interval[*ChunkView], count int) { } func (rc *ReaderCache) ReadChunkAt(ctx context.Context, buffer []byte, fileId string, cipherKey []byte, isGzipped bool, offset int64, chunkSize int, shouldCache bool) (int, error) { +retry: rc.Lock() for { @@ -151,7 +161,9 @@ func (rc *ReaderCache) ReadChunkAt(ctx context.Context, buffer []byte, fileId st if oldestFid != "" { oldDownloader := rc.downloaders[oldestFid] delete(rc.downloaders, oldestFid) + rc.Unlock() oldDownloader.destroy() + goto retry } } @@ -169,22 +181,31 @@ func (rc *ReaderCache) ReadChunkAt(ctx context.Context, buffer []byte, fileId st func (rc *ReaderCache) UnCache(fileId string) { rc.Lock() - defer rc.Unlock() - // glog.V(4).Infof("uncache %s", fileId) - if downloader, found := rc.downloaders[fileId]; found { + downloader := rc.downloaders[fileId] + delete(rc.downloaders, fileId) + rc.Unlock() + if downloader != nil { downloader.destroy() - delete(rc.downloaders, fileId) } } +func (rc *ReaderCache) remove(downloader *SingleChunkCacher) { + rc.Lock() + if rc.downloaders[downloader.chunkFileId] == downloader { + delete(rc.downloaders, downloader.chunkFileId) + } + rc.Unlock() + downloader.destroy() +} + func (rc *ReaderCache) destroy() { rc.Lock() - defer rc.Unlock() - - for _, downloader := range rc.downloaders { + downloaders := rc.downloaders + rc.downloaders = make(map[string]*SingleChunkCacher) + rc.Unlock() + for _, downloader := range downloaders { downloader.destroy() } - } func newSingleChunkCacher(parent *ReaderCache, fileId string, cipherKey []byte, isGzipped bool, chunkSize int, shouldCache bool) *SingleChunkCacher { @@ -200,27 +221,31 @@ func newSingleChunkCacher(parent *ReaderCache, fileId string, cipherKey []byte, } } -// startCaching downloads the chunk data in the background. -// It does NOT hold the lock during the HTTP download to allow concurrent readers -// to wait efficiently using the done channel. -// -// Concurrent downloads of the same chunk are already deduplicated by the -// ReaderCache.downloaders map (guarded by the ReaderCache mutex). Each fileId -// has at most one active SingleChunkCacher at any time. +// startCaching downloads a chunk shared by concurrent readers. func (s *SingleChunkCacher) startCaching() { s.wg.Add(1) - defer s.wg.Done() - defer close(s.done) // guarantee completion signal even on panic + defer func() { + close(s.done) + s.wg.Done() + if s.hasCompletedError() { + s.parent.remove(s) + } else { + s.parent.budget.complete(s) + } + }() - s.cacheStartedCh <- struct{}{} // signal that we've started + s.cacheStartedCh <- struct{}{} + if err := s.parent.budget.reserve(s); err != nil { + s.setError(err) + return + } - // Note: We intentionally use context.Background() here, NOT a request-specific context. - // The downloaded chunk is a shared resource - multiple concurrent readers may be waiting - // for this same download to complete. If we used a request context and that request was - // cancelled, it would abort the download and cause errors for all other waiting readers. - // The download should always complete once started to serve all potential consumers. - - // Lookup file ID without holding the lock + // Intentionally use context.Background(), not a request-specific context. + // The downloaded chunk is a shared resource: multiple concurrent readers may + // wait on this same download via s.done. A request-scoped context that got + // cancelled would abort the download and error every other waiting reader. + // The download always runs to completion once started; readers that cancel + // individually drop out via readChunkAt's select on ctx.Done(). urlStrings, err := s.parent.lookupFileIdFn(context.Background(), s.chunkFileId) if err != nil { s.setError(fmt.Errorf("operation LookupFileId %s failed, err: %v", s.chunkFileId, err)) @@ -295,12 +320,12 @@ func (s *SingleChunkCacher) destroy() { // wait for all reads to finish before destroying the data s.wg.Wait() s.Lock() - defer s.Unlock() - if s.data != nil { mem.Free(s.data) s.data = nil } + s.Unlock() + s.parent.budget.release(s) } // readChunkAt reads data from the cached chunk. diff --git a/weed/filer/reader_cache_budget.go b/weed/filer/reader_cache_budget.go new file mode 100644 index 000000000..fc08869f0 --- /dev/null +++ b/weed/filer/reader_cache_budget.go @@ -0,0 +1,88 @@ +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{}) + } +} diff --git a/weed/filer/reader_cache_memory_test.go b/weed/filer/reader_cache_memory_test.go new file mode 100644 index 000000000..7bdf9e072 --- /dev/null +++ b/weed/filer/reader_cache_memory_test.go @@ -0,0 +1,249 @@ +package filer + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" +) + +func TestChunkGroupReaderCacheMemory(t *testing.T) { + budget := NewReaderCacheBudget(8 << 10) + groups := make([]*ChunkGroup, 32) + for i := range groups { + group, err := NewChunkGroup(func(context.Context, string) ([]string, error) { return []string{"unused"}, nil }, newMockChunkCacheForReaderCache(), nil, 128, nil, budget) + if err != nil { + t.Fatal(err) + } + groups[i] = group + defer group.Close() + group.readerCache.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) { + buffer[0] = 42 + return len(buffer), nil + } + buffer := make([]byte, 1) + n, err := group.readerCache.ReadChunkAt(context.Background(), buffer, fmt.Sprint(i), nil, false, 0, 3<<10, false) + if err != nil || n != 1 || buffer[0] != 42 { + t.Fatalf("read %d: n=%d data=%v err=%v", i, n, buffer, err) + } + budget.Lock() + used := budget.used + budget.Unlock() + if used > 8<<10 { + t.Fatalf("shared budget used %d bytes", used) + } + } + groups[0].readerCache.Lock() + retained := len(groups[0].readerCache.downloaders) + groups[0].readerCache.Unlock() + if retained != 0 { + t.Fatalf("first file still retains %d downloaders", retained) + } + for _, group := range groups { + _ = group.Close() + } + budget.Lock() + defer budget.Unlock() + if budget.used != 0 { + t.Fatalf("closed files still reserve %d bytes", budget.used) + } +} + +func TestReaderCacheBudgetInFlight(t *testing.T) { + for _, prefetch := range []bool{false, true} { + t.Run(fmt.Sprintf("prefetch=%t", prefetch), func(t *testing.T) { + budget := NewReaderCacheBudget(8 << 10) + started := make(chan struct{}, 4) + gate := make(chan struct{}) + var readers sync.WaitGroup + var caches []*ReaderCache + for i := 0; i < 4; i++ { + rc := NewReaderCache(256, newMockChunkCacheForReaderCache(), func(context.Context, string) ([]string, error) { return []string{"unused"}, nil }, nil, budget) + caches = append(caches, rc) + defer rc.destroy() + rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) { + started <- struct{}{} + <-gate + buffer[0] = 42 + return len(buffer), nil + } + if prefetch { + rc.MaybeCache(&Interval[*ChunkView]{Value: &ChunkView{FileId: "chunk", ChunkSize: 3 << 10}}, 1) + } else { + readers.Add(1) + go func() { + defer readers.Done() + buffer := make([]byte, 1) + n, err := rc.ReadChunkAt(context.Background(), buffer, "chunk", nil, false, 0, 3<<10, false) + if err != nil || n != 1 || buffer[0] != 42 { + t.Errorf("read: n=%d data=%v err=%v", n, buffer, err) + } + }() + } + } + for i := 0; i < 2; i++ { + <-started + } + select { + case <-started: + t.Error("third download allocated before budget was released") + case <-time.After(50 * time.Millisecond): + } + budget.Lock() + used := budget.used + budget.Unlock() + if used != 8<<10 { + t.Errorf("in-flight reservations = %d, want 8192", used) + } + close(gate) + readers.Wait() + for i := 0; i < 2; i++ { + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("download did not resume after eviction") + } + } + for _, rc := range caches { + rc.destroy() + } + budget.Lock() + defer budget.Unlock() + if budget.used != 0 { + t.Errorf("reservations leaked: %d", budget.used) + } + }) + } +} + +func TestReaderCacheBudgetOversizedChunk(t *testing.T) { + rc := NewReaderCache(256, newMockChunkCacheForReaderCache(), func(context.Context, string) ([]string, error) { + t.Error("oversized chunk performed lookup") + return nil, nil + }, nil, NewReaderCacheBudget(3<<10)) + defer rc.destroy() + _, err := rc.ReadChunkAt(context.Background(), make([]byte, 1), "chunk", nil, false, 0, 3<<10, false) + if err == nil { + t.Fatal("expected pooled buffer larger than budget to be rejected") + } +} + +func TestReaderCacheEvictionDoesNotHoldCacheLock(t *testing.T) { + rc := NewReaderCache(2, newMockChunkCacheForReaderCache(), func(context.Context, string) ([]string, error) { return []string{"unused"}, nil }, nil) + defer rc.destroy() + rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) { + return len(buffer), nil + } + if _, err := rc.ReadChunkAt(context.Background(), make([]byte, 1), "chunk", nil, false, 0, 1024, false); err != nil { + t.Fatal(err) + } + rc.Lock() + downloader := rc.downloaders["chunk"] + downloader.wg.Add(1) + rc.Unlock() + evicted := make(chan struct{}) + go func() { rc.UnCache("chunk"); close(evicted) }() + deadline := time.Now().Add(5 * time.Second) + available := false + for time.Now().Before(deadline) { + if rc.TryLock() { + available = rc.downloaders["chunk"] == nil + rc.Unlock() + if available { + break + } + } + time.Sleep(time.Millisecond) + } + downloader.wg.Done() + <-evicted + if !available { + t.Fatal("cache lock held while eviction waited for a reader") + } +} + +func TestReaderCacheFailedPrefetchReleasesBudget(t *testing.T) { + for _, lookupFailure := range []bool{false, true} { + t.Run(fmt.Sprintf("lookupFailure=%t", lookupFailure), func(t *testing.T) { + budget := NewReaderCacheBudget(1024) + rc := NewReaderCache(1, newMockChunkCacheForReaderCache(), func(context.Context, string) ([]string, error) { + if lookupFailure { + return nil, fmt.Errorf("lookup failed") + } + return []string{"unused"}, nil + }, nil, budget) + defer rc.destroy() + rc.fetchChunkDataFn = func(_ context.Context, _ []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) { + return 0, fmt.Errorf("fetch failed") + } + rc.MaybeCache(&Interval[*ChunkView]{Value: &ChunkView{FileId: "failed", ChunkSize: 1024}}, 1) + deadline := time.Now().Add(5 * time.Second) + for { + rc.Lock() + count := len(rc.downloaders) + rc.Unlock() + budget.Lock() + used := budget.used + budget.Unlock() + if count == 0 && used == 0 { + break + } + if time.Now().After(deadline) { + t.Fatalf("failed prefetch retains %d slots and %d bytes", count, used) + } + time.Sleep(time.Millisecond) + } + }) + } +} + +// TestReaderCacheReReadAfterEviction verifies that a chunk evicted by budget +// pressure is transparently re-downloaded on the next read and returns the +// correct data. This is the core correctness property of eviction: a reader +// must never observe missing or stale data after a chunk has been evicted. +func TestReaderCacheReReadAfterEviction(t *testing.T) { + budget := NewReaderCacheBudget(4 << 10) // fits exactly one 4 KiB pooled chunk + rc := NewReaderCache(256, newMockChunkCacheForReaderCache(), func(context.Context, string) ([]string, error) { + return []string{"unused"}, nil + }, nil, budget) + defer rc.destroy() + + var fetchCount int32 + rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) { + n := atomic.AddInt32(&fetchCount, 1) + buffer[0] = byte(n) // each download writes a distinct value + return len(buffer), nil + } + + // Read chunk "a": triggers download #1, fills the budget. + buf := make([]byte, 1) + if n, err := rc.ReadChunkAt(context.Background(), buf, "a", nil, false, 0, 4<<10, false); err != nil || n != 1 || buf[0] != 1 { + t.Fatalf("first read of 'a': n=%d data=%d err=%v", n, buf[0], err) + } + + // Read chunk "b": budget only fits one chunk, so "a" is evicted to make room. + if n, err := rc.ReadChunkAt(context.Background(), buf, "b", nil, false, 0, 4<<10, false); err != nil || n != 1 || buf[0] != 2 { + t.Fatalf("read of 'b': n=%d data=%d err=%v", n, buf[0], err) + } + + // "a" should no longer be in the cache. + rc.Lock() + _, stillCached := rc.downloaders["a"] + rc.Unlock() + if stillCached { + t.Fatal("chunk 'a' was not evicted by budget pressure") + } + + // Re-read "a": must trigger download #3 and return the fresh value. + if n, err := rc.ReadChunkAt(context.Background(), buf, "a", nil, false, 0, 4<<10, false); err != nil || n != 1 || buf[0] != 3 { + t.Fatalf("re-read of 'a': n=%d data=%d err=%v (expected re-download with value 3)", n, buf[0], err) + } + + if got := atomic.LoadInt32(&fetchCount); got != 3 { + t.Fatalf("fetchCount=%d, want 3 (a, b, a-re-read)", got) + } +} diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 2526bce32..9275663d8 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -128,8 +128,11 @@ func (fh *FileHandle) SetEntry(entry *filer_pb.Entry) { if entry != nil { fileSize := filer.FileSize(entry) entry.Attributes.FileSize = fileSize + if fh.entryChunkGroup != nil { + _ = fh.entryChunkGroup.Close() + } var resolveManifestErr error - fh.entryChunkGroup, resolveManifestErr = filer.NewChunkGroup(fh.wfs.LookupFn(), fh.wfs.chunkCache, entry.Chunks, fh.wfs.option.ConcurrentReaders, fh.wfs.CacheInvalidator()) + fh.entryChunkGroup, resolveManifestErr = filer.NewChunkGroup(fh.wfs.LookupFn(), fh.wfs.chunkCache, entry.Chunks, fh.wfs.option.ConcurrentReaders, fh.wfs.CacheInvalidator(), fh.wfs.readerCacheBudget) if resolveManifestErr != nil { glog.Warningf("failed to resolve manifest chunks in %+v", entry) } @@ -219,6 +222,9 @@ func (fh *FileHandle) ReleaseHandle() { fhActiveLock := fh.wfs.fhLockTable.AcquireLock("ReleaseHandle", fh.fh, util.ExclusiveLock) defer fh.wfs.fhLockTable.ReleaseLock(fh.fh, fhActiveLock) + if fh.entryChunkGroup != nil { + _ = fh.entryChunkGroup.Close() + } fh.dirtyPages.Destroy() if IsDebugFileReadWrite { fh.mirrorFile.Close() diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index e16584c50..b126a3ad5 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -49,6 +49,7 @@ type Option struct { ChunkSizeLimit int64 ConcurrentWriters int ConcurrentReaders int + ReaderCacheSizeMB int64 CacheDirForRead string CacheSizeMBForRead int64 CacheDirForWrite string @@ -139,6 +140,7 @@ type WFS struct { metaCache *meta_cache.MetaCache stats statsCache chunkCache *chunk_cache.TieredChunkCache + readerCacheBudget *filer.ReaderCacheBudget writeBufferAccountant *page_writer.WriteBufferAccountant signature int32 concurrentWriters *util.LimitedConcurrentExecutor @@ -249,6 +251,7 @@ func NewSeaweedFileSystem(option *Option) *WFS { wfs := &WFS{ RawFileSystem: fuse.NewDefaultRawFileSystem(), option: option, + readerCacheBudget: filer.NewReaderCacheBudget(option.ReaderCacheSizeMB << 20), signature: util.RandomInt32(), inodeToPath: NewInodeToPath(util.FullPath(option.FilerMountRootPath), option.CacheMetaTTlSec), fhMap: NewFileHandleToInode(), diff --git a/weed/util/mem/slot_pool.go b/weed/util/mem/slot_pool.go index 70f2525b3..897034e64 100644 --- a/weed/util/mem/slot_pool.go +++ b/weed/util/mem/slot_pool.go @@ -39,6 +39,13 @@ func getSlotPool(size int) (*sync.Pool, bool) { return pools[index], true } +func AllocationSize(size int) int { + if _, found := getSlotPool(size); found { + return min_size << bitCount(size) + } + return size +} + func Allocate(size int) []byte { if pool, found := getSlotPool(size); found { slab := *pool.Get().(*[]byte) diff --git a/weed/util/mem/slot_pool_test.go b/weed/util/mem/slot_pool_test.go index d457a7490..620f2da8b 100644 --- a/weed/util/mem/slot_pool_test.go +++ b/weed/util/mem/slot_pool_test.go @@ -47,3 +47,12 @@ func TestBitCount(t *testing.T) { } } + +func TestAllocationSize(t *testing.T) { + for _, size := range []int{0, 1, 1024, 1025, 4883, 2 << 20} { + buf := Allocate(size) + assert.Equal(t, cap(buf), AllocationSize(size)) + Free(buf) + } + assert.Equal(t, 257<<20, AllocationSize(257<<20)) +}