diff --git a/weed/filer/reader_cache.go b/weed/filer/reader_cache.go index 577d6776e..62026d435 100644 --- a/weed/filer/reader_cache.go +++ b/weed/filer/reader_cache.go @@ -20,6 +20,8 @@ type CacheInvalidator interface { type fetchChunkDataFnType func(ctx context.Context, buffer []byte, urlStrings []string, cipherKey []byte, isGzipped bool, isFullChunk bool, offset int64, fileId string, refreshUrls util_http.RefreshUrlsFunc) (n int, err error) +const readerCacheMemoryLimit = 64 << 20 + type ReaderCache struct { chunkCache chunk_cache.ChunkCache lookupFileIdFn wdclient.LookupFileIdFunctionType @@ -177,6 +179,30 @@ func (rc *ReaderCache) UnCache(fileId string) { } } +func (rc *ReaderCache) trim() { + rc.Lock() + defer rc.Unlock() + + for { + var retained int64 + var oldest *SingleChunkCacher + for _, downloader := range rc.downloaders { + downloader.Lock() + size := cap(downloader.data) + downloader.Unlock() + retained += int64(size) + if size > 0 && (oldest == nil || atomic.LoadInt64(&downloader.completedTimeNew) < atomic.LoadInt64(&oldest.completedTimeNew)) { + oldest = downloader + } + } + if retained <= readerCacheMemoryLimit { + return + } + delete(rc.downloaders, oldest.chunkFileId) + oldest.destroy() + } +} + func (rc *ReaderCache) destroy() { rc.Lock() defer rc.Unlock() @@ -200,27 +226,19 @@ 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() + // Release download waiters before eviction waits for active readers. + s.parent.trim() + }() s.cacheStartedCh <- struct{}{} // signal that we've started - // 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 + // Request cancellation must not abort a download shared by other readers. 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)) diff --git a/weed/filer/reader_cache_memory_test.go b/weed/filer/reader_cache_memory_test.go new file mode 100644 index 000000000..54706aff3 --- /dev/null +++ b/weed/filer/reader_cache_memory_test.go @@ -0,0 +1,103 @@ +package filer + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + util_http "github.com/seaweedfs/seaweedfs/weed/util/http" +) + +func TestChunkGroupReaderCacheMemory(t *testing.T) { + for _, tt := range []struct { + name string + chunkSize int + chunks int + mode string + }{ + {"default chunks", 2 << 20, 40, "sequential"}, + {"pooled buffers", 3 << 20, 20, "sequential"}, + {"oversized chunk", 65 << 20, 1, "sequential"}, + {"concurrent readers", 2 << 20, 40, "concurrent"}, + {"prefetch", 2 << 20, 40, "prefetch"}, + } { + t.Run(tt.name, func(t *testing.T) { + group, err := NewChunkGroup(func(context.Context, string) ([]string, error) { + return []string{"unused"}, nil + }, newMockChunkCacheForReaderCache(), nil, 128, nil) + if err != nil { + t.Fatal(err) + } + rc := group.readerCache + defer rc.destroy() + started := make(chan struct{}, tt.chunks) + gate := make(chan struct{}) + if tt.mode == "sequential" { + close(gate) + } + 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 + } + var readers sync.WaitGroup + var views *Interval[*ChunkView] + for i := 0; i < tt.chunks; i++ { + read := func() { + buffer := make([]byte, 1) + n, err := rc.ReadChunkAt(context.Background(), buffer, fmt.Sprint(i), nil, false, 0, tt.chunkSize, false) + if err != nil || n != 1 || buffer[0] != 42 { + t.Errorf("read %d: n=%d, data=%v, err=%v", i, n, buffer, err) + } + } + switch tt.mode { + case "sequential": + read() + case "concurrent": + readers.Add(1) + go func() { defer readers.Done(); read() }() + case "prefetch": + views = &Interval[*ChunkView]{Value: &ChunkView{FileId: fmt.Sprint(i), ChunkSize: uint64(tt.chunkSize)}, Next: views} + } + } + if tt.mode == "prefetch" { + rc.MaybeCache(views, tt.chunks) + } + if tt.mode != "sequential" { + for i := 0; i < tt.chunks; i++ { + <-started + } + close(gate) + readers.Wait() + } + + deadline := time.Now().Add(time.Second) + for { + rc.Lock() + retained := 0 + completed := true + for _, downloader := range rc.downloaders { + select { + case <-downloader.done: + default: + completed = false + } + downloader.Lock() + retained += cap(downloader.data) + downloader.Unlock() + } + rc.Unlock() + if completed && retained <= 64<<20 { + break + } + if time.Now().After(deadline) { + t.Fatalf("reader cache retained %d MiB, want at most 64 MiB", retained>>20) + } + time.Sleep(time.Millisecond) + } + }) + } +}