diff --git a/weed/mount/meta_cache/listing_cache.go b/weed/mount/meta_cache/listing_cache.go new file mode 100644 index 000000000..93784f5ac --- /dev/null +++ b/weed/mount/meta_cache/listing_cache.go @@ -0,0 +1,190 @@ +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. A chunkless entry runs a few +// hundred bytes, so this is tens of megabytes. +const DefaultListingCacheEntries = 200000 + +// 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 +} diff --git a/weed/mount/meta_cache/listing_cache_test.go b/weed/mount/meta_cache/listing_cache_test.go new file mode 100644 index 000000000..1582aa6c8 --- /dev/null +++ b/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 := NewMetaCacheWithListingCache(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) + } +} diff --git a/weed/mount/meta_cache/meta_cache.go b/weed/mount/meta_cache/meta_cache.go index cb22745a9..9bd04e7b0 100644 --- a/weed/mount/meta_cache/meta_cache.go +++ b/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 @@ -103,6 +109,13 @@ type metadataApplyRequest struct { 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 { + return NewMetaCacheWithListingCache(dbFolder, uidGidMapper, root, includeSystemEntries, markCachedFn, isCachedFn, invalidateFunc, onDirectoryUpdate, DefaultListingCacheEntries) +} + +// NewMetaCacheWithListingCache sizes the directory listing cache; zero disables it. +func NewMetaCacheWithListingCache(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), + listingCacheEntries int) *MetaCache { leveldbStore, virtualStore := openMetaStore(dbFolder) mc := &MetaCache{ root: root, @@ -119,6 +132,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 +171,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 +201,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 +223,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 +244,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 +317,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 +328,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 +337,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 +349,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 +362,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 +385,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 +596,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 +607,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 +626,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 +637,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 +651,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 +662,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 +1070,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 +1112,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 {