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) }