mount: size the listing cache through NewMetaCache rather than a second constructor

A NewXWithY alongside NewX is two names for one thing, and the one that
takes the argument is the only one that ever needed to exist. Callers
pass the default explicitly.

weed/command/mount.go

@@ -3,6 +3,8 @@ package command
 import (
 	"os"
 	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/mount/meta_cache"
 )
 
 type MountOptions struct {
@@ -20,6 +22,7 @@ type MountOptions struct {
 	concurrentWriters    *int
 	concurrentReaders    *int
 	cacheMetaTtlSec      *int
+	dirListCacheEntries  *int
 	cacheDirForRead      *string
 	cacheDirForWrite     *string
 	cacheSizeMBForRead   *int64
@@ -114,6 +117,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.dirListCacheEntries = cmdMount.Flag.Int("dirListCacheEntries", meta_cache.DefaultListingCacheEntries, "directory children kept decoded in memory, so reopening a folder rereads nothing. A ceiling, not a reservation: only walked directories are held, at roughly half a kilobyte per child. 0 disables")
 	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,
+		DirListCacheEntries:         *option.dirListCacheEntries,
 		DataCenter:                  *option.dataCenter,
 		Quota:                       int64(*option.collectionQuota) * 1024 * 1024,
 		LogicalDiskUsage:            *option.logicalDiskUsage,

weed/mount/filehandle_read_remote_test.go

@@ -111,6 +111,7 @@ func TestReadUncachedRemoteEntryDoesNotDeadlock(t *testing.T) {
 			wfs.fhLockTable.ReleaseLock(fh.fh, lock)
 		},
 		nil,
+		meta_cache.DefaultListingCacheEntries,
 	)
 	wfs.inodeToPath.MarkChildrenCached(root)
 	wfs.inodeToPath.Lookup(util.FullPath("/dir"), time.Now().Unix(), true, false, 0, false)

weed/mount/meta_cache/listing_cache.go

@@ -0,0 +1,195 @@
+package meta_cache
+
+import (
+	"sync"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+// DefaultListingCacheEntries is how many directory children the mount will hold
+// decoded in memory across every cached listing.
+//
+// This is a ceiling, not a reservation: nothing is held for a directory that was
+// never walked, so a mount over ordinary directories never approaches it. A
+// chunkless child measured about 480 bytes, so reaching the default means
+// holding something near 450MB, and a mount that wants a tighter bound should
+// lower it rather than rely on never walking that far.
+const DefaultListingCacheEntries = 1000000
+
+// listingCache keeps directories that have been walked all the way through, so
+// walking one again costs no store reads and no decoding. It holds only what the
+// meta cache already holds, and is void the moment that is: every store write
+// goes through one of a handful of methods, and each drops the directory it
+// touched.
+//
+// The entries are shared with whoever listed them, and are never written to
+// after publication.
+type listingCache struct {
+	mu       sync.Mutex
+	maxTotal int
+	total    int
+	listings map[util.FullPath]*cachedListing
+	// builds are the walks in progress. A walk publishes only if its build is
+	// still the live one for that directory when it finishes, so a write landing
+	// halfway through cannot leave a stale listing behind.
+	builds map[util.FullPath]*listingBuild
+}
+
+type cachedListing struct {
+	entries    []*filer.Entry
+	lastAccess time.Time
+}
+
+type listingBuild struct {
+	entries []*filer.Entry
+	// nextStart is the name the next page has to begin at for this build to be
+	// a continuation rather than a different walk.
+	nextStart string
+}
+
+func newListingCache(maxTotal int) *listingCache {
+	return &listingCache{
+		maxTotal: maxTotal,
+		listings: make(map[util.FullPath]*cachedListing),
+		builds:   make(map[util.FullPath]*listingBuild),
+	}
+}
+
+func (lc *listingCache) enabled() bool { return lc != nil && lc.maxTotal > 0 }
+
+// lookup returns a directory's complete listing.
+func (lc *listingCache) lookup(dirPath util.FullPath) ([]*filer.Entry, bool) {
+	if !lc.enabled() {
+		return nil, false
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	cached, found := lc.listings[dirPath]
+	if !found {
+		return nil, false
+	}
+	cached.lastAccess = time.Now()
+	return cached.entries, true
+}
+
+// invalidate drops a directory's listing and kills any walk building one. Called
+// for every write the meta cache makes, so it has to stay cheap on a miss.
+func (lc *listingCache) invalidate(dirPath util.FullPath) {
+	if !lc.enabled() {
+		return
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	lc.dropLocked(dirPath)
+	delete(lc.builds, dirPath)
+}
+
+// invalidateChild drops the listing of the directory holding path.
+func (lc *listingCache) invalidateChild(path util.FullPath) {
+	if !lc.enabled() {
+		return
+	}
+	dir, _ := path.DirAndName()
+	lc.invalidate(util.FullPath(dir))
+}
+
+func (lc *listingCache) invalidateAll() {
+	if !lc.enabled() {
+		return
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	lc.listings = make(map[util.FullPath]*cachedListing)
+	lc.builds = make(map[util.FullPath]*listingBuild)
+	lc.total = 0
+}
+
+func (lc *listingCache) dropLocked(dirPath util.FullPath) {
+	if cached, found := lc.listings[dirPath]; found {
+		lc.total -= len(cached.entries)
+		delete(lc.listings, dirPath)
+	}
+}
+
+// beginOrContinue reports the build a page belongs to, or nil when this page is
+// not the next step of a walk from the directory's first child. Only a walk that
+// covers every child in order can be published.
+func (lc *listingCache) beginOrContinue(dirPath util.FullPath, startFileName string, includeStartFile bool) *listingBuild {
+	if !lc.enabled() || includeStartFile {
+		return nil
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	if startFileName == "" {
+		build := &listingBuild{}
+		lc.builds[dirPath] = build
+		return build
+	}
+	build, found := lc.builds[dirPath]
+	if !found || build.nextStart != startFileName {
+		// A seek, a second walk at a different position, or a page that follows
+		// one this build never saw. Neither walk can be trusted to be complete.
+		delete(lc.builds, dirPath)
+		return nil
+	}
+	return build
+}
+
+// carry keeps a build alive for the next page.
+func (lc *listingCache) carry(dirPath util.FullPath, build *listingBuild, nextStart string) {
+	if build == nil {
+		return
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	if lc.builds[dirPath] != build {
+		return // invalidated mid-walk
+	}
+	build.nextStart = nextStart
+}
+
+// publish installs a completed walk, unless a write invalidated it on the way or
+// it does not fit. Dropping is always safe: the next walk reads the store.
+func (lc *listingCache) publish(dirPath util.FullPath, build *listingBuild) {
+	if build == nil {
+		return
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	if lc.builds[dirPath] != build {
+		return
+	}
+	delete(lc.builds, dirPath)
+	if len(build.entries) > lc.maxTotal {
+		return
+	}
+	lc.dropLocked(dirPath)
+	lc.evictLocked(len(build.entries))
+	lc.listings[dirPath] = &cachedListing{entries: build.entries, lastAccess: time.Now()}
+	lc.total += len(build.entries)
+}
+
+// evictLocked makes room for want entries, oldest use first.
+func (lc *listingCache) evictLocked(want int) {
+	for lc.total+want > lc.maxTotal && len(lc.listings) > 0 {
+		var oldestPath util.FullPath
+		var oldest time.Time
+		for path, cached := range lc.listings {
+			if oldest.IsZero() || cached.lastAccess.Before(oldest) {
+				oldestPath, oldest = path, cached.lastAccess
+			}
+		}
+		lc.dropLocked(oldestPath)
+	}
+}
+
+func (lc *listingCache) size() (dirs, entries int) {
+	if !lc.enabled() {
+		return 0, 0
+	}
+	lc.mu.Lock()
+	defer lc.mu.Unlock()
+	return len(lc.listings), lc.total
+}

weed/mount/meta_cache/listing_cache_test.go

@@ -0,0 +1,330 @@
+package meta_cache
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+func newListingTestCache(t *testing.T, maxEntries int) *MetaCache {
+	t.Helper()
+	uidGidMapper, err := NewUidGidMapper("", "")
+	if err != nil {
+		t.Fatalf("uid/gid mapper: %v", err)
+	}
+	cached := map[util.FullPath]bool{}
+	mc := NewMetaCache(t.TempDir(), uidGidMapper, util.FullPath("/"), false,
+		func(p util.FullPath) { cached[p] = true },
+		func(p util.FullPath) bool { return cached[p] },
+		func(EntryInvalidation) {}, nil, maxEntries)
+	t.Cleanup(mc.Shutdown)
+	return mc
+}
+
+func seedDir(t *testing.T, mc *MetaCache, dir util.FullPath, names ...string) {
+	t.Helper()
+	now := time.Now()
+	ctx := context.Background()
+	if err := mc.InsertEntry(ctx, &filer.Entry{
+		FullPath: dir, Attr: filer.Attr{Mode: os.ModeDir | 0o755, Mtime: now, Crtime: now},
+	}, 0); err != nil {
+		t.Fatalf("insert dir: %v", err)
+	}
+	for _, name := range names {
+		if err := mc.InsertEntry(ctx, &filer.Entry{
+			FullPath: dir.Child(name), Attr: filer.Attr{Mode: 0o644, Mtime: now, Crtime: now, FileSize: 1},
+		}, 0); err != nil {
+			t.Fatalf("insert %s: %v", name, err)
+		}
+	}
+	mc.markCachedFn(util.FullPath("/"))
+	mc.markCachedFn(dir)
+}
+
+// walk lists a directory to exhaustion in pages, returning the names seen.
+func walk(t *testing.T, mc *MetaCache, ctx context.Context, dir util.FullPath, pageSize int64) []string {
+	t.Helper()
+	var seen []string
+	start := ""
+	for round := 0; round < 100; round++ {
+		count := 0
+		last, err := mc.ListDirectoryEntries(ctx, dir, start, false, pageSize, func(entry *filer.Entry) (bool, error) {
+			seen = append(seen, entry.Name())
+			count++
+			return true, nil
+		})
+		if err != nil {
+			t.Fatalf("list: %v", err)
+		}
+		if last == "" || last == start {
+			break
+		}
+		start = last
+		if int64(count) < pageSize && count == 0 {
+			break
+		}
+	}
+	return seen
+}
+
+func listingCtx() context.Context { return filer_pb.WithChunksOmitted(context.Background()) }
+
+func TestListingCacheServesAfterFullWalk(t *testing.T) {
+	mc := newListingTestCache(t, DefaultListingCacheEntries)
+	dir := util.FullPath("/d")
+	seedDir(t, mc, dir, "a", "b", "c")
+
+	if got := walk(t, mc, listingCtx(), dir, 100); len(got) != 3 {
+		t.Fatalf("first walk listed %v, want 3 entries", got)
+	}
+	dirs, entries := mc.ListingCacheSize()
+	if dirs != 1 || entries != 3 {
+		t.Fatalf("cache holds %d dirs / %d entries, want 1/3", dirs, entries)
+	}
+	if got := walk(t, mc, listingCtx(), dir, 100); len(got) != 3 {
+		t.Fatalf("second walk listed %v, want the same 3", got)
+	}
+}
+
+func TestListingCacheNotUsedWithoutChunksOmitted(t *testing.T) {
+	mc := newListingTestCache(t, DefaultListingCacheEntries)
+	dir := util.FullPath("/d")
+	seedDir(t, mc, dir, "a", "b")
+
+	// A listing that wants chunks must neither fill nor read the cache, since
+	// the cached entries were decoded without them.
+	if got := walk(t, mc, context.Background(), dir, 100); len(got) != 2 {
+		t.Fatalf("listed %v, want 2", got)
+	}
+	if dirs, _ := mc.ListingCacheSize(); dirs != 0 {
+		t.Fatalf("cache holds %d dirs, want none", dirs)
+	}
+}
+
+func TestListingCachePartialWalkIsNotCached(t *testing.T) {
+	mc := newListingTestCache(t, DefaultListingCacheEntries)
+	dir := util.FullPath("/d")
+	seedDir(t, mc, dir, "a", "b", "c", "d")
+
+	// Caller stops after the first entry: the rest of the directory was never
+	// seen, so nothing may be published.
+	if _, err := mc.ListDirectoryEntries(listingCtx(), dir, "", false, 100, func(entry *filer.Entry) (bool, error) {
+		return false, nil
+	}); err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if dirs, entries := mc.ListingCacheSize(); dirs != 0 {
+		t.Fatalf("cache holds %d dirs / %d entries after a stopped walk, want none", dirs, entries)
+	}
+}
+
+func TestListingCacheSeekDoesNotBuild(t *testing.T) {
+	mc := newListingTestCache(t, DefaultListingCacheEntries)
+	dir := util.FullPath("/d")
+	seedDir(t, mc, dir, "a", "b", "c")
+
+	// Starting partway through is not a walk of the whole directory.
+	if _, err := mc.ListDirectoryEntries(listingCtx(), dir, "a", false, 100, func(entry *filer.Entry) (bool, error) {
+		return true, nil
+	}); err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if dirs, _ := mc.ListingCacheSize(); dirs != 0 {
+		t.Fatalf("cache holds %d dirs after a mid-directory listing, want none", dirs)
+	}
+}
+
+func TestListingCacheInvalidation(t *testing.T) {
+	dir := util.FullPath("/d")
+	now := time.Now()
+	cases := []struct {
+		name   string
+		mutate func(t *testing.T, mc *MetaCache)
+	}{
+		{"insert a child", func(t *testing.T, mc *MetaCache) {
+			if err := mc.InsertEntry(context.Background(), &filer.Entry{
+				FullPath: dir.Child("new"), Attr: filer.Attr{Mode: 0o644, Mtime: now, Crtime: now},
+			}, 0); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{"update a child", func(t *testing.T, mc *MetaCache) {
+			if err := mc.UpdateEntry(context.Background(), &filer.Entry{
+				FullPath: dir.Child("a"), Attr: filer.Attr{Mode: 0o644, Mtime: now, Crtime: now, FileSize: 99},
+			}); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{"delete a child", func(t *testing.T, mc *MetaCache) {
+			if err := mc.DeleteEntry(context.Background(), dir.Child("a")); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{"delete the children", func(t *testing.T, mc *MetaCache) {
+			if err := mc.DeleteFolderChildren(context.Background(), dir); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{"rename into the directory", func(t *testing.T, mc *MetaCache) {
+			if err := mc.AtomicUpdateEntryFromFiler(context.Background(), "", &filer.Entry{
+				FullPath: dir.Child("moved"), Attr: filer.Attr{Mode: 0o644, Mtime: now, Crtime: now},
+			}); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{"purge the children", func(t *testing.T, mc *MetaCache) {
+			mc.PurgeDirectoryChildren(dir, func() {})
+		}},
+		{"distrust everything", func(t *testing.T, mc *MetaCache) {
+			mc.InvalidateAllListings()
+		}},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			mc := newListingTestCache(t, DefaultListingCacheEntries)
+			seedDir(t, mc, dir, "a", "b", "c")
+			walk(t, mc, listingCtx(), dir, 100)
+			if dirs, _ := mc.ListingCacheSize(); dirs != 1 {
+				t.Fatalf("expected the walk to cache the directory, got %d dirs", dirs)
+			}
+			tc.mutate(t, mc)
+			if dirs, entries := mc.ListingCacheSize(); dirs != 0 {
+				t.Errorf("cache still holds %d dirs / %d entries after %s", dirs, entries, tc.name)
+			}
+		})
+	}
+}
+
+func TestListingCacheSeesAWriteThatLandsMidWalk(t *testing.T) {
+	mc := newListingTestCache(t, DefaultListingCacheEntries)
+	dir := util.FullPath("/d")
+	seedDir(t, mc, dir, "a", "b", "c", "d")
+	ctx := listingCtx()
+
+	// First page of a two-page walk.
+	last, err := mc.ListDirectoryEntries(ctx, dir, "", false, 2, func(entry *filer.Entry) (bool, error) {
+		return true, nil
+	})
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+
+	// A write lands before the walk finishes.
+	if err := mc.InsertEntry(context.Background(), &filer.Entry{
+		FullPath: dir.Child("e"), Attr: filer.Attr{Mode: 0o644, Mtime: time.Now(), Crtime: time.Now()},
+	}, 0); err != nil {
+		t.Fatal(err)
+	}
+
+	// Finishing the walk must not publish what it started collecting.
+	if _, err := mc.ListDirectoryEntries(ctx, dir, last, false, 100, func(entry *filer.Entry) (bool, error) {
+		return true, nil
+	}); err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if dirs, entries := mc.ListingCacheSize(); dirs != 0 {
+		t.Fatalf("published %d dirs / %d entries from a walk a write interrupted", dirs, entries)
+	}
+
+	// A fresh walk sees all five.
+	if got := walk(t, mc, ctx, dir, 100); len(got) != 5 {
+		t.Fatalf("walk listed %v, want 5 entries", got)
+	}
+}
+
+func TestListingCacheAppliesExpiryOnServe(t *testing.T) {
+	mc := newListingTestCache(t, DefaultListingCacheEntries)
+	dir := util.FullPath("/d")
+	now := time.Now()
+	ctx := context.Background()
+	if err := mc.InsertEntry(ctx, &filer.Entry{
+		FullPath: dir, Attr: filer.Attr{Mode: os.ModeDir | 0o755, Mtime: now, Crtime: now},
+	}, 0); err != nil {
+		t.Fatal(err)
+	}
+	// "b" expires two seconds from now; nothing will write when it does.
+	for _, e := range []struct {
+		name   string
+		ttlSec int32
+		crtime time.Time
+	}{{"a", 0, now}, {"b", 2, now}, {"c", 0, now}} {
+		if err := mc.InsertEntry(ctx, &filer.Entry{
+			FullPath: dir.Child(e.name),
+			Attr:     filer.Attr{Mode: 0o644, Mtime: now, Crtime: e.crtime, TtlSec: e.ttlSec},
+		}, 0); err != nil {
+			t.Fatal(err)
+		}
+	}
+	mc.markCachedFn(util.FullPath("/"))
+	mc.markCachedFn(dir)
+
+	if got := walk(t, mc, listingCtx(), dir, 100); len(got) != 3 {
+		t.Fatalf("first walk listed %v, want 3", got)
+	}
+
+	// Age "b" past its TTL in place, the way the clock would, and confirm the
+	// cached listing stops reporting it without anything invalidating.
+	entry, _, err := mc.FindEntry(ctx, dir.Child("b"))
+	if err != nil {
+		t.Fatalf("find: %v", err)
+	}
+	entry.Crtime = now.Add(-time.Hour)
+	cachedEntries, found := mc.listings.lookup(dir)
+	if !found {
+		t.Fatal("expected the directory to be cached")
+	}
+	for _, e := range cachedEntries {
+		if e.Name() == "b" {
+			e.Crtime = now.Add(-time.Hour)
+		}
+	}
+
+	got := walk(t, mc, listingCtx(), dir, 100)
+	for _, name := range got {
+		if name == "b" {
+			t.Fatalf("expired entry still listed: %v", got)
+		}
+	}
+	if len(got) != 2 {
+		t.Fatalf("listed %v, want a and c", got)
+	}
+}
+
+func TestListingCacheEvictsToStayWithinBudget(t *testing.T) {
+	mc := newListingTestCache(t, 6)
+	ctx := listingCtx()
+	for d := 0; d < 4; d++ {
+		dir := util.FullPath(fmt.Sprintf("/d%d", d))
+		seedDir(t, mc, dir, "a", "b", "c")
+		if got := walk(t, mc, ctx, dir, 100); len(got) != 3 {
+			t.Fatalf("%s listed %v, want 3", dir, got)
+		}
+		_, entries := mc.ListingCacheSize()
+		if entries > 6 {
+			t.Fatalf("cache holds %d entries, over the 6 budget", entries)
+		}
+	}
+	dirs, entries := mc.ListingCacheSize()
+	if dirs != 2 || entries != 6 {
+		t.Fatalf("cache holds %d dirs / %d entries, want 2/6 after eviction", dirs, entries)
+	}
+}
+
+func TestListingCacheDisabled(t *testing.T) {
+	mc := newListingTestCache(t, 0)
+	dir := util.FullPath("/d")
+	seedDir(t, mc, dir, "a", "b")
+	if got := walk(t, mc, listingCtx(), dir, 100); len(got) != 2 {
+		t.Fatalf("listed %v, want 2", got)
+	}
+	if dirs, entries := mc.ListingCacheSize(); dirs != 0 || entries != 0 {
+		t.Fatalf("disabled cache holds %d dirs / %d entries", dirs, entries)
+	}
+}

weed/mount/meta_cache/meta_cache.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"math"
 	"os"
+	"sort"
 	"sync"
 	"time"
 
@@ -43,6 +44,11 @@ type MetaCache struct {
 	dedupRing            dedupRingBuffer
 	includeSystemEntries bool
 
+	// listings holds directories walked all the way through, so walking one
+	// again reads and decodes nothing. Every store write drops the directory it
+	// touched; see the storeX wrappers.
+	listings *listingCache
+
 	// 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
@@ -101,8 +107,10 @@ type metadataApplyRequest struct {
 	done         chan error
 }
 
+// listingCacheEntries sizes the directory listing cache; zero disables it.
 func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPath, includeSystemEntries bool,
-	markCachedFn func(path util.FullPath), isCachedFn func(path util.FullPath) bool, invalidateFunc func(EntryInvalidation), onDirectoryUpdate func(dir util.FullPath)) *MetaCache {
+	markCachedFn func(path util.FullPath), isCachedFn func(path util.FullPath) bool, invalidateFunc func(EntryInvalidation), onDirectoryUpdate func(dir util.FullPath),
+	listingCacheEntries int) *MetaCache {
 	leveldbStore, virtualStore := openMetaStore(dbFolder)
 	mc := &MetaCache{
 		root:                 root,
@@ -119,6 +127,7 @@ func NewMetaCache(dbFolder string, uidGidMapper *UidGidMapper, root util.FullPat
 		buildingDirs:         make(map[util.FullPath]*directoryBuildState),
 		dedupRing:            newDedupRingBuffer(),
 		dirVersionFloors:     make(map[util.FullPath]int64),
+		listings:             newListingCache(listingCacheEntries),
 	}
 	mc.invalidateWorker = util.NewAsyncBatchWorker(func(batch []EntryInvalidation) {
 		for _, invalidation := range batch {
@@ -157,7 +166,7 @@ func (mc *MetaCache) InsertEntry(ctx context.Context, entry *filer.Entry, versio
 }
 
 func (mc *MetaCache) doInsertEntry(ctx context.Context, entry *filer.Entry, versionTsNs int64) error {
-	if err := mc.localStore.InsertEntry(ctx, entry); err != nil {
+	if err := mc.storeInsertEntry(ctx, entry); err != nil {
 		return err
 	}
 	mc.setEntryVersionLocked(ctx, entry.FullPath, versionTsNs)
@@ -187,7 +196,7 @@ func (mc *MetaCache) atomicUpdateEntryFromFilerLocked(ctx context.Context, oldPa
 	if entry != nil && vacatingOldPath {
 		ctx = context.WithValue(ctx, "OP", "MV")
 		glog.V(3).Infof("DeleteEntry %s", oldPath)
-		if err := mc.localStore.DeleteEntry(ctx, oldPath); err != nil {
+		if err := mc.storeDeleteEntry(ctx, oldPath); err != nil {
 			return err
 		}
 	}
@@ -209,7 +218,7 @@ func (mc *MetaCache) atomicUpdateEntryFromFilerLocked(ctx context.Context, oldPa
 		newDir, _ := newEntry.DirAndName()
 		if allowUncachedInsert || mc.isCachedFn(util.FullPath(newDir)) {
 			glog.V(3).Infof("InsertEntry %s/%s", newDir, newEntry.Name())
-			if err := mc.localStore.InsertEntry(ctx, newEntry); err != nil {
+			if err := mc.storeInsertEntry(ctx, newEntry); err != nil {
 				return err
 			}
 			mc.setEntryVersionLocked(ctx, newEntry.FullPath, versionTsNs)
@@ -230,11 +239,11 @@ func (mc *MetaCache) purgeEntryLocked(ctx context.Context, fullpath util.FullPat
 	if fullpath == "" {
 		return nil
 	}
-	if err := mc.localStore.DeleteEntry(ctx, fullpath); err != nil {
+	if err := mc.storeDeleteEntry(ctx, fullpath); err != nil {
 		return err
 	}
 	if isDirectory {
-		if err := mc.localStore.DeleteFolderChildren(ctx, fullpath); err != nil {
+		if err := mc.storeDeleteFolderChildren(ctx, fullpath); err != nil {
 			return err
 		}
 	}
@@ -303,7 +312,10 @@ func (mc *MetaCache) applyMetadataResponseEnqueue(ctx context.Context, resp *fil
 	}
 }
 
+// BeginDirectoryBuild drops the cached listing: a rebuild replaces the whole
+// child set, and nothing may be served from the old one meanwhile.
 func (mc *MetaCache) BeginDirectoryBuild(ctx context.Context, dirPath util.FullPath) error {
+	mc.listings.invalidate(dirPath)
 	return mc.enqueueAndWait(ctx, metadataApplyRequest{
 		kind:      metadataBeginBuild,
 		buildPath: dirPath,
@@ -311,6 +323,7 @@ func (mc *MetaCache) BeginDirectoryBuild(ctx context.Context, dirPath util.FullP
 }
 
 func (mc *MetaCache) CompleteDirectoryBuild(ctx context.Context, dirPath util.FullPath, snapshotTsNs int64) error {
+	mc.listings.invalidate(dirPath)
 	return mc.enqueueAndWait(ctx, metadataApplyRequest{
 		kind:         metadataCompleteBuild,
 		buildPath:    dirPath,
@@ -319,6 +332,7 @@ func (mc *MetaCache) CompleteDirectoryBuild(ctx context.Context, dirPath util.Fu
 }
 
 func (mc *MetaCache) AbortDirectoryBuild(ctx context.Context, dirPath util.FullPath) error {
+	mc.listings.invalidate(dirPath)
 	return mc.enqueueAndWait(ctx, metadataApplyRequest{
 		kind:      metadataAbortBuild,
 		buildPath: dirPath,
@@ -330,6 +344,7 @@ func (mc *MetaCache) AbortDirectoryBuild(ctx context.Context, dirPath util.FullP
 // like kernel Forget don't block; see purgeDirectoryChildrenNow for why off-loop
 // callers must route through here rather than wiping the store directly.
 func (mc *MetaCache) PurgeDirectoryChildren(dirPath util.FullPath, resetFn func()) {
+	mc.listings.invalidate(dirPath)
 	_ = mc.enqueueApplyRequest(metadataApplyRequest{
 		ctx:       context.Background(),
 		kind:      metadataPurgeDir,
@@ -342,7 +357,7 @@ func (mc *MetaCache) PurgeDirectoryChildren(dirPath util.FullPath, resetFn func(
 func (mc *MetaCache) UpdateEntry(ctx context.Context, entry *filer.Entry) error {
 	mc.Lock()
 	defer mc.Unlock()
-	if err := mc.localStore.UpdateEntry(ctx, entry); err != nil {
+	if err := mc.storeUpdateEntry(ctx, entry); err != nil {
 		return err
 	}
 	mc.markEntryUnversionedLocked(ctx, entry.FullPath)
@@ -365,7 +380,7 @@ func (mc *MetaCache) TouchDirMtimeCtime(ctx context.Context, dirPath util.FullPa
 	}
 	entry.Attr.Mtime = now
 	entry.Attr.Ctime = now
-	if err := mc.localStore.UpdateEntry(ctx, entry); err != nil {
+	if err := mc.storeUpdateEntry(ctx, entry); err != nil {
 		return err
 	}
 	mc.markEntryUnversionedLocked(ctx, dirPath)
@@ -576,7 +591,7 @@ func (mc *MetaCache) deleteChildVersionRecordsLocked(ctx context.Context, dirPat
 func (mc *MetaCache) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
 	mc.Lock()
 	defer mc.Unlock()
-	if err = mc.localStore.DeleteEntry(ctx, fp); err != nil {
+	if err = mc.storeDeleteEntry(ctx, fp); err != nil {
 		return err
 	}
 	mc.clearEntryVersionLocked(ctx, fp)
@@ -587,7 +602,7 @@ func (mc *MetaCache) DeleteFolderChildren(ctx context.Context, fp util.FullPath)
 	defer mc.Unlock()
 	delete(mc.dirVersionFloors, fp)
 	mc.deleteChildVersionRecordsLocked(ctx, fp)
-	return mc.localStore.DeleteFolderChildren(ctx, fp)
+	return mc.storeDeleteFolderChildren(ctx, fp)
 }
 
 // SetPinnedChildFn installs a predicate reporting whether a child holds
@@ -606,7 +621,7 @@ func (mc *MetaCache) deleteFolderChildrenForRebuild(ctx context.Context, dirPath
 	mc.Lock()
 	defer mc.Unlock()
 	if mc.pinnedChildFn == nil {
-		return mc.localStore.DeleteFolderChildren(ctx, dirPath)
+		return mc.storeDeleteFolderChildren(ctx, dirPath)
 	}
 	var pinned []*filer.Entry
 	if _, err := mc.localStore.ListDirectoryEntries(ctx, dirPath, "", true, math.MaxInt64, func(entry *filer.Entry) (bool, error) {
@@ -617,7 +632,7 @@ func (mc *MetaCache) deleteFolderChildrenForRebuild(ctx context.Context, dirPath
 	}); err != nil {
 		return err
 	}
-	if err := mc.localStore.DeleteFolderChildren(ctx, dirPath); err != nil {
+	if err := mc.storeDeleteFolderChildren(ctx, dirPath); err != nil {
 		return err
 	}
 	if len(pinned) > 0 {
@@ -631,6 +646,10 @@ func (mc *MetaCache) deleteFolderChildrenForRebuild(ctx context.Context, dirPath
 // the store has already spent it against limit. A caller paginating by count
 // would read a short batch as the end of the directory, and one resuming from
 // the last name it saw would re-read the dropped ones forever.
+//
+// A directory already walked all the way through is served from memory. Only a
+// listing that leaves out chunk lists is cached or served, so an entry from the
+// cache can never be handed to a caller that wanted chunks.
 func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
 	mc.RLock()
 	defer mc.RUnlock()
@@ -638,17 +657,128 @@ func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.Full
 	if !mc.isCachedFn(dirPath) {
 		// if this request comes after renaming, it should be fine
 		glog.Warningf("unsynchronized dir: %v", dirPath)
+		mc.listings.invalidate(dirPath)
+		return mc.listFromStore(ctx, dirPath, startFileName, includeStartFile, limit, eachEntryFunc, nil)
 	}
 
+	cacheable := filer_pb.ChunksOmitted(ctx)
+	if cacheable {
+		if entries, found := mc.listings.lookup(dirPath); found {
+			return serveCachedListing(entries, startFileName, includeStartFile, limit, eachEntryFunc)
+		}
+	}
+
+	var build *listingBuild
+	if cacheable {
+		build = mc.listings.beginOrContinue(dirPath, startFileName, includeStartFile)
+	}
+
+	storeCount := int64(0)
+	stoppedEarly := false
+	wrapped := func(entry *filer.Entry) (bool, error) {
+		ok, err := eachEntryFunc(entry)
+		if err != nil || !ok {
+			stoppedEarly = true
+		}
+		return ok, err
+	}
+	lastFileName, err = mc.listFromStore(ctx, dirPath, startFileName, includeStartFile, limit, wrapped, func(entry *filer.Entry) {
+		storeCount++
+		if build != nil {
+			build.entries = append(build.entries, entry)
+		}
+	})
+	if err != nil {
+		mc.listings.invalidate(dirPath)
+		return lastFileName, err
+	}
+	if build != nil {
+		// A short page means the store ran out only if the caller did not stop
+		// it first. A caller whose buffer filled has seen part of a page, and
+		// the rest of the directory is still out there.
+		if storeCount < limit && !stoppedEarly {
+			mc.listings.publish(dirPath, build)
+		} else {
+			mc.listings.carry(dirPath, build, lastFileName)
+		}
+	}
+	return lastFileName, nil
+}
+
+// listFromStore reads a page, dropping expired children and mapping ids. observe,
+// when set, sees every entry the store produced, expired ones included, since
+// those are what limit was spent on.
+func (mc *MetaCache) listFromStore(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc, observe func(*filer.Entry)) (string, error) {
 	return mc.localStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *filer.Entry) (bool, error) {
-		if entry.TtlSec > 0 && entry.Crtime.Add(time.Duration(entry.TtlSec)*time.Second).Before(time.Now()) {
+		mc.mapIdFromFilerToLocal(entry)
+		if observe != nil {
+			observe(entry)
+		}
+		if isTtlExpired(entry) {
 			return true, nil
 		}
-		mc.mapIdFromFilerToLocal(entry)
 		return eachEntryFunc(entry)
 	})
 }
 
+// serveCachedListing walks an in-memory listing the way the store would: limit
+// counts every child it steps over, expired or not, and lastFileName is the last
+// one stepped over rather than the last one reported. Expiry is applied here
+// rather than at cache time, so a listing does not go stale as children age out.
+func serveCachedListing(entries []*filer.Entry, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
+	start := sort.Search(len(entries), func(i int) bool {
+		if includeStartFile {
+			return entries[i].Name() >= startFileName
+		}
+		return entries[i].Name() > startFileName
+	})
+	for i := start; i < len(entries) && int64(i-start) < limit; i++ {
+		entry := entries[i]
+		lastFileName = entry.Name()
+		if isTtlExpired(entry) {
+			continue
+		}
+		if ok, err := eachEntryFunc(entry); err != nil || !ok {
+			return lastFileName, err
+		}
+	}
+	return lastFileName, nil
+}
+
+// Every write to the local store goes through these, so a directory's cached
+// listing cannot outlive a change to it. Reaching past them to mc.localStore
+// for a write would leave the listing stale.
+func (mc *MetaCache) storeInsertEntry(ctx context.Context, entry *filer.Entry) error {
+	mc.listings.invalidateChild(entry.FullPath)
+	return mc.localStore.InsertEntry(ctx, entry)
+}
+
+func (mc *MetaCache) storeUpdateEntry(ctx context.Context, entry *filer.Entry) error {
+	mc.listings.invalidateChild(entry.FullPath)
+	return mc.localStore.UpdateEntry(ctx, entry)
+}
+
+func (mc *MetaCache) storeDeleteEntry(ctx context.Context, fp util.FullPath) error {
+	mc.listings.invalidateChild(fp)
+	return mc.localStore.DeleteEntry(ctx, fp)
+}
+
+func (mc *MetaCache) storeDeleteFolderChildren(ctx context.Context, fp util.FullPath) error {
+	mc.listings.invalidate(fp)
+	return mc.localStore.DeleteFolderChildren(ctx, fp)
+}
+
+// InvalidateAllListings distrusts every cached listing, for when a subscription
+// gap means the mount cannot know what it missed.
+func (mc *MetaCache) InvalidateAllListings() {
+	mc.listings.invalidateAll()
+}
+
+// ListingCacheSize reports the cached directories and the children they hold.
+func (mc *MetaCache) ListingCacheSize() (dirs, entries int) {
+	return mc.listings.size()
+}
+
 func (mc *MetaCache) Shutdown() {
 	done := make(chan error, 1)
 
@@ -935,7 +1065,7 @@ func (mc *MetaCache) applyMetadataResponseLocked(ctx context.Context, resp *file
 		isDelete := message.NewEntry == nil
 		isMove := message.NewEntry != nil && (message.NewParentPath != resp.Directory || message.NewEntry.Name != message.OldEntry.Name)
 		if isDelete || isMove {
-			if deleteErr := mc.localStore.DeleteFolderChildren(ctx, oldPath); deleteErr != nil {
+			if deleteErr := mc.storeDeleteFolderChildren(ctx, oldPath); deleteErr != nil {
 				glog.V(2).Infof("delete descendants of %s: %v", oldPath, deleteErr)
 			}
 		}
@@ -977,7 +1107,7 @@ func (mc *MetaCache) purgeDirectoryChildrenNow(ctx context.Context, dirPath util
 	defer mc.Unlock()
 	delete(mc.dirVersionFloors, dirPath)
 	mc.deleteChildVersionRecordsLocked(ctx, dirPath)
-	return mc.localStore.DeleteFolderChildren(ctx, dirPath)
+	return mc.storeDeleteFolderChildren(ctx, dirPath)
 }
 
 func (mc *MetaCache) completeDirectoryBuildNow(ctx context.Context, dirPath util.FullPath, snapshotTsNs int64) error {

weed/mount/meta_cache/meta_cache_apply_test.go

@@ -506,6 +506,7 @@ func newTestMetaCache(t *testing.T, cached map[util.FullPath]bool) (*MetaCache,
 		func(dir util.FullPath) {
 			notifications.record(dir)
 		},
+		DefaultListingCacheEntries,
 	)
 
 	return mc, cached, notifications, invalidations

weed/mount/meta_cache/meta_cache_deadlock_test.go

@@ -66,6 +66,7 @@ func TestApplyLoopInvalidateDoesNotDeadlockWithLockHoldingEnqueuer(t *testing.T)
 			fhLockTable.ReleaseLock(fhKey, lock)
 		},
 		func(dir util.FullPath) {},
+		DefaultListingCacheEntries,
 	)
 	defer func() {
 		// Only safe to shut down once the apply loop is unwedged.

weed/mount/weedfs.go

@@ -53,6 +53,7 @@ type Option struct {
 	CacheDirForWrite            string
 	WriteBufferSizeMB           int64
 	CacheMetaTTlSec             int
+	DirListCacheEntries         int
 	DataCenter                  string
 	Umask                       os.FileMode
 	Quota                       int64
@@ -314,7 +315,7 @@ func NewSeaweedFileSystem(option *Option) *WFS {
 			if wfs.inodeToPath.RecordDirectoryUpdate(dirPath, time.Now(), wfs.dirHotWindow, wfs.dirHotThreshold) {
 				wfs.markDirectoryReadThrough(dirPath)
 			}
-		})
+		}, option.DirListCacheEntries)
 	wfs.metaCache.SetPinnedChildFn(wfs.isLocalOnlyEntry)
 	grace.OnInterrupt(func() {
 		// grace calls os.Exit(0) after all hooks, so WaitForAsyncFlush
@@ -506,6 +507,7 @@ func (wfs *WFS) StartBackgroundTasks() error {
 		// wipe the root's stale children through the apply loop so the delete
 		// cannot strand a concurrent rebuild cached-but-empty.
 		wfs.inodeToPath.InvalidateAllChildrenCache()
+		wfs.metaCache.InvalidateAllListings()
 		wfs.purgeDirectoryCache(util.FullPath(wfs.option.FilerMountRootPath))
 	}, follower)
 	go wfs.loopCheckQuota()

weed/mount/weedfs_dir_read_bench_test.go

@@ -110,6 +110,7 @@ func newBenchWFS(tb testing.TB, dir util.FullPath, n int) *WFS {
 		func(path util.FullPath) bool { return wfs.inodeToPath.IsChildrenCached(path) },
 		func(meta_cache.EntryInvalidation) {},
 		nil,
+		meta_cache.DefaultListingCacheEntries,
 	)
 	tb.Cleanup(wfs.metaCache.Shutdown)
 

weed/mount/weedfs_dir_read_pagination_test.go

@@ -63,6 +63,7 @@ func newPagingWFS(tb testing.TB, dir util.FullPath, names []string, ttlSec int32
 		func(p util.FullPath) { wfs.inodeToPath.MarkChildrenCached(p) },
 		func(p util.FullPath) bool { return wfs.inodeToPath.IsChildrenCached(p) },
 		func(meta_cache.EntryInvalidation) {}, nil,
+		meta_cache.DefaultListingCacheEntries,
 	)
 	tb.Cleanup(wfs.metaCache.Shutdown)
 

weed/mount/weedfs_file_copy_range_test.go

@@ -348,6 +348,7 @@ func newCopyRangeTestWFSWithMetaCache(t *testing.T) *WFS {
 		},
 		func(meta_cache.EntryInvalidation) {},
 		nil,
+		meta_cache.DefaultListingCacheEntries,
 	)
 	t.Cleanup(func() {
 		wfs.metaCache.Shutdown()

weed/mount/weedfs_file_mkrm_test.go

@@ -130,6 +130,7 @@ func newCreateTestWFS(t *testing.T) (*WFS, *createEntryTestServer) {
 		},
 		func(meta_cache.EntryInvalidation) {},
 		nil,
+		meta_cache.DefaultListingCacheEntries,
 	)
 	wfs.inodeToPath.MarkChildrenCached(root)
 	t.Cleanup(func() {

weed/mount/weedfs_invalidate_open_handle_test.go

@@ -62,6 +62,7 @@ func newInvalidateTestWFS(t *testing.T) *WFS {
 		func(path util.FullPath) bool { return wfs.inodeToPath.IsChildrenCached(path) },
 		wfs.invalidateOpenFileHandle,
 		nil,
+		meta_cache.DefaultListingCacheEntries,
 	)
 	t.Cleanup(wfs.metaCache.Shutdown)
 

weed/mount/weedfs_rename_test.go

@@ -32,6 +32,7 @@ func TestHandleRenameResponseLeavesUncachedTargetOutOfCache(t *testing.T) {
 		},
 		func(meta_cache.EntryInvalidation) {},
 		nil,
+		meta_cache.DefaultListingCacheEntries,
 	)
 	defer mc.Shutdown()