From dd73fee077e97b4cf9f19dceffafe26a1a24df8e Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 7 Aug 2026 17:48:40 -0700 Subject: [PATCH] mount: read oversized directories through instead of caching them (#10631) * mount: read oversized directories through instead of caching them Visiting a directory pulls every child from the filer into the local LevelDB before the first listing returns. For a directory of a few million entries that is minutes of streaming, gigabytes of local store, and gigabytes of decoded entries in flight -- paid by a mount that may only walk the directory once. A build that crosses -cacheDirMaxEntries (default ten thousand) now stops, cleans up, and marks the directory read-through: listings stream from the filer with pagination, the way update-hot directories already do, and lookups in it consult the filer per entry as any uncached directory does. The refusal is remembered, so the next visit fails fast instead of streaming to the limit again, and an oversized ancestor is stepped over when caching its subdirectories rather than wedging every listing beneath it. The direct path keeps the same pagination state on the handle, so a walk that crosses the limit mid-flight carries on from where the cached walk reached. * mount: an ancestor found oversized must not fail its descendants Visiting a directory builds its whole uncached ancestor chain in one group, so the first discovery that an ancestor is oversized cancelled the group and surfaced as the listed directory's own refusal: the descendant build was aborted and the caller marked the descendant read-through, leaving a perfectly cacheable directory streaming from the filer until its inode was forgotten. The earlier test missed this by pre-marking the ancestor, which exercises only the fast path. The refusal of any directory other than the one being listed is now kept out of the group's result; it is already remembered for the next visit. --- weed/command/mount.go | 2 + weed/command/mount_common.go | 1 + weed/mount/meta_cache/meta_cache.go | 19 +++ .../mount/meta_cache/meta_cache_build_test.go | 10 +- weed/mount/meta_cache/meta_cache_init.go | 50 ++++++- weed/mount/meta_cache/oversized_dir_test.go | 138 ++++++++++++++++++ weed/mount/weedfs.go | 1 + weed/mount/weedfs_dir_read.go | 27 +++- .../weedfs_invalidate_open_handle_test.go | 2 +- 9 files changed, 238 insertions(+), 12 deletions(-) create mode 100644 weed/mount/meta_cache/oversized_dir_test.go diff --git a/weed/command/mount.go b/weed/command/mount.go index 6a28c2fe8..e8945ddc8 100644 --- a/weed/command/mount.go +++ b/weed/command/mount.go @@ -20,6 +20,7 @@ type MountOptions struct { concurrentWriters *int concurrentReaders *int cacheMetaTtlSec *int + cacheDirMaxEntries *int cacheDirForRead *string cacheDirForWrite *string cacheSizeMBForRead *int64 @@ -114,6 +115,7 @@ func init() { mountOptions.cacheDirForWrite = cmdMount.Flag.String("cacheDirWrite", "", "buffer writes mostly for large files") mountOptions.writeBufferSizeMB = cmdMount.Flag.Int64("writeBufferSizeMB", 0, "global cap on the per-mount write buffer (memory + swap) in MB, 0 means unlimited. Bounds /tmp growth when volume uploads stall") mountOptions.cacheMetaTtlSec = cmdMount.Flag.Int("cacheMetaTtlSec", 60, "metadata cache validity seconds") + mountOptions.cacheDirMaxEntries = cmdMount.Flag.Int("cacheDirMaxEntries", 10000, "a directory with more children than this is not cached locally but read directly from the filer; 0 caches everything") mountOptions.dataCenter = cmdMount.Flag.String("dataCenter", "", "prefer to write to the data center") mountOptions.allowOthers = cmdMount.Flag.Bool("allowOthers", true, "allows other users to access the file system") mountOptions.defaultPermissions = cmdMount.Flag.Bool("defaultPermissions", true, "enforce permissions by the operating system") diff --git a/weed/command/mount_common.go b/weed/command/mount_common.go index 0cd7450d5..8b682bf0b 100644 --- a/weed/command/mount_common.go +++ b/weed/command/mount_common.go @@ -197,6 +197,7 @@ func buildSeaweedFileSystem(option *MountOptions, p fileSystemParams) *mount.WFS CacheDirForWrite: p.cacheDirForWrite, WriteBufferSizeMB: *option.writeBufferSizeMB, CacheMetaTTlSec: *option.cacheMetaTtlSec, + CacheDirMaxEntries: *option.cacheDirMaxEntries, DataCenter: *option.dataCenter, Quota: int64(*option.collectionQuota) * 1024 * 1024, LogicalDiskUsage: *option.logicalDiskUsage, diff --git a/weed/mount/meta_cache/meta_cache.go b/weed/mount/meta_cache/meta_cache.go index cb22745a9..6ffa31cac 100644 --- a/weed/mount/meta_cache/meta_cache.go +++ b/weed/mount/meta_cache/meta_cache.go @@ -43,6 +43,11 @@ type MetaCache struct { dedupRing dedupRingBuffer includeSystemEntries bool + // oversizedDirs are directories the mount refused to cache for their size. + // Their listings read through to the filer, and a later visit fails fast + // instead of streaming to the limit again to rediscover them. + oversizedDirs map[util.FullPath]struct{} + // dirVersionFloors is each cached directory's listing snapshot: the // version of every child the listing covered, present or absent, unless // a later event gave that child its own record. One map write per build @@ -119,6 +124,7 @@ func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPat buildingDirs: make(map[util.FullPath]*directoryBuildState), dedupRing: newDedupRingBuffer(), dirVersionFloors: make(map[util.FullPath]int64), + oversizedDirs: make(map[util.FullPath]struct{}), } mc.invalidateWorker = util.NewAsyncBatchWorker(func(batch []EntryInvalidation) { for _, invalidation := range batch { @@ -649,6 +655,19 @@ func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.Full }) } +func (mc *MetaCache) markOversized(dirPath util.FullPath) { + mc.Lock() + defer mc.Unlock() + mc.oversizedDirs[dirPath] = struct{}{} +} + +func (mc *MetaCache) isOversized(dirPath util.FullPath) bool { + mc.RLock() + defer mc.RUnlock() + _, found := mc.oversizedDirs[dirPath] + return found +} + func (mc *MetaCache) Shutdown() { done := make(chan error, 1) diff --git a/weed/mount/meta_cache/meta_cache_build_test.go b/weed/mount/meta_cache/meta_cache_build_test.go index 2bb8ba21f..458111cca 100644 --- a/weed/mount/meta_cache/meta_cache_build_test.go +++ b/weed/mount/meta_cache/meta_cache_build_test.go @@ -110,7 +110,7 @@ func TestEnsureVisitedReplaysBufferedEventsAfterSnapshot(t *testing.T) { }, } - if err := EnsureVisited(mc, accessor, util.FullPath("/dir")); err != nil { + if err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 0); err != nil { t.Fatalf("ensure visited: %v", err) } if applyErr != nil { @@ -504,7 +504,7 @@ func TestEnsureVisitedPreservesLocalOnlyEntry(t *testing.T) { }}, }} - if err := EnsureVisited(mc, accessor, util.FullPath("/dir")); err != nil { + if err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 0); err != nil { t.Fatalf("ensure visited: %v", err) } if !mc.IsDirectoryCached(util.FullPath("/dir")) { @@ -550,7 +550,7 @@ func TestEnsureVisitedDropsUnpinnedStaleEntry(t *testing.T) { }}, }} - if err := EnsureVisited(mc, accessor, util.FullPath("/dir")); err != nil { + if err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 0); err != nil { t.Fatalf("ensure visited: %v", err) } if entry, _, err := mc.FindEntry(context.Background(), util.FullPath("/dir/stale.txt")); err != filer_pb.ErrNotFound || entry != nil { @@ -597,7 +597,7 @@ func TestEnsureVisitedConfirmsTransientEmptyListing(t *testing.T) { }, }} - if err := EnsureVisited(mc, accessor, util.FullPath("/dir")); err != nil { + if err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 0); err != nil { t.Fatalf("ensure visited: %v", err) } if !mc.IsDirectoryCached(util.FullPath("/dir")) { @@ -619,7 +619,7 @@ func TestEnsureVisitedCachesGenuinelyEmptyDirectory(t *testing.T) { } accessor := &buildFilerAccessor{client: client} - if err := EnsureVisited(mc, accessor, util.FullPath("/empty")); err != nil { + if err := EnsureVisited(mc, accessor, util.FullPath("/empty"), 0); err != nil { t.Fatalf("ensure visited: %v", err) } if !mc.IsDirectoryCached(util.FullPath("/empty")) { diff --git a/weed/mount/meta_cache/meta_cache_init.go b/weed/mount/meta_cache/meta_cache_init.go index a7df7f523..3f7c6d902 100644 --- a/weed/mount/meta_cache/meta_cache_init.go +++ b/weed/mount/meta_cache/meta_cache_init.go @@ -2,6 +2,7 @@ package meta_cache import ( "context" + "errors" "fmt" "time" @@ -13,7 +14,19 @@ import ( "github.com/seaweedfs/seaweedfs/weed/util" ) -func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.FullPath) error { +// DirectoryTooLargeError reports a directory the mount refuses to cache +// locally. Its listings read through to the filer instead. +type DirectoryTooLargeError struct { + Path util.FullPath +} + +func (e *DirectoryTooLargeError) Error() string { + return fmt.Sprintf("directory %s is too large to cache locally", e.Path) +} + +// maxCacheableEntries is the directory size above which a build gives up, or 0 +// to cache everything. +func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.FullPath, maxCacheableEntries int) error { // Collect all uncached paths from target directory up to root var uncachedPaths []util.FullPath currentPath := dirPath @@ -23,7 +36,15 @@ func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.Full if mc.isCachedFn(currentPath) { break } - uncachedPaths = append(uncachedPaths, currentPath) + if mc.isOversized(currentPath) { + // The directory itself reads through; an ancestor is stepped over, + // or it would wedge every listing beneath it forever. + if currentPath == dirPath { + return &DirectoryTooLargeError{Path: currentPath} + } + } else { + uncachedPaths = append(uncachedPaths, currentPath) + } // Continue to parent directory if currentPath != mc.root { @@ -44,7 +65,16 @@ func EnsureVisited(mc *MetaCache, client filer_pb.FilerClient, dirPath util.Full for _, p := range uncachedPaths { path := p // capture for closure g.Go(func() error { - return doEnsureVisited(ctx, mc, client, path) + err := doEnsureVisited(ctx, mc, client, path, maxCacheableEntries) + var tooLarge *DirectoryTooLargeError + if errors.As(err, &tooLarge) && path != dirPath { + // An ancestor found oversized just reads through; failing the + // group here would cancel the builds of its cacheable + // descendants, and the caller would treat the refusal as the + // listed directory's own. + return nil + } + return err }) } return g.Wait() @@ -60,7 +90,7 @@ const ( emptyRebuildConfirmDelay = 50 * time.Millisecond ) -func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerClient, path util.FullPath) error { +func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerClient, path util.FullPath, maxCacheableEntries int) error { // Use singleflight to deduplicate concurrent requests for the same path _, err, _ := mc.visitGroup.Do(string(path), func() (interface{}, error) { // Check for cancellation before starting @@ -116,6 +146,9 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl return nil } + if maxCacheableEntries > 0 && entryCount >= maxCacheableEntries { + return &DirectoryTooLargeError{Path: path} + } batch = append(batch, entry) entryCount++ @@ -143,6 +176,15 @@ func doEnsureVisited(ctx context.Context, mc *MetaCache, client filer_pb.FilerCl entryCount, snapshotTsNs, fetchErr := reloadFromFiler() if fetchErr != nil { + var tooLarge *DirectoryTooLargeError + if errors.As(fetchErr, &tooLarge) { + // Remember the refusal so the next visit fails fast instead of + // streaming up to the limit again to rediscover it. + mc.markOversized(path) + glog.V(0).Infof("directory %s exceeds %d entries, reading it through instead of caching", path, maxCacheableEntries) + cleanupBuild("oversized") + return nil, fetchErr + } cleanupBuild("failed") return nil, fmt.Errorf("list %s: %w", path, fetchErr) } diff --git a/weed/mount/meta_cache/oversized_dir_test.go b/weed/mount/meta_cache/oversized_dir_test.go new file mode 100644 index 000000000..4edb87ba0 --- /dev/null +++ b/weed/mount/meta_cache/oversized_dir_test.go @@ -0,0 +1,138 @@ +package meta_cache + +import ( + "context" + "errors" + "fmt" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/filer" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc" +) + +func listResponses(n int) []*filer_pb.ListEntriesResponse { + responses := make([]*filer_pb.ListEntriesResponse, 0, n) + for i := 0; i < n; i++ { + responses = append(responses, &filer_pb.ListEntriesResponse{ + Entry: &filer_pb.Entry{ + Name: fmt.Sprintf("f%05d", i), + Attributes: &filer_pb.FuseAttributes{ + Crtime: 1, Mtime: 1, FileMode: 0o644, FileSize: 3, + }, + }, + }) + } + return responses +} + +// TestEnsureVisitedRefusesOversizedDirectory checks that a directory past the +// limit is not cached, that the refusal is remembered, and that the partial +// build leaves nothing behind in the local store. +func TestEnsureVisitedRefusesOversizedDirectory(t *testing.T) { + mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{"/": true}) + defer mc.Shutdown() + + accessor := &buildFilerAccessor{client: &buildListClient{responses: listResponses(10)}} + + err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 5) + var tooLarge *DirectoryTooLargeError + if !errors.As(err, &tooLarge) { + t.Fatalf("EnsureVisited = %v, want DirectoryTooLargeError", err) + } + if tooLarge.Path != util.FullPath("/dir") { + t.Fatalf("oversized path = %s, want /dir", tooLarge.Path) + } + if mc.IsDirectoryCached(util.FullPath("/dir")) { + t.Error("oversized directory reported cached") + } + // The aborted build must leave no partial children to be served later. + count := 0 + if _, err := mc.ListDirectoryEntries(context.Background(), util.FullPath("/dir"), "", false, 100, func(e *filer.Entry) (bool, error) { + count++ + return true, nil + }); err != nil { + t.Fatalf("list: %v", err) + } + if count != 0 { + t.Errorf("local store still holds %d children of the aborted build", count) + } + + // A second visit fails fast without streaming to the limit again. + if err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 5); !errors.As(err, &tooLarge) { + t.Fatalf("second EnsureVisited = %v, want DirectoryTooLargeError", err) + } +} + +// TestEnsureVisitedStepsOverOversizedAncestor checks a huge ancestor does not +// wedge the caching of its subdirectories. +func TestEnsureVisitedStepsOverOversizedAncestor(t *testing.T) { + mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{"/": true}) + defer mc.Shutdown() + mc.markOversized(util.FullPath("/huge")) + + accessor := &buildFilerAccessor{client: &buildListClient{responses: listResponses(3)}} + if err := EnsureVisited(mc, accessor, util.FullPath("/huge/sub"), 5); err != nil { + t.Fatalf("EnsureVisited under an oversized ancestor: %v", err) + } + if !mc.IsDirectoryCached(util.FullPath("/huge/sub")) { + t.Error("subdirectory of an oversized ancestor was not cached") + } + if mc.IsDirectoryCached(util.FullPath("/huge")) { + t.Error("oversized ancestor became cached") + } +} + +// TestEnsureVisitedUnderTheLimitStillCaches pins that the gate does not change +// behaviour for ordinary directories. +func TestEnsureVisitedUnderTheLimitStillCaches(t *testing.T) { + mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{"/": true}) + defer mc.Shutdown() + + accessor := &buildFilerAccessor{client: &buildListClient{responses: listResponses(5)}} + if err := EnsureVisited(mc, accessor, util.FullPath("/dir"), 5); err != nil { + t.Fatalf("EnsureVisited: %v", err) + } + if !mc.IsDirectoryCached(util.FullPath("/dir")) { + t.Error("directory at the limit was not cached") + } +} + +// pathListClient serves canned listings per directory, so one visit can see +// directories of different sizes. +type pathListClient struct { + filer_pb.SeaweedFilerClient + perDir map[string][]*filer_pb.ListEntriesResponse +} + +func (c *pathListClient) ListEntries(ctx context.Context, in *filer_pb.ListEntriesRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[filer_pb.ListEntriesResponse], error) { + return &buildListStream{responses: c.perDir[in.Directory]}, nil +} + +// TestEnsureVisitedAncestorFoundOversizedMidVisit covers the first discovery: +// the ancestor's refusal must neither cancel the descendant's build nor be +// reported as the descendant's own. +func TestEnsureVisitedAncestorFoundOversizedMidVisit(t *testing.T) { + mc, _, _, _ := newTestMetaCache(t, map[util.FullPath]bool{"/": true}) + defer mc.Shutdown() + + accessor := &buildFilerAccessor{client: &pathListClient{perDir: map[string][]*filer_pb.ListEntriesResponse{ + "/huge": listResponses(10), + "/huge/sub": listResponses(3), + }}} + + if err := EnsureVisited(mc, accessor, util.FullPath("/huge/sub"), 5); err != nil { + t.Fatalf("EnsureVisited: %v", err) + } + if !mc.IsDirectoryCached(util.FullPath("/huge/sub")) { + t.Error("descendant of a just-discovered oversized ancestor was not cached") + } + if mc.IsDirectoryCached(util.FullPath("/huge")) { + t.Error("oversized ancestor became cached") + } + if !mc.isOversized(util.FullPath("/huge")) { + t.Error("oversized ancestor was not remembered") + } +} diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 80b2acabf..032eed7cc 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -53,6 +53,7 @@ type Option struct { CacheDirForWrite string WriteBufferSizeMB int64 CacheMetaTTlSec int + CacheDirMaxEntries int DataCenter string Umask os.FileMode Quota int64 diff --git a/weed/mount/weedfs_dir_read.go b/weed/mount/weedfs_dir_read.go index faf1b3d3b..faeb01857 100644 --- a/weed/mount/weedfs_dir_read.go +++ b/weed/mount/weedfs_dir_read.go @@ -2,6 +2,7 @@ package mount import ( "context" + "errors" "sync" "time" @@ -268,7 +269,11 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode if len(dh.entryStream) == 0 && input.Offset > dh.entryStreamOffset { skipCount := int64(input.Offset - dh.entryStreamOffset) - if err := meta_cache.EnsureVisited(wfs.metaCache, wfs, dirPath); err != nil { + if err := wfs.ensureDirectoryVisited(dirPath); err != nil { + var tooLarge *meta_cache.DirectoryTooLargeError + if errors.As(err, &tooLarge) { + return wfs.readDirectoryDirect(input, out, dh, dirPath, processEachEntryFn) + } glog.Errorf("dir ReadDirAll %s: %v", dirPath, err) return fuse.EIO } @@ -311,7 +316,13 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode } // Cache exhausted, load next batch - if err := meta_cache.EnsureVisited(wfs.metaCache, wfs, dirPath); err != nil { + if err := wfs.ensureDirectoryVisited(dirPath); err != nil { + var tooLarge *meta_cache.DirectoryTooLargeError + if errors.As(err, &tooLarge) { + // The direct path keeps the same pagination state on dh, so it + // carries on from wherever the cached walk reached. + return wfs.readDirectoryDirect(input, out, dh, dirPath, processEachEntryFn) + } glog.Errorf("dir ReadDirAll %s: %v", dirPath, err) return fuse.EIO } @@ -351,6 +362,18 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode return fuse.OK } +// ensureDirectoryVisited pulls the directory into the local cache, unless it is +// too large to cache: then the directory is marked read-through, so later +// listings go straight to the filer without re-asking. +func (wfs *WFS) ensureDirectoryVisited(dirPath util.FullPath) error { + err := meta_cache.EnsureVisited(wfs.metaCache, wfs, dirPath, wfs.option.CacheDirMaxEntries) + var tooLarge *meta_cache.DirectoryTooLargeError + if errors.As(err, &tooLarge) { + wfs.inodeToPath.MarkDirectoryReadThrough(dirPath, time.Now()) + } + return err +} + func (wfs *WFS) readDirectoryDirect(input *fuse.ReadIn, out DirEntrySink, dh *DirectoryHandle, dirPath util.FullPath, processEachEntryFn func(entry *filer.Entry, index int64) bool) fuse.Status { var lastEntryName string diff --git a/weed/mount/weedfs_invalidate_open_handle_test.go b/weed/mount/weedfs_invalidate_open_handle_test.go index af991e10c..eec95ded6 100644 --- a/weed/mount/weedfs_invalidate_open_handle_test.go +++ b/weed/mount/weedfs_invalidate_open_handle_test.go @@ -1405,7 +1405,7 @@ func TestEmptyListingTrailerSnapshotSetsAbsenceFloor(t *testing.T) { startFakeFiler(t, wfs, &fakeFilerServer{listSnapshotTrailerTsNs: 4000}) wfs.inodeToPath.Lookup(util.FullPath("/dir"), time.Now().Unix(), true, false, 0, false) - if err := meta_cache.EnsureVisited(wfs.metaCache, wfs, util.FullPath("/dir")); err != nil { + if err := meta_cache.EnsureVisited(wfs.metaCache, wfs, util.FullPath("/dir"), 0); err != nil { t.Fatalf("EnsureVisited: %v", err) }