From 12627d376d5c3404c909bef15e927671fa24c7b9 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 7 Aug 2026 12:11:27 -0700 Subject: [PATCH] mount: fix four readdir pagination bugs (#10624) * mount: size the direct listing slice from the batch, not the offset The limit passed here is skipCount+batchSize when a client resumes a fresh handle partway through a directory, so preallocating for it turns the client's cookie into an allocation: a readdir at offset 3,000,000 reserves 24MB before the first entry arrives, and an offset near the uint32 ceiling asks makeslice for ~4.29e9 elements. * mount: stop replaying a directory that shrank past the resume offset A client that opens a fresh handle and resumes at a cookie from an earlier, larger listing gets a preload that cannot reach the entry before that offset. The resume name was then left empty and the follow-up batch listed from the directory's first child again, so the client was handed every name a second time. The stream always runs from the first child, so failing to reach that entry means the directory is simply shorter than the offset. That is the end of it. * mount: page a directory from where the store reached The batch loader treated a short batch as the end of the directory, but the meta cache drops an expired child after the store has already spent it against the limit, so a batch that filled up could still deliver fewer entries than asked for. A directory with a handful of expired children would stop listing early and hide every child behind them; a whole batch of expired ones truncated the listing to nothing. ListDirectoryEntries now reports the name the store itself reached. That is both the sound end-of-directory signal -- the store returning nothing -- and the right cursor, since resuming from the last visible name would re-read the dropped children every round and never get past a batch that was entirely expired. * mount: drop the entries a directory walk has already passed entryStreamOffset was only ever written by reset, so the stream a handle holds grew for the life of the walk and a directory was retained whole even though nothing could read the entries behind the client's position again. A 10M-entry walk parked millions of entries per handle, and NFS-Ganesha opens several on the same directory. Offsets index into the stream from entryStreamOffset, so advancing the two together keeps them lined up. One entry is held back because the next batch resumes from the name immediately before the offset. Seeking back behind what is still held now restarts the directory, which is what the offset scheme can honestly support -- it previously returned nothing. --- weed/mount/meta_cache/meta_cache.go | 13 +- weed/mount/weedfs_dir_read.go | 69 ++++- weed/mount/weedfs_dir_read_chunks_test.go | 2 +- weed/mount/weedfs_dir_read_pagination_test.go | 246 ++++++++++++++++++ 4 files changed, 313 insertions(+), 17 deletions(-) create mode 100644 weed/mount/weedfs_dir_read_pagination_test.go diff --git a/weed/mount/meta_cache/meta_cache.go b/weed/mount/meta_cache/meta_cache.go index 966a2367f5..cb22745a99 100644 --- a/weed/mount/meta_cache/meta_cache.go +++ b/weed/mount/meta_cache/meta_cache.go @@ -626,7 +626,12 @@ func (mc *MetaCache) deleteFolderChildrenForRebuild(ctx context.Context, dirPath return nil } -func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) error { +// ListDirectoryEntries reports the last name the store reached, which is not +// the last name handed to eachEntryFunc: an expired child is dropped here after +// 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. +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() @@ -635,17 +640,13 @@ func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.Full glog.Warningf("unsynchronized dir: %v", dirPath) } - _, err := mc.localStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *filer.Entry) (bool, 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()) { return true, nil } mc.mapIdFromFilerToLocal(entry) return eachEntryFunc(entry) }) - if err != nil { - return err - } - return err } func (mc *MetaCache) Shutdown() { diff --git a/weed/mount/weedfs_dir_read.go b/weed/mount/weedfs_dir_read.go index 06a637047c..faf1b3d3bd 100644 --- a/weed/mount/weedfs_dir_read.go +++ b/weed/mount/weedfs_dir_read.go @@ -33,11 +33,15 @@ type DirectoryHandle struct { isFinished bool entryStream []*filer.Entry entryStreamOffset uint64 - snapshotTsNs int64 // snapshot timestamp for consistent readdir in direct mode + // lastListedName is how far the store itself reached, which runs ahead of + // the last visible entry whenever children are dropped as expired. + lastListedName string + snapshotTsNs int64 // snapshot timestamp for consistent readdir in direct mode } func (dh *DirectoryHandle) reset() { dh.isFinished = false + dh.lastListedName = "" dh.snapshotTsNs = 0 // Nil out pointers to allow garbage collection of old entries, // then reuse the slice's capacity to avoid re-allocations. @@ -161,6 +165,11 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode if input.Offset == 0 { dh.reset() + } else if input.Offset < dh.entryStreamOffset { + // Seeking back before what the handle still holds. Start the directory + // again rather than reporting nothing; the preload below refills up to + // the requested offset. + dh.reset() } else if dh.isFinished && input.Offset >= dh.entryStreamOffset { entryCurrentIndex := input.Offset - dh.entryStreamOffset if uint64(len(dh.entryStream)) <= entryCurrentIndex { @@ -241,6 +250,19 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode // Read from cache first, then load next batch if needed if input.Offset >= dh.entryStreamOffset { + // Drop what the client has walked past. Offsets are indexes into the + // stream from entryStreamOffset, so advancing the two together keeps + // them lined up; one entry is kept back because the next batch resumes + // from the name immediately before the offset. + if trim := int(input.Offset-dh.entryStreamOffset) - 1; trim > 0 && trim <= len(dh.entryStream) { + copy(dh.entryStream, dh.entryStream[trim:]) + for i := len(dh.entryStream) - trim; i < len(dh.entryStream); i++ { + dh.entryStream[i] = nil + } + dh.entryStream = dh.entryStream[:len(dh.entryStream)-trim] + dh.entryStreamOffset += uint64(trim) + } + // Handle case: new handle with non-zero offset but empty cache // This happens when NFS-Ganesha opens multiple directory handles if len(dh.entryStream) == 0 && input.Offset > dh.entryStreamOffset { @@ -252,10 +274,11 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode } // Load entries from beginning to fill cache up to the requested offset - loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, "", false, skipCount+int64(batchSize), func(entry *filer.Entry) (bool, error) { + storeLastName, loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, "", false, skipCount+int64(batchSize), func(entry *filer.Entry) (bool, error) { dh.entryStream = append(dh.entryStream, entry) return true, nil }) + dh.lastListedName = storeLastName if loadErr != nil { glog.Errorf("list meta cache: %v", loadErr) return fuse.EIO @@ -266,6 +289,13 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode entryPreviousIndex := (input.Offset - dh.entryStreamOffset) - 1 if uint64(len(dh.entryStream)) > entryPreviousIndex { lastEntryName = dh.entryStream[entryPreviousIndex].Name() + } else { + // The stream runs from the directory's first child, so failing to + // reach the entry before this offset means the directory has since + // shrunk past it. Listing on from an empty name would replay the + // directory from the start and hand the client every name twice. + dh.isFinished = true + return fuse.OK } } @@ -286,13 +316,18 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode return fuse.EIO } - // Batch loading: fetch batchSize entries starting from lastEntryName - loadedCount := 0 + // Page from where the store itself reached, not from the last entry the + // sink saw. An expired child is counted against the batch and then + // dropped, so resuming from the last visible name would re-read it every + // round and never get past a batch that was entirely expired. + if dh.lastListedName > lastEntryName { + lastEntryName = dh.lastListedName + } + bufferFull := false - loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, lastEntryName, false, int64(batchSize), func(entry *filer.Entry) (bool, error) { + storeLastName, loadErr := wfs.metaCache.ListDirectoryEntries(readdirContext, dirPath, lastEntryName, false, int64(batchSize), func(entry *filer.Entry) (bool, error) { currentIndex := int64(len(dh.entryStream)) dh.entryStream = append(dh.entryStream, entry) - loadedCount++ if !processEachEntryFn(entry, currentIndex) { bufferFull = true return false, nil @@ -303,10 +338,12 @@ func (wfs *WFS) doReadDirectory(input *fuse.ReadIn, out DirEntrySink, isPlusMode glog.Errorf("list meta cache: %v", loadErr) return fuse.EIO } + dh.lastListedName = storeLastName - // Mark finished only when loading completed normally (not buffer full) - // and we got fewer entries than requested - if !bufferFull && loadedCount < batchSize { + // The store reaching nothing is the only sound end-of-directory signal: + // a batch can come back short because entries expired, not because the + // directory ran out. + if !bufferFull && storeLastName == "" { dh.isFinished = true } } @@ -335,6 +372,11 @@ func (wfs *WFS) readDirectoryDirect(input *fuse.ReadIn, out DirEntrySink, dh *Di entryPreviousIndex := (input.Offset - dh.entryStreamOffset) - 1 if uint64(len(dh.entryStream)) > entryPreviousIndex { lastEntryName = dh.entryStream[entryPreviousIndex].Name() + } else { + // See the cached path: the directory shrank past this offset, and + // resuming from an empty name would replay it from the start. + dh.isFinished = true + return fuse.OK } } @@ -379,7 +421,14 @@ func (wfs *WFS) readDirectoryDirect(input *fuse.ReadIn, out DirEntrySink, dh *Di } func loadDirectoryEntriesDirect(ctx context.Context, client filer_pb.FilerClient, uidGidMapper *meta_cache.UidGidMapper, dirPath util.FullPath, startFileName string, includeStart bool, limit uint32, snapshotTsNs int64, includeSystemEntries bool) ([]*filer.Entry, int64, error) { - entries := make([]*filer.Entry, 0, limit) + // limit can be a client-supplied resume offset rather than a batch size, so + // preallocating for it would size the slice from where the caller happened to + // seek. Reserve a batch and let append find the rest. + prealloc := limit + if prealloc > batchSize { + prealloc = batchSize + } + entries := make([]*filer.Entry, 0, prealloc) var actualSnapshotTsNs int64 err := client.WithFilerClient(false, func(sc filer_pb.SeaweedFilerClient) error { var innerErr error diff --git a/weed/mount/weedfs_dir_read_chunks_test.go b/weed/mount/weedfs_dir_read_chunks_test.go index bf12e0e356..25a554c95c 100644 --- a/weed/mount/weedfs_dir_read_chunks_test.go +++ b/weed/mount/weedfs_dir_read_chunks_test.go @@ -27,7 +27,7 @@ func TestListDirectoryEntriesOmitsChunks(t *testing.T) { } { t.Run(tc.name, func(t *testing.T) { var seen int - err := wfs.metaCache.ListDirectoryEntries(tc.ctx, dir, "", false, 100, func(entry *filer.Entry) (bool, error) { + _, err := wfs.metaCache.ListDirectoryEntries(tc.ctx, dir, "", false, 100, func(entry *filer.Entry) (bool, error) { seen++ if got := len(entry.Chunks) > 0; got != tc.wantChunks { t.Errorf("%s: has chunks = %v, want %v", entry.Name(), got, tc.wantChunks) diff --git a/weed/mount/weedfs_dir_read_pagination_test.go b/weed/mount/weedfs_dir_read_pagination_test.go new file mode 100644 index 0000000000..00e07d7e1a --- /dev/null +++ b/weed/mount/weedfs_dir_read_pagination_test.go @@ -0,0 +1,246 @@ +package mount + +import ( + "context" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/mount/meta_cache" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// pagingSink collects names, accepting at most limit per round the way a kernel +// reply buffer does. +type pagingSink struct { + limit int + names []string + lastOff uint64 + round int +} + +func (s *pagingSink) AddEntry(entry fuse.DirEntry) bool { + if s.round >= s.limit { + return false + } + s.round++ + s.names = append(s.names, entry.Name) + s.lastOff = entry.Off + return true +} + +func (s *pagingSink) AddEntryPlus(entry fuse.DirEntry) *fuse.EntryOut { return nil } +func (s *pagingSink) TakesLookupRef() bool { return true } + +func newPagingWFS(tb testing.TB, dir util.FullPath, names []string, ttlSec int32) *WFS { + tb.Helper() + uidGidMapper, err := meta_cache.NewUidGidMapper("", "") + if err != nil { + tb.Fatalf("uid/gid mapper: %v", err) + } + root := util.FullPath("/") + wfs := &WFS{ + option: &Option{ + ChunkSizeLimit: 1024, ConcurrentReaders: 1, VolumeServerAccess: "filerProxy", + FilerAddresses: []pb.ServerAddress{pb.NewServerAddressWithGrpcPort("127.0.0.1:1", 1)}, + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + FilerMountRootPath: "/", MountUid: 99, MountGid: 100, MountMode: 0o777, + MountMtime: time.Now(), MountCtime: time.Now(), UidGidMapper: uidGidMapper, + }, + signature: 1, inodeToPath: NewInodeToPath(root, 0), + fhMap: NewFileHandleToInode(), dhMap: NewDirectoryHandleToInode(), + fhLockTable: util.NewLockTable[FileHandleId](), hardLinkLockTable: util.NewLockTable[string](), + } + wfs.metaCache = meta_cache.NewMetaCache( + filepath.Join(tb.TempDir(), "meta"), uidGidMapper, root, false, + func(p util.FullPath) { wfs.inodeToPath.MarkChildrenCached(p) }, + func(p util.FullPath) bool { return wfs.inodeToPath.IsChildrenCached(p) }, + func(meta_cache.EntryInvalidation) {}, nil, + ) + tb.Cleanup(wfs.metaCache.Shutdown) + + now := time.Now() + ctx := context.Background() + if err := wfs.metaCache.InsertEntry(ctx, &filer.Entry{ + FullPath: dir, + Attr: filer.Attr{Mode: os.ModeDir | 0o755, Mtime: now, Crtime: now}, + }, 0); err != nil { + tb.Fatalf("insert dir: %v", err) + } + for _, name := range names { + child := dir.Child(name) + attr := filer.Attr{Mode: 0o644, Mtime: now, Crtime: now, FileSize: 1, Inode: child.AsInode(now.Unix())} + if ttlSec > 0 { + // Expired well in the past, so the meta cache drops it after the + // store has already counted it against the limit. + attr.TtlSec = ttlSec + attr.Crtime = now.Add(-time.Duration(ttlSec+60) * time.Second) + } + if err := wfs.metaCache.InsertEntry(ctx, &filer.Entry{FullPath: child, Attr: attr}, 0); err != nil { + tb.Fatalf("insert %s: %v", name, err) + } + } + wfs.inodeToPath.MarkChildrenCached(root) + wfs.inodeToPath.Lookup(dir, now.Unix(), true, false, 0, true) + wfs.inodeToPath.MarkChildrenCached(dir) + return wfs +} + +// TestReadDirResumePastShrunkDirectory covers a client resuming a fresh handle +// at a cookie from an earlier, larger listing. The directory has since shrunk +// past that offset, and the readdir used to fall back to listing from the first +// child again, handing back names the client had already consumed. +func TestReadDirResumePastShrunkDirectory(t *testing.T) { + dir := util.FullPath("/d") + var names []string + for i := 0; i < 100; i++ { + names = append(names, fmt.Sprintf("f%03d", i)) + } + wfs := newPagingWFS(t, dir, names, 0) + dirInode, _ := wfs.inodeToPath.GetInode(dir) + + dhid, _ := wfs.AcquireDirectoryHandle() + defer wfs.ReleaseDirectoryHandle(dhid) + + sink := &pagingSink{limit: 4096} + status := wfs.doReadDirectory(&fuse.ReadIn{ + InHeader: fuse.InHeader{NodeId: dirInode}, + Fh: uint64(dhid), + Offset: 152, // past the end of a directory that now holds 100 + Size: 1 << 20, + }, sink, false) + if status != fuse.OK { + t.Fatalf("readdir: %v", status) + } + if len(sink.names) != 0 { + t.Errorf("resuming past the end returned %d entries (first %q), want none", len(sink.names), sink.names[0]) + } +} + +// TestReadDirWithExpiredEntries covers a batch the store fills to the limit but +// whose entries the meta cache then drops as expired. The post-filter count used +// to be read as end-of-directory, silently hiding every later child. +func TestReadDirWithExpiredEntries(t *testing.T) { + dir := util.FullPath("/d") + // One full store batch of entries that all expire, then live ones behind + // them. Nothing survives the first batch, which is the case that latched. + var names []string + for i := 0; i < batchSize; i++ { + names = append(names, fmt.Sprintf("a%05d", i)) + } + wfs := newPagingWFS(t, dir, names, 30) + + live := []string{"z001", "z002", "z003"} + now := time.Now() + for _, name := range live { + child := dir.Child(name) + if err := wfs.metaCache.InsertEntry(context.Background(), &filer.Entry{ + FullPath: child, + Attr: filer.Attr{Mode: 0o644, Mtime: now, Crtime: now, FileSize: 1, Inode: child.AsInode(now.Unix())}, + }, 0); err != nil { + t.Fatalf("insert %s: %v", name, err) + } + } + + dirInode, _ := wfs.inodeToPath.GetInode(dir) + dhid, _ := wfs.AcquireDirectoryHandle() + defer wfs.ReleaseDirectoryHandle(dhid) + + sink := &pagingSink{limit: 4096} + var offset uint64 + for round := 0; round < 20; round++ { + sink.round = 0 + status := wfs.doReadDirectory(&fuse.ReadIn{ + InHeader: fuse.InHeader{NodeId: dirInode}, + Fh: uint64(dhid), + Offset: offset, + Size: 1 << 20, + }, sink, false) + if status != fuse.OK { + t.Fatalf("readdir: %v", status) + } + if sink.lastOff <= offset { + break + } + offset = sink.lastOff + } + + var seen []string + for _, n := range sink.names { + if n != "." && n != ".." { + seen = append(seen, n) + } + } + if len(seen) != len(live) { + t.Fatalf("listed %d entries %v, want the %d live ones %v", len(seen), seen, len(live), live) + } +} + +// TestReadDirTrimsConsumedEntries checks that a walk does not accumulate the +// whole directory in the handle. Offsets index into the stream from +// entryStreamOffset, so the two have to advance together or the listing +// silently misaligns. +func TestReadDirTrimsConsumedEntries(t *testing.T) { + dir := util.FullPath("/d") + const total = 5000 + var names []string + for i := 0; i < total; i++ { + names = append(names, fmt.Sprintf("f%05d", i)) + } + wfs := newPagingWFS(t, dir, names, 0) + dirInode, _ := wfs.inodeToPath.GetInode(dir) + + dhid, dh := wfs.AcquireDirectoryHandle() + defer wfs.ReleaseDirectoryHandle(dhid) + + // A small sink forces many rounds, which is when the stream would grow. + sink := &pagingSink{limit: 64} + var offset uint64 + var seen []string + peak := 0 + for round := 0; round < 500; round++ { + sink.round = 0 + before := len(sink.names) + status := wfs.doReadDirectory(&fuse.ReadIn{ + InHeader: fuse.InHeader{NodeId: dirInode}, + Fh: uint64(dhid), + Offset: offset, + Size: 1 << 20, + }, sink, false) + if status != fuse.OK { + t.Fatalf("readdir: %v", status) + } + if n := len(dh.entryStream); n > peak { + peak = n + } + if len(sink.names) == before || sink.lastOff <= offset { + break + } + offset = sink.lastOff + } + for _, n := range sink.names { + if n != "." && n != ".." { + seen = append(seen, n) + } + } + + if len(seen) != total { + t.Fatalf("listed %d entries, want %d", len(seen), total) + } + for i, name := range seen { + if want := fmt.Sprintf("f%05d", i); name != want { + t.Fatalf("entry %d is %q, want %q -- trimming misaligned the offsets", i, name, want) + } + } + // Without trimming the handle ends up holding every entry. + if peak >= total { + t.Errorf("handle held %d entries at peak, want well under the %d in the directory", peak, total) + } +}