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