From 8db41d02177a0d6b9b4d56621d0debadb1fa4977 Mon Sep 17 00:00:00 2001 From: ssshr-66 Date: Sun, 13 Sep 2026 11:06:25 +0800 Subject: [PATCH] [Mount] Cache Chunk Manifest Resolution for Repeated File Opens (#11266) * cache resolved chunk manifests for Mount * Address PR review: per-mount cache, singleflight, reuse ResolveOneChunkManifest - Own the manifest cache per WFS mount instead of a process-global variable, so manifests from one filer backend are never served to another (Devin/CodeRabbit major bug). - Coalesce concurrent cold misses via singleflight so only one fetch runs during a cold burst (Greptile P2). - Copy cached data after releasing the mutex so a large copy does not block concurrent hits, inserts, and evictions (CodeRabbit nitpick). - Reuse the existing ResolveOneChunkManifest function name instead of introducing a new resolveOneChunkManifest wrapper. - Validate (unmarshal) manifest bytes before caching so malformed manifests do not poison the cache. - Add TestChunkGroupManifestResolutionCoalescesColdMisses covering the singleflight cold-miss path. * Address round 2 review: coalesced-miss cancellation, test overlap - Use singleflight.DoChan in fetchOrLoad and select on ctx.Done() so a caller whose context is canceled while waiting for an in-flight fetch returns ctx.Err() promptly instead of blocking for the leader's result (Devin BUG). - Add TestResolveOneChunkManifestCanceledWaiterReturnsDuringCoalescedMiss covering the canceled-waiter path. - Delay the cold-miss fixture response so the leader's fetch is still in flight when concurrent opens join the singleflight, making the one-fetch assertions reliable (CodeRabbit Minor). * Address review: keep ResolveOneChunkManifest four-argument Restore the exported ResolveOneChunkManifest to its original four-argument signature so external callers keep compiling. Move the cache-aware resolution into an unexported resolveOneChunkManifest helper that accepts the per-mount ChunkManifestCache. The exported function delegates to the helper with a nil cache, preserving the historical uncached behavior for every non-Mount caller. The Mount path (ChunkGroup.SetChunks) now calls the unexported helper with the mount-owned cache. Tests and benchmarks that exercise the cache path call the unexported helper directly. --------- Co-authored-by: Chris Lu Co-authored-by: Chris Lu --- weed/filer/filechunk_group.go | 8 +- weed/filer/filechunk_group_test.go | 4 +- weed/filer/filechunk_manifest.go | 68 +++- weed/filer/filechunk_manifest_cache.go | 146 +++++++++ ...filechunk_manifest_cache_benchmark_test.go | 72 +++++ weed/filer/filechunk_manifest_cache_test.go | 305 ++++++++++++++++++ weed/filer/reader_cache_memory_test.go | 2 +- weed/mount/filehandle.go | 2 +- weed/mount/weedfs.go | 2 + weed/mount/weedfs_attr_race_test.go | 4 +- weed/mount/weedfs_attr_unlinked_test.go | 2 +- 11 files changed, 595 insertions(+), 20 deletions(-) create mode 100644 weed/filer/filechunk_manifest_cache.go create mode 100644 weed/filer/filechunk_manifest_cache_benchmark_test.go create mode 100644 weed/filer/filechunk_manifest_cache_test.go diff --git a/weed/filer/filechunk_group.go b/weed/filer/filechunk_group.go index 6cc45a041..f920a7a7d 100644 --- a/weed/filer/filechunk_group.go +++ b/weed/filer/filechunk_group.go @@ -20,6 +20,9 @@ type ChunkGroup struct { concurrentReaders int // cacheInvalidator lets manifest resolution drop stale volume locations, as ReaderCache does for chunk reads cacheInvalidator CacheInvalidator + // manifestCache caches resolved chunk manifest bytes across repeated opens + // for the same mount. nil for non-mount callers (no caching). + manifestCache *ChunkManifestCache // resolveErr is set when chunk manifest resolution failed, guarded by // sectionsLock. Reads must fail with this error instead of silently // zero-filling the unresolved sections as if they were sparse holes. @@ -32,7 +35,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, budgets ...*ReaderCacheBudget) (*ChunkGroup, error) { +func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_cache.ChunkCache, chunks []*filer_pb.FileChunk, concurrentReaders int, cacheInvalidator CacheInvalidator, manifestCache *ChunkManifestCache, budgets ...*ReaderCacheBudget) (*ChunkGroup, error) { if concurrentReaders <= 0 { concurrentReaders = 16 } @@ -50,6 +53,7 @@ func NewChunkGroup(lookupFn wdclient.LookupFileIdFunctionType, chunkCache chunk_ readerCache: NewReaderCache(readerCacheLimit, chunkCache, lookupFn, cacheInvalidator, budgets...), concurrentReaders: concurrentReaders, cacheInvalidator: cacheInvalidator, + manifestCache: manifestCache, } err := group.SetChunks(chunks) @@ -241,7 +245,7 @@ func (group *ChunkGroup) SetChunks(chunks []*filer_pb.FileChunk) error { continue } - resolvedChunks, err := ResolveOneChunkManifest(context.Background(), group.lookupFn, chunk, group.cacheInvalidator) + resolvedChunks, err := resolveOneChunkManifest(context.Background(), group.lookupFn, chunk, group.cacheInvalidator, group.manifestCache) if err != nil { // remember the failure so ReadDataAt returns an error instead of // treating the unresolved sections as sparse holes diff --git a/weed/filer/filechunk_group_test.go b/weed/filer/filechunk_group_test.go index 4926f2809..3bfa4a7f9 100644 --- a/weed/filer/filechunk_group_test.go +++ b/weed/filer/filechunk_group_test.go @@ -421,7 +421,7 @@ func TestChunkGroup_SearchChunks(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - group, err := NewChunkGroup(nil, nil, tt.chunks, 1, nil) + group, err := NewChunkGroup(nil, nil, tt.chunks, 1, nil, nil) if !assert.NoError(t, err) { return } @@ -447,7 +447,7 @@ func TestChunkGroup_ReadDataAt_ManifestResolveFailure(t *testing.T) { {FileId: "1,1679011dc64abd40", IsChunkManifest: true, Offset: 0, Size: 1 << 20}, } - group, err := NewChunkGroup(lookupFn, nil, chunks, 1, nil) + group, err := NewChunkGroup(lookupFn, nil, chunks, 1, nil, nil) assert.Error(t, err, "manifest resolution should fail") // Reads must fail with the resolve error, not silently return zeros. diff --git a/weed/filer/filechunk_manifest.go b/weed/filer/filechunk_manifest.go index 597861a87..015bbe790 100644 --- a/weed/filer/filechunk_manifest.go +++ b/weed/filer/filechunk_manifest.go @@ -287,22 +287,69 @@ func (r *chunkManifestResolver) resolve(chunks []*filer_pb.FileChunk, startOffse return } +// ResolveOneChunkManifest fetches and decodes a single manifest chunk. It is +// the uncached, exported path used by every non-Mount caller; the Mount path +// routes through resolveOneChunkManifest so it can share a per-mount cache. +// Keeping this signature stable preserves the existing four-argument contract +// for external callers. func ResolveOneChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunk *filer_pb.FileChunk, invalidator CacheInvalidator) (dataChunks []*filer_pb.FileChunk, manifestResolveErr error) { + return resolveOneChunkManifest(ctx, lookupFileIdFn, chunk, invalidator, nil) +} + +// resolveOneChunkManifest is the cache-aware implementation. cache may be nil, +// in which case the manifest is fetched and validated on every call, matching +// the historical uncached behavior. A non-nil cache is owned by a single mount +// (WFS) and coalesces concurrent cold misses via singleflight. +func resolveOneChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunk *filer_pb.FileChunk, invalidator CacheInvalidator, cache *ChunkManifestCache) (dataChunks []*filer_pb.FileChunk, manifestResolveErr error) { if !chunk.IsChunkManifest { return } - - // IsChunkManifest - bytesBuffer := bytesBufferPool.Get().(*bytes.Buffer) - bytesBuffer.Reset() - defer bytesBufferPool.Put(bytesBuffer) - err := fetchWholeChunk(ctx, bytesBuffer, lookupFileIdFn, chunk.GetFileIdString(), chunk.CipherKey, chunk.IsCompressed, invalidator) - if err != nil { - return nil, fmt.Errorf("fail to read manifest %s: %w", chunk.GetFileIdString(), err) + if err := ctx.Err(); err != nil { + return nil, err } + + key := chunkManifestCacheKey{ + fileID: chunk.GetFileIdString(), + cipherKey: string(chunk.CipherKey), + isCompressed: chunk.IsCompressed, + } + + fetch := func() ([]byte, error) { + bytesBuffer := bytesBufferPool.Get().(*bytes.Buffer) + bytesBuffer.Reset() + defer bytesBufferPool.Put(bytesBuffer) + if err := fetchWholeChunk(ctx, bytesBuffer, lookupFileIdFn, key.fileID, chunk.CipherKey, chunk.IsCompressed, invalidator); err != nil { + return nil, fmt.Errorf("fail to read manifest %s: %w", key.fileID, err) + } + // Copy before the buffer returns to the pool so concurrent callers + // cannot overwrite the slice before it is cached. + data := append([]byte(nil), bytesBuffer.Bytes()...) + // Validate before returning so fetchOrLoad only caches well-formed + // manifests. A malformed manifest must not poison the cache. + if err := proto.Unmarshal(data, &filer_pb.FileChunkManifest{}); err != nil { + return nil, fmt.Errorf("fail to unmarshal manifest %s: %w", key.fileID, err) + } + return data, nil + } + + var manifestBytes []byte + if cache != nil { + data, err := cache.fetchOrLoad(ctx, key, fetch) + if err != nil { + return nil, err + } + manifestBytes = data + } else { + data, err := fetch() + if err != nil { + return nil, err + } + manifestBytes = data + } + m := &filer_pb.FileChunkManifest{} - if err := proto.Unmarshal(bytesBuffer.Bytes(), m); err != nil { - return nil, fmt.Errorf("fail to unmarshal manifest %s: %w", chunk.GetFileIdString(), err) + if err := proto.Unmarshal(manifestBytes, m); err != nil { + return nil, fmt.Errorf("fail to unmarshal manifest %s: %w", key.fileID, err) } // recursive @@ -310,7 +357,6 @@ func ResolveOneChunkManifest(ctx context.Context, lookupFileIdFn wdclient.Lookup return m.Chunks, nil } -// TODO fetch from cache for weed mount? func fetchWholeChunk(ctx context.Context, bytesBuffer *bytes.Buffer, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool, invalidator CacheInvalidator) error { urlStrings, err := lookupFileIdFn(ctx, fileId) if err != nil { diff --git a/weed/filer/filechunk_manifest_cache.go b/weed/filer/filechunk_manifest_cache.go new file mode 100644 index 000000000..3fc7430d9 --- /dev/null +++ b/weed/filer/filechunk_manifest_cache.go @@ -0,0 +1,146 @@ +package filer + +import ( + "container/list" + "context" + "fmt" + "sync" + + "golang.org/x/sync/singleflight" +) + +const ( + MaxMountChunkManifestCacheEntries = 256 + MaxMountChunkManifestCacheBytes = 64 << 20 +) + +type chunkManifestCacheKey struct { + fileID string + cipherKey string + isCompressed bool +} + +func (k chunkManifestCacheKey) flightKey() string { + return fmt.Sprintf("%s\x00%s\x00%t", k.fileID, k.cipherKey, k.isCompressed) +} + +type chunkManifestCacheEntry struct { + key chunkManifestCacheKey + data []byte +} + +// ChunkManifestCache is a bounded, thread-safe LRU cache for chunk manifest +// bytes. Each mount (WFS) owns its own instance so manifests fetched through +// one filer backend are never served to another. Concurrent cold misses for +// the same key are coalesced via singleflight so only one fetch runs. +type ChunkManifestCache struct { + mu sync.Mutex + maxEntries int + maxBytes int64 + bytes int64 + entries map[chunkManifestCacheKey]*list.Element + lru *list.List + flight singleflight.Group +} + +// NewChunkManifestCache creates a bounded LRU cache for chunk manifest bytes. +func NewChunkManifestCache(maxEntries int, maxBytes int64) *ChunkManifestCache { + return &ChunkManifestCache{ + maxEntries: maxEntries, + maxBytes: maxBytes, + entries: make(map[chunkManifestCacheKey]*list.Element), + lru: list.New(), + } +} + +func (c *ChunkManifestCache) get(key chunkManifestCacheKey) ([]byte, bool) { + c.mu.Lock() + element, found := c.entries[key] + if !found { + c.mu.Unlock() + return nil, false + } + c.lru.MoveToFront(element) + // entry.data is immutable after insertion; copy outside the lock so a + // large copy does not block concurrent hits, inserts, and evictions. + data := element.Value.(*chunkManifestCacheEntry).data + c.mu.Unlock() + return append([]byte(nil), data...), true +} + +func (c *ChunkManifestCache) put(key chunkManifestCacheKey, data []byte) { + if c.maxEntries <= 0 || int64(len(data)) > c.maxBytes { + return + } + + entry := &chunkManifestCacheEntry{ + key: key, + data: append([]byte(nil), data...), + } + + c.mu.Lock() + defer c.mu.Unlock() + + if old, found := c.entries[key]; found { + c.removeElement(old) + } + + element := c.lru.PushFront(entry) + c.entries[key] = element + c.bytes += int64(len(entry.data)) + + for len(c.entries) > c.maxEntries || c.bytes > c.maxBytes { + c.removeElement(c.lru.Back()) + } +} + +// fetchOrLoad returns cached manifest bytes for key, or invokes fetch and +// caches the result. Concurrent calls for the same key are coalesced via +// singleflight so only one fetch runs during a cold burst. A caller whose +// context is canceled while waiting for the in-flight fetch returns +// ctx.Err() immediately rather than blocking for the leader's result. +func (c *ChunkManifestCache) fetchOrLoad(ctx context.Context, key chunkManifestCacheKey, fetch func() ([]byte, error)) ([]byte, error) { + if data, ok := c.get(key); ok { + return data, nil + } + ch := c.flight.DoChan(key.flightKey(), func() (interface{}, error) { + // Re-check under the flight: another flight may have just populated + // the cache for this key. + if data, ok := c.get(key); ok { + return data, nil + } + data, err := fetch() + if err != nil { + return nil, err + } + c.put(key, data) + return data, nil + }) + select { + case <-ctx.Done(): + return nil, ctx.Err() + case res := <-ch: + if res.Err != nil { + return nil, res.Err + } + return res.Val.([]byte), nil + } +} + +func (c *ChunkManifestCache) clear() { + c.mu.Lock() + defer c.mu.Unlock() + c.entries = make(map[chunkManifestCacheKey]*list.Element) + c.lru.Init() + c.bytes = 0 +} + +func (c *ChunkManifestCache) removeElement(element *list.Element) { + if element == nil { + return + } + c.lru.Remove(element) + entry := element.Value.(*chunkManifestCacheEntry) + delete(c.entries, entry.key) + c.bytes -= int64(len(entry.data)) +} diff --git a/weed/filer/filechunk_manifest_cache_benchmark_test.go b/weed/filer/filechunk_manifest_cache_benchmark_test.go new file mode 100644 index 000000000..91903d925 --- /dev/null +++ b/weed/filer/filechunk_manifest_cache_benchmark_test.go @@ -0,0 +1,72 @@ +package filer + +import ( + "context" + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +func BenchmarkManifestResolutionRepeatedOpen(b *testing.B) { + for _, delay := range []time.Duration{0, 5 * time.Millisecond, 20 * time.Millisecond} { + b.Run(fmt.Sprintf("latency=%s", delay), func(b *testing.B) { + fixture := newManifestReadFixture(b, map[string][]*filer_pb.FileChunk{ + "benchmark-cached": {resolveTestData("benchmark-data", 0)}, + }, map[string]time.Duration{"benchmark-cached": delay}) + var lookups atomic.Int32 + lookup := func(ctx context.Context, fileID string) ([]string, error) { + lookups.Add(1) + return fixture.lookup(ctx, fileID) + } + chunk := newManifestCacheTestChunk("benchmark-cached") + cache := NewChunkManifestCache(MaxMountChunkManifestCacheEntries, MaxMountChunkManifestCacheBytes) + _, err := resolveOneChunkManifest(context.Background(), lookup, chunk, nil, cache) + if err != nil { + b.Fatal(err) + } + fixture.loads.Store(0) + lookups.Store(0) + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := resolveOneChunkManifest(context.Background(), lookup, chunk, nil, cache); err != nil { + b.Fatal(err) + } + } + b.StopTimer() + b.ReportMetric(float64(fixture.loads.Load())/float64(b.N), "fetches/op") + b.ReportMetric(float64(lookups.Load())/float64(b.N), "lookups/op") + }) + } +} + +func BenchmarkManifestResolutionWithoutCache(b *testing.B) { + for _, delay := range []time.Duration{0, 5 * time.Millisecond, 20 * time.Millisecond} { + b.Run(fmt.Sprintf("latency=%s", delay), func(b *testing.B) { + fixture := newManifestReadFixture(b, map[string][]*filer_pb.FileChunk{ + "benchmark-uncached": {resolveTestData("benchmark-data", 0)}, + }, map[string]time.Duration{"benchmark-uncached": delay}) + var lookups atomic.Int32 + lookup := func(ctx context.Context, fileID string) ([]string, error) { + lookups.Add(1) + return fixture.lookup(ctx, fileID) + } + chunk := newManifestCacheTestChunk("benchmark-uncached") + + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := ResolveOneChunkManifest(context.Background(), lookup, chunk, nil); err != nil { + b.Fatal(err) + } + } + b.StopTimer() + b.ReportMetric(float64(fixture.loads.Load())/float64(b.N), "fetches/op") + b.ReportMetric(float64(lookups.Load())/float64(b.N), "lookups/op") + }) + } +} diff --git a/weed/filer/filechunk_manifest_cache_test.go b/weed/filer/filechunk_manifest_cache_test.go new file mode 100644 index 000000000..af0ffd34c --- /dev/null +++ b/weed/filer/filechunk_manifest_cache_test.go @@ -0,0 +1,305 @@ +package filer + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +func newManifestCacheTestChunk(fileID string) *filer_pb.FileChunk { + return resolveTestManifest(fileID, 0) +} + +func newTestManifestCache(t testing.TB) *ChunkManifestCache { + t.Helper() + cache := NewChunkManifestCache(MaxMountChunkManifestCacheEntries, MaxMountChunkManifestCacheBytes) + t.Cleanup(cache.clear) + return cache +} + +func TestChunkGroupManifestResolutionCachesRepeatedOpens(t *testing.T) { + cache := newTestManifestCache(t) + const manifestID = "cache-repeated-open" + fixture := newManifestReadFixture(t, map[string][]*filer_pb.FileChunk{ + manifestID: {resolveTestData("cached-data", 0)}, + }, nil) + var lookups atomic.Int32 + lookup := func(ctx context.Context, fileID string) ([]string, error) { + lookups.Add(1) + return fixture.lookup(ctx, fileID) + } + + for i := 0; i < 3; i++ { + _, err := NewChunkGroup(lookup, nil, []*filer_pb.FileChunk{ + newManifestCacheTestChunk(manifestID), + }, 1, nil, cache) + require.NoError(t, err) + } + + require.Equal(t, int32(1), lookups.Load(), "repeated Mount opens should reuse the Manifest lookup") + require.Equal(t, int32(1), fixture.loads.Load(), "repeated Mount opens should reuse the Manifest fetch") +} + +func TestChunkGroupManifestResolutionDoesNotCacheFailedReads(t *testing.T) { + cache := newTestManifestCache(t) + const manifestID = "cache-failed-read" + fixture := newManifestReadFixture(t, nil, nil) + var lookups atomic.Int32 + lookup := func(ctx context.Context, fileID string) ([]string, error) { + lookups.Add(1) + return fixture.lookup(ctx, fileID) + } + chunk := newManifestCacheTestChunk(manifestID) + + _, err := NewChunkGroup(lookup, nil, []*filer_pb.FileChunk{chunk}, 1, nil, cache) + require.Error(t, err) + + fixture.manifests[manifestID] = manifestBytes(t, resolveTestData("retried-data", 0)) + _, err = NewChunkGroup(lookup, nil, []*filer_pb.FileChunk{chunk}, 1, nil, cache) + require.NoError(t, err) + + require.Equal(t, int32(2), lookups.Load(), "a failed lookup must not poison later Mount opens") + require.Equal(t, int32(2), fixture.loads.Load(), "a failed fetch must not poison later Mount opens") +} + +func TestChunkGroupManifestResolutionConcurrentWarmOpensReuseFetch(t *testing.T) { + cache := newTestManifestCache(t) + const manifestID = "cache-concurrent-warm-open" + fixture := newManifestReadFixture(t, map[string][]*filer_pb.FileChunk{ + manifestID: {resolveTestData("concurrent-data", 0)}, + }, nil) + var lookups atomic.Int32 + lookup := func(ctx context.Context, fileID string) ([]string, error) { + lookups.Add(1) + return fixture.lookup(ctx, fileID) + } + + _, err := NewChunkGroup(lookup, nil, []*filer_pb.FileChunk{ + newManifestCacheTestChunk(manifestID), + }, 1, nil, cache) + require.NoError(t, err) + + const concurrentOpens = 8 + var wg sync.WaitGroup + errs := make(chan error, concurrentOpens) + for i := 0; i < concurrentOpens; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _, openErr := NewChunkGroup(lookup, nil, []*filer_pb.FileChunk{ + newManifestCacheTestChunk(manifestID), + }, 1, nil, cache) + errs <- openErr + }() + } + wg.Wait() + close(errs) + for openErr := range errs { + require.NoError(t, openErr) + } + + require.Equal(t, int32(1), lookups.Load(), "warm concurrent Mount opens should not repeat the lookup") + require.Equal(t, int32(1), fixture.loads.Load(), "warm concurrent Mount opens should not repeat the fetch") +} + +func TestChunkGroupManifestResolutionCoalescesColdMisses(t *testing.T) { + cache := newTestManifestCache(t) + const manifestID = "cache-cold-miss" + // Delay the manifest response so the leader's fetch is still in flight + // when the other concurrent opens arrive and join the singleflight. + fixture := newManifestReadFixture(t, map[string][]*filer_pb.FileChunk{ + manifestID: {resolveTestData("cold-data", 0)}, + }, map[string]time.Duration{manifestID: 50 * time.Millisecond}) + var lookups atomic.Int32 + lookup := func(ctx context.Context, fileID string) ([]string, error) { + lookups.Add(1) + return fixture.lookup(ctx, fileID) + } + + const concurrentOpens = 8 + var wg sync.WaitGroup + errs := make(chan error, concurrentOpens) + for i := 0; i < concurrentOpens; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _, openErr := NewChunkGroup(lookup, nil, []*filer_pb.FileChunk{ + newManifestCacheTestChunk(manifestID), + }, 1, nil, cache) + errs <- openErr + }() + } + wg.Wait() + close(errs) + for openErr := range errs { + require.NoError(t, openErr) + } + + require.Equal(t, int32(1), lookups.Load(), "concurrent cold opens should coalesce into a single lookup") + require.Equal(t, int32(1), fixture.loads.Load(), "concurrent cold opens should coalesce into a single fetch") +} + +func TestResolveOneChunkManifestDoesNotUseMountCache(t *testing.T) { + cache := newTestManifestCache(t) + const manifestID = "cache-public-resolver" + fixture := newManifestReadFixture(t, map[string][]*filer_pb.FileChunk{ + manifestID: {resolveTestData("public-resolver-data", 0)}, + }, nil) + var lookups atomic.Int32 + lookup := func(ctx context.Context, fileID string) ([]string, error) { + lookups.Add(1) + return fixture.lookup(ctx, fileID) + } + chunk := newManifestCacheTestChunk(manifestID) + + for i := 0; i < 2; i++ { + _, err := ResolveOneChunkManifest(context.Background(), lookup, chunk, nil) + require.NoError(t, err) + } + + require.Equal(t, int32(2), lookups.Load(), "the general resolver must retain its uncached behavior") + require.Equal(t, int32(2), fixture.loads.Load(), "the general resolver must not use the Mount cache") + + // The shared cache must remain empty: a nil-cache caller must not pollute + // a mount-owned cache, and vice versa. + _, found := cache.get(chunkManifestCacheKey{fileID: manifestID}) + require.False(t, found, "a nil-cache resolve must not populate an unrelated cache") +} + +func TestChunkGroupManifestResolutionDoesNotCacheMalformedManifest(t *testing.T) { + cache := newTestManifestCache(t) + const manifestID = "cache-malformed-manifest" + fixture := newManifestReadFixture(t, map[string][]*filer_pb.FileChunk{ + manifestID: nil, + }, nil) + fixture.manifests[manifestID] = []byte("malformed manifest") + chunk := newManifestCacheTestChunk(manifestID) + + _, err := NewChunkGroup(fixture.lookup, nil, []*filer_pb.FileChunk{chunk}, 1, nil, cache) + require.Error(t, err) + + fixture.manifests[manifestID] = manifestBytes(t, resolveTestData("after-malformed-data", 0)) + _, err = NewChunkGroup(fixture.lookup, nil, []*filer_pb.FileChunk{chunk}, 1, nil, cache) + require.NoError(t, err) + require.Equal(t, int32(2), fixture.loads.Load(), "a malformed manifest must not poison later Mount opens") +} + +func TestChunkManifestCacheSeparatesReadParameters(t *testing.T) { + cache := NewChunkManifestCache(4, 1024) + base := chunkManifestCacheKey{fileID: "same-file"} + cache.put(base, []byte("manifest")) + + _, found := cache.get(chunkManifestCacheKey{fileID: "same-file", cipherKey: "key"}) + require.False(t, found, "cipher keys must be part of the cache key") + _, found = cache.get(chunkManifestCacheKey{fileID: "same-file", isCompressed: true}) + require.False(t, found, "compression settings must be part of the cache key") + _, found = cache.get(chunkManifestCacheKey{fileID: "different-file"}) + require.False(t, found, "FileIds must be part of the cache key") +} + +func TestChunkManifestCacheCopiesData(t *testing.T) { + cache := NewChunkManifestCache(1, 1024) + key := chunkManifestCacheKey{fileID: "copy-data"} + original := []byte("manifest") + cache.put(key, original) + original[0] = 'X' + + data, found := cache.get(key) + require.True(t, found) + require.Equal(t, []byte("manifest"), data) + data[0] = 'Y' + + data, found = cache.get(key) + require.True(t, found) + require.Equal(t, []byte("manifest"), data) +} + +func TestChunkManifestCacheEvictsLeastRecentlyUsedAndOversizedEntries(t *testing.T) { + cache := NewChunkManifestCache(2, 6) + first := chunkManifestCacheKey{fileID: "first"} + second := chunkManifestCacheKey{fileID: "second"} + third := chunkManifestCacheKey{fileID: "third"} + oversized := chunkManifestCacheKey{fileID: "oversized"} + + cache.put(first, []byte("one")) + cache.put(second, []byte("two")) + _, found := cache.get(first) + require.True(t, found, "the first entry should be present before eviction") + cache.put(third, []byte("tri")) + + _, found = cache.get(first) + require.True(t, found, "a recent entry should survive LRU eviction") + _, found = cache.get(second) + require.False(t, found, "the least recently used entry should be evicted") + _, found = cache.get(third) + require.True(t, found) + + cache.put(oversized, []byte("too large")) + _, found = cache.get(oversized) + require.False(t, found, "an entry over the byte limit must not be cached") +} + +func TestResolveOneChunkManifestHonorsCanceledContextOnCacheHit(t *testing.T) { + cache := NewChunkManifestCache(1, 1024) + chunk := newManifestCacheTestChunk("cache-canceled-hit") + cache.put(chunkManifestCacheKey{fileID: chunk.GetFileIdString()}, manifestBytes(t, resolveTestData("canceled-data", 0))) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + lookupCalled := false + lookup := func(context.Context, string) ([]string, error) { + lookupCalled = true + return nil, errors.New("lookup should not be called") + } + + _, err := resolveOneChunkManifest(ctx, lookup, chunk, nil, cache) + require.ErrorIs(t, err, context.Canceled) + require.False(t, lookupCalled, "a canceled cache hit must not issue a lookup") +} + +func TestResolveOneChunkManifestCanceledWaiterReturnsDuringCoalescedMiss(t *testing.T) { + cache := newTestManifestCache(t) + const manifestID = "cache-canceled-waiter" + // Delay the manifest response so the leader's fetch is still in flight + // when the waiter arrives and cancels. + fixture := newManifestReadFixture(t, map[string][]*filer_pb.FileChunk{ + manifestID: {resolveTestData("canceled-waiter-data", 0)}, + }, map[string]time.Duration{manifestID: 100 * time.Millisecond}) + chunk := newManifestCacheTestChunk(manifestID) + + // Leader starts the fetch with a live context. + leaderCtx, leaderCancel := context.WithCancel(context.Background()) + defer leaderCancel() + leaderDone := make(chan error, 1) + go func() { + _, err := resolveOneChunkManifest(leaderCtx, fixture.lookup, chunk, nil, cache) + leaderDone <- err + }() + + // Waiter cancels its own context while the leader's fetch is still in + // flight; it must return context.Canceled promptly instead of blocking + // for the leader's result. + waiterCtx, waiterCancel := context.WithCancel(context.Background()) + waiterCancel() + _, err := resolveOneChunkManifest(waiterCtx, fixture.lookup, chunk, nil, cache) + require.ErrorIs(t, err, context.Canceled) + + // The leader must still complete successfully and populate the cache. + require.NoError(t, <-leaderDone) + data, found := cache.get(chunkManifestCacheKey{fileID: manifestID}) + require.True(t, found, "the leader's successful fetch must populate the cache") + require.NotEmpty(t, data) +} + +func manifestBytes(t testing.TB, chunks ...*filer_pb.FileChunk) []byte { + t.Helper() + data, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: chunks}) + require.NoError(t, err) + return data +} diff --git a/weed/filer/reader_cache_memory_test.go b/weed/filer/reader_cache_memory_test.go index 7bdf9e072..e1a647163 100644 --- a/weed/filer/reader_cache_memory_test.go +++ b/weed/filer/reader_cache_memory_test.go @@ -15,7 +15,7 @@ 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) + group, err := NewChunkGroup(func(context.Context, string) ([]string, error) { return []string{"unused"}, nil }, newMockChunkCacheForReaderCache(), nil, 128, nil, nil, budget) if err != nil { t.Fatal(err) } diff --git a/weed/mount/filehandle.go b/weed/mount/filehandle.go index 9275663d8..8514b505e 100644 --- a/weed/mount/filehandle.go +++ b/weed/mount/filehandle.go @@ -132,7 +132,7 @@ func (fh *FileHandle) SetEntry(entry *filer_pb.Entry) { _ = 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.wfs.readerCacheBudget) + fh.entryChunkGroup, resolveManifestErr = filer.NewChunkGroup(fh.wfs.LookupFn(), fh.wfs.chunkCache, entry.Chunks, fh.wfs.option.ConcurrentReaders, fh.wfs.CacheInvalidator(), fh.wfs.manifestCache, fh.wfs.readerCacheBudget) if resolveManifestErr != nil { glog.Warningf("failed to resolve manifest chunks in %+v", entry) } diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 4ebacac71..a636ded60 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -142,6 +142,7 @@ type WFS struct { stats statsCache chunkCache *chunk_cache.TieredChunkCache readerCacheBudget *filer.ReaderCacheBudget + manifestCache *filer.ChunkManifestCache writeBufferAccountant *page_writer.WriteBufferAccountant signature int32 concurrentWriters *util.LimitedConcurrentExecutor @@ -253,6 +254,7 @@ func NewSeaweedFileSystem(option *Option) *WFS { RawFileSystem: fuse.NewDefaultRawFileSystem(), option: option, readerCacheBudget: filer.NewReaderCacheBudget(option.ReaderCacheSizeMB << 20), + manifestCache: filer.NewChunkManifestCache(filer.MaxMountChunkManifestCacheEntries, filer.MaxMountChunkManifestCacheBytes), signature: util.RandomInt32(), inodeToPath: NewInodeToPath(util.FullPath(option.FilerMountRootPath), option.CacheMetaTTlSec), fhMap: NewFileHandleToInode(), diff --git a/weed/mount/weedfs_attr_race_test.go b/weed/mount/weedfs_attr_race_test.go index f78b28782..805e4261b 100644 --- a/weed/mount/weedfs_attr_race_test.go +++ b/weed/mount/weedfs_attr_race_test.go @@ -35,7 +35,7 @@ func TestAttrChunkRace(t *testing.T) { Name: "sample.txt", Attributes: &filer_pb.FuseAttributes{FileMode: 0644}, } - chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1, nil) + chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1, nil, nil) if err != nil { t.Fatalf("NewChunkGroup: %v", err) } @@ -115,7 +115,7 @@ func TestReadFromChunksRace(t *testing.T) { Name: "sample.txt", Attributes: &filer_pb.FuseAttributes{FileMode: 0644}, } - chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1, nil) + chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1, nil, nil) if err != nil { t.Fatalf("NewChunkGroup: %v", err) } diff --git a/weed/mount/weedfs_attr_unlinked_test.go b/weed/mount/weedfs_attr_unlinked_test.go index 1da8885e5..97920e58f 100644 --- a/weed/mount/weedfs_attr_unlinked_test.go +++ b/weed/mount/weedfs_attr_unlinked_test.go @@ -33,7 +33,7 @@ func newUnlinkedOpenFile(t *testing.T) (*WFS, uint64, *FileHandle) { Name: "file", Attributes: &filer_pb.FuseAttributes{FileMode: 0644}, } - chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1, nil) + chunkGroup, err := filer.NewChunkGroup(nil, nil, nil, 1, nil, nil) if err != nil { t.Fatalf("NewChunkGroup: %v", err) }