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

@@ -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")

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,

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)
 

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")) {

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

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

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

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
 

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